【Agent】OpenManus-Agent-BaseAgent详细分析
概述
BaseAgent 是一个抽象基类，用于管理Agent状态和执行流程。它提供了状态转换、内存管理和基于步骤的执行循环的基础功能。子类必须实现 step 方法来定义具体行为。
Class 参数
| 参数名称 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| name | str | 必填 | agent 的唯一名称 |
| description | Optional[str] | None | agent 的可选描述 |
| system_prompt | Optional[str] | None | 系统级指令提示 |
| next_step_prompt | Optional[str] | None | 确定下一步行动的提示 |
| llm | LLM | LLM() | LLM 实例 |
| memory | Memory | Memory() | Agent 的内存存储，见 xxx 详细设计 |
| state | AgentState | AgentState.IDLE | 当前Agent状态 |
| max_steps | int | 10 | 允许 agent 轮次最大步骤数 |
| current_step | int | 0 | 执行中的当前步骤 |
| duplicate_threshold | int | 2 | 用来判断agent是否卡住的轮次次数，如果这次 message 的 role 和 content 跟 Memory message 的 role 和 content 一致，就会认为是重复，重复次数到 threshold，就会认为是卡住。 |
Optional[str]是 Python 类型注解的一种写法，表示一个值可以是str类型（字符串），也可以是None。它是typing模块中Optional类型的用法。
Field是pydantic库中的一个函数，用于为模型的字段提供额外的元数据和配置。name字段是必填的（...表示必填），并且有一个描述信息。
description、system_prompt、next_step_prompt 主要是由实现类来定义
类配置 (Config)
| 配置项 | 值 | 描述 |
|---|---|---|
| arbitrary_types_allowed | True | 允许任意类型的字段，用于支持灵活的子类扩展。 |
| extra | “allow” | 允许子类中添加额外的字段，提供更大的灵活性。 |
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。