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.

python

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.

Diagram

python

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, or a small fixed list of steps run in order each iteration
`Discussion`	N-agent conversation loop	Multiple agents take turns on the same thread

Import

python

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

Quick Start

Import workflow helpers

python

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

Combine patterns in one workflow

python

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

Diagram

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:

python

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:

python

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

Context pass-through: each branch sees the exact upstream output by default. See Context pass-through to branches for the opt-in token-saving modes (context="truncate" / "summarize").

📖 Full Documentation →

3. Loop (Iterate Over Data)

Process each item in a list or file:

python

# 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")
])

Context pass-through: each parallel iteration sees the exact upstream output by default. See Context pass-through to parallel iterations for the opt-in context="truncate" token-saving mode.

📖 Full Documentation →

4. Repeat (Evaluator-Optimizer)

Repeat until a condition is met:

python

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

📖 Full Documentation →

5. Discussion (N-Agent Conversation Loop)

Have N agents take turns on the same thread:

python

from praisonaiagents import Discussion

workflow = AgentFlow(steps=[
    Discussion(
        [critic, author],
        rounds=3,
        until=lambda ctx: "AGREED" in (ctx.previous_result or ""),
    )
])

📖 Full Documentation →

Combining Patterns

Patterns can be combined for complex workflows:

python

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

python

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

python

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

Batch Processing Pipeline

python

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

python

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

Pattern Selection Guide

Diagram

Pattern Tips

1. Start Simple

python

# 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

python

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

python

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

4. Track State with Variables

python

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, max_workers=None, on_failure="partial_ok", timeout=None, context="full")`
`loop()`	`loop(step, over=None, from_csv=None, from_file=None, var_name="item", parallel=False, context="full")`
`repeat()`	`repeat(step, until=None, max_iterations=10)`
`Discussion(...)`	`Discussion(agents=[...], rounds=3, until=None, select=None, name="discussion")`

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.

Workflow Patterns

Quick Comparison

Import

Quick Start

How It Works

Pattern Overview

1. Routing (Decision Branching)

2. Parallel (Concurrent Execution)

3. Loop (Iterate Over Data)

4. Repeat (Evaluator-Optimizer)

5. Discussion (N-Agent Conversation Loop)

Combining Patterns

Common Workflow Architectures

Orchestrator-Worker

Fan-Out/Fan-In

Batch Processing Pipeline

Self-Improving Agent

Pattern Selection Guide

Pattern Tips

1. Start Simple

2. Handle Errors

3. Use Verbose Mode

4. Track State with Variables

API Reference

Best Practices

See Also