Parallel Workflow

Run independent steps concurrently with Parallel and pass their combined output to the next step.

Example use cases: Multi-source research, parallel analysis, concurrent data processing

Steps inside a Parallel block run concurrently. Their outputs are aggregated in the configured step order and passed to the next step in the workflow.

Example

parallel_workflow.py

from agno.workflow import Parallel, Step, StepInput, StepOutput, Workflow


def research_hackernews(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"HackerNews notes for {step_input.input}")


def research_web(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"Web notes for {step_input.input}")


def research_papers(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"Paper notes for {step_input.input}")


def synthesize(step_input: StepInput) -> StepOutput:
    combined = step_input.previous_step_content or "No research returned"
    return StepOutput(content=f"Synthesis:\n{combined}")

workflow = Workflow(
    name="Parallel Research Pipeline",
    steps=[
        Parallel(
            Step(name="HackerNews Research", executor=research_hackernews),
            Step(name="Web Research", executor=research_web),
            Step(name="Academic Research", executor=research_papers),
            name="Research Step",
        ),
        Step(name="Synthesis", executor=synthesize),
    ],
)

workflow.print_response("Write about the latest AI developments", markdown=True)

Handling Session State Data in Parallel Steps

Custom Python functions can accept a run_context parameter and update run_context.session_state.

Parallel branches share that session-state dictionary. Coordinate writes to the same keys, or assign separate keys to each branch, to avoid races.

Developer Resources

• Parallel Steps Workflow
• Session State in a Function

Reference

For complete API documentation, see Parallel Steps Reference.