Skip to main content
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())

​
How It Works

PhaseWhat happens
1. Fan-outAgentFlow dispatches all workers in parallel (thread pool)
2. CollectAll StepResult objects gathered into ctx.variables["parallel_outputs"]
3. AggregateNext step (aggregator) receives the combined list and produces one response
PatternWhen to use
parallel([a, b, c])Tasks do not depend on each other’s output
Sequential stepsEach step needs the previous result
Aggregator agentCombine 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,
])
OptionTypeDefaultDescription
WorkersList[Agent | Callable]—Agents or functions run concurrently
Context keystr"parallel_outputs"Where aggregated results are stored
Asyncastart()—Use workflow.astart() for async execution

​
Best Practices

Only parallelise work that does not need another worker’s output mid-flight. Route dependent steps after the aggregator.
A dedicated agent (or function step) after parallel() turns multiple raw outputs into one coherent response.
When workers call external APIs, use await workflow.astart() so concurrent I/O does not block the event loop.
Parallel LLM calls multiply token usage and API requests. Cap worker count or add a rate limiter if needed.

Orchestrator Worker

Route tasks to specialised workers from a central orchestrator

AgentFlow

Build multi-step agent pipelines