企业官网建设流程全解析

OpenManus BaseFlow 深度解析：用 Flow 编排多步骤多智能体任务（PlanningFlow 实战）

【免费下载链接】Tutorial-Codebase-KnowledgePocket Flow: Codebase to Tutorial项目地址: https://gitcode.com/gh_mirrors/tu/Tutorial-Codebase-Knowledge

在 OpenManus 中，BaseAgent 解决的是"单个智能体如何思考与行动"，Tool / ToolCollection 解决的是"智能体如何获得具体技能"。而本文要讲的Flow（流程）则位于更高一层：当一个任务需要分阶段、跨技能、甚至跨多个智能体协同完成时，Flow 就是那个统筹全局的"项目经理"。读完本文，你将掌握 OpenManus 中BaseFlow的设计思想、FlowFactory的创建方式，以及PlanningFlow从"制定计划"到"逐条执行"再到"汇总结果"的完整运行链路，能够直接用代码搭建自己的多步骤任务编排流程。

单步任务的局限：为什么需要 Flow？

先看一个单步任务：给一个配备了WebSearch工具的智能体提问"法国的首都是哪里？"——它调用工具、拿到答案、返回"巴黎"，一次交互即可完成。

但如果是这样一个任务："调研电动车的优缺点，然后写一篇简短的博客文章总结它们"？它显然不是一次行动，而是一连串相互依赖的步骤：

1. 规划：拆解步骤（搜索优点 → 搜索缺点 → 搭建文章结构 → 撰写草稿 → 审阅草稿）。
2. 执行第 1 步：用搜索工具找优点。
3. 执行第 2 步：用搜索工具找缺点。
4. 执行第 3 步：调用 LLM 的"大脑"列出博客大纲。
5. 执行第 4 步：基于调研结果和大纲撰写全文。
6. 执行第 5 步：最终审阅。

一个非常复杂的 BaseAgent 理论上或许能独自完成这一切，但当流程变长、步骤变多时，由专门的编排器（orchestrator）来统管整个过程会清晰得多、也更容易维护。这正是 Flow 的职责：BaseFlow是这些编排器的蓝图，它定义了一种能够管理多个智能体、并按照特定策略（例如遵循预先制定的计划）协调它们去完成更大目标的结构。

一个典型的用例：还是"先研究后写作"的任务，我们需要一个东西来管理整体进程。PlanningFlow（基于BaseFlow构建的一种具体 Flow）正合适——它先创建计划（如上面列出的步骤），然后逐步执行，必要时还能把不同的步骤分派给不同的专业智能体。

核心概念：Flow、Agents 与执行策略

OpenManus 的 Flow 体系由三个层次的概念组成：

1. BaseFlow（app/flow/base.py）：所有 Flow 的抽象蓝图。可以把它理解为一份"项目经理岗位说明书"——它规定了一名经理需要知道自己的团队（agents）、需要有一个运行项目的方式（execute方法），但不规定经理具体"怎么管"。它最主要的职责是持有一个agents字典，存放可供 Flow 调用的智能体。你通常不会直接使用BaseFlow，而是使用它的具体实现。

2. 具体 Flow（如PlanningFlow，位于app/flow/planning.py）：这些是具体的项目管理策略，它们继承自BaseFlow。PlanningFlow的策略是：

   • 接收总目标；
   • 使用 LLM 与一个专门的PlanningTool把目标拆解为一串步骤（即"计划"）；
   • 逐条执行计划中的步骤，通常通过调用某个合适的 BaseAgent 的run()方法来完成；
   • 跟踪每个步骤的状态（未开始、进行中、已完成）。

3. Flow 内的 Agents：被 Flow 管理的"工人"或"专家"。一个 Flow 持有一个或多个 BaseAgent 实例。在PlanningFlow中，可能指定一个主智能体（primary agent，常负责协助生成计划），其余（或同一个）智能体则作为计划步骤的"执行者"。Flow 负责决定每一步该交给哪个智能体。

用盖房子来类比最直观：

• BaseFlow是"总承包商"这个概念本身；
• PlanningFlow是某种具体的总承包商——他总是先绘制详细的建筑图纸（计划），再为每个阶段聘请专家；
• agents是各工种专家：水管工、电工、木匠……；
• 总目标（"盖一栋房子"）交给PlanningFlow（承包商）；
• PlanningFlow制定计划（地基、框架、水电……），然后针对计划的每一步调用合适的agent（专家）。

实战：用 FlowFactory 创建 PlanningFlow

在 OpenManus 中，创建 Flow 的标准方式是使用FlowFactory，并把 Flow 所需的智能体交给它。下面的示例创建一个带有一个名为 "Manus"（OpenManus 的通用型智能体）的PlanningFlow：

# 导入必要的类 from app.agent.manus import Manus # 一个能力全面的智能体 from app.flow.flow_factory import FlowFactory, FlowType import asyncio # 异步执行所需 # 1. 创建我们想让 Flow 管理的智能体 # 可以给智能体在 Flow 内分配特定的键（名称） agents_for_flow = { "research_writer": Manus() # 用 Manus 智能体处理所有任务 } # 2. 使用工厂创建 Flow # 指定类型（PLANNING）并提供智能体 planning_flow_instance = FlowFactory.create_flow( flow_type=FlowType.PLANNING, agents=agents_for_flow, # 可选：指定哪个智能体是主智能体（如果不使用第一个） # primary_agent_key="research_writer" ) print(f"Created a {type(planning_flow_instance).__name__}") print(f"Primary agent: {planning_flow_instance.primary_agent.name}") # 3. 定义 Flow 的总体目标 overall_goal = "Research the main benefits of solar power and write a short summary." # 定义运行 Flow 的异步函数 async def run_the_flow(): print(f"\nExecuting flow with goal: '{overall_goal}'") # 4. 用目标执行 Flow final_result = await planning_flow_instance.execute(overall_goal) print("\n--- Flow Execution Finished ---") print(f"Final Result:\n{final_result}") # 运行异步函数 # asyncio.run(run_the_flow()) # 取消注释即可运行

逐步拆解这段代码：

1. 导入要用的智能体（Manus）以及FlowFactory、FlowType；
2. 创建字典agents_for_flow，把键"research_writer"映射到一个Manus实例——这告诉 Flow 有哪些"工人"可用；
3. 调用FlowFactory.create_flow()，指定FlowType.PLANNING并传入agents_for_flow，工厂负责正确构造PlanningFlow对象；
4. 定义高层任务overall_goal；
5. await planning_flow_instance.execute(overall_goal)——这是魔法发生的地方，PlanningFlow接管后续一切。

预期输出（高层视角）：运行后你不会立刻得到答案，而是会看到类似这样的过程：

• 正在创建计划（例如：Step 1: 搜索优点，Step 2: 综合发现，Step 3: 撰写摘要）；
• "research_writer"智能体开始执行 Step 1，可能伴随它调用网络搜索工具的输出；
• 智能体继续 Step 2、Step 3，期间可能展示 LLM 的思考或写作输出；
• 最终，execute调用返回一个字符串，其中包含各步骤的结果以及（可能的）最终总结。

整个多步骤过程由PlanningFlow基于初始目标自动管理——这正是它区别于单步 Agent 的核心价值。

关于 FlowFactory 与 FlowType

FlowFactory是一个静态工厂：它接收FlowType枚举值与agents字典，通过内部映射找到对应的 Flow 类并实例化。从教程文档所描述的 OpenManus 教程 可以看出，FlowType目前至少包含PLANNING = "planning"一项，且该映射是可扩展的——新增 Flow 类型时只需在flows映射表中登记新类即可。如果传入未注册的flow_type，工厂会抛出ValueError(f"Unknown flow type: {flow_type}")。

源码走查：PlanningFlow.execute 的执行链路

调用execute(input_text)后到底发生了什么？以下是高层完整走查：

1. 接收目标：execute方法拿到input_text（即总体目标）。
2. 创建计划（_create_initial_plan）：
   • 为 LLM 构造消息，包括要求它扮演"规划师"角色的系统消息；
   • 告诉 LLM 存在PlanningTool（一个专门用于创建和管理计划的 Tool）；
   • 调用 LLM 的ask_tool方法，实质上是在问："请使用 PlanningTool 为这个目标创建计划：{input_text}"；
   • PlanningTool（在被 LLM 调用时）把生成的步骤（如["Search benefits", "Write summary"]）与一个唯一的plan_id关联并存储起来。
3. 执行循环：Flow 进入循环逐条执行计划步骤：
   • 取下一步（_get_current_step_info）：用PlanningTool检查已存计划，找到第一个未标记为"completed"的步骤，取得其文本与索引；
   • 检查是否完成：若找不到未完成步骤，说明计划已执行完毕，跳出循环；
   • 选择执行者（get_executor）：决定当前步骤由哪个智能体执行。在简单示例里它总是选择"research_writer"；更复杂的 Flow 可以根据步骤类型选择（例如"[CODE]"步骤交给编码智能体）；
   • 执行步骤（_execute_step）：为选中的执行者智能体准备提示词（包含当前计划状态与当前步骤的具体指令，如"You are working on step 0: 'Search benefits'. Please execute this step."），然后调用await executor.run(step_prompt)——智能体随之展开工作（可能使用它自己的工具、记忆与 LLM）；拿到run()返回的结果；
   • 标记完成（_mark_step_completed）：让PlanningTool把当前步骤状态更新为 "completed"；
   • 循环：回到"取下一步"，继续下一轮。
4. 收尾（_finalize_plan）：循环结束后，可能生成已完成计划的最终总结（可能再次借助 LLM）。
5. 返回结果：把所有步骤累计得到的结果作为字符串返回。

时序图

核心代码片段解析

以下代码片段是教程文档基于 OpenManus 源码给出的简化版本，用于说明三个文件各自承担的角色。

app/flow/base.py：蓝图的骨架，只持有 agents

# 简化片段（源自 app/flow/base.py） from abc import ABC, abstractmethod from typing import Dict, List, Optional, Union from pydantic import BaseModel from app.agent.base import BaseAgent class BaseFlow(BaseModel, ABC): """Base class for execution flows supporting multiple agents""" agents: Dict[str, BaseAgent] # 持有所有可用智能体 primary_agent_key: Optional[str] = None # 主智能体的键 # ... __init__ 负责构建 agents 字典 ... @property def primary_agent(self) -> Optional[BaseAgent]: """获取 Flow 的主智能体""" return self.agents.get(self.primary_agent_key) @abstractmethod # 子类必须实现 execute async def execute(self, input_text: str) -> str: """以给定输入执行 Flow""" pass

关键点：BaseFlow同时继承 Pydantic 的BaseModel与 Python 的ABC，因此字段具备数据校验能力，而execute被声明为抽象方法——子类必须给出自己的编排逻辑。primary_agent是一个 property，按primary_agent_key从agents字典中取出主智能体；若未指定primary_agent_key，通常会默认取字典的第一个键（这也是上面示例代码中注释"if not first"的原因）。

app/flow/flow_factory.py：工厂，创建具体 Flow

# 简化片段（源自 app/flow/flow_factory.py） from enum import Enum from app.agent.base import BaseAgent from app.flow.base import BaseFlow from app.flow.planning import PlanningFlow # 导入具体 Flow class FlowType(str, Enum): PLANNING = "planning" # 在此扩展其他 Flow 类型 class FlowFactory: @staticmethod def create_flow(flow_type: FlowType, agents, **kwargs) -> BaseFlow: flows = { # 类型枚举到具体类的映射 FlowType.PLANNING: PlanningFlow, } flow_class = flows.get(flow_type) if not flow_class: raise ValueError(f"Unknown flow type: {flow_type}") # 创建 PlanningFlow(agents, **kwargs) 实例 return flow_class(agents, **kwargs)

关键点：FlowType是继承自str的枚举，因此FlowType.PLANNING本身就可以直接当作字符串"planning"使用。工厂的核心是"枚举 → 类"的映射字典，kwargs会把primary_agent_key等额外配置透传给具体 Flow 的构造函数。

app/flow/planning.py：计划与执行的核心逻辑

# 简化片段（源自 app/flow/planning.py） from app.flow.base import BaseFlow from app.tool import PlanningTool from app.agent.base import BaseAgent from app.schema import Message class PlanningFlow(BaseFlow): planning_tool: PlanningTool = Field(default_factory=PlanningTool) # ... 其他字段，如 llm、active_plan_id ... async def execute(self, input_text: str) -> str: """使用智能体执行计划流程。""" # 1. 提供输入时创建计划 if input_text: await self._create_initial_plan(input_text) result_accumulator = "" while True: # 2. 获取要执行的下一步 step_index, step_info = await self._get_current_step_info() # 3. 没有更多步骤则退出 if step_index is None: result_accumulator += await self._finalize_plan() break # 4. 获取执行该步骤的智能体 executor_agent = self.get_executor(step_info.get("type")) # 5. 让智能体执行步骤 step_result = await self._execute_step(executor_agent, step_info) result_accumulator += step_result + "\n" return result_accumulator async def _create_initial_plan(self, request: str): """使用 LLM 和 PlanningTool 创建计划。""" logger.info(f"Creating plan for: {request}") system_msg = Message.system_message("You are a planner...") user_msg = Message.user_message(f"Create a plan for: {request}") # 请 LLM 使用规划工具 response = await self.llm.ask_tool( messages=[user_msg], system_msgs=[system_msg], tools=[self.planning_tool.to_param()], # 提供工具规格 # tool_choice=ToolChoice.AUTO # 或指定规划工具名称 ) logger.info("Plan created.") async def _execute_step(self, executor: BaseAgent, step_info: dict) -> str: """让执行者智能体执行单个步骤。""" step_text = step_info.get("text", "Current step") plan_status = await self._get_plan_text() # 获取当前计划状态 # 为智能体构造提示词 step_prompt = f"Current Plan:\n{plan_status}\n\nYour Task:\nExecute step: {step_text}" # 调用智能体的 run 方法！ step_result = await executor.run(step_prompt) # 执行后标记步骤完成 await self._mark_step_completed() return step_result async def _mark_step_completed(self): """更新当前步骤的规划工具状态。""" if self.current_step_index is not None: await self.planning_tool.execute( command="mark_step", plan_id=self.active_plan_id, step_index=self.current_step_index, step_status="completed" ) logger.info(f"Step {self.current_step_index} marked complete.")

对这段核心逻辑的逐点说明：

• execute是主控循环：先创建计划，然后进入while True循环——取下一步、判断是否结束、选执行者、执行步骤并累计结果，直到没有剩余步骤时调用_finalize_plan收尾。result_accumulator把所有步骤结果拼成一个字符串返回，这正是我们在实战示例中打印的final_result。
• _create_initial_plan打通了 LLM 与工具：它把"规划师"系统提示、用户目标封装成 Message，再通过self.llm.ask_tool(..., tools=[self.planning_tool.to_param()])让 LLM 以工具调用（而不是纯文本回复）的方式生成计划——to_param()把PlanningTool的描述与参数规格转成 LLM API 需要的格式，这与 ToolCollection 的 to_params() 是同一套机制。
• _execute_step是"委托"的关键：它把当前计划全文 + 本步骤指令拼成step_prompt，然后await executor.run(step_prompt)。这里的run()正是 BaseAgent 提供的标准执行循环——智能体内部会自行经历"检查状态 → 置为 RUNNING → 循环调用 step() → 归位 IDLE"的完整过程。也就是说，Flow 负责"安排做什么"，Agent 负责"具体怎么做"，两层职责被干净地切分。
• _mark_step_completed维护计划状态：通过planning_tool.execute(command="mark_step", ...)把当前步骤标记为completed，配合_get_current_step_info读取"下一个未完成步骤"，形成"取一步 → 执行 → 标记 → 再取一步"的状态推进闭环。

与前面章节的串联：Flow 如何站在 Agent 与 Tool 之上

把第 1～5 章的知识串起来，就能看到 OpenManus 的分层协作全景：

1. LLM 的ask提供纯文本"思考"，ask_tool则允许 LLM 返回工具调用指令；
2. Message / Memory 为这些思考与调用提供结构化的对话上下文；
3. BaseAgent 的run()循环把"思考 + 工具调用"组装成一个可重复执行的智能体个体；
4. Tool / ToolCollection 通过to_params()和execute()让智能体获得可被发现、可被执行的技能；
5. 本文的BaseFlow/PlanningFlow站在最顶层：_create_initial_plan用 LLM +PlanningTool生成计划，执行循环则反复调用执行者智能体的run()——而这些智能体内部又各自使用 LLM、Memory 与 Tools。

于是，OpenManus 的能力边界从"单个智能体的一次性问答"扩展到了"由计划驱动的、可跨智能体的多步骤项目"。这也解释了 OpenManus 教程首页 架构图中BaseFlow -- Orchestrates Agents --> BaseAgent与BaseFlow -- Uses Tools --> Tool / ToolCollection两条连线：Flow 既编排智能体，也间接消费工具能力。

总结与下一步

本文完整覆盖了BaseFlow的定位与使用方式：

• BaseFlow是所有 Flow 的抽象基类，本质是一个持有agents字典、声明抽象execute方法的编排蓝图；
• PlanningFlow是最核心的具体实现，采用"先生成计划、再逐条顺序执行"的策略，通过PlanningTool维护计划的创建、读取与状态标记；
• FlowFactory+FlowType是创建 Flow 的标准入口，新增流程类型只需扩展枚举与映射表；
• 执行链路由_create_initial_plan、_get_current_step_info、get_executor、_execute_step、_mark_step_completed、_finalize_plan六个环节构成，最终把累计结果以字符串返回。

它让 OpenManus 能够处理比单个智能体独自作战更大的目标。而要让这些 Flow、Agent、Tool 之间顺畅地传递数据（例如工具参数的格式、智能体的状态定义），还需要一套"官方表格"来约束它们——这正是 Chapter 6: Schema 要解决的问题：了解 OpenManus 如何定义与校验这些数据结构。

【免费下载链接】Tutorial-Codebase-KnowledgePocket Flow: Codebase to Tutorial项目地址: https://gitcode.com/gh_mirrors/tu/Tutorial-Codebase-Knowledge