Chapter 11 · Workflow Basics: Step Orchestration

Learning objectives: Understand the fundamental difference between Workflow and Team, and learn to use Workflow + Step to orchestrate Agents, Teams, and plain functions into a deterministic pipeline.

11.1 Why You Need Workflow

Teams are "model-driven": the leader decides on the fly whom to assign and what task to delegate. This brings flexibility, but also unpredictability — the same question may follow different execution paths. Many business processes, however, cannot tolerate improvisation:

• Content production: research must come first, then writing, then review — the order must not be violated;
• Data processing: clean → analyze → report, with fixed inputs and outputs at each step;
• Compliance scenarios: certain steps must execute unconditionally, without the model deciding to skip them.

The official Agno positioning of the two:

	Team	Workflow
Control	Leader model自主决策	Developer defines control flow with code
Suitable for	Role collaboration, task dispatch	Deterministic flows, structured pipelines
Primitives	mode / delegation / task lists	Step / Parallel / Condition / Loop / Router

11.2 Your First Two-Step Workflow: Draft → Review

Workflow manages a single run; the steps list executes in order. Each Step wraps exactly one executor: an Agent, a Team, a plain Python function, or even a nested Workflow.

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

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

writer = Agent(
    name="Writer",
    model=model,
    instructions=["Write an essay of no more than 300 words on the given topic."],
)

reviewer = Agent(
    name="Reviewer",
    model=model,
    instructions=[
        "Review the article from the previous step: check factual accuracy, grammar, and structure.",
        "Output the revised final version directly.",
    ],
)

# Draft -> Review; order is guaranteed by code
workflow = Workflow(
    name="writing-pipeline",
    steps=[
        Step(name="draft", executor=writer),
        Step(name="review", executor=reviewer),
    ],
)

workflow.print_response(
    "Topic: Why Automated Testing Is Worth the Investment", markdown=True
)

At runtime, the output of the draft step automatically becomes the input of the review step — no manual wiring needed.

11.3 Three Forms of Step Executors

from agno.team import Team
from agno.workflow import Step, StepInput, StepOutput, Workflow


def data_preprocessor(step_input: StepInput) -> StepOutput:
    """Function executor: performs deterministic processing without consuming tokens"""
    research = step_input.previous_step_content or "No research content received"
    return StepOutput(content=f"Research materials for analysis:\n{research}")


research_team = Team(name="Research Team", members=[...], model=model)

workflow = Workflow(
    name="mixed-pipeline",
    steps=[
        research_team,        # Form 1: directly place Agent/Team (auto-wrapped as Step)
        data_preprocessor,    # Form 2: directly place a function
        Step(                 # Form 3: explicit Step wrapper, recommended when you need naming/descriptions
            name="analysis",
            executor=writer,
            description="Generate an analysis report based on research findings",
        ),
    ],
)

All three forms can be freely mixed. Naming steps (name=) is not just for readability: event tracking and output lookup both use the step name as the key.

11.4 StepInput and StepOutput: The Data Contract Between Steps

Function executors follow a standard signature — accept StepInput, return StepOutput:

def summarize(step_input: StepInput) -> StepOutput:
    # input: the original workflow input (always preserved)
    original = str(step_input.input)
    # previous_step_content: the body of the previous step's output
    prev = step_input.previous_step_content or ""
    # previous_step_outputs: if the upstream is a parallel block, you can get the output dict of each named step
    return StepOutput(
        content=f"Original requirement: {original}\n\nMaterial summary:\n{prev[:500]}",
    )

Quick reference for common fields:

StepInput field	Meaning
input	The workflow's original user input
previous_step_content	The content text output by the previous step
previous_step_outputs	The collection of outputs from upstream steps (especially useful after parallel aggregation)
additional_data	Custom data attached at runtime

The Value of Function Steps

Inserting a pure function step between two Agents is the most common Workflow pattern: format conversion, length truncation, result validation — none of this should burn tokens; let Python handle it.

11.5 Running and Parsing Output

print_response() is great for debugging; for production code use run() to get structured results:

result = workflow.run("Topic: Vector Database Selection")

# workflow_output.content is the output of the last step
print(result.content)

# Finer-grained: inspect each step's execution result
for step_result in (result.step_results or []):
    print(step_result.step_name, "->", str(step_result.content)[:80])

You can also use arun() for async execution, or iterate the event stream with run(..., stream=True) to observe each step's start and end events.

11.6 Chapter Summary

• Team = model-driven flexible collaboration; Workflow = code-defined deterministic control flow;
• A Step's executor can be an Agent, Team, function, or nested Workflow — all three can be mixed in the same steps list;
• Data flows downstream automatically via StepInput.previous_step_content; StepOutput(content=...) is the standard exit for function steps;
• In production, use run() and iterate step_results to parse each step's output, rather than parsing printed text.

🧪 随堂测验

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

1. Regarding the division of labor between Team and Workflow in Agno, which is correct?

2. Which of the following is NOT a valid executor for a Step in a Workflow?

3. In a function step, how do you access "the output of the previous step"?

4. Why is it recommended to insert a pure Python function step between two LLM steps?

🛠️ Hands-On Exercises

1. Take the research team from Chapter 9 and connect it as the first step in this chapter's writing pipeline, implementing a three-step flow: "Research → Draft → Review".
2. Add a function step word_limit between drafting and reviewing that truncates drafts exceeding 400 words and appends a notice.
3. Run the entire pipeline with run() and print each step_result's name and the first 100 characters.

Next chapter unlocks all Workflow control-flow primitives: Parallel, Branch & Loop.