Nested Workflows

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

How It Works

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.

Quick Start

Simple Usage

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

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

With Configuration

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 when()

Loop Inside Loop

Process nested data structures by placing one loop inside another.

Python
YAML

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

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.

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.

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) — 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.

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.

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.

Runs that previously reported completed while containing a failed step now correctly report failed. See PR #4194 — the old behaviour silently swallowed nested failures.

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

Nested workflows have a maximum depth of 5 levels to prevent infinite recursion and stack overflow errors.

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

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

How It Works

Quick Start

Supported Nesting Patterns

Loop in Loop

Parallel in Loop

Route in Loop

when() in Loop

Loop Inside Loop

Parallel Inside Loop

Route Inside Loop

Retry, guardrails, and `output_file` inside nested steps

Failure propagation in nested patterns

Depth Limit Protection

Best Practices

Conditional Branching

Workflow Loops

Parallel Execution

Routing

​How It Works

​Quick Start

​Supported Nesting Patterns

Loop in Loop

Parallel in Loop

Route in Loop

when() in Loop

​Loop Inside Loop

​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

Conditional Branching

Workflow Loops

Parallel Execution

Routing

How It Works

Quick Start

Supported Nesting Patterns

Loop Inside Loop

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