深入解析 OpenManus:一个通用型 AI Agent 框架的设计哲学

如果你想构建一个能够自主思考、调用工具、协同工作的 AI Agent,这篇文章会给你一个清晰的架构蓝图。

引言

在 AI 应用开发的浪潮中,Agent 无疑是今年最火热的话题之一。从 AutoGPT 到 LangChain Agent,从单工具调用到多 Agent 协作,技术演进的速度令人眼花缭乱。

今天,我们来深入解析 OpenManus——一个设计精良的通用型 AI Agent 框架。它没有追求花哨的功能堆砌,而是回归本质:模块化、可扩展、安全隔离


一、整体架构:五层清晰分层

OpenManus 采用经典的分层架构设计,从上到下依次是:

┌─────────────────────────────────────────────────────────────────────────┐
│                              Entry Layer                                │
│                    ┌──────────┐    ┌──────────┐                         │
│                    │ main.py  │    │run_flow.py│                        │
│                    └────┬─────┘    └─────┬────┘                         │
└─────────────────────────┼────────────────┼──────────────────────────────┘
                          │                │
                          ▼                ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                           Flow Layer                                     │
│  ┌────────────────┐    ┌──────────────────┐    ┌───────────────────┐   │
│  │  FlowFactory   │───▶│   PlanningFlow   │───▶│   BaseFlow        │   │
│  └────────────────┘    └──────────────────┘    └───────────────────┘   │
│                                │                                         │
│                                │ execute()                               │
│                                ▼                                         │
└─────────────────────────────────┼───────────────────────────────────────┘
                                  │
                                  ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                           Agent Layer                                    │
│                                                                          │
│    ┌──────────────┐     ┌──────────────┐     ┌──────────────┐          │
│    │   Manus      │────▶│ ToolCallAgent│────▶│  ReActAgent  │          │
│    └──────────────┘     └──────────────┘     └──────────────┘          │
│           │                    │                    │                   │
│           │                    │                    ▼                   │
│           │                    │           ┌──────────────┐            │
│           │                    └──────────▶│  BaseAgent   │            │
│           │                                 └──────────────┘            │
│           │                                                              │
│    ┌──────┴──────┐     ┌──────────────┐     ┌──────────────┐           │
│    │BrowserAgent │     │  SWEAgent    │     │MCPAgent      │           │
│    └─────────────┘     └──────────────┘     └──────────────┘           │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘
                                  │
                    ┌─────────────┼─────────────┐
                    │             │             │
                    ▼             ▼             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                         Tool Layer                                       │
│                                                                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐    │
│  │BaseTool     │  │ToolCollection│ │PlanningTool │  │MCPClientTool│    │
│  └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘    │
│        △                                                                │
│        │ implements                                                     │
│        ▼                                                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐    │
│  │PythonExecute│  │BrowserUseTool│ │StrReplaceEd │  │AskHuman     │    │
│  └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘    │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘
                                  │
                                  ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                        Infrastructure Layer                              │
│                                                                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐    │
│  │    LLM      │  │   Config    │  │   Sandbox   │  │   Logger    │    │
│  └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘    │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

为什么要这样分层?

每一层只依赖下一层,职责清晰:

层级 职责 一句话总结
Entry 处理用户输入 “入口”
Flow 任务编排,多 Agent 协调 “指挥官”
Agent 核心决策与执行逻辑 “大脑”
Tool 具体能力实现 “手脚”
Infrastructure 基础设施支持 “地基”

二、Agent 设计:继承的艺术

Agent 是 OpenManus 的核心。它没有采用平铺式设计,而是通过继承层次逐步增强能力:

BaseAgent (抽象基类)
    │
    ├── ReActAgent (实现 think/act 循环)
    │       │
    │       └── ToolCallAgent (工具调用能力)
    │               │
    │               ├── Manus (主 Agent,集成所有能力)
    │               ├── BrowserAgent (浏览器操作专家)
    │               ├── SWEAgent (软件工程专家)
    │               └── MCPAgent (MCP 协议专家)

2.1 BaseAgent:定义生命周期

class BaseAgent(BaseModel, ABC):
    name: str                    # Agent 名称
    llm: LLM                     # LLM 实例
    memory: Memory               # 记忆存储
    max_steps: int = 10          # 最大步数
    state: AgentState            # 运行状态

    async def run(self, request: str) -> str:
        """主执行循环"""
        async with self.state_context(AgentState.RUNNING):
            while self.current_step < self.max_steps:
                step_result = await self.step()  # 子类实现
                if self.is_stuck():              # 卡住检测
                    self.handle_stuck_state()

状态机设计

Agent 定义了四种运行状态,通过 state_context 上下文管理器保证状态转换安全:

class AgentState(str, Enum):
    """Agent 执行状态"""
    IDLE = "IDLE"           # 空闲
    RUNNING = "RUNNING"     # 运行中
    FINISHED = "FINISHED"   # 已完成
    ERROR = "ERROR"         # 错误

状态转换图:

                    ┌─────────────────────────────────────┐
                    │                                     │
                    ▼                                     │
              ┌─────────┐                                 │
              │  IDLE   │◄────────────────────────────┐   │
              └────┬────┘                             │   │
                   │ run()                            │   │
                   ▼                                  │   │
              ┌─────────┐   达到 max_steps            │   │
              │ RUNNING │─────────────────────────────┘   │
              └────┬────┘                                 │
                   │                                      │
         ┌─────────┼─────────┐                            │
         │         │         │                            │
         ▼         ▼         ▼                            │
   ┌──────────┐ ┌──────────┐ ┌──────────┐                │
   │ FINISHED │ │  ERROR   │ │  卡住检测 │                │
   │ (完成)   │ │ (异常)   │ │ is_stuck │                │
   └──────────┘ └──────────┘ └────┬─────┘                │
                                  │                      │
                                  │ handle_stuck_state() │
                                  └──────────────────────┘

三种终止条件

条件 触发方式 结果状态 说明
达到最大步数 current_step >= max_steps IDLE 防止无限循环的安全机制
正常完成 调用 Terminate 工具 FINISHED LLM 主动结束任务
异常发生 抛出异常 ERROR state_context 捕获

设计亮点

  • state_context 上下文管理器保证状态转换安全,异常时自动设置为 ERROR
  • is_stuck() 检测循环死锁并自动干预,避免 Agent 陷入无限重复

2.2 ReActAgent:思考-行动循环

ReAct(Reasoning + Acting)模式是 Agent 的核心:

class ReActAgent(BaseAgent, ABC):
    async def think(self) -> bool:
        """思考:分析当前状态,决定下一步行动"""

    async def act(self) -> str:
        """行动:执行决定的操作"""

    async def step(self) -> str:
        should_act = await self.think()
        if not should_act:
            return "Thinking complete - no action needed"
        return await self.act()

这个设计有多优雅?

思考与行动分离,让 Agent 能够:

  1. 先分析当前局势
  2. 再决定是否需要执行操作
  3. 最后执行并获取反馈

2.3 ToolCallAgent:让 Agent 拥有"手"

class ToolCallAgent(ReActAgent):
    available_tools: ToolCollection     # 可用工具集合
    tool_calls: List[ToolCall] = []     # 待执行的工具调用

    async def think(self) -> bool:
        response = await self.llm.ask_tool(
            messages=self.messages,
            tools=self.available_tools.to_params(),
        )
        self.tool_calls = response.tool_calls
        return bool(self.tool_calls)

三、Tool 模块:能力即插即用

工具是 Agent 与世界交互的桥梁。OpenManus 的工具设计同样遵循"简单即美"的原则。

3.1 BaseTool:一切工具的起点

class BaseTool(ABC, BaseModel):
    name: str                    # 工具名称
    description: str             # 工具描述
    parameters: Optional[dict]   # 参数 Schema

    async def execute(self, **kwargs) -> Any:
        """执行工具逻辑"""

    def to_param(self) -> Dict:
        """转换为 OpenAI Function Calling 格式"""

3.2 内置工具一览

工具名称 功能描述 应用场景
PythonExecute 执行 Python 代码 数据分析、脚本执行
BrowserUseTool 浏览器自动化 网页爬取、自动操作
StrReplaceEditor 文件字符串替换 代码修改
AskHuman 请求人工输入 需要确认的场景
Terminate 终止任务 任务完成

3.3 MCP:动态扩展工具生态

OpenManus 支持 MCP(Model Context Protocol) 协议,可以动态接入外部工具:

class MCPClients(ToolCollection):
    async def connect_sse(self, server_url: str, server_id: str):
        """通过 SSE 连接 MCP 服务器"""
        streams = await sse_client(url=server_url)
        session = ClientSession(*streams)
        await session.initialize()

        # 获取服务器提供的工具列表
        response = await session.list_tools()
        for tool in response.tools:
            server_tool = MCPClientTool(name=f"mcp_{server_id}_{tool.name}")
            self.tool_map[tool.name] = server_tool

只需一行配置,就能接入全新的工具能力:

{
  "mcpServers": {
    "my-server": {
      "type": "stdio",
      "command": "python",
      "args": ["-m", "my_mcp_server"]
    }
  }
}

四、Flow 模块:多 Agent 协作的指挥官

当任务变得复杂时,单个 Agent 可能力不从心。Flow 的作用就是编排多个 Agent 协同工作。

4.1 PlanningFlow:规划驱动的执行流

class PlanningFlow(BaseFlow):
    async def execute(self, input_text: str) -> str:
        # 1. 创建初始计划
        await self._create_initial_plan(input_text)

        # 2. 循环执行每个步骤
        while True:
            step_index, step_info = await self._get_current_step_info()
            if step_index is None:
                break

            # 3. 选择合适的 Agent 执行
            executor = self.get_executor(step_info.get("type"))
            result = await self._execute_step(executor, step_info)

            # 4. 标记步骤完成
            await self._mark_step_completed()

执行流程图

用户输入: "帮我分析 sales.csv 文件并生成报告"
                    │
                    ▼
        ┌───────────────────────┐
        │ LLM 生成计划           │
        │                       │
        │ 1. [DATA] 读取数据文件 │
        │ 2. [DATA] 分析数据     │
        │ 3. [MANUS] 生成报告    │
        └───────────┬───────────┘
                    │
                    ▼
        ┌───────────────────────┐
        │ 逐步执行,状态追踪      │
        │ [→] in_progress        │
        │ [✓] completed          │
        └───────────────────────┘

五、两种运行模式:隐式 vs 显式规划

OpenManus 提供两种运行模式,适应不同复杂度的任务。

5.1 简单模式(main.py)

# main.py
async def main():
    agent = await Manus.create()
    await agent.run(prompt)  # Agent 自主决策执行

特点:LLM 在"脑子里"规划,用户看不到步骤结构。

5.2 Flow 模式(run_flow.py)

# run_flow.py
async def run_flow():
    flow = FlowFactory.create_flow(flow_type=FlowType.PLANNING)
    result = await flow.execute(prompt)  # 先规划再执行

特点:显式调用 PlanningTool,用户可见步骤状态。

5.3 对比总结

维度 简单模式 Flow 模式
执行方式 Agent 自主决策 先规划,再逐步执行
计划可见性 不可见 可见步骤状态
Agent 数量 单个 支持多 Agent 协作
适用任务 简单直接任务 复杂多步骤任务

六、沙箱机制:安全执行代码

让 AI 执行代码是一件"危险"的事情。OpenManus 通过 Docker 沙箱 解决这个问题:

┌─────────────────────────────────────────────┐
│              Docker Container                │
│                                             │
│  python:3.12-slim (默认镜像)                 │
│                                             │
│  /workspace/     ← 工作目录                  │
│                                             │
│  资源限制:                                   │
│  ├── 内存: 512MB                            │
│  ├── CPU: 1.0 核                            │
│  ├── 网络: 禁用                             │
│  └── 超时: 300秒                            │
└─────────────────────────────────────────────┘

安全特性

  • CPU/内存资源限制
  • 网络隔离(可配置)
  • 超时控制
  • 自动清理

七、设计亮点:那些值得学习的细节

7.1 控制反转(IoC):Tool 不应该知道 Agent

先看一段代码,你可能会觉得奇怪:

class Terminate(BaseTool):
    """终止任务的工具"""
    name: str = "terminate"
    description: str = "当任务完成时调用此工具"

    async def execute(self) -> str:
        return "任务已完成"

问题来了:这个工具几乎什么都没做,只是返回了一个字符串。为什么不在这里直接设置 agent.state = FINISHED

答案:这是**控制反转(IoC)**的体现。

# Tool 的职责:执行并返回结果
class Terminate(BaseTool):
    async def execute(self) -> str:
        return "任务已完成"  # 只负责返回

# Agent 的职责:根据结果决定状态
class ToolCallAgent:
    async def _handle_special_tool(self, name: str, result: Any):
        if self._is_special_tool(name):
            self.state = AgentState.FINISHED  # Agent 负责状态管理

为什么这样设计?

原因 说明
解耦 工具保持独立,可复用,不需要知道 Agent 的存在
灵活性 不同 Agent 可以有不同的终止逻辑(比如先保存进度)
可测试性 测试 Tool 不需要 mock Agent,测试 Agent 不需要考虑 Tool 内部
单一职责 Tool 负责执行,Agent 负责状态管理,职责清晰

架构关系图:

┌─────────────────┐          ┌─────────────────┐
│      Tool       │          │      Agent      │
│                 │   依赖    │                 │
│  - 独立组件     │◀───────── │  - 状态管理       │
│  - 可复用       │           │  - 调用工具      │
│  - 不知道 Agent │           │  - 处理结果      │
└─────────────────┘          └─────────────────┘

7.2 按需上下文注入:节省 Token

async def think(self) -> bool:
    # 检查最近是否使用了浏览器
    browser_in_use = self._check_browser_in_recent_messages()

    if browser_in_use:
        # 只有在需要时才注入浏览器状态和截图
        self.next_step_prompt = await self.format_browser_context()

效果:相比每次都注入,节省约 75% 的 Token 消耗。

7.3 资源安全清理

async def cleanup(self):
    if self.browser_context_helper:
        await self.browser_context_helper.cleanup_browser()
    if self._initialized:
        await self.disconnect_mcp_server()

八、总结

OpenManus 的设计哲学可以概括为:

模块化架构 + ReAct 模式 + 工具生态 + 安全沙箱 + 多 Agent 协作

它没有追求"大而全",而是把每一层都做到职责清晰、易于扩展。对于想构建 AI Agent 应用的开发者来说,这是一个很好的学习范本。

核心优势总结

  1. 模块化架构:Agent、Tool、Flow 清晰分层,易于扩展
  2. ReAct 模式:思考-行动循环,实现复杂任务分解
  3. 工具生态:内置丰富工具,支持 MCP 动态扩展
  4. 安全执行:Docker 沙箱隔离,资源可控
  5. 多 Agent 协作:Flow 机制编排复杂工作流
  6. 控制反转设计:Tool 与 Agent 解耦,保持单一职责

参考

  • OpenManus 项目地址:https://github.com/FoundationAgents/OpenManus
Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐