Chapter 12 · Workflow Advanced: Parallel / Branch / Loop

Learning objectives: Master the four control-flow primitives — Parallel, Condition, Loop, Router — and combine them to build a content-production pipeline with a quality feedback loop.

12.1 Parallel: Concurrent Execution of Independent Steps

Steps inside a Parallel block run concurrently; their outputs are aggregated in configured order and passed to the next step:

import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow import Parallel, Step, StepInput, StepOutput, Workflow

model = OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

hackernews_researcher = Agent(name="HN Researcher", role="Find tech trends from HackerNews", model=model)
web_researcher = Agent(name="Web Researcher", role="Search the web and official docs", model=model)
paper_researcher = Agent(name="Paper Researcher", role="Find academic paper perspectives", model=model)

workflow = Workflow(
    name="parallel-research",
    steps=[
        Parallel(
            Step(name="hn", executor=hackernews_researcher),
            Step(name="web", executor=web_researcher),
            Step(name="papers", executor=paper_researcher),
            name="research",
        ),
        Step(name="synthesis", executor=writer),  # receives the aggregated result from all three
    ],
)

The three research directions are independent of each other; running them in parallel compresses total time to that of the slowest branch. Note: if parallel function steps need to write to session state, each branch should write to different keys to avoid race conditions.

12.2 Condition: Conditional Branching

Condition uses an evaluator function to decide which path to take, and supports else_steps for the reverse branch:

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


def is_technical_issue(step_input: StepInput) -> bool:
    text = str(step_input.input or "").lower()
    return any(kw in text for kw in ["bug", "crash", "timeout", "error"])


def diagnose(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"Technical diagnosis: {step_input.input}")


def general_support(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"General support reply: {step_input.input}")


workflow = Workflow(
    name="support-router",
    steps=[
        Condition(
            name="triage",
            evaluator=is_technical_issue,
            steps=[Step(name="diagnose", executor=diagnose)],          # True branch
            else_steps=[Step(name="general", executor=general_support)],  # False branch
        ),
        Step(name="follow_up", executor=log_and_reply),
    ],
)

The evaluator can be a synchronous/asynchronous Python function, a boolean value, or a CEL expression string. Unlike Team's "the model decides whom to delegate to", the branching logic here is code you write — testable and reproducible.

12.3 Loop: Iterate Until Quality Standards Are Met

Loop repeats its internal steps until end_condition returns True or max_iterations (default 3) is reached:

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


def draft_section(step_input: StepInput) -> StepOutput:
    prev = step_input.previous_step_content or str(step_input.input)
    return StepOutput(content=f"{prev}\n[Add a more specific argument]")


def quality_check(outputs: list[StepOutput]) -> bool:
    """Consider it done if any iteration output exceeds 120 characters"""
    return any(len(str(o.content or "")) > 120 for o in outputs)


def final_edit(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"Final version:\n{step_input.previous_step_content}")


workflow = Workflow(
    name="quality-loop",
    steps=[
        Loop(
            name="drafting-loop",
            steps=[Step(name="expand", executor=draft_section)],
            end_condition=quality_check,
            max_iterations=3,
            forward_iteration_output=True,  # next round receives the previous round's output as previous_step_content
        ),
        Step(name="edit", executor=final_edit),
    ],
)

forward_iteration_output=True is the key to iterative accumulation: by default each round restarts from the original input; when enabled, the next round's previous_step_content is the previous round's result, achieving "progressively longer and refined writing."

12.4 Router: Dynamically Select Execution Paths

The Router's selector function returns a list of steps to execute, chosen from choices:

from typing import List
from agno.workflow import Router, Step
from agno.workflow.types import StepInput

tech_step = Step(name="tech_research", executor=hackernews_researcher)
general_step = Step(name="general_research", executor=web_researcher)


def research_router(step_input: StepInput) -> List[Step]:
    topic = (step_input.previous_step_content or step_input.input or "").lower()
    if any(kw in topic for kw in ["ai", "programming", "startup", "software"]):
        return [tech_step]     # returns a list — multiple steps are allowed
    return [general_step]


workflow = Workflow(
    name="router-demo",
    steps=[
        Router(
            name="strategy",
            selector=research_router,
            choices=[tech_step, general_step],
        ),
        publish_step,
    ],
)

Rule of thumb: use Condition for boolean checks, Router for multi-path selection. Both support nesting — for example, a Loop can be placed inside a branch selected by a Router.

12.5 Combined Practice: Content Pipeline + Step-Level Structured I/O

String together the primitives from this chapter into a real pipeline, and have each LLM step output Pydantic objects via output_schema:

from pydantic import BaseModel, Field


class Draft(BaseModel):
    title: str = Field(description="Article title")
    body: str = Field(description="Article body")
    word_count: int


class Review(BaseModel):
    approved: bool
    comments: list[str]


drafter = Agent(name="Drafter", model=model, output_schema=Draft,
                instructions=["Output the first draft according to the schema"])
review_agent = Agent(name="Reviewer", model=model, output_schema=Review,
                     instructions=["Review the draft; approved indicates whether it passes"])

content_workflow = Workflow(
    name="content-pipeline",
    steps=[
        Parallel(Step(name="collect_hn", executor=hackernews_researcher),
                 Step(name="collect_web", executor=web_researcher),
                 name="collect"),
        Step(name="draft", executor=drafter),
        Loop(
            name="revise-loop",
            steps=[Step(name="review", executor=review_agent)],
            end_condition=lambda outputs: all(
                getattr(o.content, "approved", False) for o in outputs
            ),
            max_iterations=3,
            forward_iteration_output=True,
        ),
        Step(name="publish", executor=publisher),
    ],
)

Each step's output is a Pydantic-validated object — downstream steps no longer receive plain text where you "pray the format is correct." This is the dividing line between a production-grade Workflow and a demo.

Class Executors

When you need initialization configuration or state maintenance across runs, implement a class with __call__(self, step_input) -> StepOutput as the executor; the instance is reused across multiple runs.

12.6 Chapter Summary

• Parallel runs independent steps concurrently and aggregates outputs in configured order; use different keys when parallel branches write shared state;
• Condition(evaluator, steps, else_steps) implements testable deterministic branching;
• Loop converges via end_condition + max_iterations; forward_iteration_output=True makes iterations accumulate based on the previous round's result;
• Router.selector returns the list of Steps to execute — designed for mutually exclusive multi-path selection;
• Step-level output_schema lets structured data flow through the entire pipeline.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. When Parallel steps output to the next step, what order are the outputs in?

2. When does a Loop stop iterating?

3. To make Loop's next iteration continue from the previous round's output, which parameter should you set?

4. For "select one of three research strategies based on the topic", which primitive is most appropriate?

🛠️ Hands-On Exercises

1. Add a Parallel research phase (two researchers using different tools) to the Chapter 11 writing pipeline, and verify the concatenation order of aggregated outputs.
2. Implement a "rewrite if word count is insufficient" Loop: end_condition checks whether the draft exceeds 200 characters, max_iterations=3. Observe the behavior difference with and without the forward flag.
3. Use Router to implement "technical question → HN Researcher / lifestyle question → General Researcher"; submit both types of questions to verify routing results.

Once the pipeline runs end-to-end, we'll fill in multimodal, session, and data capabilities; in Chapter 23: AgentOS we'll publish it as an HTTP service.