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 步用搜索工具找缺点。执行第 3 步调用 LLM 的大脑列出博客大纲。执行第 4 步基于调研结果和大纲撰写全文。执行第 5 步最终审阅。一个非常复杂的 BaseAgent 理论上或许能独自完成这一切但当流程变长、步骤变多时由专门的编排器orchestrator来统管整个过程会清晰得多、也更容易维护。这正是 Flow 的职责BaseFlow是这些编排器的蓝图它定义了一种能够管理多个智能体、并按照特定策略例如遵循预先制定的计划协调它们去完成更大目标的结构。一个典型的用例还是先研究后写作的任务我们需要一个东西来管理整体进程。PlanningFlow基于BaseFlow构建的一种具体 Flow正合适——它先创建计划如上面列出的步骤然后逐步执行必要时还能把不同的步骤分派给不同的专业智能体。核心概念Flow、Agents 与执行策略OpenManus 的 Flow 体系由三个层次的概念组成BaseFlowapp/flow/base.py所有 Flow 的抽象蓝图。可以把它理解为一份项目经理岗位说明书——它规定了一名经理需要知道自己的团队agents、需要有一个运行项目的方式execute方法但不规定经理具体怎么管。它最主要的职责是持有一个agents字典存放可供 Flow 调用的智能体。你通常不会直接使用BaseFlow而是使用它的具体实现。具体 Flow如PlanningFlow位于app/flow/planning.py这些是具体的项目管理策略它们继承自BaseFlow。PlanningFlow的策略是接收总目标使用 LLM 与一个专门的PlanningTool把目标拆解为一串步骤即计划逐条执行计划中的步骤通常通过调用某个合适的 BaseAgent 的run()方法来完成跟踪每个步骤的状态未开始、进行中、已完成。Flow 内的 Agents被 Flow 管理的工人或专家。一个 Flow 持有一个或多个 BaseAgent 实例。在PlanningFlow中可能指定一个主智能体primary agent常负责协助生成计划其余或同一个智能体则作为计划步骤的执行者。Flow 负责决定每一步该交给哪个智能体。用盖房子来类比最直观BaseFlow是总承包商这个概念本身PlanningFlow是某种具体的总承包商——他总是先绘制详细的建筑图纸计划再为每个阶段聘请专家agents是各工种专家水管工、电工、木匠……总目标盖一栋房子交给PlanningFlow承包商PlanningFlow制定计划地基、框架、水电……然后针对计划的每一步调用合适的agent专家。实战用 FlowFactory 创建 PlanningFlow在 OpenManus 中创建 Flow 的标准方式是使用FlowFactory并把 Flow 所需的智能体交给它。下面的示例创建一个带有一个名为 ManusOpenManus 的通用型智能体的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_typeFlowType.PLANNING, agentsagents_for_flow, # 可选指定哪个智能体是主智能体如果不使用第一个 # primary_agent_keyresearch_writer ) print(fCreated a {type(planning_flow_instance).__name__}) print(fPrimary 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(fFinal Result:\n{final_result}) # 运行异步函数 # asyncio.run(run_the_flow()) # 取消注释即可运行逐步拆解这段代码导入要用的智能体Manus以及FlowFactory、FlowType创建字典agents_for_flow把键research_writer映射到一个Manus实例——这告诉 Flow 有哪些工人可用调用FlowFactory.create_flow()指定FlowType.PLANNING并传入agents_for_flow工厂负责正确构造PlanningFlow对象定义高层任务overall_goalawait 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 与 FlowTypeFlowFactory是一个静态工厂它接收FlowType枚举值与agents字典通过内部映射找到对应的 Flow 类并实例化。从教程文档所描述的 OpenManus 教程 可以看出FlowType目前至少包含PLANNING planning一项且该映射是可扩展的——新增 Flow 类型时只需在flows映射表中登记新类即可。如果传入未注册的flow_type工厂会抛出ValueError(fUnknown flow type: {flow_type})。源码走查PlanningFlow.execute 的执行链路调用execute(input_text)后到底发生了什么以下是高层完整走查接收目标execute方法拿到input_text即总体目标。创建计划_create_initial_plan为 LLM 构造消息包括要求它扮演规划师角色的系统消息告诉 LLM 存在PlanningTool一个专门用于创建和管理计划的 Tool调用 LLM 的ask_tool方法实质上是在问请使用 PlanningTool 为这个目标创建计划{input_text}PlanningTool在被 LLM 调用时把生成的步骤如[Search benefits, Write summary]与一个唯一的plan_id关联并存储起来。执行循环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循环回到取下一步继续下一轮。收尾_finalize_plan循环结束后可能生成已完成计划的最终总结可能再次借助 LLM。返回结果把所有步骤累计得到的结果作为字符串返回。时序图核心代码片段解析以下代码片段是教程文档基于 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(fUnknown 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_factoryPlanningTool) # ... 其他字段如 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(fCreating plan for: {request}) system_msg Message.system_message(You are a planner...) user_msg Message.user_message(fCreate 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_choiceToolChoice.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 fCurrent 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( commandmark_step, plan_idself.active_plan_id, step_indexself.current_step_index, step_statuscompleted ) logger.info(fStep {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(commandmark_step, ...)把当前步骤标记为completed配合_get_current_step_info读取下一个未完成步骤形成取一步 → 执行 → 标记 → 再取一步的状态推进闭环。与前面章节的串联Flow 如何站在 Agent 与 Tool 之上把第 15 章的知识串起来就能看到 OpenManus 的分层协作全景LLM 的ask提供纯文本思考ask_tool则允许 LLM 返回工具调用指令Message / Memory 为这些思考与调用提供结构化的对话上下文BaseAgent 的run()循环把思考 工具调用组装成一个可重复执行的智能体个体Tool / ToolCollection 通过to_params()和execute()让智能体获得可被发现、可被执行的技能本文的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维护计划的创建、读取与状态标记FlowFactoryFlowType是创建 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考