跳到主要内容

计划流：把一个大任务拆成步骤、逐步派给 agent 执行

这章讲什么： 单个 agent 靠 ReAct 循环能干不少事，但面对"要好多阶段"的复杂任务，直接扔进循环容易跑偏。PlanningFlow 加一层显式的计划：先让 LLM 列出步骤清单，再一步步派 agent 去做、逐步打勾。这是 OpenManus 的"多 agent 编排"入口（run_flow.py）。

1. 它要解决的小问题​

ReAct 循环是"走一步看一步"，没有全局路线图。对"先调研、再写代码、最后画图"这种多阶段任务，缺乏计划会导致：忘了某个阶段、反复横跳、无法把不同步骤派给不同专长的 agent。

PlanningFlow 的思路很直接：

先规划（LLM 生成步骤清单）→ 再执行（逐步派 agent 做）→ 最后汇总

2. 两个角色：Flow 与 Tool​

角色是什么职责
PlanningTool一个工具存储计划：steps 列表 + 每步状态；提供 create/update/mark_step/get 等命令
PlanningFlow一个流程调度：调 LLM 生成计划、循环取步、派 agent、标记进度、收尾汇总
  • PlanningTool（app/tool/planning.py:14）把计划存在内存字典 plans[plan_id] 里，每个计划含 steps / step_statuses / step_notes 三条平行列表。状态有四种：not_started / in_progress / completed / blocked。
  • PlanningFlow（app/flow/planning.py:45）继承 BaseFlow（app/flow/base.py:9），持有一组 agent（agents 字典）和一个 PlanningTool。

3. 主流程：execute​

PlanningFlow.execute（app/flow/planning.py:94）是这层的主循环：

execute(input_text):
  1. _create_initial_plan(input_text)      # 让 LLM 生成计划
     若计划没建成 → 返回失败
  2. while True:
        current_step_index, step_info = _get_current_step_info()  # 找第一个未完成步
        若没有未完成步 → _finalize_plan() 汇总 → break
        executor = get_executor(step_type)   # 选一个 agent 来干这步
        step_result = _execute_step(executor, step_info)  # 派它执行
        若该 agent 状态变成 FINISHED → break
  3. 返回累计结果

下面拆开三个关键环节。

3.1 生成计划：_create_initial_plan​

_create_initial_plan（app/flow/planning.py:136）让 LLM 扮演"规划助手"，并强制它调用 PlanningTool（tool_choice=AUTO，工具只给了 planning 这一个）来产出结构化步骤：

  • 若有多个 agent 可选，会把每个 agent 的名字+描述塞进系统提示，并要求 LLM 在步骤里用 [agent_name] 格式标注该步该谁做（:154）。
  • 若 LLM 没正确调工具，兜底创建一个默认三步计划："Analyze request / Execute task / Verify results"（:204）。

3.2 找下一步：_get_current_step_info​

_get_current_step_info（app/flow/planning.py:213）遍历 steps，找第一个状态是 not_started 或 in_progress 的步：

  • 找到后立刻把它标成 in_progress。
  • 顺手用正则 \[([A-Z_]+)\] 从步骤文本里抠出"类型标签"（如 [SEARCH]、[CODE]），存进 step_info["type"]（:245）——这个类型决定派哪个 agent。
  • 没有未完成步 → 返回 (None, None)，主循环据此收尾。

3.3 选执行者：get_executor​

get_executor（app/flow/planning.py:77）根据步骤类型选 agent：

get_executor(step_type):
  若 step_type 恰好是某个 agent 的 key → 用那个 agent
  否则 → 用第一个可用 executor，再不行退回 primary_agent

这就是"把不同步骤派给不同专长 agent"的落点。例如 run_flow.py 里注册了 manus 和（可选）data_analysis 两个 agent，标了 [data_analysis] 的步就会派给数据分析 agent。

3.4 执行一步：_execute_step​

_execute_step（app/flow/planning.py:277）把"当前计划全貌 + 当前这一步是什么"拼成一段提示词，然后直接调 executor.run(step_prompt)——也就是说，每一步的执行，本身就是跑一遍完整的 ReAct 循环（§01/§02 那套）。执行成功后把该步标成 completed（_mark_step_completed，:306）。

关键理解：计划流是套在单 agent 循环外面的一层。计划的每一步 = 一次完整的 agent.run。两层循环嵌套：外层走计划步骤，内层走 ReAct 想-做。

3.5 收尾：_finalize_plan​

所有步骤完成后，_finalize_plan（app/flow/planning.py:406）用 LLM 直接（llm.ask，非工具模式）对最终计划状态做一段总结汇报；若失败则退回让 agent 生成总结。

4. 计划怎么显示：状态标记​

PlanningTool._format_plan（app/tool/planning.py:322）和 PlanStepStatus.get_status_marks（app/flow/planning.py:35）把状态渲染成醒目的符号，让人和 LLM 都能一眼看懂进度：

状态标记
completed[✓]
in_progress[→]
blocked[!]
not_started[ ]

渲染出来大致长这样（供 LLM 每步执行前参考）：

Plan: 分析销售数据 (ID: plan_1710000000)
Progress: 1/3 steps completed (33.3%)
Steps:
0. [✓] 读取 CSV
1. [→] [DATA_ANALYSIS] 画趋势图
2. [ ] 写总结

5. 巧妙之处 / 边界​

  • 计划=外循环，agent=内循环：编排逻辑（PlanningFlow）和执行逻辑（agent.run）彻底解耦——换 agent、换工具都不影响调度层。
  • 容错优先：标记步骤状态失败时，_get_current_step_info / _mark_step_completed 都有"直接改内存里 step_statuses"的兜底路径（:257、:324），不让一次工具调用失败卡死整个流程。
  • 局限：
    • 计划一次性生成，执行中不会自动重规划（步骤做砸了不会自动插新步）。
    • 只有 FlowType.PLANNING 一种流（app/flow/flow_factory.py:9）——"多种 flow"的抽象搭好了但目前只实现了这一种。
    • 步骤派发靠正则抠 [TYPE] 标签，依赖 LLM 老实按格式写。

6. 代码地图​

主题文件路径符号名
流程主循环app/flow/planning.pyPlanningFlow.execute
生成计划app/flow/planning.pyPlanningFlow._create_initial_plan
找下一步app/flow/planning.pyPlanningFlow._get_current_step_info
选执行者app/flow/planning.pyPlanningFlow.get_executor
执行一步app/flow/planning.pyPlanningFlow._execute_step
收尾汇总app/flow/planning.pyPlanningFlow._finalize_plan
步骤状态枚举app/flow/planning.pyPlanStepStatus
计划存储工具app/tool/planning.pyPlanningTool.execute, PlanningTool._format_plan
流程基类app/flow/base.pyBaseFlow, BaseFlow.primary_agent
流程工厂app/flow/flow_factory.pyFlowFactory.create_flow, FlowType
入口run_flow.pyrun_flow