Skip to main content
Execute multiple steps concurrently and combine their results. This pattern is ideal for independent tasks that can run simultaneously. 
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

ParameterTypeDefaultDescription
stepsList—List of steps to execute concurrently
max_workersOptional[int]NoneCap 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:
VariableTypeDescription
parallel_outputsList[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")

​
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 CaseDescription
Multi-source ResearchQuery multiple APIs simultaneously
Data ProcessingProcess independent data chunks
Report GenerationGenerate sections in parallel
ValidationRun multiple validators concurrently
A/B ComparisonRun different approaches and compare

​
How It Works

PhaseWhat happens
1. Fan-outparallel() submits each step to a thread pool simultaneously
2. ExecuteSteps run concurrently; each receives the same workflow context
3. CollectAll results gather into ctx.variables["parallel_outputs"] in completion order
4. ContinueNext workflow step (aggregator) receives the merged outputs

​
Best Practices

Branches must not depend on each other’s output in the same parallel group.
Rate-limited tools may need sequential execution or throttling despite parallel support.
Read the aggregated error list after ParallelExecutionError before retrying.
Large parallel results inflate context — summarise before merging downstream.

​
Failure Handling

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

​
Failure Strategies

StrategyBehaviorUse Case
partial_okContinue with partial results. Failed branches return "Error: <msg>"Data aggregation where some sources may be unavailable
fail_fastCancel remaining branches and raise WorkflowStepError on first failureCritical workflows where any failure invalidates results
fail_allWait for all branches, then raise WorkflowStepError if any failedComprehensive error reporting and debugging

​
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"]
    
    # Filter successful results
    successful = [o for o in outputs if not o.startswith("Error:")]
    errors = [o for o in outputs if o.startswith("Error:")]
    
    if len(successful) >= 2:  # Minimum threshold
        return StepResult(output=f"Processed {len(successful)} sources")
    else:
        return StepResult(output=f"Insufficient data: {len(errors)} failures")

​
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']}")

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