【Agent】OpenManus-Agent-BaseAgent详细分析

概述

BaseAgent 是一个抽象基类，用于管理Agent状态和执行流程。它提供了状态转换、内存管理和基于步骤的执行循环的基础功能。子类必须实现 step 方法来定义具体行为。

Class 参数

Optional[str] 是 Python 类型注解的一种写法，表示一个值可以是 str 类型（字符串），也可以是 None。它是 typing 模块中 Optional 类型的用法。

Field 是 pydantic 库中的一个函数，用于为模型的字段提供额外的元数据和配置。name 字段是必填的（... 表示必填），并且有一个描述信息。

description、system_prompt、next_step_prompt 主要是由实现类来定义

类配置 (Config)

Code trace

initialize_agent

@model_validator(mode="after")
def initialize_agent(self) -> "BaseAgent":

功能：初始化Agent，如果未提供默认设置则使用默认设置。主要是针对 LLM 和 memory。

state_context

@asynccontextmanager
async def state_context(self, new_state: AgentState):

功能：安全管理Agent状态转换的上下文管理器。

设计理念：

• 使用上下文管理器模式确保状态的安全转换
• 在异常情况下自动将状态设置为 ERROR
• 在上下文退出时恢复先前状态，确保状态一致性
• 提供了一种优雅的方式来临时更改Agent状态

update_memory

def update_memory(self, role: ROLE_TYPE, content: str, **kwargs) -> None:

功能：向Agent的内存中添加消息。

调用地址：baseAgent 里面的 run 方法，user request 的时候用的

设计理念：

• 使用工厂模式创建不同类型的消息
• 支持所有标准角色（用户、系统、助手、工具）
• 通过 kwargs 提供灵活性，特别是对于工具消息
• 验证角色以防止无效消息

run

async def run(self, request: Optional[str] = None) -> str:
        if self.state != AgentState.IDLE:
            raise RuntimeError(f"Cannot run agent from state: {self.state}")

        if request:
            self.update_memory("user", request)

        results: List[str] = [] #agent 执行的结果 str 数组
        async with self.state_context(AgentState.RUNNING):
            while (
                # 检查是否超过最大步骤数
                self.current_step < self.max_steps and self.state != AgentState.FINISHED
            ):
                self.current_step += 1
                logger.info(f"Executing step {self.current_step}/{self.max_steps}")
                step_result = await self.step()

                # 检查是否卡住
                if self.is_stuck():
                    self.handle_stuck_state()

                results.append(f"Step {self.current_step}: {step_result}")

            if self.current_step >= self.max_steps:
                self.current_step = 0
                self.state = AgentState.IDLE
                results.append(f"Terminated: Reached max steps ({self.max_steps})")

        return "\n".join(results) if results else "No steps executed"

功能：异步执行Agent的主循环。

设计理念：

• 使用状态上下文管理器确保状态一致性
• 实现有限步骤循环以防止无限执行
• 在每一步检查是否陷入循环
• 收集并返回所有步骤的结果
• 支持可选的初始用户请求

step

@abstractmethod
async def step(self) -> str:

功能：执行Agent工作流中的单个步骤。

实现的 subClass： ReActAgent

设计理念：

• 使用抽象方法强制子类实现特定行为
• 允许不同类型的Agent定义自己的步骤逻辑
• 返回字符串结果以便于日志记录和结果收集

handle_stuck_state

def handle_stuck_state(self):
    stuck_prompt = "\
        Observed duplicate responses. Consider new strategies and avoid repeating ineffective paths already attempted."

功能：通过添加提示来改变策略，处理卡住的状态。

is_stuck

def is_stuck(self) -> bool:

功能：通过检测重复内容来检查Agent是否陷入循环。当前 message 的 role 和 message 和上一条 Memory 中 message 的 role 和 message

属性方法

@property
def messages(self) -> List[Message]:

@messages.setter
def messages(self, value: List[Message]):

功能：提供对Agent内存中消息的访问和修改。

设计理念：

• 使用属性装饰器提供对内部状态的受控访问
• 允许直接设置消息列表，同时保持封装
• 简化对消息历史的访问

总结

BaseAgent 提供了构建 Agent 的基础，包括状态管理、内存存储、执行控制和错误处理。它的设计允许创建各种类型的Agent，从简单的对话Agent到复杂的工具使用Agent。通过继承这个基类并实现 step 方法，可以创建具有特定行为的自定义Agent。