跳到主要内容

数据截至 (上游 commit ae57a2357745)

第 3 章 · 规划 agent：把大任务拆开做

本章讲：复杂度被判 HIGH 之后接手的那个 agent——它怎么让弱模型稳定吐出计划、怎么在子 agent 之间传数据、怎么在中途改主意。

3.1 它要解决的小问题​

有些请求单个 agent 干不了：

「上网找几篇 CRISPR 最新论文，然后做个网页应用把它们列出来。」

这至少要三步：查网 → 建项目目录 → 写代码。而且第三步要用到第一步的产出。

PlannerAgent 就是接这种活的。它自己不写代码、不上网，只做两件事：拆和调度。

3.2 它手下有谁​

构造时它把四个子 agent 各自新建一份（sources/agents/planner_agent.py:25-30）：

self.agents = {
    "coder":  CoderAgent(name, "prompts/base/coder_agent.txt", provider, verbose=False),
    "file":   FileAgent(name, "prompts/base/file_agent.txt", provider, verbose=False),
    "web":    BrowserAgent(name, "prompts/base/browser_agent.txt", provider, verbose=False, browser=browser),
    "casual": CasualAgent(name, "prompts/base/casual_agent.txt", provider, verbose=False)
}

两点值得注意：

  • 这些是独立实例，和路由层顶层的那些 agent 不是同一批。所以 planner 的子 agent 有自己独立的 memory——顶层 coder 的对话历史，planner 里的 coder 看不见。
  • 浏览器是共用的：browser 对象从外面传进来（api.py:128-137 把同一个 Browser 实例同时给了顶层 BrowserAgent 和 PlannerAgent），毕竟一台 Chrome 就够了。
  • planner 自己没有 casual 之外的对外角色：它的 role = "planification"，type = "planner_agent"。

3.3 计划长什么样​

prompts/base/planner_agent.txt 要求模型输出**「人话小标题 + ```json 计划」**两部分。样例（来自该 prompt 文件）：

## Task 1: I will search for available weather api with the help of the web agent.

## Task 2: I will create an api key for the weather api using the web agent

```json
{
  "plan": [
    { "agent": "Web",  "id": "1", "need": [],          "task": "Search for reliable weather APIs" },
    { "agent": "Web",  "id": "2", "need": ["1"],       "task": "Obtain API key from the selected service" },
    { "agent": "File", "id": "3", "need": [],          "task": "Create and setup a web app folder…" },
    { "agent": "Coder","id": "4", "need": ["2", "3"], "task": "Develop a Python application using the API…" }
  ]
}
```

字段含义：

字段意思
agent派给谁（Coder / File / Web / Casual）
id这一步的编号，供别人 need 引用
need依赖哪几步的产出（id 列表）
task给那个 agent 的具体指令

为什么要「人话小标题 + JSON」两份？ 小标题喂给用户看和朗读（“I will search for available weather api…”），JSON 喂给机器执行。get_task_names() 专门去捞那些以 ## 开头或以数字开头的行（sources/agents/planner_agent.py:39-61）。

如果两边数量对不上，代码会降级用 JSON 里的 task 字段当标题（sources/agents/planner_agent.py:105-108）：

if len(tasks_names) != len(tasks):
    names = [task['task'] for task in tasks]
    return list(map(list, zip(names, tasks)))
return list(map(list, zip(tasks_names, tasks)))

这是对弱模型的一处务实妥协——它经常忘了写小标题，或者写的条数对不上。

3.4 让弱模型吐出合法 JSON：重试到成功​

make_plan() 是个带上限的死磕循环（sources/agents/planner_agent.py:150-182）：

  while not ok:
      retries >= 4 ? → 放弃，返回 []
      memory.push('user', prompt)
      answer = llm_request()
      答案含 "NO_UPDATE" → 返回 []          （重规划时用，见 3.6）
      agents_tasks = parse_agent_tasks(answer)
      解析空？ → 换一段更凶的 prompt，retries += 1，继续
      解析出来了 → 展示计划，ok = True

重试用的那句 prompt 很值得抄（sources/agents/planner_agent.py:175）：

“Failed to parse the tasks. Please write down your task followed by a json plan within ```json. Do not ask for clarification.”

最后那句是关键——小模型解析失败后特别爱反问「你想要什么样的应用？」，反问就永远出不来计划。

解析时的三重校验​

parse_agent_tasks() 对每个 JSON 块做（sources/agents/planner_agent.py:63-108）：

校验不过关就
json.loads 能解析记 warning，返回 [] 触发重试
task['agent'] 在四个已知 agent 里（忽略大小写）打印 “Agent X does not exist.”，返回 []
agent / id / task 三个字段齐全捕获 KeyError，返回 []

注意任何一步不过关都是整块作废，不做部分修补。宁可重来，也不要执行半个残缺的计划。

3.5 依赖是怎么传的​

没有共享黑板、没有向量库——就是一个 id → 上一步的答案文本 的字典。

  agents_work_result = {}          # 全局累积

  执行第 i 步时：
      required_infos = { k: agents_work_result[k] for k in task['need'] }
           ↓
      make_prompt(task['task'], required_infos)
           ↓
      "You are given informations from your AI friends work:
              - According to agent 1:
              <第 1 步的完整答案>
        Your task is:
              <本步任务>"
           ↓
      子 agent.process(prompt)
           ↓
      agents_work_result[task['id']] = 本步答案

对应 get_work_result_agent（sources/agents/planner_agent.py:257-260）和 make_prompt（110-132）。

存进去的不是纯文字，而是「答案 + 代码块执行结果」的拼接（sources/agents/planner_agent.py:249）：

agent_answer = self.agents[task['agent'].lower()].raw_answer_blocks(answer)
...
agent_answer += "\nAgent succeeded with task." if success else "\nAgent failed with task (Error detected)."

raw_answer_blocks 会把 block:N 占位符换回真实的代码与执行反馈（sources/agents/agent.py:193-208），所以下游 agent 看到的是上一步实际跑出了什么，不只是它嘴上说了什么。末尾还追加一句明确的成败标记，给下游一个直白的信号。

3.6 每步之后重新规划​

这是 planner 最有意思的部分。主循环长这样（sources/agents/planner_agent.py:262-303）：

  agents_tasks = make_plan(goal)
  i = 0; steps = len(agents_tasks)

  while i < steps and not stop:
      取第 i 步
      收集依赖 required_infos
      answer, success = start_agent_process(task, required_infos)
      agents_work_result[task['id']] = answer
      ┌──────────────────────────────────────────┐
      │ agents_tasks = update_plan(...)          │  ← 每一步都问
      │ steps = len(agents_tasks)                │  ← 计划长度可变
      └──────────────────────────────────────────┘
      i += 1

update_plan() 组一段专门的 prompt 再走一次 make_plan（sources/agents/planner_agent.py:208-225），核心几句：

  • “Agent {id} work was a {success/failure} according to system interpreter.”
  • “If agent work was good: answer NO_UPDATE”
  • “You need to rewrite the whole plan, but only change the tasks after task {id}.”
  • “Make the plan the same length as the original one or with only one additional step.”

三条约束各有用意：

约束防的是什么
顺利就回 NO_UPDATE省一次完整计划的生成，也避免无谓改动
只改 id 之后的任务防止模型把已经做完的步骤改掉
长度不变或只加一步防止计划无限膨胀，每次失败加一堆补救步骤

NO_UPDATE 的返回路径是：make_plan 里检测到该字符串就返回 []（sources/agents/planner_agent.py:170-171），update_plan 见到 [] 就原样保留旧计划（226-228）。

一个被注释掉的分支，透露了设计的演进​

#if success:
#    return agents_tasks # we only update the plan if last task failed, for now

（sources/agents/planner_agent.py:206-207）

曾经的策略是「只在失败时重规划」，现在改成了每步都问。代价是每一步多一次 LLM 调用，收益是成功但跑偏的步骤也能被纠正。

3.7 原理演示：整个 planner 的骨架​

# 示意，非源码
async def run(goal):
    plan = await make_plan(goal)              # 让 LLM 吐 JSON，最多重试 4 次
    results, i = {}, 0
    while i < len(plan):
        name, task = plan[i]
        needed = {k: results[k] for k in task.get("need", [])}   # 只取声明依赖的那几步
        answer, ok = await run_sub_agent(task["agent"], task["task"], needed)
        results[task["id"]] = answer
        plan = await update_plan(goal, plan, results, task["id"], ok)  # 每步都重问
        i += 1
    return answer

重点看两处：依赖只取 need 里声明的那几项（不是把所有历史一股脑塞进去，省上下文），以及循环边界每轮重算（计划长度可能变）。

3.8 关键细节与坑​

坑 1：只返回最后一步的答案​

process() 的返回值是循环里最后一次赋值的 answer（sources/agents/planner_agent.py:303），也就是最后一个子 agent 说了什么。没有全局汇总。这就是为什么 planner 的 prompt 里带了个 Casual agent，并在示例计划的最后一步用它「resume what has been done and conclude」——总结这件事是靠 prompt 约定，而不是靠代码保证的。

坑 2：need 是软约束​

need 只影响「给下游看什么信息」，不影响执行顺序。任务永远按数组下标从 0 顺跑。如果模型写出 id=1 依赖 id=3，代码不会报错，只会在 get_work_result_agent 里因为 if k in agents_work_result 静默漏掉那一项（sources/agents/planner_agent.py:258）。

坑 3：官方示例本身 id 就是重复的​

看 prompts/base/planner_agent.txt 里那份样例计划——最后一步 Casual 的 "id": "3"，和前面的 File 步骤撞号。因为 agents_work_result 用 id 当键，撞号会互相覆盖。这是 prompt 里的既有瑕疵，模型有很大概率照抄。

坑 4：planner 的记忆只增不减​

PlannerAgent 的 Memory 关了压缩（memory_compression=False，sources/agents/planner_agent.py:33-36）。每一步的 update_plan 都往里 push 一段包含上一步完整产出的长 prompt。步数一多，上下文会迅速胀大——这也是为什么 README 的硬件表说 7B「planner agents will likely fail」。

坑 5：会话恢复跳过 planner​

Interaction.load_last_session() 明确跳过 planner 类型（sources/interaction.py:184-189）：

for agent in self.agents:
    if agent.type == "planner_agent":
        continue
    agent.memory.load_memory(agent.type)

因为半截的计划状态恢复出来没有意义。但 save_session() 没有做同样的过滤（191-194），planner 的记忆照存不误。

3.9 本章代码地图​

主题文件符号
主循环sources/agents/planner_agent.pyPlannerAgent.process
生成计划 + 重试sources/agents/planner_agent.pymake_plan
JSON 解析与校验sources/agents/planner_agent.pyparse_agent_tasks、get_task_names
重规划sources/agents/planner_agent.pyupdate_plan
依赖注入sources/agents/planner_agent.pymake_prompt、get_work_result_agent
调度子 agentsources/agents/planner_agent.pystart_agent_process
计划展示sources/agents/planner_agent.pyshow_plan
把块结果拼回文本sources/agents/agent.pyAgent.raw_answer_blocks
系统提示词prompts/base/planner_agent.txt—
解析测试tests/test_planner_agent_parsing.py—