发布时间: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 篇(第二季 · 工程实现)。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐