OpenManus 开源解读，深入剖析背后的 AI Agent 架构-首席AI分享圈

🚀邀请体验：中国首家 AI IDE 智能编程软件 Trae 中文版下载，不限量使用 DeepSeek-R1 和 Doubao-pro!

开篇：Manus 的火爆与 OpenManus 的破局

最近，AI 圈子里出了一件大事，那就是 Manus AI Agent 的横空出世。Manus 以其强大的功能和灵活的使用方式，迅速吸引了无数目光。简单来说，Manus 就像一个全能助手，无论是编程、查资料、处理文件，还是上网冲浪，它都能帮你搞定。

不过，想用 Manus 可没那么容易，得有邀请码才行。这就把很多开发者和研究者挡在了门外。正当大家一筹莫展的时候，开源社区出手了！MetaGPT 团队的几位大佬@mannaandpoem、@XiangJinyu、@MoshiQAQ 和 @didiforgithub，花了短短 3 小时，就搞出了一个叫 OpenManus 的开源项目。这下好了，不用邀请码，也能体验 Manus 的强大功能了！更令人兴奋的是 OpenManus 是一个开源项目，这意味着大家可以根据自己的需求对它进行修改和扩展！

OpenManus：MetaGPT 推出的开源版 Manus-1

OpenManus 的出现，不仅让更多人有机会体验到 AI Agent 的魅力，也为 AI Agent 的发展注入了新的活力。对于咱们这些搞技术的人来说，OpenManus 不仅是一个好用的工具，更是一个绝佳的学习资源。通过研究它的代码，我们可以深入了解 AI Agent 的框架设计和实现细节。

AI Agent 框架：OpenManus 的设计哲学

OpenManus 的代码结构非常清晰，采用了模块化的设计，就像搭积木一样，把不同的功能模块组合在一起。这种设计的好处是，代码复用性高、扩展性强，而且每个模块的职责都很明确。

OpenManus 的核心组件包括：

OpenManus
├── Agent (代理层)
│   ├── BaseAgent (基础抽象类)
│   ├── ReActAgent (思考-行动模式)
│   ├── ToolCallAgent (工具调用能力)
│   ├── PlanningAgent (规划能力)
│   ├── SWEAgent (软件工程能力)
│   └── Manus (通用代理)
├── LLM (语言模型层)
├── Memory (记忆层)
├── Tool (工具层)
│   ├── BaseTool (工具基类)
│   ├── PlanningTool (规划工具)
│   ├── PythonExecute (Python 执行)
│   ├── GoogleSearch (搜索工具)
│   ├── BrowserUseTool (浏览器工具)
│   └── ... (其他工具)
├── Flow (工作流层)
│   ├── BaseFlow (基础流程)
│   └── PlanningFlow (规划流程)
└── Prompt (提示词层)

LLM 组件：Agent 的大脑

如果把 Agent 比作一个人，那么 LLM（Large Language Model，大型语言模型）就是 Agent 的大脑。它负责理解用户的指令，生成回复，并做出决策。OpenManus 通过 LLM 类来封装与语言模型的交互。

class LLM:
_instances: Dict[str, "LLM"] = {}  # 单例模式实现
def __init__(
self, config_name: str = "default", llm_config: Optional[LLMSettings] = None
):
if not hasattr(self, "client"):  # 只初始化一次
llm_config = llm_config or config.llm
llm_config = llm_config.get(config_name, llm_config["default"])
self.model = llm_config.model
self.max_tokens = llm_config.max_tokens
self.temperature = llm_config.temperature
self.client = AsyncOpenAI(
api_key=llm_config.api_key, base_url=llm_config.base_url
)

LLM 类提供了两个核心方法：

ask: 发送普通对话请求
ask_tool: 发送带工具调用的请求

async def ask_tool(
self,
messages: List[Union[dict, Message]],
system_msgs: Optional[List[Union[dict, Message]]] = None,
timeout: int = 60,
tools: Optional[List[dict]] = None,
tool_choice: Literal["none", "auto", "required"] = "auto",
temperature: Optional[float] = None,
**kwargs,
):
# 格式化消息
if system_msgs:
system_msgs = self.format_messages(system_msgs)
messages = system_msgs + self.format_messages(messages)
else:
messages = self.format_messages(messages)
# 发送请求
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature or self.temperature,
max_tokens=self.max_tokens,
tools=tools,
tool_choice=tool_choice,
timeout=timeout,
**kwargs,
)

Memory 组件：Agent 的记忆

Memory 组件就像 Agent 的笔记本，负责记录和管理 Agent 的对话历史。有了 Memory，Agent 才能记住之前说过的话，保持对话的连贯性。

class Memory(BaseModel):
"""Stores and manages agent's conversation history."""
messages: List[Message] = Field(default_factory=list)
def add_message(self, message: Union[Message, dict]) -> None:
"""Add a message to memory."""
if isinstance(message, dict):
message = Message(**message)
self.messages.append(message)
def get_messages(self) -> List[Message]:
"""Get all messages in memory."""
return self.messages

Memory 组件是 Agent 的核心组成部分，通过 BaseAgent 的 update_memory 方法添加新消息：

def update_memory(
self,
role: Literal["user", "system", "assistant", "tool"],
content: str,
**kwargs,
) -> None:
"""Add a message to the agent's memory."""
message_map = {
"user": Message.user_message,
"system": Message.system_message,
"assistant": Message.assistant_message,
"tool": lambda content, **kw: Message.tool_message(content, **kw),
}
if role not in message_map:
raise ValueError(f"Unsupported message role: {role}")
msg_factory = message_map[role]
msg = msg_factory(content, **kwargs) if role == "tool" else msg_factory(content)
self.memory.add_message(msg)

Tools 组件：Agent 的工具箱

Tools 组件是 Agent 与外部世界交互的桥梁。OpenManus 实现了一个灵活的工具系统，让 Agent 能够调用各种工具来完成任务。

class BaseTool(ABC, BaseModel):
name: str
description: str
parameters: Optional[dict] = None
async def __call__(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
return await self.execute(**kwargs)
    @abstractmethod
async def execute(self, **kwargs) -> Any:
"""Execute the tool with given parameters."""
def to_param(self) -> Dict:
"""Convert tool to function call format."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}

工具的执行结果用 ToolResult 类来表示：

class ToolResult(BaseModel):
"""Represents the result of a tool execution."""
output: Any = Field(default=None)
error: Optional[str] = Field(default=None)
system: Optional[str] = Field(default=None)

OpenManus 提供了很多内置工具，比如 PlanningTool：

class PlanningTool(BaseTool):
"""
A planning tool that allows the agent to create and manage plans for solving complex tasks.
The tool provides functionality for creating plans, updating plan steps, and tracking progress.
"""
name: str = "planning"
description: str = _PLANNING_TOOL_DESCRIPTION
parameters: dict = {
"type": "object",
"properties": {
"command": {
"description": "The command to execute. Available commands: create, update, list, get, set_active, mark_step, delete.",
"enum": [
"create",
"update",
"list",
"get",
"set_active",
"mark_step",
"delete",
],
"type": "string",
},
# 其他参数...
},
"required": ["command"],
}

Planning 组件：Agent 的规划能力

Planning 组件是 OpenManus 实现复杂任务处理的关键。它可以让 Agent 制定计划，把复杂的任务分解成一步一步的小任务，然后逐个完成。

Planning 组件主要包括两部分：

PlanningTool：提供计划的创建、更新和跟踪功能。
PlanningAgent：使用 PlanningTool 来进行任务规划和执行。

class PlanningAgent(ToolCallAgent):
"""
An agent that creates and manages plans to solve tasks.
This agent uses a planning tool to create and manage structured plans,
and tracks progress through individual steps until task completion.
"""
name: str = "planning"
description: str = "An agent that creates and manages plans to solve tasks"
system_prompt: str = PLANNING_SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(PlanningTool(), Terminate())
)
# 步骤执行跟踪器
step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: Optional[int] = None

PlanningAgent 的核心方法包括：

async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:n{await self.get_plan()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))
# 获取当前步骤索引
self.current_step_index = await self._get_current_step_index()
result = await super().think()
# 关联工具调用与当前步骤
if result and self.tool_calls:
# ...关联逻辑...
return result

Flow 组件：Agent 的协作能力

Flow 组件的作用是协调多个 Agent 一起工作，完成更复杂的任务。

class BaseFlow(BaseModel, ABC):
"""Base class for execution flows supporting multiple agents"""
agents: Dict[str, BaseAgent]
tools: Optional[List] = None
primary_agent_key: Optional[str] = None
    @property
def primary_agent(self) -> Optional[BaseAgent]:
"""Get the primary agent for the flow"""
return self.agents.get(self.primary_agent_key)
    @abstractmethod
async def execute(self, input_text: str) -> str:
"""Execute the flow with given input"""

PlanningFlow 是一个具体的 Flow 实现，用于规划和执行任务：

class PlanningFlow(BaseFlow):
"""A flow that manages planning and execution of tasks using agents."""
llm: LLM = Field(default_factory=lambda: LLM())
planning_tool: PlanningTool = Field(default_factory=PlanningTool)
executor_keys: List[str] = Field(default_factory=list)
active_plan_id: str = Field(default_factory=lambda: f"plan_{int(time.time())}")
current_step_index: Optional[int] = None
async def execute(self, input_text: str) -> str:
"""Execute the planning flow with agents."""
try:
# 创建初始计划
if input_text:
await self._create_initial_plan(input_text)
# 执行计划步骤
while await self._has_next_step():
# 获取当前步骤
step_info = await self._get_current_step()
# 选择合适的执行者
executor = self.get_executor(step_info.get("type"))
# 执行步骤
result = await self._execute_step(executor, step_info)
# 更新步骤状态
await self._update_step_status(step_info["index"], "completed")
# 完成计划
return await self._finalize_plan()
except Exception as e:
# 处理异常
return f"Error executing flow: {str(e)}"

OpenManus 的 Agent 实现：分层架构

OpenManus 的 Agent 采用层次化架构，从基础功能到专业应用逐层构建。这种设计的好处是，代码复用性高、扩展性强，而且每个层次的职责都很明确。

BaseAgent (抽象基类)
└── ReActAgent (思考-行动模式)
└── ToolCallAgent (工具调用能力)
├── PlanningAgent (规划能力)
├── SWEAgent (软件工程能力)
└── Manus (通用代理)

BaseAgent：基础中的基础

BaseAgent 是整个框架的基础，它定义了 Agent 的核心属性和方法：

class BaseAgent(BaseModel, ABC):
"""Abstract base class for managing agent state and execution."""
# 核心属性
name: str = Field(..., description="Unique name of the agent")
description: Optional[str] = Field(None, description="Optional agent description")
# 提示词
system_prompt: Optional[str] = Field(None, description="System-level instruction prompt")
next_step_prompt: Optional[str] = Field(None, description="Prompt for determining next action")
# 依赖组件
llm: LLM = Field(default_factory=LLM, description="Language model instance")
memory: Memory = Field(default_factory=Memory, description="Agent's memory store")
state: AgentState = Field(default=AgentState.IDLE, description="Current agent state")
# 执行控制
max_steps: int = Field(default=10, description="Maximum steps before termination")
current_step: int = Field(default=0, description="Current step in execution")

ReActAgent：会思考的 Agent

ReActAgent 实现了“思考-行动”模式，把 Agent 的执行过程分为两个阶段：

class ReActAgent(BaseAgent, ABC):
    @abstractmethod
async def think(self) -> bool:
"""Process current state and decide next action"""
    @abstractmethod
async def act(self) -> str:
"""Execute decided actions"""
async def step(self) -> str:
"""Execute a single step: think and act."""
should_act = await self.think()
if not should_act:
return "Thinking complete - no action needed"
return await self.act()

ToolCallAgent：能用工具的 Agent

ToolCallAgent 给 Agent 增加了使用工具的能力：

class ToolCallAgent(ReActAgent):
"""Base agent class for handling tool/function calls with enhanced abstraction"""
available_tools: ToolCollection = ToolCollection(
CreateChatCompletion(), Terminate()
)
tool_choices: Literal["none", "auto", "required"] = "auto"
async def think(self) -> bool:
# 获取 LLM 响应和工具选择
response = await self.llm.ask_tool(
messages=self.messages,
system_msgs=[Message.system_message(self.system_prompt)]
if self.system_prompt
else None,
tools=self.available_tools.to_params(),
tool_choice=self.tool_choices,
)
self.tool_calls = response.tool_calls
# 处理响应和工具调用
# ...
async def act(self) -> str:
# 执行工具调用
results = []
for command in self.tool_calls:
result = await self.execute_tool(command)
# 添加工具响应到内存
# ...
results.append(result)
return "nn".join(results)

PlanningAgent：会做计划的 Agent

PlanningAgent 实现了任务规划和执行跟踪：

class PlanningAgent(ToolCallAgent):
"""
An agent that creates and manages plans to solve tasks.
This agent uses a planning tool to create and manage structured plans,
and tracks progress through individual steps until task completion.
"""
# 步骤执行跟踪器
step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict)
current_step_index: Optional[int] = None
async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:n{await self.get_plan()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))
# 获取当前步骤索引
self.current_step_index = await self._get_current_step_index()
result = await super().think()
# 关联工具调用与当前步骤
if result and self.tool_calls:
# ...关联逻辑...
return result

Manus：全能 Agent

Manus 是 OpenManus 的核心 Agent，它集成了各种工具和能力，可以处理各种各样的任务：

class Manus(ToolCallAgent):
"""
A versatile general-purpose agent that uses planning to solve various tasks.
This agent extends PlanningAgent with a comprehensive set of tools and capabilities,
including Python execution, web browsing, file operations, and information retrieval
to handle a wide range of user requests.
"""
name: str = "manus"
description: str = "A versatile general-purpose agent"
system_prompt: str = SYSTEM_PROMPT
next_step_prompt: str = NEXT_STEP_PROMPT
available_tools: ToolCollection = Field(
default_factory=lambda: ToolCollection(
PythonExecute(), GoogleSearch(), BrowserUseTool(), FileSaver(), Terminate()
)
)

Prompt：Agent 的行为指南

Prompt 在构建 Agent 系统中起着至关重要的作用，它就像一份说明书，告诉 Agent 应该如何行动。

系统 Prompt：定义 Agent 的角色

系统 Prompt 给 Agent 设定了基本的角色和行为准则：

SYSTEM_PROMPT = "You are OpenManus, an all-capable AI assistant, aimed at solving any task presented by the user. You have various tools at your disposal that you can call upon to efficiently complete complex requests. Whether it's programming, information retrieval, file processing, or web browsing, you can handle it all."

这个 Prompt 告诉 Agent，它是一个全能的 AI 助手，可以利用各种工具来完成用户的请求。

规划 Prompt：指导 Agent 做计划

规划 Prompt 告诉 Agent 如何把复杂的任务分解成小任务，并制定执行计划：

PLANNING_SYSTEM_PROMPT = """
You are an expert Planning Agent tasked with solving complex problems by creating and managing structured plans.
Your job is:
1. Analyze requests to understand the task scope
2. Create clear, actionable plans with the `planning` tool
3. Execute steps using available tools as needed
4. Track progress and adapt plans dynamically
5. Use `finish` to conclude when the task is complete
Available tools will vary by task but may include:
- `planning`: Create, update, and track plans (commands: create, update, mark_step, etc.)
- `finish`: End the task when complete
Break tasks into logical, sequential steps. Think about dependencies and verification methods.
"""

这个 Prompt 告诉 Agent，它是一个规划专家，需要使用 planning 工具来创建、更新和跟踪计划。

工具使用 Prompt：告诉 Agent 怎么用工具

工具使用 Prompt 详细介绍了每个工具的功能和使用场景，帮助 Agent 选择合适的工具：

NEXT_STEP_PROMPT = """You can interact with the computer using PythonExecute, save important content and information files through FileSaver, open browsers with BrowserUseTool, and retrieve information using GoogleSearch.
PythonExecute: Execute Python code to interact with the computer system, data processing, automation tasks, etc.
FileSaver: Save files locally, such as txt, py, html, etc.
BrowserUseTool: Open, browse, and use web browsers.If you open a local HTML file, you must provide the absolute path to the file.
GoogleSearch: Perform web information retrieval
Based on user needs, proactively select the most appropriate tool or combination of tools. For complex tasks, you can break down the problem and use different tools step by step to solve it. After using each tool, clearly explain the execution results and suggest the next steps.
"""

动态 Prompt：让 Agent 更灵活

OpenManus 的 Prompt 不仅可以是静态的，还可以动态生成。比如，在 PlanningAgent 中，系统会把当前的计划状态添加到 Prompt 中：

async def think(self) -> bool:
"""Decide the next action based on plan status."""
prompt = (
f"CURRENT PLAN STATUS:n{await self.get_plan()}nn{self.next_step_prompt}"
if self.active_plan_id
else self.next_step_prompt
)
self.messages.append(Message.user_message(prompt))

这种动态 Prompt 让 Agent 能够根据当前的情况做出更合理的决策。

总结：OpenManus 的启示

通过对 OpenManus 代码的分析，我们可以总结出 AI Agent 框架的几个关键组成部分：

Agent：从基础到专业的层次化设计，实现不同级别的能力。
- BaseAgent：提供基础的状态管理和执行循环。
- ReActAgent：实现思考-行动模式。
- ToolCallAgent：添加工具调用能力。
- 专业 Agent：如 PlanningAgent、SWEAgent 和 Manus。
LLM：封装与大型语言模型的交互，提供对话和工具调用能力。
- 支持普通对话和工具调用。
- 实现重试机制和错误处理。
- 支持流式响应。
Memory：管理对话历史和上下文。
- 存储和检索消息。
- 维护对话上下文。
Tool：提供与外部世界交互的接口。
- 基础工具抽象。
- 多种专业工具实现。
- 工具结果处理。
Planning：实现任务规划和执行跟踪。
- 计划创建和管理。
- 步骤状态跟踪。
- 动态调整计划。
Flow：管理多个 Agent 的协作。
- 任务分配。
- 结果整合。
- 流程控制。
Prompt：指导 Agent 的行为和决策。
- 系统 Prompt 定义角色。
- 专业 Prompt 指导决策。
- 动态 Prompt 生成。