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.

Diagram

How It Works

Diagram

python

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.

Diagram

Quick Start

Simple Usage

Nested Loops

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

workflow:
  steps:
    - loop:
        over: rows
        var_name: row
        steps:
          - loop:
              over: columns
              var_name: col
              steps:
                - agent: processor
                  action: "Process {{row}}-{{col}}"

With Configuration

python

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")

Supported Nesting Patterns

Loop in Loop

Process multi-dimensional data like matrices or hierarchical structures

Parallel in Loop

Run concurrent tasks for each item in a collection

Route in Loop

Make routing decisions for each item based on conditions

when() in Loop

Apply conditional logic to each item during iteration using <code>when()</code>

Loop Inside Loop

Process nested data structures by placing one loop inside another.

Python

python

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']

YAML

yaml

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}}"

Parallel Inside Loop

Run multiple tasks concurrently for each item in a loop.

As of PraisonAI #4932, a nested Loop(..., output_variable="x") or parallel(...) inside a parallel branch now propagates its aggregate variable back to the workflow scope — previously that write was lost. As of #4947, the loop body’s output_variable writes now also propagate out of the loop (and, when the loop sits inside a parallel() branch, out of that branch too). See Named outputs from branches and Named outputs from a loop body.

Diagram

python

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

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, steps inside a nested pattern (parallel, loop, route, if_, repeat, Discussion) — 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

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")

Failure propagation in nested patterns

As of PraisonAI #4194, a failing step inside any nested pattern (loop, parallel, when/if_, route, repeat) propagates the same way it does at the top level: the run is marked status="failed" and the exception text is never handed forward as the next step’s input.

Diagram

on_error="stop" (the Task default) — a failing inner step raises a stop signal that the pattern executor propagates all the way out to the top-level workflow. The enclosing workflow aborts, not just the pattern iteration, and downstream top-level steps do not run.
on_error="continue" — the workflow keeps going past the failing step, but the final result["status"] is "failed", and the error is not fed forward to the next step.

As of PraisonAI #4203, a nested on_error inside parallel() now behaves the same as inside route() / loop() / repeat() / if_() — a stopping branch halts the enclosing workflow even under on_failure="partial_ok". See Stop propagation from nested steps.

python

from praisonaiagents import Task, AgentFlow, WorkflowContext, StepResult, loop

def boom(ctx: WorkflowContext) -> StepResult:
    raise RuntimeError("boom")

def after(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="this never runs")

workflow = AgentFlow(
    steps=[
        loop(steps=[Task(name="boom", handler=boom, max_retries=0)], over="items"),
        Task(name="after", handler=after),
    ],
    variables={"items": ["only"]},
)
result = workflow.start("go")

print(result["status"])                                     # "failed"
print(any(s["step"] == "after" for s in result["steps"]))   # False — stop propagated out

Depth Limit Protection

python

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

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.

Use Meaningful Variable Names

Use descriptive var_name values like customer, order, item instead of generic names like x, y, z.

Consider Parallel for Performance

When processing independent items, use parallel=True in loops to speed up execution.

Handle Errors Gracefully

Add error handling in your step functions if you want per-item recovery inside a loop.

As of PraisonAI #4194, a raising iteration inside loop(parallel=True) is recorded as {"step": "loop_{idx}", "output": None, "error": str(e)} — the exception text is no longer folded into output. The run is marked status="failed" as soon as any iteration fails, even if the loop keeps processing later items. Sequential loop iterations now honour on_error="stop" (the Task default): a failing step halts the current iteration and, via the propagated stop signal, the enclosing workflow, so downstream top-level steps do not run. Add try/except inside your step function only when you want to convert an error into a valid StepResult that downstream steps can consume normally.

Conditional Branching

Use if/then/else patterns for dynamic workflow paths

Workflow Loops

Learn the basics of loop patterns

Parallel Execution

Run steps concurrently for better performance

Routing

Route workflow execution based on conditions

Nested Workflows

How It Works

Quick Start

Supported Nesting Patterns

Loop Inside Loop

Python

YAML

Parallel Inside Loop

Route Inside Loop

Retry, guardrails, and output_file inside nested steps

Failure propagation in nested patterns

Depth Limit Protection

Best Practices

Related

Retry, guardrails, and `output_file` inside nested steps