当前位置：首页 > news >正文

【Agent】OpenManus-Flow组件详细分析

news 来源：原创 2025/8/14 8:37:41

1. Flow架构概述

OpenManus 的Flow组件实现了一个灵活的工作流管理系统，主要用于协调多个智能体的协作，以完成复杂任务。Flow组件的核心是基于计划的执行模型，它将任务分解为一系列步骤，然后逐步执行这些步骤，直到任务完成。

2. 核心组件设计

2.1 BaseFlow (基础Flow)

设计特点：

使用 Pydantic 模型实现，支持数据验证和序列化
支持多智能体管理，可以灵活配置主要智能体和执行者
提供了智能体访问和管理的接口
定义了 execute 抽象方法，要求子类实现

关键属性：

agents: 智能体字典，存储Flow中可用的所有智能体
tools: 可选的工具列表
primary_agent_key: 主要智能体的键名

初始化逻辑：

支持多种智能体输入方式：单个智能体、智能体列表或智能体字典
自动设置主要智能体，如果未指定则使用第一个智能体

关键方法：

primary_agent: 属性方法，获取主要智能体
get_agent: 根据键名获取特定智能体
add_agent: 添加新智能体到Flow中
execute: 抽象方法，执行Flow

2.2 FlowType (Flow类型)

设计特点：

使用枚举类定义可用的Flow类型
目前只有 PLANNING 类型，但设计支持扩展

2.3 PlanStepStatus (计划步骤状态)

设计特点：

使用枚举类定义计划步骤的可能状态
提供了状态管理和显示的辅助方法

状态类型：

NOT_STARTED: 未开始
IN_PROGRESS: 进行中
COMPLETED: 已完成
BLOCKED: 已阻塞

辅助方法：

get_all_statuses: 获取所有可能的状态值
get_active_statuses: 获取活动状态值（未开始或进行中）
get_status_marks: 获取状态标记符号映射

2.4 FlowFactory (Flow工厂)

设计特点：

实现了工厂模式，用于创建不同类型的Flow实例
支持根据Flow类型和智能体配置创建相应的Flow

关键方法：

create_flow: 静态方法，根据Flow类型和智能体创建Flow实例

3. PlanningFlow (规划Flow)

3.1 设计特点

继承自 BaseFlow，实现了基于计划的任务执行Flow
使用 PlanningTool 管理计划的创建、更新和执行
支持多智能体协作，可以根据步骤类型选择合适的执行者
实现了计划的自动创建、步骤执行和状态跟踪
提供了计划可视化和总结功能

3.2 关键属性

llm: 语言模型实例，用于计划创建和总结
planning_tool: 计划工具实例，用于管理计划
executor_keys: 执行者智能体的键名列表
active_plan_id: 当前活动计划的 ID
current_step_index: 当前执行步骤的索引

3.3 初始化逻辑

处理执行者配置和计划 ID
初始化计划工具
如果未指定执行者，则使用所有可用智能体作为执行者

3.4 执行Flow

主要执行Flow (execute 方法)：

验证主要智能体是否可用
如果提供了输入文本，创建初始计划
循环执行以下步骤，直到计划完成或无更多步骤：
- 获取当前要执行的步骤信息
- 如果没有更多步骤，完成计划并退出
- 根据步骤类型选择合适的执行者
- 执行当前步骤
- 检查执行者状态，如果为 FINISHED 则退出
返回执行结果

计划创建Flow (_create_initial_plan 方法)：

创建系统消息和用户消息
使用 LLM 和 PlanningTool 创建计划
处理工具调用结果
如果创建失败，创建默认计划

步骤执行Flow (_execute_step 方法)：

准备计划状态和步骤文本
创建步骤提示
使用选定的执行者执行步骤
标记步骤为已完成
返回执行结果

计划完成Flow (_finalize_plan 方法)：

获取最终计划文本
使用 LLM 创建计划总结
如果 LLM 总结失败，尝试使用智能体创建总结
返回计划完成信息和总结

3.5 步骤管理

步骤获取 (_get_current_step_info 方法)：

验证计划是否存在
从计划工具存储中获取计划数据
查找第一个未完成的步骤
提取步骤类型（如果有）
将步骤标记为进行中
返回步骤索引和信息

步骤完成 (_mark_step_completed 方法)：

使用计划工具将当前步骤标记为已完成
如果工具调用失败，直接更新计划工具存储中的步骤状态

3.6 计划可视化

计划文本获取 (_get_plan_text 方法)：

尝试使用计划工具获取计划文本
如果失败，从存储中直接生成计划文本

计划文本生成 (_generate_plan_text_from_storage 方法)：

从计划工具存储中获取计划数据
确保步骤状态和注释与步骤数量匹配
计算各状态的步骤数量和完成进度
生成格式化的计划文本，包括标题、进度和步骤列表
使用状态标记显示每个步骤的状态

4. PlanningTool (计划工具)

4.1 设计特点

实现了计划的创建、更新、查询和管理功能
使用内存存储管理计划数据
支持多个计划的并行管理
提供了计划步骤状态的跟踪和更新

4.2 关键属性

plans: 字典，存储所有计划数据
_current_plan_id: 当前活动计划的 ID

4.3 命令支持

create: 创建新计划
update: 更新现有计划
list: 列出所有计划
get: 获取特定计划
set_active: 设置活动计划
mark_step: 更新步骤状态
delete: 删除计划

5. 调用Flow分析

5.1 Flow创建和初始化

# 创建智能体
agents = {"manus": Manus()}# 使用工厂创建Flow
flow = FlowFactory.create_flow(flow_type=FlowType.PLANNING,agents=agents,
)

首先创建所需的智能体实例
使用 FlowFactory 创建特定类型的Flow
Flow初始化过程中设置智能体和其他配置

5.2 Flow执行

# 执行Flow
result = await flow.execute(prompt)

调用Flow的 execute 方法，传入用户输入
Flow内部创建初始计划
循环执行计划步骤，直到完成
返回执行结果

5.3 计划创建

# 内部计划创建Flow
await self._create_initial_plan(input_text)

使用 LLM 和 PlanningTool 创建初始计划
设置计划 ID 和标题
定义计划步骤

5.4 步骤执行

# 获取当前步骤
self.current_step_index, step_info = await self._get_current_step_info()# 选择执行者
executor = self.get_executor(step_type)# 执行步骤
step_result = await self._execute_step(executor, step_info)

获取当前未完成的步骤
根据步骤类型选择合适的执行者
使用执行者执行步骤
标记步骤为已完成

5.5 计划完成

# 完成计划
result += await self._finalize_plan()

获取最终计划状态
使用 LLM 创建总结
返回计划完成信息

6. 设计模式应用

6.1 工厂模式

FlowFactory 实现了工厂模式，用于创建不同类型的Flow

6.2 策略模式

不同的Flow实现不同的执行策略
可以根据需要选择不同的Flow类型

6.3 命令模式

PlanningTool 实现了命令模式，支持不同的计划操作命令

6.4 观察者模式

计划状态变化和事件处理

6.5 模板方法模式

BaseFlow 定义了算法骨架
PlanningFlow 实现特定步骤

7. 扩展性设计

7.1 Flow扩展

可以通过继承 BaseFlow 创建新的Flow类型
只需实现 execute 方法

7.2 智能体集成

Flow支持多种智能体集成方式
可以根据步骤类型选择不同的执行者

7.3 计划工具扩展

PlanningTool 支持添加新的命令和功能
可以扩展计划数据结构和状态类型

8. 错误处理机制

8.1 异常捕获

在主要方法中捕获异常
记录错误并提供友好的错误消息

8.2 超时处理

在 run_flow.py 中实现了执行超时处理
防止长时间运行的任务阻塞系统

8.3 状态恢复

在计划工具操作失败时提供直接更新存储的备选方案
确保计划状态的一致性

9. Flow使用示例

9.1 基本使用

# 创建智能体和Flow
agents = {"manus": Manus()}
flow = FlowFactory.create_flow(flow_type=FlowType.PLANNING, agents=agents)# 执行Flow
result = await flow.execute("创建一个网站爬虫")

9.2 多智能体协作

# 创建多个智能体
agents = {"planner": PlanningAgent(),"coder": SWEAgent(),"researcher": Manus()
}# 创建Flow并指定执行者
flow = FlowFactory.create_flow(flow_type=FlowType.PLANNING,agents=agents,primary_agent_key="planner",executors=["coder", "researcher"]
)# 执行Flow
result = await flow.execute("分析市场数据并创建可视化")

OpenManus 的Flow组件通过分离计划创建、步骤执行和状态管理的职责，它能够支持复杂任务的分解和执行，同时保持代码的可维护性和可扩展性。核心的基于计划的执行模型和多智能体协作机制使Flow能够处理各种复杂任务，从而实现更高级的自动化和问题解决能力。

1. Flow架构概述

2. 核心组件设计

2.1 BaseFlow (基础Flow)

2.2 FlowType (Flow类型)

2.3 PlanStepStatus (计划步骤状态)

2.4 FlowFactory (Flow工厂)

3. PlanningFlow (规划Flow)

3.1 设计特点

3.2 关键属性

3.3 初始化逻辑

3.4 执行Flow

3.5 步骤管理

3.6 计划可视化

4. PlanningTool (计划工具)

4.1 设计特点

4.2 关键属性

4.3 命令支持

5. 调用Flow分析

5.1 Flow创建和初始化

5.2 Flow执行

5.3 计划创建

5.4 步骤执行

5.5 计划完成

6. 设计模式应用

6.1 工厂模式

6.2 策略模式

6.3 命令模式

6.4 观察者模式

6.5 模板方法模式

7. 扩展性设计

7.1 Flow扩展

7.2 智能体集成

7.3 计划工具扩展

8. 错误处理机制

8.1 异常捕获

8.2 超时处理

8.3 状态恢复

9. Flow使用示例

9.1 基本使用

9.2 多智能体协作

相关文章：