当 AI 开始“写计划”：OpenManus 多 Agent 协作背后的编排设计

etic2009

498人浏览 · 2026-03-07 23:22:18

etic2009 · 2026-03-07 23:22:18 发布

搞 AI Agent 的都知道，单个 Agent 干活还行，一旦涉及多 Agent 协作，事情就变得复杂了：谁先谁后？怎么分工？出了问题谁负责？

我翻 OpenManus 源码的时候，发现它有个 run_flow.py 的入口，跟 main.py 不一样——这个是"先规划再执行"模式。研究了下背后的 Planning 架构，不得不说，这套设计有点东西。今天就来聊聊。

一、整体流程概览

先上个整体流程，看完心里就有数了：

整个流程分三个阶段：LLM 生成计划 → 状态驱动执行 → 执行总结。

二、入口代码分析

async def run_flow():
    agents = {
        "manus": Manus(),
    }
    if config.run_flow_config.use_data_analysis_agent:
        agents["data_analysis"] = DataAnalysis()

    flow = FlowFactory.create_flow(
        flow_type=FlowType.PLANNING,
        agents=agents,
    )

    result = await flow.execute(prompt)

看起来简单，但有几个值得注意的点：

1. agents 是字典结构

为什么用字典而不是列表？因为后面要根据步骤类型来选择对应的 Agent。比如 [DATA_ANALYSIS] 步骤可以路由到数据分析 Agent，[MANUS] 步骤路由到通用 Agent。

2. FlowFactory 工厂模式

目前只有 FlowType.PLANNING 一种类型，但设计上预留了扩展空间。以后加个 FlowType.CHAIN 或 FlowType.PARALLEL 都很容易。

三、核心模块设计（静态架构）

在深入执行流程之前，先了解三个核心模块的职责和设计。

3.1 PlanningTool：计划管理器

这是整个 Planning 架构的基石，提供了完整的计划 CRUD 操作。

支持的操作：

命令	功能	关键参数
create	创建新计划	plan_id, title, steps
update	更新计划	plan_id, title?, steps?
get	获取计划详情	plan_id
list	列出所有计划	无
mark_step	标记步骤状态	plan_id, step_index, step_status
set_active	设置活跃计划	plan_id
delete	删除计划	plan_id

核心数据结构：

plan = {
    "plan_id": "plan_1234567890",
    "title": "分析数据并生成报告",
    "steps": [
        "[DATA_ANALYSIS] 读取 CSV 文件",
        "[MANUS] 生成可视化图表",
        "[MANUS] 输出分析报告"
    ],
    "step_statuses": ["completed", "in_progress", "not_started"],
    "step_notes": ["", "处理中...", ""]
}

划重点：step_statuses 和 step_notes 与 steps 一一对应，状态变更会同步更新。

状态流转：

not_started → in_progress → completed
                  ↓
               blocked

3.2 PlanningFlow：编排执行流程

这是"大脑"，负责协调整个规划-执行过程。

核心属性：

class PlanningFlow(BaseFlow):
    llm: LLM                          # 独立的 LLM 实例
    planning_tool: PlanningTool        # 计划管理器
    executor_keys: List[str]           # 可执行 Agent 列表
    active_plan_id: str               # 当前活跃计划 ID
    current_step_index: Optional[int]  # 当前执行步骤索引

execute() 方法的骨架：

async def execute(self, input_text: str) -> str:
    # 阶段一：创建计划
    await self._create_initial_plan(input_text)

    # 阶段二：循环执行步骤
    while True:
        self.current_step_index, step_info = await self._get_current_step_info()
        if self.current_step_index is None:
            break
        executor = self.get_executor(step_type)
        await self._execute_step(executor, step_info)

    # 阶段三：执行总结
    return await self._finalize_plan()

3.3 BaseFlow：抽象基类

class BaseFlow(BaseModel, ABC):
    agents: Dict[str, BaseAgent]
    tools: Optional[List] = None
    primary_agent_key: Optional[str] = None

    @abstractmethod
    async def execute(self, input_text: str) -> str:
        """Execute the flow with given input"""

精妙之处：

agents 参数的多态处理 - 可以传单个 Agent、列表、或字典：

# 这三种写法都合法
flow = PlanningFlow(Manus())
flow = PlanningFlow([Manus(), DataAnalysis()])
flow = PlanningFlow({"manus": Manus(), "data": DataAnalysis()})

primary_agent 的自动推断 - 没指定就取第一个。
抽象方法强制子类实现 execute() - 保证了所有 Flow 都有统一的调用入口。

四、完整执行流程（动态流程）

接下来按时间顺序，详细分析三个阶段的执行过程。

4.1 阶段一：LLM 生成计划

这部分是整个架构最核心的地方：计划不是硬编码的，是 LLM 理解、决策并生成的。

4.1.1 完整调用流程

async def _create_initial_plan(self, request: str) -> None:
    # 1. 构建 system message，告诉 LLM 有哪些 Agent 可用
    system_message_content = (
        "You are a planning assistant. Create a concise, actionable plan..."
    )

    # 2. 把 Agent 信息加进去
    agents_description = []
    for key in self.executor_keys:
        agents_description.append({
            "name": key.upper(),
            "description": self.agents[key].description,
        })

    system_message_content += f"\nNow we have {agents_description} agents..."

    # 3. 调用 LLM，把 PlanningTool 作为可用工具传进去
    response = await self.llm.ask_tool(
        messages=[user_message],
        system_msgs=[system_message],
        tools=[self.planning_tool.to_param()],  # 关键：工具定义
        tool_choice=ToolChoice.AUTO,
    )

    # 4. LLM 返回工具调用，解析并执行
    if response.tool_calls:
        args = json.loads(tool_call.function.arguments)
        args["plan_id"] = self.active_plan_id
        result = await self.planning_tool.execute(**args)

4.1.2 LLM 收到的信息

System Message（部分）：

You are a planning assistant. Create a concise, actionable plan with clear steps.

Now we have [{'name': 'MANUS', 'description': 'A versatile agent...'},
             {'name': 'DATA_ANALYSIS', 'description': 'An agent for data...'}] agents.
When creating steps, please specify the agent names using the format '[agent_name]'.

User Message：

Create a reasonable plan with clear steps to accomplish the task: 帮我分析 sales.csv 并生成报告

工具定义（PlanningTool.to_param()）：

{
    "type": "function",
    "function": {
        "name": "planning",
        "description": "A planning tool that allows the agent to create and manage plans...",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {"enum": ["create", "update", "list", ...]},
                "plan_id": {"type": "string"},
                "title": {"type": "string"},
                "steps": {"type": "array", "items": {"type": "string"}},
                ...
            },
            "required": ["command"]
        }
    }
}

4.1.3 LLM 的响应

LLM 理解了意图后，会返回一个工具调用：

{
    "tool_calls": [{
        "id": "call_abc123",
        "type": "function",
        "function": {
            "name": "planning",
            "arguments": {
                "command": "create",
                "plan_id": "plan_1709876543",
                "title": "分析 sales.csv 并生成报告",
                "steps": [
                    "[DATA_ANALYSIS] 读取并分析 sales.csv 文件结构",
                    "[MANUS] 根据分析结果生成可视化图表",
                    "[MANUS] 输出完整的分析报告"
                ]
            }
        }
    }]
}

4.1.4 为什么这样设计是"智能"的？

1. 计划根据任务动态调整

同样是"分析数据"，如果用户说"帮我分析 sales.csv"，LLM 会生成一个简单的 3 步计划。如果用户说"帮我深入分析销售数据，找出异常，预测下季度趋势"，LLM 会生成更复杂的计划（取决于 LLM 的"智能"）。

2. Agent 选择是 LLM 决定的

LLM 看到 [DATA_ANALYSIS] 的描述是"数据分析师"，它就会在数据相关的步骤里标记 [DATA_ANALYSIS]。这比写死规则灵活多了。

3. 步骤粒度可控

通过调整 system_message 里的提示词，可以让 LLM 生成粗粒度或细粒度的计划。比如加上 “Break tasks into detailed sub-steps”，LLM 就会拆得更细。

4.2 阶段二：状态驱动执行

计划创建好后，进入执行阶段。核心是一个 while 循环：

while True:
    # 1. 找到下一个要执行的步骤
    self.current_step_index, step_info = await self._get_current_step_info()

    # 2. 没有步骤了，退出
    if self.current_step_index is None:
        result += await self._finalize_plan()
        break

    # 3. 执行当前步骤
    executor = self.get_executor(step_type)
    step_result = await self._execute_step(executor, step_info)

    # 4. 检查是否要终止
    if executor.state == AgentState.FINISHED:
        break

下面按执行顺序，依次分析：步骤选择 → Agent 路由 → 步骤执行。

4.2.1 步骤选择：_get_current_step_info()

核心逻辑：

async def _get_current_step_info(self) -> tuple[Optional[int], Optional[dict]]:
    # 从 PlanningTool 读取计划
    plan_data = self.planning_tool.plans[self.active_plan_id]
    steps = plan_data.get("steps", [])
    step_statuses = plan_data.get("step_statuses", [])

    # 遍历 steps，找第一个状态为 not_started 或 in_progress 的步骤
    for i, step in enumerate(steps):
        status = step_statuses[i] if i < len(step_statuses) else "not_started"

        if status in ["not_started", "in_progress"]:
            step_info = {"text": step}

            # 用正则提取 Agent 标记
            type_match = re.search(r"\[([A-Z_]+)\]", step)
            if type_match:
                step_info["type"] = type_match.group(1).lower()

            # 标记为 in_progress
            await self.planning_tool.execute(
                command="mark_step",
                plan_id=self.active_plan_id,
                step_index=i,
                step_status="in_progress",
            )

            return i, step_info

    return None, None  # 没有未完成的步骤

关键：状态驱动

这样设计的好处：不需要维护执行队列，每次循环都从"计划状态"推导下一步。

4.2.2 Agent 路由：正则提取 + get_executor()

正则提取步骤类型：

# app/flow/planning.py:245
type_match = re.search(r"\[([A-Z_]+)\]", step)
if type_match:
    step_info["type"] = type_match.group(1).lower()

正则 r"\[([A-Z_]+)\]" 的含义：

\[ 和 \] - 匹配方括号本身
([A-Z_]+) - 捕获组，匹配一个或多个大写字母或下划线
整体匹配 [MANUS]、[DATA_ANALYSIS]、[SEARCH] 等格式

提取示例：

输入: "[DATA_ANALYSIS] 读取并分析 sales.csv 文件结构"
    ↓
re.search(r"\[([A-Z_]+)\]") → 匹配成功
    ↓
type_match.group(1) → "DATA_ANALYSIS"
    ↓
.lower() → "data_analysis"
    ↓
step_info["type"] = "data_analysis"

get_executor() 路由逻辑：

def get_executor(self, step_type: Optional[str] = None) -> BaseAgent:
    # 精确匹配
    if step_type and step_type in self.agents:
        return self.agents[step_type]

    # 默认返回第一个可用 Agent
    for key in self.executor_keys:
        if key in self.agents:
            return self.agents[key]

    return self.primary_agent  # 最终兜底

完整路由链路：

步骤文本: "[DATA_ANALYSIS] 分析数据"
    ↓
re.search(r"\[([A-Z_]+)\]") 匹配 → "DATA_ANALYSIS"
    ↓
.lower() → "data_analysis"
    ↓
step_info["type"] = "data_analysis"
    ↓
get_executor("data_analysis") → agents["data_analysis"]

边界情况处理：

没有标记：type_match 为 None，走默认逻辑，返回第一个可用 Agent
标记的 Agent 不存在：精确匹配失败，同样走默认逻辑

4.2.3 步骤执行：_execute_step()

这个方法负责把计划中的每个步骤交给合适的 Agent 去执行。

源码分析：

async def _execute_step(self, executor: BaseAgent, step_info: dict) -> str:
    # 1. 获取当前计划的整体状态
    plan_status = await self._get_plan_text()

    # 2. 获取当前步骤的文本描述
    step_text = step_info.get("text", f"Step {self.current_step_index}")

    # 3. 构建发送给 Agent 的提示词
    step_prompt = f"""
    CURRENT PLAN STATUS:
    {plan_status}

    YOUR CURRENT TASK:
    You are now working on step {self.current_step_index}: "{step_text}"

    Please only execute this current step using the appropriate tools.
    When you're done, provide a summary of what you accomplished.
    """

    # 4. 调用 Agent 执行
    try:
        step_result = await executor.run(step_prompt)

        # 5. 标记步骤完成
        await self._mark_step_completed()

        return step_result
    except Exception as e:
        logger.error(f"Error executing step {self.current_step_index}: {e}")
        return f"Error executing step {self.current_step_index}: {str(e)}"

Agent 收到的提示词示例：

假设正在执行第二步"生成可视化图表"：

CURRENT PLAN STATUS:
Plan: 分析 sales.csv 并生成报告 (ID: plan_1709876543)
================================================

Progress: 1/3 steps completed (33.3%)
Status: 1 completed, 1 in progress, 0 blocked, 1 not started

Steps:
0. [✓] [DATA_ANALYSIS] 读取并分析 sales.csv 文件结构
1. [→] [MANUS] 根据分析结果生成可视化图表
2. [ ] [MANUS] 输出完整的分析报告


YOUR CURRENT TASK:
You are now working on step 1: "[MANUS] 根据分析结果生成可视化图表"

Please only execute this current step using the appropriate tools.
When you're done, provide a summary of what you accomplished.

为什么要传整个计划状态？

让 Agent 知道：

前面做了什么（已完成步骤）
现在要做什么（当前步骤）
后面还要做什么（待执行步骤）

这样 Agent 可以基于上下文做出更好的决策，而不是盲目执行。

Agent 内部执行过程：

当 Agent 收到 step_prompt 后，会进入自己的执行循环（上一篇文章已经介绍过：OpenManus 入口设计：37 行代码里的架构智慧）：

# Agent.run() 内部逻辑（简化版）
async def run(self, prompt: str) -> str:
    self.update_memory("user", prompt)

    while self.state != AgentState.FINISHED and self.current_step < self.max_steps:
        await self.think()   # 调用 LLM 思考
        if self.tool_calls:
            await self.act()  # 执行工具
        self.current_step += 1

    return self.get_final_result()

步骤完成的判定：

两种方式：

Agent 自己判断 - LLM 没有返回工具调用，而是返回总结性文字
显式调用 terminate 工具 - Agent 有一个 terminate 工具可以显式标记完成

class Terminate(BaseTool):
    name: str = "terminate"
    description: str = """Terminate the interaction when the request is met..."""
    parameters: dict = {
        "type": "object",
        "properties": {
            "status": {"type": "string", "enum": ["success", "failure"]}
        },
        "required": ["status"],
    }

状态标记的实现：

async def _mark_step_completed(self) -> None:
    try:
        # 通过 PlanningTool 更新状态
        await self.planning_tool.execute(
            command="mark_step",
            plan_id=self.active_plan_id,
            step_index=self.current_step_index,
            step_status="completed",
        )
    except Exception as e:
        # 降级处理：直接修改内存中的状态
        plan_data = self.planning_tool.plans[self.active_plan_id]
        plan_data["step_statuses"][self.current_step_index] = "completed"

完整的执行时序：

4.3 阶段三：执行总结

所有步骤执行完毕后，调用 _finalize_plan() 生成总结：

async def _finalize_plan(self) -> str:
    plan_text = await self._get_plan_text()

    system_message = Message.system_message(
        "You are a planning assistant. Your task is to summarize the completed plan."
    )

    user_message = Message.user_message(
        f"The plan has been completed. Here is the final plan status:\n\n{plan_text}\n\n"
        f"Please provide a summary of what was accomplished."
    )

    response = await self.llm.ask(
        messages=[user_message], system_msgs=[system_message]
    )

    return f"Plan completed:\n\n{response}"

五、容错设计

计划创建失败怎么办？

# 如果 LLM 没返回工具调用，创建一个默认计划
await self.planning_tool.execute(
    command="create",
    plan_id=self.active_plan_id,
    title=f"Plan for: {request[:50]}...",
    steps=["Analyze request", "Execute task", "Verify results"],
)

步骤执行失败怎么办？

try:
    step_result = await executor.run(step_prompt)
    await self._mark_step_completed()
except Exception as e:
    logger.error(f"Error executing step {self.current_step_index}: {e}")
    return f"Error executing step {self.current_step_index}: {str(e)}"

错误会被记录，但不会中断整个流程。PlanningTool 的状态可以追踪哪些步骤失败了。

六、实战示例

假设用户输入：“帮我分析 sales.csv 并生成报告”

✨ Creating initial plan with ID: plan_1709876543
📋 Plan creation result: Plan created successfully with ID: plan_1709876543

📍 Current step: 0 - [DATA_ANALYSIS] 读取并分析 sales.csv
🔧 Executing with agent: data_analysis
...
✓ Step 0 completed

📍 Current step: 1 - [MANUS] 生成可视化图表
🔧 Executing with agent: manus
...
✓ Step 1 completed

📍 Current step: 2 - [MANUS] 输出分析报告
🔧 Executing with agent: manus
...
✓ Step 2 completed

🎉 Plan completed. Generating summary...