> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
In[Goal] --> Compose[Compose patterns]
Compose --> Run[AgentFlow runs steps]
Run --> Out[Result]
classDef input fill:#8B0000,stroke:#7C90A0,color:#fff
classDef process fill:#189AB4,stroke:#7C90A0,color:#fff
classDef output fill:#10B981,stroke:#7C90A0,color:#fff
class In input
class Compose,Run process
class Out output
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, WorkflowContext, StepResult
from praisonaiagents import route, parallel, loop, repeat
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant User
participant AgentFlow
participant Pattern
participant Agent
User->>AgentFlow: start("goal")
AgentFlow->>Pattern: Run the composed step
Pattern->>Agent: Dispatch work (route / parallel / loop / repeat)
Agent-->>Pattern: Step result
Pattern-->>AgentFlow: Merged output
AgentFlow-->>User: Final result
```
| 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = AgentFlow(steps=[
classifier,
route({
"approve": [approve_handler],
"reject": [reject_handler],
"default": [fallback]
})
])
```
📖 [Full Documentation →](/features/workflow-routing)
### 2. Parallel (Concurrent Execution)
Execute multiple steps at the same time:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = AgentFlow(steps=[
parallel([research_a, research_b, research_c]),
aggregator # Combines results
])
```
📖 [Full Documentation →](/features/workflow-parallel)
### 3. Loop (Iterate Over Data)
Process each item in a list or file:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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 →](/features/workflow-loop)
### 4. Repeat (Evaluator-Optimizer)
Repeat until a condition is met:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = AgentFlow(steps=[
repeat(
generator,
until=lambda ctx: "done" in ctx.previous_result,
max_iterations=5
)
])
```
📖 [Full Documentation →](/features/workflow-repeat)
## Combining Patterns
Patterns can be combined for complex workflows:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = AgentFlow(steps=[
splitter, # Splits work into parts
parallel([processor_1, processor_2, processor_3]),
aggregator # Combines results
])
```
### Batch Processing Pipeline
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
workflow = AgentFlow(steps=[
initial_generator,
repeat(
improve_output,
until=quality_check,
max_iterations=5
),
final_polish
])
```
## Pattern Selection Guide
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
A[What do you need?] --> B{Multiple paths?}
B -->|Yes| C[route]
B -->|No| D{Independent tasks?}
D -->|Yes| E[parallel]
D -->|No| F{Process list/file?}
F -->|Yes| G[loop]
F -->|No| H{Iterate until done?}
H -->|Yes| I[repeat]
H -->|No| J[Sequential steps]
classDef decision fill:#6366F1,stroke:#7C90A0,color:#fff
classDef option fill:#10B981,stroke:#7C90A0,color:#fff
class A,B,D,F,H decision
class C,E,G,I,J option
```
## Pattern Tips
### 1. Start Simple
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
result = workflow.start("input", )
# Shows step-by-step progress
```
### 4. Track State with Variables
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
Combine route, parallel, loop, and repeat in readable sequences — avoid deeply nested workflow trees.
Use `ctx.variables` for counts and flags instead of global Python state.
Step-by-step progress output helps validate routing and loop behaviour before production.
Unexpected classifier output should fall back to a safe path, not fail silently.
## See Also
Decision-based branching
Concurrent step execution
Iterate over data
Evaluator-optimizer pattern