解构OpenManus

一、程序结构解读

1. 核心架构分层

BaseAgent (抽象基类)
├── ReActAgent (反应式代理基类)
│   └── ToolCallAgent (工具调用代理基类)
│       ├── ManusAgent (通用多工具代理)
│       ├── PlanningAgent (带计划管理的代理)
│       └── SWEAgent (编程专用代理)
└── [其他扩展代理]

2. 关键模块说明

• Agent 核心层

  • BaseAgent：定义代理生命周期、状态机、记忆系统
  • ReActAgent：实现经典的思考-行动循环（Think-Act cycle）
  • ToolCallAgent：扩展工具调用能力，支持复杂工具编排

• 工具系统

  • ToolCollection：工具注册中心
  • ToolCall：工具调用元数据（包含参数、调用ID等）

• 记忆系统

  • Memory：对话历史存储器
  • Message：结构化消息对象（用户/系统/助手/工具消息）

• 执行控制

  • AgentState：状态枚举（IDLE/RUNNING/ERROR/FINISHED）
  • state_context：异步状态机上下文管理器
  • step_execution_tracker：步骤执行跟踪器（PlanningAgent特有）

3. 典型执行流程

---

二、核心数据结构

1. 消息系统（Memory）

class Message(BaseModel):
    role: Literal["user", "system", "assistant", "tool"]
    content: str
    tool_call_id: Optional[str] = None  # 工具调用关联ID
    name: Optional[str] = None          # 工具名称
    
class Memory:
    messages: List[Message]             # 消息历史栈
    add_message(msg: Message)           # 添加消息
    get_context(window=5)               # 获取最近N条上下文

2. 工具调用系统

class ToolCall(BaseModel):
    id: str                            # 唯一调用标识符
    function: FunctionCall             # 调用参数
    created_at: datetime               # 调用时间戳

class FunctionCall:
    name: str                         # 工具名称
    arguments: dict                   # JSON格式参数
    
class ToolCollection:
    tool_map: Dict[str, BaseTool]     # 工具注册表
    add_tool(tool: BaseTool)          # 注册新工具
    execute(name, input)              # 执行指定工具

3. 代理状态机

class AgentState(Enum):
    IDLE = "idle"
    RUNNING = "running"
    ERROR = "error"
    FINISHED = "finished"

class StateTransition:
    previous_state: AgentState
    current_state: AgentState
    transition_time: datetime

4. 计划管理（PlanningAgent特有）

class PlanTracker:
    plan_id: str                      # 唯一计划标识符
    steps: List[PlanStep]             # 步骤列表
    current_step: int                 # 当前步骤索引
    
class PlanStep:
    index: int
    description: str
    status: Literal["pending", "in_progress", "completed", "failed"]
    tool_calls: List[ToolCall]        # 关联的工具调用
    dependencies: List[int]           # 步骤依赖关系

5. 异常处理结构

class AgentError:
    code: int                        # 错误代码
    message: str                     # 可读错误信息
    stack_trace: str                 # 异常堆栈
    recovery_suggestion: str         # 恢复建议

---

关键设计特点

1. 分层扩展机制：

   • 通过继承链实现能力叠加（Base → ReAct → ToolCall → 具体代理）
   • 新增代理类型只需继承并实现step()方法

2. 工具热插拔：

   # 添加新工具示例
   class NewTool(BaseTool):
       def execute(self, input):
           return process(input)
   
   agent.available_tools.add_tool(NewTool())
   

3. 状态感知执行：

   async with self.state_context(AgentState.RUNNING):
       # 在此上下文中执行受保护的操作
       # 自动处理状态转换和错误回滚
   

4. 语义跟踪能力：

   • 通过step_execution_tracker实现语义级执行跟踪
   • 每个工具调用关联到具体计划步骤

---