> ## Documentation Index
> Fetch the complete documentation index at: https://docs.praison.ai/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

# Workflow Patterns

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

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TD
    subgraph "Pattern Types"
        A[route] --> A1[Decision Branching]
        B[parallel] --> B1[Concurrent Execution]
        C[loop] --> C1[Iterate Over Data]
        D[repeat] --> D1[Until Condition]
    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
# Or use Pipeline (alias for Workflow)
from praisonaiagents import Pipeline
from praisonaiagents import route, parallel, loop, repeat
```

<Tip>
  `Pipeline` and `Workflow` are the same class. Use whichever term you prefer!
</Tip>

## 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 TD
    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]
```

## Best Practices

### 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)`                           |

## See Also

<CardGroup cols={2}>
  <Card title="Workflow Routing" icon="route" href="/features/workflow-routing">
    Decision-based branching
  </Card>

  <Card title="Parallel Execution" icon="arrows-split-up-and-left" href="/features/workflow-parallel">
    Concurrent step execution
  </Card>

  <Card title="Loop Processing" icon="arrows-rotate" href="/features/workflow-loop">
    Iterate over data
  </Card>

  <Card title="Repeat Pattern" icon="rotate" href="/features/workflow-repeat">
    Evaluator-optimizer pattern
  </Card>
</CardGroup>