Workflow Parallel Execution

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

A real agent fanning its output into three branch workers. By default (context="full") each branch receives the full answer the planner produced, not a lossy excerpt.

from praisonaiagents import Agent, AgentFlow, parallel

planner  = Agent(name="Planner",  instructions="Produce a detailed plan.")
reviewer = Agent(name="Reviewer", instructions="Review the plan critically.")
writer   = Agent(name="Writer",   instructions="Rewrite the plan clearly.")
scorer   = Agent(name="Scorer",   instructions="Score the plan 0–100 with reasons.")

flow = AgentFlow(steps=[
    planner,
    parallel([reviewer, writer, scorer]),   # context="full" by default
])

flow.start("Draft a GTM plan for a developer tool launch.")

# Token-saving opt-in — WARNS whenever it drops the middle:
parallel([reviewer, writer, scorer], context="truncate")

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

Quick Start

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)

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",
    timeout: Optional[float] = None,
    context: str = "full",
) -> 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
`timeout`	`Optional[float]`	`None`	Wall-clock bound (seconds) for the whole parallel block. Must be positive, or `None` for no bound (`0` / negatives raise `ValueError`). Raises `WorkflowStepError` if any branch is still running when the clock elapses. The clock starts when the block begins, including context-preparation work.
`context`	`"full" \| "truncate" \| "summarize"`	`"full"`	How the upstream step’s output reaches each branch. `"full"` (default) passes the exact bytes to every branch — no data loss. `"truncate"` reduces to a head+tail excerpt when the output exceeds ~3,000 chars (middle dropped). `"summarize"` LLM-summarises for large fan-outs, falls back to truncate for small ones. Both reducing modes emit a `logger.warning` whenever they drop content. Invalid values raise `ValueError`.

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 isolated copy of the variables scope, so branches never race on one shared dict and a branch’s in-place mutation of a list or dict is invisible to its siblings. Values that cannot be deep-copied — a threading.Lock, a DB connection, an HTTP or LLM client, an open file, a live Agent, a generator — are shared by reference across branches instead of aborting the run (as of PraisonAI #5717). 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.

The same silent-loss defect existed inside loop() bodies and was fixed by the sibling PR #4947. See Named outputs from a loop body — loop bodies now propagate writes the same way, with the one difference that per-iteration collisions are not warned about (that would emit N-1 warnings for a correct workflow). A threading.Lock, DB/HTTP/LLM client, open file, live Agent, or generator can sit in variables even when a branch reads it — the un-copyable leaf is shared by reference, everything around it is still isolated.

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

client = threading.Lock()   # stands in for an HTTP/DB/LLM client

def append_row(label):
    def run(ctx: WorkflowContext) -> StepResult:
        # The surrounding dict and its 'rows' list ARE branch-local.
        ctx.variables["state"]["rows"].append(label)
        # The 'client' leaf is SHARED by reference across branches.
        with ctx.variables["state"]["client"]:
            pass
        return StepResult(
            output=label,
            variables={f"rows_{label}": list(ctx.variables["state"]["rows"])},
        )
    return run

wf = AgentFlow(
    steps=[parallel([
        Task(name="a", handler=append_row("A")),
        Task(name="b", handler=append_row("B")),
    ])],
    variables={"state": {"client": client, "rows": []}},
)
result = wf.start("go")
# Each branch saw only its own append — rows_A == ["A"], rows_B == ["B"].
# The lock in state["client"] is the one real object, usable for coordination.

Branches share the un-copyable leaf by reference. A Lock/connection/client is designed to be shared; a mutable list or dict that happens to be un-copyable (rare) is also shared — mutations through it will be visible to siblings. Keep ordinary mutable state in a copyable container (plain list / dict) so the isolation guarantee applies to it.

Values that degrade to shared-by-reference

Pass any of these through AgentFlow(variables=...) and sibling branches share one reference instead of crashing.

Example value	Why it couldn’t be deep-copied
`threading.Lock()` / `threading.RLock()` / `multiprocessing.Lock()`	C-level primitives, un-picklable
A `psycopg` / `sqlalchemy` / `sqlite3` connection or cursor	Owns an OS socket/file handle
`httpx.Client()` / `requests.Session()` / an OpenAI/Anthropic LLM client	Holds a pool, event loop refs, locks
An open file (`open(path, "r")`)	Owns an OS fd
A live `Agent` with mid-conversation state	ContextVar / locks inside the LLM
A generator / iterator	Un-picklable by Python

Cross-variable aliasing inside one branch is preserved. If variables={"records": shared, "state": {"records": shared}} passes the same list under both names, a branch still sees one shared list — a write through records is visible through state["records"] within that branch.

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.

Inject shared resources through variables

A Lock, DB connection, HTTP/LLM client, open file, or live Agent can be injected via AgentFlow(variables=...) and used from branch steps. The un-copyable leaf is shared by reference, so coordination (locking, connection pooling) works as expected; everything around it is still branch-local, so your ordinary mutable state stays isolated. See Sharing resources across branches.

Bound external calls with timeout

LLM / HTTP / tool calls can hang. parallel(..., timeout=N) is the cheapest hung-branch bound — the run raises WorkflowStepError at the deadline instead of waiting forever on a stuck thread. See Timing out hung branches.

Prefer fail_fast for latency-critical fan-outs

Now that fail_fast actually fails fast (post-#5718), it is the right default for user-facing paths where returning quickly on first error beats collecting every error.

Context pass-through to branches

Every branch sees the exact upstream output by default. Opt in to token savings with context="truncate" or context="summarize" — those drop the middle of large outputs and warn when they do.

# Default — lossless (new default as of #5719)
parallel([worker_a, worker_b, worker_c])
# equivalent to:
parallel([worker_a, worker_b, worker_c], context="full")

# Opt in to the head+tail excerpt (and warn on drop)
parallel([worker_a, worker_b, worker_c], context="truncate")

# Opt in to the LLM summariser (large fan-outs); falls back to truncate
parallel([worker_a, worker_b, worker_c], context="summarize")

Mode	What each branch receives	When it reduces	Warns on drop
`"full"` (default)	The exact upstream output, byte-for-byte	Never	—
`"truncate"`	Head+tail excerpt (middle dropped)	Output > ~3,000 chars / ~1,000 tokens	✅
`"summarize"`	LLM summary for ≥ 3 branches; else truncate	Large fan-outs; falls back to truncate on failure or small fan-outs	✅

Before PraisonAI #5719: large previous_output was always reduced to a head+tail excerpt before fan-out, with no opt-out and no default-level warning — branches silently received partial data. Upgrade to get lossless-by-default; if you relied on the token savings, opt in with context="truncate" or context="summarize".

The two paths differ only in what each branch receives — lossless by default, excerpted (with a WARNING) when you opt in.

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`	Raise `WorkflowStepError` as soon as any branch fails. Remaining branches are cancelled where possible; threads already running cannot be force-killed, so they are abandoned but the caller is not blocked on them. As of PraisonAI #5718, the raise happens in completion order — a failure in a later-declared branch no longer waits for earlier siblings to finish.	Latency-critical workflows where any failure invalidates results
`fail_all`	Wait for all branches, then raise `WorkflowStepError` if any failed	Comprehensive error reporting and debugging

Before PraisonAI #5718: fail_fast consumed results in submission order and left the thread pool via a with block whose __exit__ called shutdown(wait=True), so the exception was held until every earlier-declared and every still-running branch finished. If you observe a fail_fast block hanging on a slow sibling, upgrade.

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

Timing out hung branches

Bound the whole parallel block with timeout so a stuck branch no longer blocks the run.

from praisonaiagents import Agent, AgentFlow, parallel

# Bound the whole block at 10 seconds — a stuck branch no longer blocks the run.
flow = AgentFlow(steps=[
    parallel([researcher, analyst, writer], timeout=10),
    summarizer,
])
flow.start("Analyse Q3 numbers")

The clock starts when the block begins, including context preparation (the summarisation path can make a blocking litellm.completion call — a deadline started only at submission time would let timeout=N wait far longer than N before its clock even begins).
timeout combines with any on_failure mode. fail_fast + timeout is the common “give up on the whole fan-out quickly” pattern.
A timed-out branch is cancelled where possible; threads running user code cannot be force-killed in Python, so the branch may keep running in the background. The caller is no longer blocked on it, and its late failure cannot corrupt a later run (see Running the same workflow more than once).
timeout=0 and negatives raise ValueError at construction — the only way to mean “no bound” is None.

Running the same workflow more than once

Reusing one Workflow instance across many inputs is safe even when a previous run abandoned a branch — a late failure from an abandoned branch cannot flip a later run’s status.

# Same Workflow instance used across many inputs
wf = AgentFlow(steps=[
    parallel([slow_branch, quick_fail_branch], on_failure="fail_fast"),
    report,
])

wf.start("job 1")   # fail_fast raises; slow_branch is abandoned
wf.start("job 2")   # run 2 begins; the abandoned branch finally fails afterwards
# Result: run 2 completes with status="completed".
# The late failure from run 1's abandoned branch cannot flip run 2's status.

Each run() stamps a monotonic “generation” into a contextvars.ContextVar; branches capture it at submit time.
When a branch finally raises after its owning run has moved on, the engine suppresses the shared-status write.
This matters when you reuse one Workflow instance across many inputs (common for long-running services) and combine it with fail_fast or timeout.

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

Workflow Discussion

N-agent conversation loop

​Quick Start

​API Reference

​parallel()

​Parameters

​Accessing Results

​Named outputs from branches

​Two branches wrote the same variable

​Sharing resources across branches

​Values that degrade to shared-by-reference

​Examples

​With Agents

​Mixed Steps

​Nested Parallel

​Performance

​Use Cases

​How It Works

​Best Practices

​Context pass-through to branches

​Failure Handling

​Failure Strategies

​Examples

​Error Handling with partial_ok

​Stop propagation from nested steps

​Exception Handling with fail_fast/fail_all

​Timing out hung branches

​Running the same workflow more than once

​Related

Workflow Patterns

Workflow Routing

Workflow Loop

Workflow Repeat

Workflow Discussion

Quick Start

API Reference

parallel()

Parameters

Accessing Results

Named outputs from branches

Two branches wrote the same variable

Sharing resources across branches

Values that degrade to shared-by-reference

Examples

With Agents

Mixed Steps

Nested Parallel

Performance

Use Cases

How It Works

Best Practices

Context pass-through to branches

Failure Handling

Failure Strategies

Examples

Error Handling with partial_ok

Stop propagation from nested steps

Exception Handling with fail_fast/fail_all

Timing out hung branches

Running the same workflow more than once

Related