AI Agent 工程实践(11):Workflow——Agent 如何完成复杂任务?
发布时间:2026-07-12
标签:AI Agent|LLM|Workflow|LangGraph|DAG|工程实现
系列导航
上一篇:AI Agent 工程实践(10):Memory 不只是聊天记录——长期记忆架构设计
下一篇:AI Agent 工程实践(12):为什么很多 Multi-Agent 项目最后都失败了
本文是 [AI Agent 工程实践] 系列的第 11 篇(第二季 · 工程实现)。
单步 Agent 能做的事,早就不是问题了。
"帮我写个排序函数"——秒出。"审查这段代码"——秒出。"解释一下 async/await"——秒出。
可我第一次让它"从头做完一个需求"时,问题全来了:规划到一半忘了后半截、执行出错不知道怎么回退、跑到第三步时第一步的上下文已经丢了。
我突然意识到:Agent 的智商在模型,执行力在 Workflow。 模型决定单步质量,Workflow 决定能不能把多步串成一次完整交付。
第 08 篇讲了"什么时候该用 Workflow",这一篇讲"Workflow 本身怎么落地"——从规划到执行到审查,一条链路串到底。
本文你将学到
✓ 为什么单步 Agent 做不了复杂任务——三个不可逾越的限制
✓ Workflow 的核心链路:User → Planner → Execute → Review → Finish
✓ 四种实现范式:Pipeline / DAG / State Machine / LangGraph,各自适用场景
✓ 用 LangGraph 落地一个真实 Workflow 的完整代码
适合阅读
✓ 做过单步 Agent、想让它能"独立跑完项目"的人
✓ 在选型 Workflow 框架(LangGraph / Temporal / Prefect)的开发者
✓ 想理解 DAG / State Machine / Pipeline 在 Agent 里怎么用的工程师
问题背景
单步 Agent(一问一答)的能力边界非常清晰。它做不了复杂任务,原因不是模型差,是三个结构性问题:
上下文溢出。 一个需求从规划到上线可能涉及 10+ 步。全塞进一个 prompt,模型对中间步骤的遵循率断崖式下降——又是 Lost in the Middle。
步间信息丢失。 第三步的输出是第四步的输入,但第三步跑完,上下文被新的 prompt 覆盖,第一步的关键约束早被冲走了。Agent "走着走着忘了在干什么"。
错误无回退。 第五步执行失败,要回到第三步重新来——但没 Workflow 的 Agent 不知道"回退"是什么意思,它只会继续往下跑,或者直接报一个无上下文的错。
一句话:模型能做任何单步,但做不了任何多步。 Workflow 就是那根把单步串成多步的骨架。
错误尝试
第一次:超长 prompt,把所有步骤写进去
设计一个 3000 字的 system prompt:"你先做 A,然后根据 A 的结果做 B,如果 B 失败就做 C……"
结果:前十步还像样子,后面开始丢步骤。模型并不按你的 prompt 所列的顺序严格执行——它只是在"续写文本",不是在执行流程。把 prompt 当流程引擎,是把水彩画当蓝图。
第二次:手动接力——我做编排
Agent 跑完一步,我人工检查输出,手动传给下一步。流程是跑通了,但我成了 Workflow 引擎。
结果:一个需求来回人工接 5 次力,和直接手写代码差不多累。而且人会忘——漏传一个中间输出的概率和忘记手动指定 task_type 一样高(第 05 篇的老问题)。人当编排器,不可靠、不可持续。
两次尝试指向同一个结论:复杂任务必须有一个独立于人和模型的"编排器"——它负责步骤顺序、状态传递、异常回退、条件分支。这个编排器就是 Workflow。
关键观察
我把"Agent 成功完成的复杂任务"和"失败的"做了对比,差异不在模型,在编排方式:
| 任务特征 | 无 Workflow | 有 Workflow |
|---|---|---|
| 步骤 ≤2 | 能完成 | 能完成 |
| 步骤 3-5 且线性 | 偶尔能(看运气) | 稳定完成 |
| 步骤 ≥5 或有分支 | 几乎必失败 | 稳定完成 |
| 某步需要重试 | 不知道回退 | 自动回退到上一步 |
pie title 复杂任务失败原因
"上下文溢出 · 丢步骤" : 35
"步间信息丢失" : 30
"错误无回退机制" : 25
"其他" : 10
Agent 的智商在模型,执行力在 Workflow。
模型再好,没有编排也是"每次从头教它怎么做项目"。Workflow 把"怎么做"固化成骨架,模型只负责每步里的"做多好"。
问题不是"模型不够强",而是"步骤之间的连接"没有固化。正确做法是把"编排逻辑"从 prompt 里抽出来,交给一个独立引擎——Workflow 管步骤顺序和状态,模型只负责每步的生成。
最终方案:Workflow 链路 + 四种实现范式
核心链路:User → Planner → Execute → Review → Finish
这就是你给的链路——一个通用 Workflow 的最简骨架:
每个节点职责清晰:
- Planner:把用户需求拆成可执行的步骤清单(不是模型自由发挥,而是产出结构化 plan)
- Execute:按步骤清单逐条执行,产出中间结果
- Review:检查执行结果——不通过就回到 Execute(回退就是 Workflow 的价值)
- Finish:汇总通过审查的结果,交付用户
Planner-Execute-Review 不是三个 Agent(08 篇讲过),是 Workflow 的三个节点,同一个 Agent 换 skill 执行。 Workflow 管的是"节点怎么串、状态怎么传、失败怎么回"。
四种实现范式
同样的链路,不同复杂度用不同实现方式:
| 范式 | 适合什么 | 本质 | Agent 场景举例 |
|---|---|---|---|
| Pipeline | 固定步骤、无分支 | 线性串联 | "翻译 → 润色 → 输出" |
| DAG | 有分支/并行 | 有向无环图 | "同时研究方案A和B → 合并 → 决策" |
| State Machine | 需要回退/重试/异常 | 状态 + 事件驱动 | Planner→Execute→Review(失败回 Execute) |
| LangGraph | 以上三者的工程落地 | Graph + State + 条件边 | 任意复杂 Workflow 的生产实现 |
选型原则:步骤固定且 >2 步 → Pipeline / 有分支并行 → DAG / 需要回退重试 → State Machine / 真正写代码落地 → 用 LangGraph 表达前三种的任一种。
stateDiagram-v2
[*] --> Pipeline: 步骤固定,线性
Pipeline --> DAG: 出现分支/并行需求
DAG --> StateMachine: 需要回退/重试
StateMachine --> LangGraph: 需要工程化落地
四种不是互斥的——LangGraph 是前三种的工程实现载体,不是"第四种范式"。
实际收益
| 指标 | 超长 Prompt | 手动接力 | Workflow 编排 |
|---|---|---|---|
| 步骤完成率 | ~40%(常丢) | ~80%(人漏传) | ~95% |
| 错误恢复 | 无 | 靠人 | 自动回退重试 |
| 人参与度 | 低(但质量低) | 高(累) | 低(自动化) |
架构图 / 流程图
带有回退的完整 Workflow(State Machine 视角)

关键:fail · 需重新规划 这条回退边,是 Workflow 区别于 Pipeline 的核心——Pipeline 只能往前,Workflow 能回头。
代码或配置示例
Planner-Execute-Review Workflow(LangGraph 实现)
from langgraph.graph import StateGraph, END
# 1. 定义状态(Workflow 的"通行证",随节点流转)
class WorkflowState(TypedDict):
task: str # 用户需求
plan: list[str] # Planner 产出的步骤清单
results: list[str] # 每步的执行结果
step_index: int # 当前执行到第几步
review_pass: bool # Review 是否通过
# 2. 定义节点(Planner / Execute / Review)
def planner(state: WorkflowState) -> WorkflowState:
state["plan"] = agent.run(state["task"], skill="planning")
state["step_index"] = 0
return state
def executor(state: WorkflowState) -> WorkflowState:
step = state["plan"][state["step_index"]]
result = agent.run(step, context=state["results"])
state["results"].append(result)
state["step_index"] += 1
return state
def reviewer(state: WorkflowState) -> WorkflowState:
state["review_pass"] = agent.run(state["results"], skill="review")
return state
# 3. Workflow 定义(节点 + 边 + 条件)
wf = StateGraph(WorkflowState)
wf.add_node("planner", planner)
wf.add_node("executor", executor)
wf.add_node("reviewer", reviewer)
wf.set_entry_point("planner")
wf.add_edge("planner", "executor")
# 关键:条件边——Review 决定往前走还是回头
wf.add_conditional_edges("reviewer", lambda s: "pass" if s["review_pass"] and s["step_index"] >= len(s["plan"])
else ("execute" if s["step_index"] < len(s["plan"]) else "executor"))
wf.add_edge("executor", "reviewer")
# 4. 编译运行
app = wf.compile()
result = app.invoke({"task": "实现用户登录模块"})
代码不长,但 Workflow 的全部核心都在里面:状态在节点间传递、条件边决定去向、同一 Agent 换了三个 Skill 跑 Planner/Execute/Review。 这就是第 05 篇 Router 和第 07 篇 Skills 在 Workflow 里的交汇点。
设计权衡
| 候选方案 | 优点 | 缺点 | 不选的原因 |
|---|---|---|---|
| 超长 Prompt 硬塞多步 | 零工程成本 | 丢步骤、无回退 | prompt 不是流程引擎 |
| 手动接力 | 灵活 | 人不可靠、不可持续 | 等于没自动化 |
| Pipeline(纯线性) | 简单可靠 | 无分支、无回退 | 复杂任务必有分支和重试 |
| State Machine Workflow | 有回退、有分支、有状态 | 需工程化 | 选择理由:唯一能处理"出错后怎么办"的编排方式 |
Workflow 不是越复杂越好。 两步以内的任务不需要 Workflow,和第 08 篇同一条原则——不要为"两句话的事"上编排引擎。Workflow 的价值在步骤 ≥3 或有分支/回退需求时才体现。
总结
✅ 单步 Agent 的三个天花板:上下文溢出、步间信息丢失、无回退——Workflow 逐一破解。
✅ 核心链路:User → Planner(拆任务)→ Execute(逐步执行)→ Review(质量检查+回退决策)→ Finish(交付)。
✅ 四种实现范式:Pipeline(线性)/ DAG(分支并行)/ State Machine(回退重试)/ LangGraph(工程落地)。
✅ 状态是 Workflow 的灵魂——没有状态传递,节点之间就是孤岛。
✅ Workflow 的价值在步骤 ≥3 或有分支/回退时体现,两步任务别上。
参考资料
- LangGraph 官方文档 — StateGraph & Conditional Edges → Workflow 工程实现的直接依据,本文代码示例的技术来源
- Temporal / Prefect 文档 → 传统 Workflow 引擎的故障恢复与回退机制参考
- 第 08 篇:Workflow / Multi-Agent → "什么时候该用 Workflow",本文的前置决策篇
- 第 05 篇:Rule Router → Router 在 Workflow 节点间如何动态选择 Skill
- 第 07 篇:Skills vs Prompt → Planner/Execute/Review 各节点调用的 Skill 定义
系列导航
上一篇:AI Agent 工程实践(10):Memory 不只是聊天记录——长期记忆架构设计
下一篇:AI Agent 工程实践(12):为什么很多 Multi-Agent 项目最后都失败了
本文是 [AI Agent 工程实践] 系列的第 11 篇(第二季 · 工程实现)。
更多推荐


所有评论(0)