Skip to main content
AgentFlow is the workflow orchestration system for creating deterministic, multi-step pipelines. It supports sequential execution, conditional branching, loops, and parallel processing.

​
Quick Start

1

Simple Pipeline

Create a basic sequential workflow:
from praisonaiagents import Agent, AgentFlow

# Define agents as steps
writer = Agent(
    name="Writer",
    instructions="Write content based on the input"
)

editor = Agent(
    name="Editor", 
    instructions="Edit and improve the content"
)

# Create flow
flow = AgentFlow(
    steps=[writer, editor]
)

result = flow.start("Write about artificial intelligence")
print(result["output"])
2

With Variables

Pass variables between steps:
from praisonaiagents import Agent, AgentFlow

flow = AgentFlow(
    steps=[researcher, writer, publisher],
    variables={
        "topic": "Machine Learning",
        "word_count": 500,
        "tone": "professional"
    }
)

result = flow.start()
3

Planning Mode

Enable AI-powered planning for complex workflows:
from praisonaiagents import Agent, AgentFlow

# Option 1: Enable planning with default model
flow = AgentFlow(
    steps=[agent1, agent2, agent3],
    planning=True,
    output="verbose"
)

# Option 2: Specify a planning model (also enables planning)
flow = AgentFlow(
    steps=[agent1, agent2, agent3],
    planning="gpt-4o",
    output="verbose"
)

result = flow.start("Complete this complex task")
4

With Loops

Iterate over items:
from praisonaiagents import Agent, AgentFlow
from praisonaiagents.workflows import loop

processor = Agent(instructions="Process the item: {{item}}")

flow = AgentFlow(
    steps=[
        loop(processor, over="items", parallel=True)
    ],
    variables={
        "items": ["item1", "item2", "item3"]
    }
)

result = flow.start()

​
Flow Patterns

​
Sequential

Steps execute in order, each receiving the previous step’s output. 
flow = AgentFlow(steps=[step1, step2, step3])

​
Conditional Branching

Route to different steps based on conditions. 
from praisonaiagents import AgentFlow, when

flow = AgentFlow(steps=[
    classifier,
    when(
        condition="{{category}} == technical",
        then_steps=[technical_writer],
        else_steps=[general_writer]
    )
])

​
Parallel Execution

Execute steps simultaneously. 
from praisonaiagents import AgentFlow, parallel

flow = AgentFlow(steps=[
    parallel([analyst1, analyst2, analyst3]),
    aggregator
])

​
Loops

Iterate over collections. 
from praisonaiagents import AgentFlow, loop

flow = AgentFlow(steps=[
    loop(processor, over="items"),           # Sequential
    loop(processor, over="items", parallel=True),  # Parallel
    loop(processor, from_csv="data.csv"),    # From CSV file
])

​
Configuration Options

ParameterTypeDefaultDescription
stepsListRequiredWorkflow steps (Agent, function, pattern)
variablesDict{}Variables passed to all steps
llmstr"gpt-4o-mini"Default LLM for agents
processstr"sequential"Process type
outputstr/ConfigNoneOutput configuration
planningbool/strFalseEnable/configure planning
memorybool/ConfigNoneMemory configuration
hooksConfigNoneLifecycle callbacks
historyboolFalseTrack execution history
contextboolTrueContext management

​
Using Functions as Steps

You can use regular Python functions as steps: 
def validate(context):
    """Validate the output from previous step."""
    previous = context.get("previous_result", "")
    if len(previous) < 100:
        return "Content too short, please expand"
    return previous

def format_output(context):
    """Format the final output."""
    result = context.get("previous_result", "")
    return f"# Final Report\n\n{result}"

flow = AgentFlow(steps=[
    writer_agent,
    validate,       # Function step
    editor_agent,
    format_output   # Function step
])

​
Execution History

Enable history for debugging: 
flow = AgentFlow(
    steps=[step1, step2, step3],
    history=True  # Track execution
)

result = flow.start("Process this")

# Get execution trace
history = flow.get_history()
for record in history:
    print(f"{record['step']}: {record['success']}")

​
Best Practices

Name your agents clearly for better debugging:
flow = AgentFlow(steps=[
    Agent(name="DataFetcher", ...),
    Agent(name="DataProcessor", ...),
    Agent(name="ReportGenerator", ...),
])
Use {{variable}} syntax in agent instructions:
Agent(instructions="Write about {{topic}} in {{word_count}} words")
Use hooks for error handling:
def on_error(step, error):
    logging.error(f"Step {step} failed: {error}")

flow = AgentFlow(
    steps=[...],
    hooks=WorkflowHooksConfig(on_step_error=on_error)
)

​
Key Methods

MethodDescription
start(input)Execute the workflow
run(input)Alias for start()
arun(input)Async execution
get_history()Get execution trace
to_dict()Serialize workflow
from_template(uri)Create from template

AgentTeam

Multi-agent coordination

Agent

Individual AI agents

Recipes

Pre-built workflow templates

Process

Execution patterns