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.

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

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	All results gather into ctx.variables["parallel_outputs"] in completion order
4. Continue	Next workflow step (aggregator) receives the merged outputs

---

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.

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 return "Error: <msg>"	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

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

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