02 · Workflow 编排：路径写死反而更好

零、开始之前

这篇的目标：学会判断「这个需求到底该不该用 Agent」，并且能把不该用 Agent 的那部分，用三种基本形状搭出来。

需要的前置知识：读过 01 ReAct，知道 Agent 循环长什么样。会用 Python 的 ThreadPoolExecutor（不会也没关系，本篇会讲）。

读完你会明白：

• 为什么线上很多「AI Agent」其实是 Workflow，而且这通常是对的
• 链式、并行、路由三种形状分别解决什么问题
• Workflow 和 Agent 的分界线具体在哪一行代码上
• 为什么 Workflow 场景要把 temperature 调低

先给一个可能反直觉的结论：

上一篇教你把方向盘交给模型。这一篇教你什么时候要把方向盘拿回来 —— 而且大多数时候你都该拿回来。

一、先看一个真实问题

你要做一个客服工单处理系统。用户发来一句话，系统要给出回复。

工单大概分三类：账单问题、技术故障、退款申请。三类问题的处理方式完全不同 —— 账单要查系统、技术要问日志、退款要走审批。

用 01 的 ReAct 怎么做？ 给模型三个工具（查账单、查日志、发起退款），让它自己判断该调哪个。能跑。但上线之后你会遇到：

问题	具体表现
贵	每个工单至少 2 次模型调用（判断 + 回答），复杂的 5-6 次
慢	每一轮都是一次完整的网络往返，P99 延迟根本控不住
不可预测	同一个工单，今天走了 3 步，明天走了 5 步
没法测	你没办法写单元测试，因为路径每次都不一样
出错难查	客户投诉「回复驴唇不对马嘴」，你要翻一长串对话历史才知道哪步歪了

而实际上，这个业务的路径是完全确定的：判断类型 → 走对应流程 → 出回复。三条路，就这三条。

既然路径是确定的，为什么要让模型每一步都重新决定一次？

二、那直接写一个大 prompt 行不行

你可能想：那我把三种情况都写进一个 prompt，让模型一次搞定。

prompt = """
你是客服。判断用户问题属于账单/技术/退款哪一类，
然后按对应的规范回复。账单类要包含账期和金额，
技术类要给出排查步骤，退款类要说明审批流程和时限。
用户问题：{input}
"""

这也能跑，但质量会明显不如拆开。原因很实在：

一次要求模型做太多事，它每件都会做得敷衍。 你让它同时「判断类型 + 遵守该类型的格式规范 + 组织语言」，它的注意力被摊薄了。实测下来，把它拆成「先判断，再按专门的 prompt 回复」，两步各自的质量都会上升。

这就引出了 Workflow 的核心思想：

把一个大任务拆成几个小任务，每次只让模型做一件事；至于这些小任务怎么串起来，由你的代码决定，不由模型决定。

三、概念：三种基本形状

Anthropic 在《Building Effective Agents》里把这件事讲得最清楚。他们先划了一条分界线：

	定义
Workflow	路径预先定义好的系统，模型只在每个格子里干活
Agent	模型自己决定路径的系统

然后归纳出五种基本模式。本篇讲其中三种最基础的，另外两种因为已经越过分界线，放到后面单独讲：

模式	中文	在哪讲
Prompt Chaining	链式	本篇
Parallelization	并行	本篇
Routing	路由	本篇
Orchestrator-Workers	主管-工人	05
Evaluator-Optimizer	评审-优化	04

三种形状用一句话概括：

判断该用哪个的口诀：

• 后一步需要前一步的结果 → 链式
• 几件事互不依赖 → 并行
• 只走其中一条路 → 路由

四、动手：三种形状分别搭出来

拆解用的是 anthropics/claude-cookbooks（51,826★，MIT 协议，可以直接抄进你的项目）里的 patterns/agents/basic_workflows.ipynb。

三个模式加起来不到 40 行代码。少到会让人怀疑「这也算模式？」—— 但形状简单不代表不值钱，值钱的是知道什么时候用哪个。

4.0 先准备一个最小的底座

三种形状共用同一个函数，就是「调一次模型，拿到文本」：

import os, re
from anthropic import Anthropic

def llm_call(prompt: str, system_prompt: str = "", model="claude-sonnet-4-6") -> str:
    client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
    response = client.messages.create(
        model=model,
        max_tokens=4096,
        system=system_prompt,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,          # ← 这个值待会儿单独讲
    )
    return response.content[0].text

注意和 01 的区别：这里没有 tools，也没有循环。 就是一问一答。Workflow 的每个格子都是这样一次性的调用。

4.1 链式：五行

图 2-1 链式（Prompt Chaining）的官方示意

图片来源：Anthropic — Building Effective Agents

先看代码有多简单：

def chain(input: str, prompts: list[str]) -> str:
    result = input
    for i, prompt in enumerate(prompts, 1):
        result = llm_call(f"{prompt}\nInput: {result}")
    return result

整个模式就是 result 被反复覆盖。 第一步的输出变成第二步的输入，以此类推。

怎么用它：

prompts = [
    "从这段文字里提取所有数字和它们的含义，一行一个：",
    "把上面的内容转成两列的 Markdown 表格：",
    "按数值从大到小重新排序：",
]
result = chain(用户输入的一大段文字, prompts)

为什么这么拆有用：一次让模型「提取 + 制表 + 排序」，它经常会漏掉几个数字，或者表格格式乱掉。拆成三步，每步只有一个要求，每步都能做对。

这段代码有个必须知道的性质：

result = llm_call(f"{prompt}\nInput: {result}")
#        ↑ 注意这里，result 被覆盖了

原始输入丢了。 第三步只能看到第二步的输出，看不到用户最初说了什么。所以链式的每一步 prompt 都必须是自包含的 —— 不能写「按用户要求排序」，因为这一步根本不知道用户要求是什么。

链式的风险：中间某步出错，后面全错，而且错误会被后续步骤当成事实继续加工。链越长越危险。经验值是别超过 4 步，超过就要在中间加校验。

4.2 并行：三行

图 2-2 并行（Parallelization）的官方示意

图片来源：Anthropic — Building Effective Agents

先想清楚它解决什么：你要给一份产品改动写「对不同利益方的影响分析」—— 对用户、对运营、对开发各写一份。这三份互不依赖，串行跑就是三倍时间。

from concurrent.futures import ThreadPoolExecutor

def parallel(prompt: str, inputs: list[str], n_workers: int = 3) -> list[str]:
    with ThreadPoolExecutor(max_workers=n_workers) as executor:
        futures = [executor.submit(llm_call, f"{prompt}\nInput: {x}") for x in inputs]
        return [f.result() for f in futures]

逐行读一遍（不熟悉 ThreadPoolExecutor 的话）：

行	在干什么
with ThreadPoolExecutor(max_workers=3)	开一个最多同时跑 3 个任务的线程池
executor.submit(llm_call, ...)	把任务丢进去就返回，不等它跑完；返回一个「凭证」（future）
[f.result() for f in futures]	拿着凭证挨个取结果，这一步才会阻塞等待

为什么用线程而不是进程：因为调 API 是 I/O 等待，不吃 CPU。线程足够了。

注意它的形状是「一个 prompt，多个 input」。 这是 Anthropic 归纳的两种并行里的第一种，叫 sectioning（分片）—— 把大任务切成互不依赖的小块。

还有第二种叫 voting（投票）：同一个 input 跑多次，取多数。代码形状几乎一样，只要把 inputs 换成 [x] * 5。用在「判断这段代码有没有安全漏洞」这类高风险分类上很实用 —— 单次调用会有假阴性，跑五次取多数能压下来。

三个必须处理的坑：

① n_workers=3 是保守值。 真正的上限不是你的 CPU，是 API 的速率限制。调高之前先去看厂商的 RPM（每分钟请求数）和 TPM（每分钟 token 数）配额，不然会满屏 429。

② 顺序是有保证的。 [f.result() for f in futures] 按提交顺序取，所以返回列表和 inputs 一一对应。别改成 as_completed，除非你确实不在乎顺序。

③ 一个失败会拖垮全部。 f.result() 遇到异常会把异常抛到主线程，前面已经成功的结果全丢了。生产上要这么写：

results = []
for f in futures:
    try:
        results.append({"ok": True, "value": f.result()})
    except Exception as e:
        results.append({"ok": False, "error": str(e)})   # 失败的单独标记，不影响别人

并行是三种形状里性价比最高的。 链式和路由改的是质量，并行直接把等待时间除以 N。

4.3 路由：三种形状里唯一有「决策」的

图 2-3 路由（Routing）的官方示意

图片来源：Anthropic — Building Effective Agents

现在回到开头那个客服工单的问题。

第一步：让模型做分类。

selector_prompt = f"""
Analyze the input and select the most appropriate support team from these options: {list(routes.keys())}
First explain your reasoning, then provide your selection in this XML format:

<reasoning>
Brief explanation of why this ticket should be routed to a specific team.
Consider key terms, user intent, and urgency level.
</reasoning>

<selection>
The chosen team name
</selection>

Input: {input}"""

这个 prompt 里有一个最关键的设计，必须理解：

它强制模型先写 <reasoning>，再写 <selection>。

为什么这个顺序如此重要？因为大模型是自回归的 —— 它一个 token 一个 token 往外吐，后面的 token 会受前面已经吐出来的内容约束。

• 先写理由再给结论：理由会真实地引导它得出结论，准确率明显提升
• 先给结论再写理由：结论已经说出口了，后面的"理由"只是在给它编个说法，完全没有作用

顺手还有个好处：reasoning 是免费的可观测性。线上分类错了，日志里直接能看到它当时是怎么想的。

第二步：把分类结果解析出来。

def extract_xml(text: str, tag: str) -> str:
    match = re.search(f"<{tag}>(.*?)</{tag}>", text, re.DOTALL)
    return match.group(1) if match else ""

route_response = llm_call(selector_prompt)
reasoning = extract_xml(route_response, "reasoning")
route_key = extract_xml(route_response, "selection").strip().lower()

re.DOTALL 这个参数不能少 —— 默认情况下正则的 . 不匹配换行符，而 reasoning 是多行的，少了它一个字都提不出来。

第三步：分发到专用 prompt。

selected_prompt = routes[route_key]
return llm_call(f"{selected_prompt}\nInput: {input}")

这一行有个真 bug，必须修：

selected_prompt = routes[route_key]     # 模型返回 "Billing Team" → KeyError

模型返回的字符串不一定精确匹配你的键 —— 多个空格、带引号、大小写不对、甚至自创一个类别。示例代码可以这样写，上生产必须加兜底：

selected_prompt = routes.get(route_key) or routes["人工兜底"]

「分类失败时怎么办」这条路径，往往比分类逻辑本身更需要设计。

还有一个细节值得注意：第二次调用只传了原始 input，没有传第一步的 reasoning。这是刻意的 —— 专家 prompt 应该基于原始输入独立判断，不该被分类器的措辞带偏。

路由最大的实际价值其实是省钱。 把它反过来用：先用便宜的小模型分类，简单问题让小模型直接答，只有难的才升级到贵模型。这叫 model cascade，是线上最常见的成本优化手段。

五、把两个容易忽略的决定讲透

5.1 为什么 temperature 要调到 0.1

temperature=0.1

temperature 控制模型输出的随机性：高了有创意但不稳定，低了稳定但呆板。

• ReAct 场景常用 0.7~1.0 —— 你希望它探索不同的解决路径
• Workflow 场景应该用 0.0~0.2 —— 你希望同样的输入，永远得到同样的输出

因为 Workflow 的价值就是可预测。一个分类器今天把工单分到账单、明天分到技术，那整个流程的可测试性就没了。

换了编排范式，采样参数也得跟着换。 这一点很多人会忽略，直接沿用默认值。

5.2 为什么用 XML 标签而不是 JSON

上面用的是 <reasoning>...</reasoning> 这种 XML 标签，而不是让模型返回 JSON。两个原因：

1. Claude 对 XML 标签的遵循度明显更好。 JSON 容易出现引号转义、尾逗号之类的问题，一个字符错了整个 parse 就失败。
2. XML 可以边收边切。 流式输出时，</reasoning> 一出现你就能拿到完整的理由段并开始展示；JSON 必须等整个对象收完才能解析。

不过要注意 extract_xml 提不到时返回的是空字符串，不是抛异常。好处是流程不中断，坏处是错误会静默地传到下一步。生产上建议改成显式失败：

def extract_xml(text: str, tag: str, required: bool = True) -> str:
    match = re.search(f"<{tag}>(.*?)</{tag}>", text, re.DOTALL)
    if match:
        return match.group(1)
    if required:
        raise ValueError(f"模型输出里没有 <{tag}> 标签，原始输出：{text[:200]}")
    return ""

六、Workflow 和 Agent 的分界线，具体在哪

三种形状看完，可以画出这条线了：

	谁决定下一步	可能的执行路径有几条
链式	代码	1 条
并行	代码	1 条（只是宽了 N 倍）
路由	模型选一次	N 条，但每条都是你写的
ReAct	模型每步都选	无穷多条

路由就站在分界线上。 模型有了选择权，但选项集是封闭的、你定义的、可枚举的。

再往前一步 —— 让模型自己决定「要不要再来一轮」（04 Reflection）、「派几个人去干」（05 Supervisor）—— 就进入 Agent 领域了。

Anthropic 给的建议很直白，值得原样记住：

先用最简单的方案，只有在简单方案确实不够时，才增加复杂度。

多数团队犯的错不是 Workflow 用得太多，而是一上来就上多智能体。

七、常见故障与排查

你看到的现象	大概率的原因	怎么修
链式跑到第 3 步结果就离谱了	第 2 步的小错被当成事实放大	步骤间加校验；链长控制在 4 步内
链式某步「不按用户要求来」	该步 prompt 不是自包含的，拿不到原始需求	把关键约束写进每一步的 prompt
路由直接崩了，报 KeyError	模型返回了不在选项里的类别	加默认路由（4.3）
分类准确率忽高忽低	让模型先给结论后给理由，或者根本没要理由	强制 reasoning 在 selection 之前
满屏 429	并行度超了 API 配额	按厂商 RPM/TPM 调 n_workers，加退避重试
并行跑了一半全崩	一个 future 抛异常带走了全部结果	每个 future 单独 try（4.2 ③）
同样的输入结果不一样	temperature 太高	调到 0.1 以下
解析不到内容但程序没报错	extract_xml 静默返回了空串	改成显式失败（5.2）
流程图画得很复杂，但线上只走一条路	分支是想象出来的，不是真实需求	看日志里各分支的真实命中率，砍掉从没走过的

八、全局定位

Workflow 在最左端：最便宜、最快、最可预测、最好测试，代价是路径必须能穷举。

什么时候不该用 Workflow：

• 路径真的没法穷举 —— 「帮我把这个 bug 修了」，你写不出分支图，别硬写，用 ReAct
• 分支超过十几条还在长 —— 说明你在用 if-else 模拟推理，该换 Agent 了
• 每加一个场景都要发版 —— 加一个客服类别就要改代码上线，这个成本迟早压垮你

反过来，只要路径能穷举，就别上 Agent。

九、小结与检查清单

核心三句话：

1. Workflow 和 Agent 的区别只有一个：路径是你写的，还是模型定的
2. 三种形状对应三种关系：有依赖用链式、无依赖用并行、选一条用路由
3. 分类任务必须让模型先写理由再给结论，顺序反了等于没写

自查清单：

• temperature 是不是调到 0.2 以下了？
• 链式的每一步 prompt 是不是自包含的（不依赖看不到的上文）？
• 链子有没有超过 4 步？中间有没有校验？
• 路由有没有默认兜底分支？
• 分类 prompt 里，reasoning 是不是排在 selection 前面？
• 并行的 n_workers 是按 API 配额定的，还是随手写的 3？
• 并行任务里单个失败，会不会拖垮整批？
• 解析失败时，是静默返回空串还是显式报错？

参考资料

资料	位置	协议
本篇拆解的源码	patterns/agents/basic_workflows.ipynb，33KB	MIT
共用底座	patterns/agents/util.py，45 行	MIT
概念出处	Anthropic — Building Effective Agents	——

下一篇：03 · Plan-and-Execute —— 路径没法写死，但又不能让模型走一步看一步，怎么办。