数据截至 (上游 commit 3309bf4e416f)
04 — PlanningFlow:先出计划,再逐步派活
这章讲什么: README 里那个「不稳定的多智能体版本」(
python run_flow.py)到底怎么转。 结论先行:它是外层调度器,不是新的 agent 范式——每一步仍然是让某个ToolCallAgent跑一遍完整的 ReAct 循环。
1. 要解决的小问题
单个 ReAct 循环有个天然弱点:模型每一步只看得见「记忆里的历史」, 没有一张全局的任务地图。任务一长就容易跑偏、遗漏、或者反复做同一件事。
计划流的思路很朴素:
- 先让模型把任务拆成 3-6 个步骤,写进一张「计划表」。
- 每次只让智能体做一个步骤,并且把整张计划表贴给它看。
- 做完一步就打勾,再取下一步。
2. 全景图
run_flow.py
│ 组装 {"manus": Manus(), "data_analysis": DataAnalysis()}
▼
FlowFactory.create_flow(PLANNING, agents)
▼
┌──────────── PlanningFlow.execute ────────────┐
│ ① _create_initial_plan:让 LLM 调 planning │
│ 工具产出步骤列表 │
│ ┌──── while True ──────────────────────────┐ │
│ │ ② 取第一个未完成步骤 → 标记 in_progress │ │
│ │ ③ 按 [TAG] 选执行者 agent │ │
│ │ ④ agent.run(计划全文 + 本步任务) │ │
│ │ ⑤ 无条件标记 completed │ │
│ └──────────────────────────────────────────┘ │
│ ⑥ 没有未完成步骤了 → 让 LLM 写总结 │
└──────────────────────────────────────────────┘
主循环在 app/flow/planning.py:112-131。
3. 计划是什么:一个字典
PlanningTool 把所有计划存在一个实例字典里(app/tool/planning.py:69),
每个计划长这样(app/tool/planning.py:145-151):
plan = {
"plan_id": plan_id,
"title": title,
"steps": steps, # ["[SEARCH] 查资料", ...]
"step_statuses": ["not_started"] * len(steps), # 平行数组
"step_notes": [""] * len(steps),
}
三个平行数组,靠下标对齐。没有数据库、没有持久化——进程退出计划就没了。
3.1 七个命令
PlanningTool.execute 是个命令分发器(app/tool/planning.py:72-118):
| 命令 | 干什么 | 私有方法 |
|---|---|---|
create | 新建计划(id 冲突直接报错) | _create_plan |
update | 改标题/步骤 | _update_plan |
list | 列出所有计划及进度 | _list_plans |
get | 取某个计划的格式化文本 | _get_plan |
set_active | 切换活跃计划 | _set_active_plan |
mark_step | 改某步的状态和备注 | _mark_step |
delete | 删掉计划 | _delete_plan |
3.2 一个体贴的细节:改计划不丢进度
_update_plan 换步骤列表时,会逐位比对新旧步骤文本
(app/tool/planning.py:192-199):
for i, step in enumerate(steps):
if i < len(old_steps) and step == old_steps[i]:
new_statuses.append(old_statuses[i]) # 位置和文本都没变 → 保留状态
else:
new_statuses.append("not_started") # 变了 → 重置
妙在哪: 允许模型中途修订计划(比如在第 3 步后面插一步), 而已经做完的前几步不会被误重置。代价是判定条件严格到「同位置同文本」, 插一步会让后面全部重置。
3.3 状态可视化
_format_plan(app/tool/planning.py:322-363)输出的文本会带进度百分比和符号:
Plan: 分析销售数据 (ID: plan_1730000000)
=========================================
Progress: 1/3 steps completed (33.3%)
Status: 1 completed, 1 in progress, 0 blocked, 1 not started
Steps:
0. [✓] [CODE] 读取 CSV 并清洗
1. [→] [CODE] 按月聚合
2. [ ] 写报告
四个符号定义在两处:工具里 app/tool/planning.py:352-357,
流程里 app/flow/planning.py:35-42 的 PlanStepStatus.get_status_marks——
同一套符号写了两遍,是重复代码。
4. 计划怎么生成
_create_initial_plan(app/flow/planning.py:136-211)做三件事:
① 把可选的智能体写进系统提示。 当执行者多于一个时(:154-160):
system_message_content += (
f"\nNow we have {agents_description} agents. "
f"The infomation of them are below: {json.dumps(agents_description)}\n"
"When creating steps in the planning tool, please specify the agent names using the format '[agent_name]'."
)
于是模型被要求在步骤文本里写 [DATA_ANALYSIS] 这样的标签。
② 只给模型一个工具。 tools=[self.planning_tool.to_param()](:174)——
这一轮它只能调 planning,不会跑偏去干活。
③ 兜底默认计划。 如果模型没调工具,就硬造一个三步计划
(:204-211):["Analyze request", "Execute task", "Verify results"]。
5. 怎么挑执行者
步骤类型是从文本里正则抠出来的(app/flow/planning.py:242-247):
type_match = re.search(r"\[([A-Z_]+)\]", step)
if type_match:
step_info["type"] = type_match.group(1).lower()
然后 get_executor 拿这个小写字符串去 self.agents 字典里查
(app/flow/planning.py:77-92):
步骤文本 "[DATA_ANALYSIS] 画趋势图"
│ 正则抠出 DATA_ANALYSIS
▼ 转小写
data_analysis ──在 agents 字典里?──▶ 是 → 用它
└▶ 否 → 用 executor_keys 里第一个
└▶ 再不行 → primary_agent
路由完全靠字符串巧合。 模型写 [DATA ANALYSIS](空格)或 [ANALYSIS]
就匹配不上,静默退回默认智能体。
6. 一步怎么执行
_execute_step(app/flow/planning.py:277-304)拼的提示词是这样:
CURRENT PLAN STATUS:
<整张计划表的格式化文本>
YOUR CURRENT TASK:
You are now working on step 2: "[CODE] 按月聚合"
Please only execute this current step using the appropriate tools. ...
然后 await executor.run(step_prompt)——注意这是一次完整的 ReAct 循环,
里面可能跑十几步、调很多次工具。计划流只负责喂题目和收答案。
7. 已知的三个坑
这一节是本章最值得记住的部分。README 自己也说这是「unstable multi-agent version」。
7.1 步骤总是被标成 completed
_execute_step 的写法是(app/flow/planning.py:295-304):
try:
step_result = await executor.run(step_prompt)
await self._mark_step_completed() # ← 只要 run() 没抛异常就打勾
return step_result
except Exception as e:
return f"Error executing step ..."
智能体跑满 20 步没做成、或者工具一路报错但没抛出去——只要 run() 正常返回,
这一 步就被标成完成。计划表上的 ✓ 不代表任务真的做成了。
blocked 这个状态在整个流程里没有任何代码会写入。
7.2 「智能体想提前收工」的检查失效
app/flow/planning.py:126-129:
if hasattr(executor, "state") and executor.state == AgentState.FINISHED:
break
但 01 章 讲过,state_context 的 finally 会在 run() 返回前
把状态还原成进入时的 IDLE(app/agent/base.py:81-82)。
所以这个分支在当前代码路径下不会命中,循环只能靠「没有未完成步骤」退出。
7.3 步数预算跨步骤累加
current_step 只在撞上 max_steps 时才归零(app/agent/base.py:149-151)。
PlanningFlow 反复对同一个 Manus 实例调 run(),所以:
第 1 步:current_step 0 → 8 (可用 20 步)
第 2 步:current_step 8 → 15 (只剩 12 步)
第 3 步:current_step 15 → 20 (撞上限,归零并追加 "Terminated: Reached max steps")
第 4 步:current_step 0 → … (又满血)
越靠后的步骤越容易「饿死」,直到某次撞上限触发重置。
8. FlowFactory:目前只有一种流
FlowType 枚举只有 PLANNING 一个值(app/flow/flow_factory.py:9-10),
工厂里的字典也只有一项(:22-24)。工厂模式在这里是为未来预留的骨架。
BaseFlow.__init__(app/flow/base.py:19-40)倒是有个实用的小设计:
传单个 agent、agent 列表、或 agent 字典都能接受,内部统一归一成字典,
并把第一个当 primary_agent。
9. 代码地图
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 计划流入口 | run_flow.py | run_flow |
| 流基类 / agent 归一化 | app/flow/base.py | BaseFlow、BaseFlow.primary_agent |
| 流工厂 | app/flow/flow_factory.py | FlowFactory、FlowType |
| 主调度循环 | app/flow/planning.py | PlanningFlow.execute |
| 生成初始计划 | app/flow/planning.py | PlanningFlow._create_initial_plan |
取当前步骤 + 抠 [TAG] | app/flow/planning.py | PlanningFlow._get_current_step_info |
| 选执行者 | app/flow/planning.py | PlanningFlow.get_executor |
| 执行一步 | app/flow/planning.py | PlanningFlow._execute_step |
| 收尾总结 | app/flow/planning.py | PlanningFlow._finalize_plan |
| 步骤状态枚举 | app/flow/planning.py | PlanStepStatus |
| 计划工具(七命令) | app/tool/planning.py | PlanningTool、PlanningTool.execute |
| 改计划保留进度 | app/tool/planning.py | PlanningTool._update_plan |
| 计划渲染 | app/tool/planning.py | PlanningTool._format_plan |
| 计划提示词 | app/prompt/planning.py | PLANNING_SYSTEM_PROMPT |
| runflow 开关 | app/config.py | RunflowSettings |