ReWOO 跑多步规划时,Planner 和 Solver 各发一次 LLM 调用,run_rewoo 和 run_react_mock 的字符统计会把 ReAct 的二次膨胀暴露得很清楚。把 ScriptedPlanner/ScriptedSolver 换成真实模型,密钥管理最容易变成杂活。TaoToken 的统一接入方式在这里很顺手:先打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 创建 YOUR_API_KEY,再把 OpenAI 兼容客户端的 base_url 填成 https://taotoken.net/api,让 Planner 的 plan_for 和 Solver 的 solve 共用同一把 Key。下面按原文节奏走:先看 ReAct 为什么贵,再看三角色怎么解耦,最后把真实 LLM 配置落到代码里。
1. 从 run_react_mock 的 history_chars 看 ReAct 的二次膨胀
1.1 ReAct 每步把原始 prompt 再带一遍
ReAct 的循环是思考、行动、观察、再思考。这个循环在短任务里很自然,但它的 prompt 结构有个隐藏成本:每一步都要把原始问题、之前所有 Thought、Action、Observation 重新塞进上下文。run_react_mock 没有真的跑模型,它只是手算这条增长曲线,把 prompt_chars 和 history_chars 分开累加。
代码里 history_chars 每轮都会增加“工具名 + 参数 + 观察结果 + 分隔符”的长度,而下一步的总字符又等于 prompt_chars + history_chars + 新动作。于是第 1 步带一次原始问题,第 2 步又带一次,第 10 步还在带。原始 prompt 被重复发送,历史又线性变长,合起来就接近二次增长。真正跑生产任务时,这部分会直接变成账单。
def run_react_mock(question, tool_registry, trajectory): prompt_chars = len(question) history_chars = 0 total = 0 for tool_name, args in trajectory: total += prompt_chars + history_chars + len(tool_name) + len(str(args)) observation = tool_registry.dispatch(tool_name, args) history_chars += len(tool_name) + len(str(args)) + len(observation) + 40 total += prompt_chars + history_chars return total这段 mock 的重点不是模拟 ReAct 的智能,而是把“每步都带全历史”这件事量化。ReWOO 要解决的第一个痛点就是它。
1.2 失败卡在循环中间,恢复路径会污染后续思考链
ReAct 的第二个痛点在失败恢复。假设第三步工具报错,循环不会停下来等你修,它会把这个错误观察继续喂给模型,让模型在流中间重新推理整条计划。前文越长,模型越容易忘记自己原来的计划是什么,可能把搜索方向、参数顺序、目标问题全改掉。
更麻烦的是,错误观察会污染后面的思考链。一个 timeout、一个空结果、一个字段缺失,都可能让模型在后续步骤里反复解释这个错误,而不是继续完成原任务。调试时你看到的是“按步骤”的失败,但根因可能藏在前面某一步的观察里。
ReWOO 把失败粒度改成“按节点”。Worker 3 报错就写一个 error 字符串进 evidence 字典,Planner 完全看不到 Observation,Solver 拿到的是原计划加一条错误证据,它可以在保有全局计划的情况下优雅降级。重跑时也只需要重跑那个节点,不用把整条链重新推理一遍。
1.3 ReWOO 的下注:计划在执行前就能写出来
ReWOO 的核心判断是:大多数任务的计划在执行前就能写出来,真正需要反应式调整的步骤其实很少。Planner 只拿 user_question,输出一个有向无环图,每个节点说明用哪个工具、参数是什么、依赖哪些更早节点。节点之间用#E1、#E2这种证据引用连接。
这样做的代价是计划静态,灵活性下降;收益是 token 暴降、失败定位清晰、规划器可以被蒸馏。对于工具已知、结构化、对 token 敏感、证据可并行的任务,这笔账通常划得来。反过来,环境完全未知、需要边走边反应的短任务,ReAct 仍然更合适。Anthropic 在 2024 年 12 月的建议也是从最简单的开始:一次工具调用加一段汇总,别搭 ReWOO;四十步的研究作业,别只用 ReAct。
2. Planner、Worker、Solver 三段解耦:Planner 和 Solver 才是两次 LLM 调用
2.1 Planner 只看 question,吐出 PlanStep DAG
Planner 的输入只有原问题,看不到任何 Observation。它输出的是“该怎么做”的草图,不是“做完了什么”的复盘。代码里可以用两个纯数据类承载结构,PlanStep 负责单个节点,Plan 负责整个步骤列表。字段不复杂,但拆开以后扩展空间更大,以后加 meta、version、max_parallel 都不会破坏调用方。
from dataclasses import dataclass, field from typing import Any @dataclass class PlanStep: id: str tool: str args: dict[str, Any] @dataclass class Plan: steps: list[PlanStep]Planner 的输出通常是一段 JSON,里面每个 step 有 id、tool、args。args 里的值可以包含#E1这样的占位符,表示“这个参数要等 E1 的输出”。Planner 输入空间小,所以它很适合被蒸馏:用大模型产出的规划轨迹微调一个小模型,推理时小模型规划、大模型求解,这是生产 Agent 里常见的省钱架构。
2.2 Worker 按拓扑序跑工具,evidence 字典隔离错误
Worker 不调用 LLM,它只是按拓扑序执行节点。先把没有依赖的节点跑掉,再把依赖已经完成的节点跑掉,同一层内的节点理论上可以并行。每个节点执行三件事:绑参、分派、存证。绑参是把字符串参数里的#E1换成真实值,分派是调用工具,存证是把结果写进 evidence 字典。
import re REF = re.compile(r"#E(\d+)") def resolve_refs(value, evidence): if not isinstance(value, str): return value return REF.sub(lambda m: evidence.get(f"E{m.group(1)}", m.group(0)), value) def run_workers(plan, tools): done = set() evidence = {} pending = list(plan.steps) while pending: ready, rest = [], [] for step in pending: refs = REF.findall(str(step.args)) if all(f"E{r}" in done for r in refs): ready.append(step) else: rest.append(step) if not ready: raise RuntimeError("cyclic plan or unresolved reference") for step in ready: bound = {k: resolve_refs(v, evidence) for k, v in step.args.items()} evidence[step.id] = tools.dispatch(step.tool, bound) done.add(step.id) pending = rest return evidence这里有个 Python 细节容易踩:all([])返回 True,所以没有依赖的节点会立刻被排出来,这正是我们想要的效果。用 set 记录已完成节点,查询是 O(1),比 list 快得多。
2.3 Solver 拿原问题 + 计划 + evidence 做最后一次调用
Solver 拿到原问题、计划和所有 evidence,做最后一次 LLM 调用产出答案。它的 prompt 里包含原计划,所以即使某个工具返回 error,它也知道原计划是什么,可以解释“E3 因为工具超时没有拿到,但基于 E1、E2 可以给出部分结论”。这就是 ReWOO 比 ReAct 稳健的地方。
从 LLM 调用次数看,Planner 一次、Solver 一次,Worker 零次。这两次调用就是本篇要统一认证的地方。把 ScriptedPlanner/ScriptedSolver 换成真实模型时,你不需要给规划和求解分别维护不同服务商的 Key,也不需要在一个脚本里塞两套 base_url。两处都走同一把 TaoToken Key,认证管理会简单很多。
3. run_rewoo 与 run_react_mock 的字符账:worker prompt 不再带全历史
3.1 run_rewoo 的三个计数:planner_chars、worker_chars、solver_chars
玩具代码里用字符数近似 token 数,因为 token 数依赖 tokenizer,字符数更容易在离线环境比较。run_rewoo 把一次运行拆成三个计数:planner_chars 是问题加计划描述,worker_chars 是每个工具的参数加输出,solver_chars 是问题加全部 evidence 加答案。
@dataclass class ReWOORun: question: str plan: Plan evidence: dict[str, str] = field(default_factory=dict) answer: str = "" planner_chars: int = 0 worker_chars: int = 0 solver_chars: int = 0 def run_rewoo(question, planner_fn, tool_registry, solver_fn): plan = planner_fn(question) evidence = run_workers(plan, tool_registry) answer = solver_fn(question, plan, evidence) planner_chars = len(question) + sum(len(s.tool) + len(str(s.args)) for s in plan.steps) worker_chars = sum(len(str(s.args)) + len(evidence[s.id]) for s in plan.steps) solver_chars = len(question) + worker_chars + len(answer) return ReWOORun(question, plan, evidence, answer, planner_chars, worker_chars, solver_chars)关键差异在 worker_chars。Worker 执行每个节点时,只带这个工具需要的入参,不带完整历史。E2 的 prompt 里只有“population of Paris”,没有 E1 的 Thought、Action、Observation 全记录。省下来的就是 ReAct 里被重复发送的那部分。
3.2 run_react_mock 的 history_chars 为什么单调递增
ReAct 的 mock 里,history_chars 每一步都在增加,而且下一步的总消耗又把它算进去。你可以把 run_react_mock 和 run_rewoo 的总字符数打印出来对比,ratio 就是字符层面的倍数。注意这个倍数不是生产账单的精确预测,它只是让你直观看到“worker prompt 不再携带全历史”带来的结构差异。
react_chars = run_react_mock( question, tools, [ ("search", {"query": "capital of France"}), ("search", {"query": "population of Paris"}), ("round_million", {"text": "11.2 million metro"}), ], ) rewoo_chars = run.planner_chars + run.worker_chars + run.solver_chars ratio = react_chars / max(rewoo_chars, 1)论文在 HotpotQA 上测到 token 约少 5 倍,准确率绝对值 +4,那是论文的实验结论。玩具里的字符统计只看结构,不能把 ratio 直接当成你的生产节省比例。真实 token 以模型广场当时列表和实际返回 usage 为准。
3.3 失败粒度从步骤变成节点,重跑成本也降了
ReAct 失败时,你面对的是“第 3 步到第 7 步这条链要重来”。ReWOO 失败时,你面对的是“E3 这个节点返回了 error”。重跑只需要重跑 E3,或者把 E3 的参数改好再跑。Planner 完全没看见错误,原计划不变,Solver 仍然可以在完整计划上下文里给出降级答案。
这种结构还有一个好处:evidence 字典是显式的。每个节点的输出都留痕,调试时不用从一长串 Thought 里翻找某次搜索到底返回了什么。对于多工具编排,显式证据比隐式历史更好排查。
4. 把 ScriptedPlanner/ScriptedSolver 换成真 LLM:Planner 和 Solver 共用 TaoToken Key
4.1 先打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 创建 YOUR_API_KEY
准备材料不复杂:Python 3.10 以上、openai 包、一把 API Key。打开 TaoToken 注册并创建 API Key,复制出来先用占位符 YOUR_API_KEY 代指。模型 ID 不要凭记忆写,去 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 的模型广场看当时列表,把真实的模型 ID 复制到环境变量里。
pip install openai export TAOTOKEN_API_KEY="YOUR_API_KEY" export TAOTOKEN_PLANNER_MODEL="YOUR_MODEL_ID" export TAOTOKEN_SOLVER_MODEL="YOUR_MODEL_ID"Planner 和 Solver 可以用不同模型:Planner 负责规划,输入小,可以选便宜的;Solver 负责缝证据和长推理,可以选更强的。但认证不需要两套 Key,同一个 client 就能覆盖两次调用。
4.2 OpenAI 兼容客户端:base_url 填 https://taotoken.net/api,末尾不要 /v1
Python 侧用 OpenAI 兼容客户端接入。注意 base_url 填https://taotoken.net/api,末尾不要加/v1。Key 从官网创建,不要写到代码仓库里,用环境变量读取。
import os import json from openai import OpenAI client = OpenAI( api_key=os.environ.get("TAOTOKEN_API_KEY", "YOUR_API_KEY"), base_url="https://taotoken.net/api", ) PLANNER_MODEL = os.environ.get("TAOTOKEN_PLANNER_MODEL", "YOUR_MODEL_ID") SOLVER_MODEL = os.environ.get("TAOTOKEN_SOLVER_MODEL", "YOUR_MODEL_ID")这个 client 只创建一次。Planner 的 plan_for 和 Solver 的 solve 都复用它。这样两处调用的认证走同一把 Key,Base URL 也只有一处需要维护。
4.3 Planner 的 plan_for 和 Solver 的 solve 各发一次调用,认证走同一个 client
Planner 的职责是输出结构化 JSON。系统提示要明确要求只输出 JSON,不要解释,不要 markdown。用户消息里给出输出格式示例,但示例里的模型 ID、工具名要按你注册的工具来写。解析失败时,可以只重试 Planner,不要动 Worker。
class LLMPlanner: def __init__(self, client, model): self.client = client self.model = model def plan_for(self, question: str) -> Plan: resp = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": "你是 ReWOO Planner。只输出 JSON,不要解释。"}, { "role": "user", "content": ( f"问题:{question}\n" "输出格式:{\"steps\":[{\"id\":\"E1\",\"tool\":\"search\"," "\"args\":{\"query\":\"...\"}}]}" ), }, ], temperature=0, ) raw = resp.choices[0].message.content or "" data = json.loads(raw) return Plan([PlanStep(s["id"], s["tool"], s["args"]) for s in data["steps"]])Solver 的职责是缝证据。把原问题、计划、evidence 一起塞进 prompt,要求它只根据证据给答案。Solver 不需要看到 Worker 的中间思考,只看到结构化结果,这比 ReAct 的历史链干净得多。
class LLMSolver: def __init__(self, client, model): self.client = client self.model = model def solve(self, question: str, plan: Plan, evidence: dict[str, str]) -> str: payload = { "question": question, "plan": [{"id": s.id, "tool": s.tool, "args": s.args} for s in plan.steps], "evidence": evidence, } resp = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": "你是 ReWOO Solver,只根据 evidence 给出最终答案。"}, {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}, ], temperature=0, ) return resp.choices[0].message.content这两段代码里没有任何“中转”“共享额度”之类的动作,只是把 OpenAI 兼容客户端的 base_url 指向统一 API 通道,让两个调用共用同一套认证。Planner 和 Solver 仍然各发一次 LLM 调用,控制流和原版 ReWOO 完全一致。
4.4 Worker 不调 LLM,但工具返回值要能塞进 evidence
Worker 部分不需要改 LLM 配置,它只负责执行。ToolRegistry 捕获异常,把错误变成字符串写进 evidence,这样 Solver 不会因为某个工具失败就崩掉。错误被隔离在 evidence 字典里,Planner 完全没看见。
class ToolRegistry: def __init__(self): self._tools = {} def register(self, name, fn): self._tools[name] = fn def dispatch(self, name, args): fn = self._tools.get(name) if fn is None: return f"error: unknown tool {name!r}" try: return fn(**args) except Exception as exc: return f"error: {type(exc).__name__}: {exc}"把真实工具注册进去时,搜索、RAG、数据库查询都可以。只要返回值是字符串,就能进 evidence。整个 ReWOO 跑多步规划的结构不变,变的只是 Planner 和 Solver 从脚本化变成真实模型。
5. 验证、401 与模型 ID:用 run_rewoo 的同一把 Key 跑通两次调用
5.1 保留 Scripted 版本做对照,先只换 Planner
不要一上来就把 Planner 和 Solver 都换掉。先保留 ScriptedSolver,只把 Planner 换成 LLMPlanner,跑一次 run_rewoo,看计划 JSON 能不能解析、节点依赖能不能被拓扑排序。Planner 跑通后再换 Solver。这样出错时你能立刻判断是规划格式问题,还是求解拼接问题。
如果 Planner 返回的 JSON 带 markdown 围栏,可以在解析前做一次清洗。更稳的做法是在系统提示里强调“只输出 JSON”,并把 temperature 设成 0。重试时只重试 plan_for,不要重跑整个 Worker,否则工具可能被重复调用。
5.2 打印 usage,确认 Planner 和 Solver 都真的发了请求
OpenAI 兼容客户端通常会在响应里带 usage。把 prompt_tokens、completion_tokens 打印出来,你可以确认 Planner 和 Solver 两次调用都被记录。注意玩具里的字符数和这里的 token 数不是同一个量,字符数只帮你看结构,token 数才是账单口径。
resp = client.chat.completions.create( model=PLANNER_MODEL, messages=[{"role": "user", "content": "test"}], temperature=0, ) print(resp.usage)如果 usage 为空,先检查 SDK 版本和通道返回。不要把字符统计的 ratio 当成真实节省比例,那只是教学近似。
5.3 401、模型不存在、多了 /v1 的对照表
| 现象 | 常见原因 | 处理 |
|---|---|---|
| 401 或 authentication failed | Key 没填、复制不完整、环境变量没传进当前进程 | 去 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 重新创建 YOUR_API_KEY,确认 shell 和代码读的是同一个变量 |
| model not found 或模型不存在 | 模型 ID 和模型广场当时列表不一致 | 去模型广场复制真实 ID,不要自己拼日期后缀 |
| 路径 404 或接口不匹配 | base_url 后面多了/v1 | 改成https://taotoken.net/api,末尾不要加/v1 |
| Planner JSON 解析失败 | 模型输出带了解释或 markdown | 系统提示强调只输出 JSON,temperature 设 0,解析前清洗围栏 |
排障顺序建议从 Planner 开始。因为 Planner 是第一次调用,401、模型 ID、Base URL 这三类问题会先暴露在它这里。Planner 通过后再看 Solver,如果 Solver 报错,多半是 evidence 太长或者 JSON 序列化问题。
5.4 去控制台看这次 Planner/Solver 调用
跑通一次完整的 run_rewoo 后,打开 https://taotoken.net/?utm_source=taotoken_aicg_blog_end 看用量记录。确认 Planner 和 Solver 两次调用都记在同一把 Key 下。如果只看到一次,检查是不是 Solver 的 client 被单独初始化了,或者环境变量被覆盖。
ReWOO 的多步规划在 Agent/Harness 视角下就是典型的多工具编排:Planner 发一次,Worker 跑工具,Solver 发一次。认证统一后,你改 Planner 模型或 Solver 模型都不用碰 Key,只需要在模型广场选好 ID。
6. 跑通之后:模型对话、Coding Plan 与 Claude Code 接入文档
6.1 用同一把 Key 在模型对话里发一条测试
配完这两次调用,去 TaoToken 模型对话 用同一把 Key 发一条测试消息。这样能快速确认模型 ID 和 Base URL 没填错,也能顺便看看模型广场里有哪些模型适合做 Planner、哪些适合做 Solver。
6.2 长期跑多步规划看 Coding Plan
如果你准备把 ReWOO 跑在真实任务上,Planner 和 Solver 的调用频次会上去。可以打开 Coding Plan 看套餐是否够用。别一开始就选最重的,先用小模型 Planner 加按量 Solver 跑通结构。
6.3 创建 Key 与 Claude Code 文档
Key 在 控制台 API Keys 创建和管理。如果你平时也用 Claude Code 写代码,环境变量对照可以看 接入文档。ReWOO 这边只要记住:Planner 的 plan_for 和 Solver 的 solve 共用同一个 OpenAI 兼容 client,base_url 始终是 https://taotoken.net/api,Key 从官网落地页创建。