第 20 章 · Workflow 入门：Steps 步骤编排
本章目标：理解 Workflow 与 Team 的本质区别，学会用
Workflow+Step把 Agent、Team 和普通函数编排成确定性的流水线。
20.1 为什么需要 Workflow
Team 是"模型主导"的：leader 临场决定派谁、派什么任务。这带来灵活性，也带来不确定性——同样的问题可能走出不同的执行路径。而很多业务流程恰恰不允许临场发挥：
- 内容生产：必须先调研、再撰写、后审校，顺序不能乱；
- 数据处理：清洗 → 分析 → 报告，每步输入输出固定；
- 合规场景：某些步骤必须无条件执行，不能由模型决定跳过。
Agno 官方对两者的定位是：
| Team | Workflow | |
|---|---|---|
| 控制权 | Leader 模型自主决策 | 开发者用代码定义控制流 |
| 适用 | 角色协作、任务分派 | 确定性流程、结构化管道 |
| 原语 | mode / 委派 / 任务清单 | Step / Parallel / Condition / Loop / Router |
20.2 你的第一个两步工作流：撰写 → 审校
Workflow 管理一次运行，steps 列表按顺序执行。每个 Step 包裹恰好一个执行器：Agent、Team、普通 Python 函数，甚至嵌套的 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=["根据给定主题写一篇 300 字以内的短文。"],
)
reviewer = Agent(
name="Reviewer",
model=model,
instructions=[
"审校上一步的文章：检查事实表述、语法和结构。",
"直接输出修改后的最终版本。",
],
)
# 撰写 -> 审校，顺序由代码保证
workflow = Workflow(
name="writing-pipeline",
steps=[
Step(name="draft", executor=writer),
Step(name="review", executor=reviewer),
],
)
workflow.print_response(
"主题：为什么自动化测试值得投入", markdown=True
)运行时 draft 步骤的输出会自动成为 review 步骤的输入——这条数据流不需要你手动接线。
20.3 Step 执行器的三种形态
from agno.team import Team
from agno.workflow import Step, StepInput, StepOutput, Workflow
def data_preprocessor(step_input: StepInput) -> StepOutput:
"""函数执行器：做确定性加工，不消耗 token"""
research = step_input.previous_step_content or "没有收到调研内容"
return StepOutput(content=f"待分析的调研材料:\n{research}")
research_team = Team(name="Research Team", members=[...], model=model)
workflow = Workflow(
name="mixed-pipeline",
steps=[
research_team, # 形态一：直接放 Agent/Team（自动包装成 Step）
data_preprocessor, # 形态二：直接放函数
Step( # 形态三：显式 Step 包装，推荐用于需要命名/描述的场景
name="analysis",
executor=writer,
description="基于调研产出分析报告",
),
],
)三种形态可以自由混用。给步骤命名（name=）不只是为了可读性：事件追踪和输出查找都以步骤名为键。
20.4 StepInput 与 StepOutput：步骤间的数据契约
函数执行器遵循一个标准签名——接收 StepInput，返回 StepOutput：
def summarize(step_input: StepInput) -> StepOutput:
# input：本次工作流的原始输入（始终保留）
original = str(step_input.input)
# previous_step_content：上一个步骤输出的正文
prev = step_input.previous_step_content or ""
# previous_step_outputs：若上游是并行块，可拿到各命名步骤的输出字典
return StepOutput(
content=f"原始需求: {original}\n\n材料摘要:\n{prev[:500]}",
)常用字段速查：
| StepInput 字段 | 含义 |
|---|---|
input | 工作流的原始用户输入 |
previous_step_content | 上一步骤输出的内容文本 |
previous_step_outputs | 上游各步骤输出的集合（并行聚合后尤其有用） |
additional_data | 运行时附加的自定义数据 |
函数步骤的价值
在两个 Agent 之间插入纯函数步骤是 Workflow 最常用的模式：格式转换、长度截断、结果校验都不该烧 token，交给 Python 即可。
20.5 运行与解析输出
print_response() 适合调试；生产代码用 run() 拿到结构化结果：
result = workflow.run("主题：向量数据库选型")
# workflow_output.content 是最后一个步骤的输出
print(result.content)
# 更细粒度：逐步查看每个步骤的执行结果
for step_result in (result.step_results or []):
print(step_result.step_name, "->", str(step_result.content)[:80])也可以用 arun() 异步执行，或在 run(..., stream=True) 时迭代事件流观察每个步骤的开始与结束事件。
20.6 本章小结
- Team = 模型主导的灵活协作；Workflow = 代码定义的确定性控制流；
Step的执行器可以是 Agent、Team、函数或嵌套 Workflow，三者可在同一个 steps 列表混用；- 数据通过
StepInput.previous_step_content自动向下游流动，StepOutput(content=...)是函数步骤的标准出口； - 生产代码用
run()并遍历step_results解析各步骤输出，而不是解析打印文本。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 关于 Agno 中 Team 和 Workflow 的分工，正确的是？
2. Workflow 中一个 Step 的合法执行器不包括以下哪种？
3. 函数步骤中想获取「上一个步骤的输出」，应该读取？
4. 为什么推荐在两个 LLM 步骤之间插入纯 Python 函数步骤？
🛠️ 动手实践
- 把第 18 章的研究团队作为第一个步骤接入本章的写作流水线，实现"调研→撰写→审校"三步。
- 在撰写和审校之间加一个函数步骤
word_limit，把超过 400 字的草稿截断并附提示。 - 用
run()跑通整条流水线，并打印每个step_result的名称与前 100 字符。
下一章解锁 Workflow 的全部控制流原语：并行、分支与循环。