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.

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

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

   workflow:
     steps:
       - loop:
           over: rows
           var_name: row
           steps:
             - loop:
                 over: columns
                 var_name: col
                 steps:
                   - agent: processor
                     action: "Process {{row}}-{{col}}"
   
2. 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

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

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

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 to prevent one failed item from stopping the entire workflow.

Related

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