Skip to content

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. 动手前先问四个问题 ​

  1. 步骤固定吗? 固定 → Prompt Chaining / Routing / Parallelization;
  2. 子任务数量运行时才知道吗? 是 → Orchestrator-Worker(Send);
  3. 质量需要反复打磨吗? 是 → Evaluator-Optimizer(记得设重写上限);
  4. 连“下一步干什么”都要模型自己定吗? 是 → Agent(工具循环)。

同一个业务往往要组合。例如报告生成:先并行收集资料,再顺序汇总,最后评估打回。组合时给每一层规定输入输出,避免所有节点共享一个不断膨胀的状态对象。

设计评审时用最坏情况走一遍:模型分类失败、某个工具超时、用户中途取消、循环达到上限。这些情况下仍有明确结果,结构才算完整。

从数据依赖判断结构也很有用:画出每项任务需要哪些输入、是否依赖前一步、失败后能否单独重试。独立任务才并行;固定步骤才串联;输入类别清楚才路由;答案需要多轮修订才循环。

📁 本节示例代码:code/examples/09_workflow_patterns_demo.py —— Routing / Orchestrator-Worker / Evaluator-Optimizer 三张可运行图。


扩展阅读 ​


下一节: Checkpointer 解决的是一次对话内的记忆。用户隔天再来,Agent 就失忆了。跨线程的 Store,以及 Time Travel 回放与改道,是记忆的另一半。