当前位置: 首页 > article >正文

【Agent】OpenManus-Flow-BaseFlow详细分析

1. 概述

BaseFlow 是一个抽象基类,用于支持多个Agent的执行流程管理。它提供了Agent集合的管理、主要Agent的指定以及流程执行的基础框架。子类必须实现 execute 方法来定义具体的流程执行逻辑。

2. Class 参数

BaseAgent 见文档 xxxx

参数名称类型默认值描述
agentsDict[str, BaseAgent]必填Agent映射字典,键为Agent标识符,值为Agent实例
toolsOptional[List]None可选的工具列表,可供流程中的Agent使用
primary_agent_keyOptional[str]None主要Agent的标识符,如果未指定则使用第一个Agent

类配置 (Config)

配置项描述
arbitrary_types_allowedTrue允许任意类型的字段,用于支持复杂对象如Agent实例

3. 枚举类型

3.1. FlowType

class FlowType(str, Enum):
    PLANNING = "planning"

功能:定义流程类型的枚举,目前只有规划(PLANNING)类型。

3.2. PlanStepStatus

class PlanStepStatus(str, Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    BLOCKED = "blocked"

功能:定义计划步骤状态的枚举,用于跟踪计划执行进度。

状态描述
NOT_STARTED步骤尚未开始
IN_PROGRESS步骤正在执行中
COMPLETED步骤已完成
BLOCKED步骤被阻塞

4. Code trace

4.1. init

def __init__(
    self, agents: Union[BaseAgent, List[BaseAgent], Dict[str, BaseAgent]], **data
):

功能:初始化流程,处理不同方式提供的Agent。

实现细节

  • 支持三种方式提供Agent:单个Agent、Agent列表或Agent字典
  • 如果提供单个Agent,将其添加为key为 “default” 的字典项
  • 如果提供Agent列表,为每个Agent生成键名 “agent_0”, “agent_1” 等
  • 如果未指定主要Agent,使用字典中的第一个Agent作为主要Agent

4.2. primary_agent

@property
def primary_agent(self) -> Optional[BaseAgent]:

功能:获取流程的主要Agent。

返回值

  • Optional[BaseAgent] - 主要Agent实例,如果未设置则返回 None

4.3. get_agent

def get_agent(self, key: str) -> Optional[BaseAgent]:

功能:根据键获取特定的Agent。

参数

  • key: str - Agent的标识符

返回值

  • Optional[BaseAgent] - 指定的Agent实例,如果不存在则返回 None

4.4. add_agent

def add_agent(self, key: str, agent: BaseAgent) -> None:

功能:向流程中添加新Agent。

参数

  • key: str - 新Agent的标识符
  • agent: BaseAgent - 要添加的Agent实例

4.5. execute

@abstractmethod
async def execute(self, input_text: str) -> str:

功能:执行流程,处理 prompt。

参数

  • input_text: str - 要处理的prompt

返回值

  • str - 流程执行的结果

设计理念

  • 使用抽象方法强制子类实现具体的执行逻辑
  • 提供统一的执行接口,简化调用
  • 支持异步执行,提高性能

5. PlanStepStatus 辅助方法

5.1. get_all_statuses
@classmethod
def get_all_statuses(cls) -> list[str]:

功能:返回所有可能的步骤状态值列表。

返回值

  • list[str] - 所有状态的值列表
5.2. get_active_statuses
@classmethod
def get_active_statuses(cls) -> list[str]:

功能:返回表示活动状态(未开始或进行中)的值列表。

返回值

  • list[str] - 活动状态的值列表
5.3. get_status_marks
@classmethod
def get_status_marks(cls) -> Dict[str, str]:

功能:返回状态到其标记符号的映射。

返回值

  • Dict[str, str] - 状态值到标记符号的映射字典
状态标记
completed[✓]
in_progress[→]
blocked[!]
not_started[ ]

6. 设计理念分析

6.1. 多Agent协作框架

BaseFlow 的核心设计理念是提供一个框架,使多个Agent能够协同工作:

  • Agent集合管理:通过字典结构管理多个Agent,支持按名称访问
  • 主要Agent概念:定义主要Agent作为流程的主控制点
  • 灵活的Agent配置:支持多种方式提供和配置Agent

这种设计允许创建复杂的工作流,其中不同的Agent可以负责不同的任务或步骤。

6.2. 流程抽象与扩展

BaseFlow 提供了流程的抽象基础,同时支持灵活扩展:

  • 抽象执行方法:通过抽象的 execute 方法定义流程接口
  • 状态跟踪:通过 PlanStepStatus 枚举支持计划执行状态的跟踪
  • 类型安全:使用 Pydantic 模型确保类型安全和数据验证

这种设计使得可以创建各种类型的流程,从简单的顺序执行到复杂的条件分支和并行处理。

6.3. 计划执行支持

BaseFlow 特别支持计划执行类型的流程:

  • 计划步骤状态:定义了完整的计划步骤状态集
  • 状态可视化:提供状态标记映射,支持计划状态的可视化
  • 活动状态识别:支持识别需要处理的活动状态

这种设计使得 BaseFlow 特别适合实现基于计划的任务执行,如 PlanningAgent 所需的流程。

6.4. 与 BaseAgent 的关系

BaseFlow 与 BaseAgent 形成互补关系:

  • BaseAgent 负责单个Agent的行为:定义了Agent的状态管理、内存存储和执行循环
  • BaseFlow 负责多Agent的协调:管理多个Agent的集合,协调它们的执行
  • 共享的抽象模式:两者都使用抽象方法定义核心功能,强制子类实现具体逻辑

这种设计允许在保持单个Agent自主性的同时,实现复杂的多Agent协作流程。

7. 使用示例

class MyFlow(BaseFlow):
    async def execute(self, input_text: str) -> str:
        # 获取主要Agent
        agent = self.primary_agent
        if not agent:
            return "No primary agent available"
            
        # 执行主要Agent
        result = await agent.run(input_text)
        
        # 根据结果决定是否使用其他Agent
        if "complex task" in result.lower():
            specialist_agent = self.get_agent("specialist")
            if specialist_agent:
                return await specialist_agent.run(result)
                
        return result

# 创建并使用流程
flow = MyFlow(agents={
    "main": MainAgent(),
    "specialist": SpecialistAgent()
})
result = await flow.execute("Solve this problem")

8. Flow 工厂

返回BaseFlow类

class FlowFactory:
    """Factory for creating different types of flows with support for multiple agents"""

    @staticmethod
    def create_flow(
        flow_type: FlowType,
        agents: Union[BaseAgent, List[BaseAgent], Dict[str, BaseAgent]],
        **kwargs,
    ) -> BaseFlow:
        flows = {
            FlowType.PLANNING: PlanningFlow,
        }

        flow_class = flows.get(flow_type)
        if not flow_class:
            raise ValueError(f"Unknown flow type: {flow_type}")

        return flow_class(agents, **kwargs)

9. 总结

BaseFlow 提供了一个强大而灵活的框架,用于管理多个Agent的协作执行。它的设计支持各种流程类型,特别是基于计划的执行流程。通过继承 BaseFlow 并实现 execute 方法,可以创建自定义的流程,协调多个Agent共同完成复杂任务。

BaseFlow 与 BaseAgent 共同构成了 OpenManus 系统的核心架构,前者管理多Agent协作,后者定义单个Agent行为,两者结合提供了构建复杂 AI 系统的坚实基础。


http://www.kler.cn/a/589370.html

相关文章:

  • element-ui progress 组件源码分享
  • 蓝牙技术联盟中国实体成立!华为、小米发声支持本土化战略
  • 实战ansible-playbook
  • 《C#上位机开发从门外到门内》3-3:基于USB的设备管理系统
  • MCP 开放协议
  • Visual Studio里的“公共语言运行时支持”各选项的作用是什么,分别适用于哪些场景?
  • 基于CPLD+MCU的3U机箱数字量输入采集板DI,主要针对标准DC110V开关量信号进行采集处理
  • LINUX驱动学习之IIC驱动-----以AP3216C为例
  • ZED X系列双目3D相机的耐用性与创新设计解析
  • 基于python+django+vue.js开发的停车管理系统运行-期末作业
  • 基于Python的tkinter开发的一个工具,解析图片文件名并将数据自动化导出为Excel文件
  • Spring 原生启动过程
  • Jenkins 快讯
  • A - 整数的简单问题/A - A Simple Problem with Integers
  • Linux centos7误删/boot拯救方法
  • 【系统架构设计师】操作系统 - 文件管理 ③ ( 树形目录结构 | 文件属性 | 绝对路径 与 相对路径 )
  • 大型智慧城市、交通枢纽、跨区域联网监控视频平台使用LiveGBS国标GB28181平台如何实现分布式集群及负载均衡
  • 线性表的顺序表示
  • Mock接口编写教程-axios-mock-adapter(React)
  • react18 核心知识点杂记1