主题
10.9 工作流设计模式:先选型,再动手
学到这里,工具箱已经很满:状态图、条件路由、Send 并行、工具循环。新手最容易犯的错,是接到需求就堆节点——画一张二十个格子的巨图,最后发现三行 if/else 就够了。
LangGraph 官方在 Workflows and agents 里给了一张模式菜单。这一节把名字和适用场景钉牢。你会发现前面各节其实已经把这些模式练过一遍,缺的只是对号入座。
1. 先分清:Workflow 与 Agent 是两种东西
官方的分界只有一条:下一步由代码写死,还是由模型自己定?
- Workflow(工作流):开发者预先规定每一步,模型只是其中的工人。像工厂质检线:收件 → 检测 → 贴标 → 出库,顺序不变。
- Agent(智能体):模型自己决定调用哪个工具、何时结束。像交给同事一句“把这份竞品调研做完”,路径由他现场判断。
选型口诀:流程越确定,越要 Workflow(可控、便宜、好调试);任务越开放,才越要 Agent(灵活,但费 Token、易跑偏)。很多生产系统是外壳 Workflow、内芯 Agent——大框架写死,只在真正需要理解自然语言的点上请模型上场。
四个问题都答“不需要模型自己定下一步”,也许你根本不需要 LangGraph。一段普通 Python 更稳、更便宜。这也是一种成熟。
2. 模式速查表
官方正文里并列的是五类工作流 + Agent 循环。下面把它们和前面练过的节次对齐:
| 模式 | 一句话 | 生活里像什么 | 前面哪节 |
|---|---|---|---|
| Prompt Chaining(提示链) | A 的输出接给 B,B 再交给 C | 译稿先直译、再润色、最后校对 | 02 节多节点串联 |
| Routing(路由) | 先分类,再按类别走不同分支 | 医院分诊、客服一级分类 | 03 节条件路由 |
| Parallelization(并行) | 多个独立子任务同时开工 | 同一天把调研、报价、合规发给三个同事 | 04 节 Fan-out / Fan-in |
| Orchestrator-Worker(编排者-工人) | 主管先拆任务,动态派给 N 个工人,再汇总 | 项目经理按需求单派活 | 04 节 Send(Map-Reduce) |
| Evaluator-Optimizer(评估者-优化者) | 生成器写稿,评估器打分,不合格打回 | 作者与编辑改稿 | 03 节决策树 |
| Agent(工具循环) | 模型自己决定用哪个工具、何时结束 | 顾问 + 办事员来回配合 | 08 节 |
3. 三个最值得展开的模式
3.1 Orchestrator-Worker:工人数量运行时才知道
它和普通并行的差别只有一句:N 不是编译期写死的。 “把这篇文章摘要翻译成几种语言”要先读文档才知道几种。官方在这一模式里推荐的,正是 Send:
python
from langgraph.types import Send
def orchestrator(state: State):
"""主管节点:先让模型(或规则)产出 langs,再由条件边按清单派发。"""
return {"langs": state["langs"], "source_text": state["text"]}
def dispatch(state: State):
return [
Send("translate", {"lang": lang, "text": state["source_text"]})
for lang in state["langs"]
]主管负责拆任务并写入 State;dispatch 是条件边,不是节点。工人返回的片段用 reducer 拼起来,汇总节点等所有工人结束后再跑——和 04 节多城市比价是同一套闭环。
和 12 节 Subagents / Supervisor 的差别:Orchestrator-Worker 通常一次拆完、并行收活;Supervisor 可以根据中间结果跨多轮继续派活。
3.2 Evaluator-Optimizer:改稿必须有上限
两个角色、一条回路:Generator 生成,Evaluator 用结构化输出打分,不合格就带批注重写。两个保险丝不能省:
python
from typing import Literal
from pydantic import BaseModel, Field
class Feedback(BaseModel):
grade: Literal["ok", "needs_improvement"]
comment: str
def route_after_eval(state: State) -> Literal["rewrite", "pass", "force_pass"]:
if state["revision"] >= 3:
return "force_pass"
return "rewrite" if state["feedback"].grade == "needs_improvement" else "pass"评估器输出必须是结构,不能是一段随口的“再改改”。路由函数只读字段。重写次数写进 State,超过上限强制放行——否则编辑和作者会把 Token 烧光。
3.3 Routing:最容易被低估的性价比之王
Routing 看起来简单,却常常是生产系统里最划算的一刀:大部分请求其实不需要重型 Agent。一次廉价分类,就能把 80% 的流量分给轻量分支,只有复杂请求才进大循环。旅行助手的主助理,本质就是 Routing:用转交工具把意图分给四个子助理。
分类失败怎么办,03 节已经强调过:未知标签要有澄清或人工出口,不要“猜一个最近的科室”。
4. 动手前先问四个问题
- 步骤固定吗? 固定 → Prompt Chaining / Routing / Parallelization;
- 子任务数量运行时才知道吗? 是 → Orchestrator-Worker(
Send); - 质量需要反复打磨吗? 是 → Evaluator-Optimizer(记得设重写上限);
- 连“下一步干什么”都要模型自己定吗? 是 → Agent(工具循环)。
同一个业务往往要组合。例如报告生成:先并行收集资料,再顺序汇总,最后评估打回。组合时给每一层规定输入输出,避免所有节点共享一个不断膨胀的状态对象。
设计评审时用最坏情况走一遍:模型分类失败、某个工具超时、用户中途取消、循环达到上限。这些情况下仍有明确结果,结构才算完整。
从数据依赖判断结构也很有用:画出每项任务需要哪些输入、是否依赖前一步、失败后能否单独重试。独立任务才并行;固定步骤才串联;输入类别清楚才路由;答案需要多轮修订才循环。
📁 本节示例代码:code/examples/09_workflow_patterns_demo.py —— Routing / Orchestrator-Worker / Evaluator-Optimizer 三张可运行图。
扩展阅读
- Workflows and agents(本节官方母本):workflows-agents
- 图 API 概念(条件边 / Send / Command):graph-api
下一节: Checkpointer 解决的是一次对话内的记忆。用户隔天再来,Agent 就失忆了。跨线程的 Store,以及 Time Travel 回放与改道,是记忆的另一半。