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
Import workflow helpers
from praisonaiagents import AgentFlow , WorkflowContext , StepResult from praisonaiagents import route , parallel , loop , repeat
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
