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.

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.

Quick Start

1. Simple Usage

   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"])
   
2. With Configuration

   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())
   
3. 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.

   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

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

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.

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

---

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.

---

Related

Orchestrator Worker

Route tasks to specialised workers from a central orchestrator

AgentFlow

Build multi-step agent pipelines