Workflow Patterns

Overview of all workflow patterns: routing, parallel, loop, and repeat

PraisonAI provides four powerful workflow patterns that can be combined to create complex, production-ready workflows.

from praisonaiagents import Agent, AgentFlow

researcher = Agent(name="researcher", instructions="Gather facts")
writer = Agent(name="writer", instructions="Draft the article")
flow = AgentFlow(agents=[researcher, writer], steps=[researcher, writer])
flow.start("Write a short article about renewable energy")

The user picks a pattern; AgentFlow wires sequential, parallel, or routed steps accordingly.

from praisonaiagents import Agent, AgentFlow

planner = Agent(name="Planner", instructions="Break work into steps")
workflow = AgentFlow(agents=[planner], steps=["plan", "execute", "review"])
workflow.start("Ship the feature")

The user describes a goal; the workflow runs the composed pattern end to end.

Quick Comparison

Pattern	Purpose	Use When
route()	Decision-based branching	Output determines next steps
parallel()	Concurrent execution	Independent tasks can run together
loop()	Iterate over data	Processing lists, CSV files
repeat()	Repeat until condition	Iterative improvement

Import

from praisonaiagents import AgentFlow, WorkflowContext, StepResult
from praisonaiagents import route, parallel, loop, repeat

AgentFlow is the primary class for deterministic pipelines (v1.0+). Workflow and Pipeline are silent aliases kept for backward compatibility — new code should use AgentFlow.

Quick Start

1. Import workflow helpers

   from praisonaiagents import AgentFlow, WorkflowContext, StepResult
   from praisonaiagents import route, parallel, loop, repeat
   
2. Combine patterns in one workflow

   workflow = AgentFlow(steps=[
       parallel([research_market, research_competitors, research_customers]),
       analyze_findings,
       route({
           "positive": [expand_analysis],
           "negative": [summarize_concerns],
           "default": [standard_report]
       }),
       loop(process_recommendation, over="recommendations"),
       repeat(refine_output, until=meets_quality_threshold, max_iterations=3),
       format_final_report
   ])
   
   result = workflow.start("Analyse quarterly performance")
   

How It Works

Phase	What happens
1. Compose	You order route, parallel, loop, and repeat steps in AgentFlow
2. Dispatch	Each pattern step runs its agents in the right execution shape
3. Merge	Step results flow into the next step until the workflow completes

Pattern Overview

1. Routing (Decision Branching)

Route to different steps based on previous output:

workflow = AgentFlow(steps=[
    classifier,
    route({
        "approve": [approve_handler],
        "reject": [reject_handler],
        "default": [fallback]
    })
])

📖 Full Documentation →

2. Parallel (Concurrent Execution)

Execute multiple steps at the same time:

workflow = AgentFlow(steps=[
    parallel([research_a, research_b, research_c]),
    aggregator  # Combines results
])

📖 Full Documentation →

3. Loop (Iterate Over Data)

Process each item in a list or file:

# From list
workflow = AgentFlow(
    steps=[loop(processor, over="items")],
    variables={"items": ["a", "b", "c"]}
)

# From CSV
workflow = AgentFlow(steps=[
    loop(processor, from_csv="data.csv")
])

📖 Full Documentation →

4. Repeat (Evaluator-Optimizer)

Repeat until a condition is met:

workflow = AgentFlow(steps=[
    repeat(
        generator,
        until=lambda ctx: "done" in ctx.previous_result,
        max_iterations=5
    )
])

📖 Full Documentation →

Combining Patterns

Patterns can be combined for complex workflows:

workflow = AgentFlow(steps=[
    # Step 1: Parallel research from multiple sources
    parallel([
        research_market,
        research_competitors,
        research_customers
    ]),
    
    # Step 2: Route based on findings
    analyze_findings,
    route({
        "positive": [expand_analysis],
        "negative": [summarize_concerns],
        "default": [standard_report]
    }),
    
    # Step 3: Process each recommendation
    loop(process_recommendation, over="recommendations"),
    
    # Step 4: Refine until quality threshold
    repeat(
        refine_output,
        until=meets_quality_threshold,
        max_iterations=3
    ),
    
    # Step 5: Final output
    format_final_report
])

Common Workflow Architectures

Orchestrator-Worker

workflow = AgentFlow(steps=[
    orchestrator,  # Decides which workers to use
    route({
        "type_a": [worker_a],
        "type_b": [worker_b],
        "type_c": [worker_c]
    }),
    synthesizer  # Combines worker outputs
])

Fan-Out/Fan-In

workflow = AgentFlow(steps=[
    splitter,  # Splits work into parts
    parallel([processor_1, processor_2, processor_3]),
    aggregator  # Combines results
])

Batch Processing Pipeline

workflow = AgentFlow(steps=[
    loop(validate_item, from_csv="input.csv"),
    loop(transform_item, over="validated_items"),
    loop(load_item, over="transformed_items"),
    generate_report
])

Self-Improving Agent

workflow = AgentFlow(steps=[
    initial_generator,
    repeat(
        improve_output,
        until=quality_check,
        max_iterations=5
    ),
    final_polish
])

Pattern Selection Guide

Pattern Tips

1. Start Simple

# Start with sequential
workflow = AgentFlow(steps=[step1, step2, step3])

# Add patterns as needed
workflow = AgentFlow(steps=[
    step1,
    parallel([step2a, step2b]),  # Optimize with parallel
    step3
])

2. Handle Errors

def safe_step(ctx: WorkflowContext) -> StepResult:
    try:
        # Your logic
        return StepResult(output="Success")
    except Exception as e:
        return StepResult(output=f"Error: {e}")

3. Use Verbose Mode

result = workflow.start("input", )
# Shows step-by-step progress

4. Track State with Variables

def my_step(ctx: WorkflowContext) -> StepResult:
    count = ctx.variables.get("count", 0) + 1
    return StepResult(
        output=f"Count: {count}",
        variables={"count": count}
    )

API Reference

Function	Signature
route()	route(routes: Dict[str, List], default: Optional[List] = None)
parallel()	parallel(steps: List)
loop()	loop(step, over=None, from_csv=None, from_file=None, var_name="item")
repeat()	repeat(step, until=None, max_iterations=10)

Best Practices

Compose patterns instead of nesting deeply

Combine route, parallel, loop, and repeat in readable sequences — avoid deeply nested workflow trees.

Track state with workflow variables

Use ctx.variables for counts and flags instead of global Python state.

Enable verbose mode while designing

Step-by-step progress output helps validate routing and loop behaviour before production.

Always define a default route

Unexpected classifier output should fall back to a safe path, not fail silently.

See Also

Workflow Routing

Decision-based branching

Parallel Execution

Concurrent step execution

Loop Processing

Iterate over data

Repeat Pattern

Evaluator-optimizer pattern

Workflow Input Resolution

How the value passed to start(...) and {{input}} is resolved