> ## 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.

# AgentFlow

> Deterministic workflow orchestration for multi-step pipelines

AgentFlow is the workflow orchestration system for creating deterministic, multi-step pipelines. It supports sequential execution, conditional branching, loops, and parallel processing.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart LR
    subgraph AgentFlow["AgentFlow Pipeline"]
        direction LR
        S1["Step 1"] --> S2["Step 2"]
        S2 --> S3["Step 3"]
    end
    
    Input["Input"] --> AgentFlow
    AgentFlow --> Output["Output"]
    
    classDef input fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef process fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef output fill:#10B981,stroke:#7C90A0,color:#fff
    
    class Input input
    class S1,S2,S3 process
    class Output output
```

## Quick Start

<Steps>
  <Step title="Simple Pipeline">
    Create a basic sequential workflow:

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import Agent, AgentFlow

    # Define agents as steps
    writer = Agent(
        name="Writer",
        instructions="Write content based on the input"
    )

    editor = Agent(
        name="Editor", 
        instructions="Edit and improve the content"
    )

    # Create flow
    flow = AgentFlow(
        steps=[writer, editor]
    )

    result = flow.start("Write about artificial intelligence")
    print(result["output"])
    ```
  </Step>

  <Step title="With Variables">
    Pass variables between steps:

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import Agent, AgentFlow

    flow = AgentFlow(
        steps=[researcher, writer, publisher],
        variables={
            "topic": "Machine Learning",
            "word_count": 500,
            "tone": "professional"
        }
    )

    result = flow.start()
    ```
  </Step>

  <Step title="Planning Mode">
    Enable AI-powered planning for complex workflows:

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import Agent, AgentFlow

    # Option 1: Enable planning with default model
    flow = AgentFlow(
        steps=[agent1, agent2, agent3],
        planning=True,
        output="verbose"
    )

    # Option 2: Specify a planning model (also enables planning)
    flow = AgentFlow(
        steps=[agent1, agent2, agent3],
        planning="gpt-4o",
        output="verbose"
    )

    result = flow.start("Complete this complex task")
    ```
  </Step>

  <Step title="With Loops">
    Iterate over items:

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import Agent, AgentFlow
    from praisonaiagents.workflows import loop

    processor = Agent(instructions="Process the item: {{item}}")

    flow = AgentFlow(
        steps=[
            loop(processor, over="items", parallel=True)
        ],
        variables={
            "items": ["item1", "item2", "item3"]
        }
    )

    result = flow.start()
    ```
  </Step>
</Steps>

## Flow Patterns

### Sequential

Steps execute in order, each receiving the previous step's output.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart LR
    subgraph Sequential["Sequential Flow"]
        direction LR
        A["Step 1"] --> B["Step 2"]
        B --> C["Step 3"]
    end
    
    style A fill:#6366F1,stroke:#7C90A0,color:#fff
    style B fill:#F59E0B,stroke:#7C90A0,color:#fff
    style C fill:#10B981,stroke:#7C90A0,color:#fff
```

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flow = AgentFlow(steps=[step1, step2, step3])
```

***

### Conditional Branching

Route to different steps based on conditions.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart TB
    subgraph Conditional["Conditional Branching"]
        A["Classifier"] --> B{Condition?}
        B -->|"technical"| C["Technical Writer"]
        B -->|"general"| D["General Writer"]
        C --> E["Output"]
        D --> E
    end
    
    style A fill:#6366F1,stroke:#7C90A0,color:#fff
    style B fill:#F59E0B,stroke:#7C90A0,color:#fff
    style C fill:#189AB4,stroke:#7C90A0,color:#fff
    style D fill:#189AB4,stroke:#7C90A0,color:#fff
    style E fill:#10B981,stroke:#7C90A0,color:#fff
```

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, when

flow = AgentFlow(steps=[
    classifier,
    when(
        condition="{{category}} == technical",
        then_steps=[technical_writer],
        else_steps=[general_writer]
    )
])
```

***

### Parallel Execution

Execute steps simultaneously.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart TB
    subgraph Parallel["Parallel Execution"]
        A["Input"] --> B["Analyst 1"]
        A --> C["Analyst 2"]
        A --> D["Analyst 3"]
        B --> E["Aggregator"]
        C --> E
        D --> E
        E --> F["Output"]
    end
    
    style A fill:#6366F1,stroke:#7C90A0,color:#fff
    style B fill:#F59E0B,stroke:#7C90A0,color:#fff
    style C fill:#F59E0B,stroke:#7C90A0,color:#fff
    style D fill:#F59E0B,stroke:#7C90A0,color:#fff
    style E fill:#189AB4,stroke:#7C90A0,color:#fff
    style F fill:#10B981,stroke:#7C90A0,color:#fff
```

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, parallel

flow = AgentFlow(steps=[
    parallel([analyst1, analyst2, analyst3]),
    aggregator
])
```

***

### Loops

Iterate over collections.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart LR
    subgraph Loop["Loop Pattern"]
        A["Items"] --> B["Loop"]
        B --> C["Process Item 1"]
        B --> D["Process Item 2"]
        B --> E["Process Item N"]
        C --> F["Results"]
        D --> F
        E --> F
    end
    
    style A fill:#6366F1,stroke:#7C90A0,color:#fff
    style B fill:#F59E0B,stroke:#7C90A0,color:#fff
    style C fill:#189AB4,stroke:#7C90A0,color:#fff
    style D fill:#189AB4,stroke:#7C90A0,color:#fff
    style E fill:#189AB4,stroke:#7C90A0,color:#fff
    style F fill:#10B981,stroke:#7C90A0,color:#fff
```

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import AgentFlow, loop

flow = AgentFlow(steps=[
    loop(processor, over="items"),           # Sequential
    loop(processor, over="items", parallel=True),  # Parallel
    loop(processor, from_csv="data.csv"),    # From CSV file
])
```

## Configuration Options

| Parameter   | Type        | Default         | Description                               |
| ----------- | ----------- | --------------- | ----------------------------------------- |
| `steps`     | List        | Required        | Workflow steps (Agent, function, pattern) |
| `variables` | Dict        | `{}`            | Variables passed to all steps             |
| `llm`       | str         | `"gpt-4o-mini"` | Default LLM for agents                    |
| `process`   | str         | `"sequential"`  | Process type                              |
| `output`    | str/Config  | None            | Output configuration                      |
| `planning`  | bool/str    | `False`         | Enable/configure planning                 |
| `memory`    | bool/Config | None            | Memory configuration                      |
| `hooks`     | Config      | None            | Lifecycle callbacks                       |
| `history`   | bool        | `False`         | Track execution history                   |
| `context`   | bool        | `True`          | Context management                        |

## Using Functions as Steps

You can use regular Python functions as steps:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
def validate(context):
    """Validate the output from previous step."""
    previous = context.get("previous_result", "")
    if len(previous) < 100:
        return "Content too short, please expand"
    return previous

def format_output(context):
    """Format the final output."""
    result = context.get("previous_result", "")
    return f"# Final Report\n\n{result}"

flow = AgentFlow(steps=[
    writer_agent,
    validate,       # Function step
    editor_agent,
    format_output   # Function step
])
```

## Execution History

Enable history for debugging:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flow = AgentFlow(
    steps=[step1, step2, step3],
    history=True  # Track execution
)

result = flow.start("Process this")

# Get execution trace
history = flow.get_history()
for record in history:
    print(f"{record['step']}: {record['success']}")
```

## Best Practices

<AccordionGroup>
  <Accordion title="Step Naming">
    Name your agents clearly for better debugging:

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    flow = AgentFlow(steps=[
        Agent(name="DataFetcher", ...),
        Agent(name="DataProcessor", ...),
        Agent(name="ReportGenerator", ...),
    ])
    ```
  </Accordion>

  <Accordion title="Variable Templating">
    Use `{{variable}}` syntax in agent instructions:

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    Agent(instructions="Write about {{topic}} in {{word_count}} words")
    ```
  </Accordion>

  <Accordion title="Error Handling">
    Use hooks for error handling:

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    def on_error(step, error):
        logging.error(f"Step {step} failed: {error}")

    flow = AgentFlow(
        steps=[...],
        hooks=WorkflowHooksConfig(on_step_error=on_error)
    )
    ```
  </Accordion>
</AccordionGroup>

## Key Methods

| Method               | Description          |
| -------------------- | -------------------- |
| `start(input)`       | Execute the workflow |
| `run(input)`         | Alias for start()    |
| `arun(input)`        | Async execution      |
| `get_history()`      | Get execution trace  |
| `to_dict()`          | Serialize workflow   |
| `from_template(uri)` | Create from template |

## Related

<CardGroup cols={2}>
  <Card title="AgentTeam" icon="users" href="/docs/concepts/agentteam">
    Multi-agent coordination
  </Card>

  <Card title="Agent" icon="robot" href="/docs/concepts/agents">
    Individual AI agents
  </Card>

  <Card title="Recipes" icon="book" href="/docs/concepts/recipes">
    Pre-built workflow templates
  </Card>

  <Card title="Process" icon="gears" href="/docs/concepts/process">
    Execution patterns
  </Card>
</CardGroup>