Agentic Parallelisation

Run multiple agents in parallel and aggregate their results.

Run independent agent tasks at the same time, then combine their outputs in one workflow.

python

from praisonaiagents import Agent, AgentFlow, parallel

researcher = Agent(name="Researcher", instructions="Research the topic.")
writer = Agent(name="Writer", instructions="Write a summary.")
aggregator = Agent(name="Aggregator", instructions="Combine findings.")

workflow = AgentFlow(steps=[
    parallel([researcher, writer]),
    aggregator,
])
result = workflow.start("Research AI trends")

The user sends one request; parallel agents run at the same time before an aggregator merges results.

Diagram

Quick Start

Simple Usage

python

from praisonaiagents import Agent, AgentFlow, parallel

market = Agent(name="Market", instructions="Analyse market trends.")
competitor = Agent(name="Competitor", instructions="Analyse competitors.")
aggregator = Agent(name="Aggregator", instructions="Synthesise all findings.")

workflow = AgentFlow(steps=[
    parallel([market, competitor]),
    aggregator,
])

result = workflow.start("Research the AI industry")
print(result["output"])

With Configuration

python

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

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

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

def combine(ctx: WorkflowContext) -> StepResult:
    outputs = ctx.variables.get("parallel_outputs", [])
    return StepResult(output=f"Combined {len(outputs)} results")

workflow = AgentFlow(steps=[parallel([worker_a, worker_b]), combine])

async def run():
    return await workflow.astart("Start parallel work")

result = asyncio.run(run())

Name each branch's output

Every branch can name its result with output_variable; the write survives the block and is readable by any later step.

python

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

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

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

combine = Task(action="Combine {{finding_a}} and {{finding_b}}")

workflow = AgentFlow(steps=[
    parallel([
        Task(name="research_a", handler=research_a, output_variable="finding_a"),
        Task(name="research_b", handler=research_b, output_variable="finding_b"),
    ]),
    combine,
])
result = workflow.start("Research the topic")

How It Works

Diagram

Phase	What happens
1. Fan-out	AgentFlow dispatches all workers in parallel (thread pool)
2. Collect	All `StepResult` objects gathered into `ctx.variables["parallel_outputs"]`
3. Aggregate	Next step (aggregator) receives the combined list and produces one response

Diagram

Pattern	When to use
`parallel([a, b, c])`	Tasks do not depend on each other’s output
Sequential steps	Each step needs the previous result
Aggregator agent	Combine parallel outputs into one answer

Configuration Options

parallel() accepts a list of agents or step callables. Each worker receives the same workflow context; results are collected in ctx.variables["parallel_outputs"] for the next step.

python

from praisonaiagents import AgentFlow, parallel

workflow = AgentFlow(steps=[
    parallel([agent_a, agent_b, agent_c]),
    final_agent,
])

Option	Type	Default	Description
Workers	`List[Agent \	Callable]`	—	Agents or functions run concurrently
Context key	`str`	`"parallel_outputs"`	Where aggregated results are stored
Async	`astart()`	—	Use `workflow.astart()` for async execution
`output_variable` (on each branch)	`str`	`None`	Name the branch’s result so later steps can read it by name
`context`	`”full” \	“truncate” \	“summarize”`	`"full"`	How the upstream step’s output reaches each branch. `"full"` (default) passes the exact bytes — no data loss. `"truncate"`/`"summarize"` opt in to token savings and warn when they drop content. See Context pass-through to branches.

By default each branch receives the exact upstream output. As of PraisonAI #5719, token-saving excerpting is opt-in via context="truncate" or context="summarize" — see Context pass-through to branches for the full semantics.

Best Practices

Keep parallel tasks independent

Only parallelise work that does not need another worker’s output mid-flight. Route dependent steps after the aggregator.

Use an aggregator for synthesis

A dedicated agent (or function step) after parallel() turns multiple raw outputs into one coherent response.

Prefer astart for I/O-heavy workers

When workers call external APIs, use await workflow.astart() so concurrent I/O does not block the event loop.

Watch rate limits

Parallel LLM calls multiply token usage and API requests. Cap worker count or add a rate limiter if needed.

Prefer named outputs over positional aggregation when order matters

Name each branch’s result with output_variable and read it by name downstream, rather than relying on the position of parallel_outputs. See Named outputs from branches.

Orchestrator Worker

Route tasks to specialised workers from a central orchestrator

AgentFlow

Build multi-step agent pipelines

Agentic Parallelisation

Quick Start

How It Works

Configuration Options

Best Practices

Related