Workflow Parallel Execution

Execute multiple steps concurrently using the parallel() helper

Execute multiple steps concurrently and combine their results. This pattern is ideal for independent tasks that can run simultaneously.

Steps inside this pattern inherit the same max_retries, guardrails, and output_file policies as top-level steps. See Nested workflows → Retry, guardrails, and output_file.

As of PraisonAI #4932, an output_variable set inside a branch is preserved after the parallel block. Earlier releases silently discarded it.

from praisonaiagents import Agent, AgentFlow, parallel, WorkflowContext, StepResult

coordinator = Agent(name="Coordinator", instructions="Merge parallel results")

def task_a(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Result A")

def task_b(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="Result B")

workflow = AgentFlow(agents=[coordinator], steps=[parallel([task_a, task_b])])
workflow.start("Run independent tasks in parallel")

The user starts one workflow; independent steps run concurrently and merge results.

Quick Start

1. Define parallel workers

   from praisonaiagents import AgentFlow, WorkflowContext, StepResult
   from praisonaiagents import parallel
   import time
   
   def research_market(ctx: WorkflowContext) -> StepResult:
       time.sleep(0.1)  # Simulate work
       return StepResult(output="📊 Market: Growth 15% YoY")
   
   def research_competitors(ctx: WorkflowContext) -> StepResult:
       time.sleep(0.1)  # Simulate work
       return StepResult(output="🏢 Competitors: 3 major players")
   
   def research_customers(ctx: WorkflowContext) -> StepResult:
       time.sleep(0.1)  # Simulate work
       return StepResult(output="👥 Customers: 85% satisfaction")
   
   # Aggregator
   def summarize(ctx: WorkflowContext) -> StepResult:
       outputs = ctx.variables.get("parallel_outputs", [])
       summary = "📋 SUMMARY:\n" + "\n".join(f"  • {o}" for o in outputs)
       return StepResult(output=summary)
   
2. Run parallel workflow

   workflow = AgentFlow(steps=[
       parallel([research_market, research_competitors, research_customers]),
       summarize
   ])
   
   result = workflow.start("Analyze business")
   print(result["output"])
   

Output:

⚡ Running 3 steps in parallel...
✅ Parallel complete: 3 results
✅ summarize: 📋 SUMMARY:...

📋 SUMMARY:
  • 📊 Market: Growth 15% YoY
  • 🏢 Competitors: 3 major players
  • 👥 Customers: 85% satisfaction

API Reference

parallel()

parallel(
    steps: List, 
    max_workers: Optional[int] = None,
    on_failure: str = "partial_ok"
) -> Parallel

Parameters

Parameter	Type	Default	Description
steps	List	—	List of steps to execute concurrently
max_workers	Optional[int]	None	Cap on ThreadPoolExecutor workers. Defaults to min(3, len(steps))
on_failure	"partial_ok" | "fail_fast" | "fail_all"	"partial_ok"	Failure-handling strategy

Accessing Results

After parallel execution, results are available in ctx.variables:

Variable	Type	Description
parallel_outputs	List[str]	List of all outputs in order

def aggregator(ctx: WorkflowContext) -> StepResult:
    outputs = ctx.variables["parallel_outputs"]
    # outputs = ["Result A", "Result B", "Result C"]
    return StepResult(output=f"Combined: {len(outputs)} results")

Index alignment under partial_ok. A failed branch contributes None at the same index as its step, so parallel_outputs[i] is always the result of parallel_step.steps[i]. As of PraisonAI #4203, a downstream reader of parallel_outputs[i] can no longer silently receive a later branch’s result once an earlier branch fails.

Named outputs from branches

Every step inside parallel([...]) may set output_variable. As of PraisonAI #4932, that write is visible to any step after the parallel block, keyed by that exact name — not just as a positional entry in parallel_outputs.

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

def branch_a(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="A")

def branch_b(ctx: WorkflowContext) -> StepResult:
    return StepResult(output="B")

def after(ctx: WorkflowContext) -> StepResult:
    a = ctx.variables["branch_a_output"]
    b = ctx.variables["branch_b_output"]
    return StepResult(output=f"{a} + {b}")

workflow = AgentFlow(steps=[
    parallel([
        Task(name="branch_a", handler=branch_a, output_variable="branch_a_output"),
        Task(name="branch_b", handler=branch_b, output_variable="branch_b_output"),
    ]),
    Task(name="after", handler=after),
])
result = workflow.start("go")
print(result["variables"]["branch_a_output"])  # "A"
print(result["variables"]["branch_b_output"])  # "B"

The branch’s own output_variable is stored under that exact name; the ordered parallel_outputs list is still there for aggregators that want positional access.

Each branch runs against its own deep copy of the variables, so branches never race on one shared dict. After the block, only the keys a branch actually wrote merge back into the shared scope — untouched variables keep the parent’s own object.

Two branches wrote the same variable

When two branches write the same output_variable, declaration order wins — the later branch overwrites the earlier, matching what running the same steps sequentially would produce. The result never depends on which thread finishes first.

workflow = AgentFlow(steps=[
    parallel([
        Task(name="first",  handler=lambda ctx: StepResult(output="FROM-FIRST"),
             output_variable="shared"),
        Task(name="second", handler=lambda ctx: StepResult(output="FROM-SECOND"),
             output_variable="shared"),
    ]),
])
result = workflow.start("go")
print(result["variables"]["shared"])  # "FROM-SECOND"
# WARNING logged: Parallel branches 0 and 1 both wrote variable 'shared'...

A WARNING names the key and both branches. Silent last-writer-wins was the exact bug #4932 removes, so a collision is reported rather than hidden.

If two branches want to write the same concept, give them distinct output_variable names (branch_a_score, branch_b_score) and combine them in the aggregator.

Examples

With Agents

from praisonaiagents import Agent

researcher = Agent(name="Researcher", role="Research topics")
analyst = Agent(name="Analyst", role="Analyze data")
writer = Agent(name="Writer", role="Write content")

workflow = AgentFlow(steps=[
    parallel([researcher, analyst, writer]),
    final_aggregator
])

Mixed Steps

workflow = AgentFlow(steps=[
    parallel([
        my_function,           # Function
        Agent(name="Bot"),     # Agent
        Task(...)      # Task
    ]),
    aggregator
])

Nested Parallel

workflow = AgentFlow(steps=[
    parallel([
        parallel([task_a1, task_a2]),  # Group A
        parallel([task_b1, task_b2])   # Group B
    ]),
    final_aggregator
])

Performance

Parallel execution uses Python’s ThreadPoolExecutor:

• Concurrent I/O: Ideal for API calls, file operations
• Thread-safe: Each step gets its own copy of variables
• Automatic joining: All results collected before next step

# Performance comparison
# Sequential: 3 steps × 1s each = 3s total
# Parallel:   3 steps × 1s each = ~1s total (concurrent)

Use Cases

Use Case	Description
Multi-source Research	Query multiple APIs simultaneously
Data Processing	Process independent data chunks
Report Generation	Generate sections in parallel
Validation	Run multiple validators concurrently
A/B Comparison	Run different approaches and compare

How It Works

Phase	What happens
1. Fan-out	parallel() submits each step to a thread pool simultaneously
2. Execute	Steps run concurrently; each receives the same workflow context
3. Collect	Results gather into ctx.variables["parallel_outputs"], index-aligned with the branches; named output_variable writes merge back in declaration order
4. Continue	Next workflow step (aggregator) receives the merged outputs and named variables

---

Best Practices

Parallelise only independent steps

Branches must not depend on each other’s output in the same parallel group.

Cap concurrency for external APIs

Rate-limited tools may need sequential execution or throttling despite parallel support.

Collect and inspect branch errors

Read the aggregated error list after ParallelExecutionError before retrying.

Keep branch outputs small

Large parallel results inflate context — summarise before merging downstream.

Give sibling branches distinct output_variable names

Two branches writing the same output_variable collide — the later-declared branch silently overwrites the earlier one (with a warning). That class of silent loss is exactly what #4932 fixes; distinct names (branch_a_score, branch_b_score) avoid it entirely.

Failure Handling

Choose how a parallel block reacts when a branch fails using the on_failure parameter.

Failure Strategies

Strategy	Behavior	Use Case
partial_ok	Continue with partial results. Failed branches contribute None at the same index in parallel_outputs, index-aligned with the step.	Data aggregation where some sources may be unavailable
fail_fast	Cancel remaining branches and raise WorkflowStepError on first failure	Critical workflows where any failure invalidates results
fail_all	Wait for all branches, then raise WorkflowStepError if any failed	Comprehensive error reporting and debugging

Variable merging on failure. Under partial_ok, the variables a failed branch had already written before it failed are preserved — its own output_variable is never set (the failure path returns before that write), so a downstream reader cannot pick up a value the failure invented. Under fail_fast / fail_all, nothing from any branch is merged: the block raises before the merge runs. See Named outputs from branches.

Examples

from praisonaiagents import Agent, AgentFlow, parallel, WorkflowStepError

# partial_ok (default) — continue even if one branch fails
workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c]),
    aggregator,
])

# fail_fast — abort on first failure
workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c], on_failure="fail_fast"),
    aggregator,
])

# fail_all — gather all errors then fail
workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c], on_failure="fail_all"),
    aggregator,
])

Error Handling with partial_ok

def aggregator(ctx: WorkflowContext) -> StepResult:
    outputs = ctx.variables["parallel_outputs"]

    # A failed branch contributes None at its index
    successful = [o for o in outputs if o is not None]
    failures = [o for o in outputs if o is None]

    if len(successful) >= 2:  # Minimum threshold
        return StepResult(output=f"Processed {len(successful)} sources")
    else:
        return StepResult(output=f"Insufficient data: {len(failures)} failures")

Stop propagation from nested steps

A branch step with on_error="stop" halts the whole workflow, not just the parallel block — even under on_failure="partial_ok". As of PraisonAI #4203, parallel(...) propagates a nested stop up to the enclosing workflow, matching route() / loop() / repeat() / if_().

from praisonaiagents import Task, parallel, AgentFlow

risky = Task(name="charge_customer", handler=charge, on_error="stop")   # halts on failure
safe  = Task(name="log_event", handler=log_event)

wf = AgentFlow(steps=[
    parallel([risky, safe], on_failure="partial_ok"),
    Task(name="ship_order", handler=ship),   # must NOT run if charge_customer failed
])
wf.run(input=order)
# charge_customer fails → workflow halts with status="failed"; ship_order never runs.
# Printed: 🛑 Workflow stopped by nested parallel step

The interaction between the parallel block’s on_failure mode and a branch step’s on_error is:

on_failure (parallel)	Branch step’s on_error	Result
partial_ok	continue (default)	Branch error recorded, other branches keep running, workflow proceeds
partial_ok	stop	Branch error recorded, workflow halts with status="failed" after the parallel block
fail_fast	any	First failure raises WorkflowStepError, workflow halts
fail_all	any	All branches run, then WorkflowStepError raised if any failed

Exception Handling with fail_fast/fail_all

try:
    result = workflow.start("Process all branches")
except WorkflowStepError as e:
    print(f"Workflow failed: {e}")
    print(f"Root cause: {e.cause}")
    for error in e.errors:
        print(f"  Branch {error['step']}: {error['error']}")

Related

Workflow Patterns

Overview of routing, parallel, loop, and repeat

Workflow Routing

Decision-based branching

Workflow Loop

Iterate over lists and files

Workflow Repeat

Repeat until a condition is met