AgentFlow

Deterministic workflow orchestration for multi-step pipelines

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, 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

Parameter	Type	Default	Description
steps	List	Required	Workflow steps (Agent, function, pattern)
variables	Dict	{}	Variables passed to all steps
llm	str	"gpt-4o-mini"	Default LLM for agents
process	str	"sequential"	Process type
output	str/Config	None	Output configuration
planning	bool/str	False	Enable/configure planning
memory	bool/Config	None	Memory configuration
hooks	Config	None	Lifecycle callbacks
history	bool	False	Track execution history
context	bool	True	Context 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

Step Naming

Name your agents clearly for better debugging:

flow = AgentFlow(steps=[
    Agent(name="DataFetcher", ...),
    Agent(name="DataProcessor", ...),
    Agent(name="ReportGenerator", ...),
])

Variable Templating

Use {{variable}} syntax in agent instructions:

Agent(instructions="Write about {{topic}} in {{word_count}} words")

Error Handling

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

Method	Description
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

Related

AgentTeam

Multi-agent coordination

Agent

Individual AI agents

Recipes

Pre-built workflow templates

Process

Execution patterns