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

# Nested Workflows

> Build complex data pipelines with nested loops, parallel execution, and routing patterns

Nested workflows combine loops, parallel execution, and routing to process multi-dimensional data like matrices, hierarchies, or batched items.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    subgraph "Nested Workflows"
        Request[📋 User Request] --> Process[⚙️ Nested Workflows]
        Process --> Result[✅ Result]
    end

    classDef input fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef process fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef output fill:#10B981,stroke:#7C90A0,color:#fff

    class Request input
    class Process process
    class Result output
```

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    subgraph "Nested Workflows"
        In[📝 Input] --> Outer[🔁 Outer Loop]
        Outer --> Inner[⚙️ Inner Loop / Parallel]
        Inner --> Agent[🤖 AgentFlow]
        Agent --> Out[✅ Results]
    end

    classDef input fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef process fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef output fill:#10B981,stroke:#7C90A0,color:#fff

    class In input
    class Outer,Inner process
    class Agent agent
    class Out output
```

## How It Works

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
    participant U as 👤 User
    participant W as AgentFlow
    participant OL as Outer Loop
    participant IL as Inner Loop / Parallel

    U->>W: workflow.start("input")
    W->>OL: iterate over outer list
    loop for each outer item
        OL->>IL: iterate / run parallel
        IL-->>OL: results per item
    end
    OL-->>W: all results
    W-->>U: result["output"]
```

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

runner = Agent(name="Pipeline", instructions="Run nested loop workflows from YAML config.")
runner.start("Process the 3x3 matrix pipeline")
```

The user runs a pipeline; nested loops and parallel steps process each item in order.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart TD
    subgraph Outer["Outer Loop"]
        direction TB
        A[Item 1] --> B[Inner Loop]
        B --> C[Item 2]
        C --> D[Inner Loop]
    end
    
    subgraph Inner["Inner Loop"]
        E[Process A] --> F[Process B]
    end
    
    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff
    class Outer,A,C agent
    class Inner,E,F tool
```

## Quick Start

<Steps>
  <Step title="Simple Usage">
    <CodeGroup>
      ```python Nested Loops theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
      from praisonaiagents import AgentFlow, WorkflowContext, StepResult, loop

      def process_cell(ctx: WorkflowContext) -> StepResult:
          row = ctx.variables.get("row")
          col = ctx.variables.get("col")
          return StepResult(output=f"Processed {row}-{col}")

      # Inner loop processes columns
      inner = loop(process_cell, over="columns", var_name="col")

      # Outer loop processes rows
      outer = loop(steps=[inner], over="rows", var_name="row")

      workflow = AgentFlow(
          steps=[outer],
          variables={
              "rows": ["A", "B"],
              "columns": ["1", "2", "3"]
          }
      )
      workflow.start("Process matrix")
      ```

      ```yaml YAML theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
      workflow:
        steps:
          - loop:
              over: rows
              var_name: row
              steps:
                - loop:
                    over: columns
                    var_name: col
                    steps:
                      - agent: processor
                        action: "Process {{row}}-{{col}}"
      ```
    </CodeGroup>
  </Step>

  <Step title="With Configuration">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import AgentFlow, WorkflowContext, StepResult, loop, parallel

    def security_scan(ctx: WorkflowContext) -> StepResult:
        project = ctx.variables.get("project")
        return StepResult(output=f"Security scan: {project}")

    def run_tests(ctx: WorkflowContext) -> StepResult:
        project = ctx.variables.get("project")
        return StepResult(output=f"Tests passed: {project}")

    parallel_tasks = parallel([security_scan, run_tests])

    workflow = AgentFlow(
        steps=[loop(steps=[parallel_tasks], over="projects", var_name="project")],
        variables={"projects": ["api", "web", "mobile"]},
    )
    workflow.start("Analyse all projects")
    ```
  </Step>
</Steps>

## Supported Nesting Patterns

<CardGroup cols={2}>
  <Card title="Loop in Loop" icon="arrows-rotate">
    Process multi-dimensional data like matrices or hierarchical structures
  </Card>

  <Card title="Parallel in Loop" icon="bolt">
    Run concurrent tasks for each item in a collection
  </Card>

  <Card title="Route in Loop" icon="code-branch">
    Make routing decisions for each item based on conditions
  </Card>

  <Card title="when() in Loop" icon="code-compare">
    Apply conditional logic to each item during iteration using <code>when()</code>
  </Card>
</CardGroup>

## Loop Inside Loop

Process nested data structures by placing one loop inside another.

<Tabs>
  <Tab title="Python">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    from praisonaiagents import AgentFlow, WorkflowContext, StepResult, loop

    results = []

    def process_member(ctx: WorkflowContext) -> StepResult:
        team = ctx.variables.get("team", {})
        member = ctx.variables.get("member")
        results.append(f"{team['name']}/{member}")
        return StepResult(output=f"Processed {member}")

    # Inner: process each member
    member_loop = loop(process_member, over="members", var_name="member")

    # Outer: process each team
    team_loop = loop(steps=[member_loop], over="teams", var_name="team")

    workflow = AgentFlow(
        steps=[team_loop],
        variables={
            "teams": [
                {"name": "Frontend", "members": ["Alice", "Bob"]},
                {"name": "Backend", "members": ["Charlie"]}
            ],
            "members": ["Dev1", "Dev2"]  # Default members
        }
    )
    workflow.start("Process organization")
    # Results: ['Frontend/Dev1', 'Frontend/Dev2', 'Backend/Dev1', 'Backend/Dev2']
    ```
  </Tab>

  <Tab title="YAML">
    ```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    workflow:
      variables:
        teams:
          - name: Frontend
          - name: Backend
        members: ["Dev1", "Dev2"]
      
      steps:
        - loop:
            over: teams
            var_name: team
            steps:
              - loop:
                  over: members
                  var_name: member
                  steps:
                    - agent: processor
                      action: "Process {{member}} from {{team.name}}"
    ```
  </Tab>
</Tabs>

## Parallel Inside Loop

Run multiple tasks concurrently for each item in a loop.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart LR
    subgraph Loop["For Each Project"]
        direction TB
        P1[Project 1] --> PA1
        P2[Project 2] --> PA2
    end
    
    subgraph PA1["Parallel Tasks"]
        direction LR
        S1[Security] 
        T1[Tests]
        R1[Review]
    end
    
    subgraph PA2["Parallel Tasks"]
        direction LR
        S2[Security]
        T2[Tests]
        R2[Review]
    end
    
    style Loop fill:#8B0000,color:#fff
    style PA1 fill:#189AB4,color:#fff
    style PA2 fill:#189AB4,color:#fff
```

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

def security_scan(ctx: WorkflowContext) -> StepResult:
    project = ctx.variables.get("project")
    return StepResult(output=f"Security scan: {project}")

def run_tests(ctx: WorkflowContext) -> StepResult:
    project = ctx.variables.get("project")
    return StepResult(output=f"Tests passed: {project}")

def code_review(ctx: WorkflowContext) -> StepResult:
    project = ctx.variables.get("project")
    return StepResult(output=f"Review complete: {project}")

# Parallel tasks for each project
parallel_tasks = parallel([security_scan, run_tests, code_review])

workflow = AgentFlow(
    steps=[
        loop(
            steps=[parallel_tasks],
            over="projects",
            var_name="project"
        )
    ],
    variables={"projects": ["api", "web", "mobile"]}
)
workflow.start("Analyze all projects")
```

## Route Inside Loop

Apply different processing logic to each item based on conditions.

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

def process_image(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Image processed")

def process_video(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Video processed")

def get_file_type(ctx: WorkflowContext) -> StepResult:
    file = ctx.variables.get("file", {})
    return StepResult(output=file.get("type", "unknown"))

# Route based on file type
type_router = route({
    "image": [process_image],
    "video": [process_video]
})

workflow = AgentFlow(
    steps=[
        loop(
            steps=[get_file_type, type_router],
            over="files",
            var_name="file"
        )
    ],
    variables={
        "files": [
            {"name": "photo.jpg", "type": "image"},
            {"name": "movie.mp4", "type": "video"}
        ]
    }
)
workflow.start("Process files")
```

## Retry, guardrails, and `output_file` inside nested steps

As of PraisonAI [#3086](https://github.com/MervinPraison/PraisonAI/pull/3086), steps inside a nested pattern (`parallel`, `loop`, `route`, `if_`, `repeat`) — and steps inside hierarchical mode — inherit the same policy machinery the top-level workflow uses:

* **`max_retries`** (default `3`) with exponential backoff `2 ** (attempt - 1)` seconds between attempts.
* **`is_retryable`** on the raised exception is honoured — set it `False` on your exception class to short-circuit retries for known non-retryable failures.
* **`guardrails`** (canonical name; `guardrail` still works) — on a failed check, the step retries with the validator's feedback merged into the next attempt as `"Previous attempt feedback: {feedback}"`.
* **`output_file`** — the step's output is written to disk after a successful run, with `{{variable}}` substitution from the workflow variables.

Before this fix, these policies were only applied to top-level steps; nested-pattern steps silently dropped them. If you had a `parallel(...)` step with `max_retries=5` and it never retried, that was the bug — upgrade to pick up the fix.

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

worker = Agent(name="Flaky", instructions="Sometimes fails, sometimes succeeds")

flow = AgentFlow(steps=[
    parallel(
        worker,             # inherits max_retries / guardrails / output_file from the step spec
        worker,
    ),
])
flow.start("Do work in parallel — each branch retries independently")
```

## Depth Limit Protection

<Warning>
  Nested workflows have a maximum depth of **5 levels** to prevent infinite recursion and stack overflow errors.
</Warning>

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

print(f"Maximum nesting depth: {MAX_NESTING_DEPTH}")  # 5

# This will raise ValueError if depth exceeds 5
try:
    # Creating 6+ levels of nesting
    inner = loop(handler, over="items")
    for _ in range(5):
        inner = loop(steps=[inner], over="items")
    
    workflow = AgentFlow(steps=[inner], variables={"items": ["a"]})
    workflow.start("test")
except ValueError as e:
    print(f"Error: {e}")
    # "Maximum nesting depth (5) exceeded..."
```

## Best Practices

<AccordionGroup>
  <Accordion title="Keep Nesting Shallow">
    Aim for 2-3 levels of nesting maximum. If you need more, consider restructuring your workflow or breaking it into sub-workflows.
  </Accordion>

  <Accordion title="Use Meaningful Variable Names">
    Use descriptive `var_name` values like `customer`, `order`, `item` instead of generic names like `x`, `y`, `z`.
  </Accordion>

  <Accordion title="Consider Parallel for Performance">
    When processing independent items, use `parallel=True` in loops to speed up execution.
  </Accordion>

  <Accordion title="Handle Errors Gracefully">
    Add error handling in your step functions to prevent one failed item from stopping the entire workflow.
  </Accordion>
</AccordionGroup>

## Related

<CardGroup cols={2}>
  <Card title="Conditional Branching" icon="code-branch" href="/docs/features/conditional-branching">
    Use if/then/else patterns for dynamic workflow paths
  </Card>

  <Card title="Workflow Loops" icon="arrows-rotate" href="/docs/features/workflow-loop">
    Learn the basics of loop patterns
  </Card>

  <Card title="Parallel Execution" icon="bolt" href="/docs/features/workflow-parallel">
    Run steps concurrently for better performance
  </Card>

  <Card title="Routing" icon="route" href="/docs/features/workflow-routing">
    Route workflow execution based on conditions
  </Card>
</CardGroup>