OpenManus 是由 MetaGPT 团队孵化的开源通用 AI 智能体框架，核心目标是无需邀请码即可实现任意创意的智能体开发，支持多工具扩展、多智能体协作、强化学习（RL）调优（OpenManus-RL），并兼容 Google A2A 协议生态。
项目的定位是通用LLM智能体开发框架，支持远程和本地的工具调用。
项目代码的仓库地址：https://github.com/FoundationAgents/OpenManus
整个项目使用分层架构设计
交互层： （终端输入，A2A Client， HuggingFace Demo） 接受用户任务输入，输出执行结果
启动层： （main.py run_mcp.py run_flow.py） 提供不同运行模式，单智能体，MCP工具，多智能体流程版
流程调度层： （多智能体规划 PlanningFlow） 任务规划，多智能体调度，步骤状态管理
智能体核心层： （Manus（通用智能体），DataAnalysis（专用智能体）） 通用智能体核心，管理工具集，MCP连接，浏览器上下文，决策逻辑
工具/协议扩展层： （本地工具，MCP远程工具， A2A协议工具） 内置本地工具，远程MCP工具，A2A协议工具
One line for run OpenManus:
python main.py
Then input your idea via terminal!
For MCP tool version, you can run:
python run_mcp.py
For unstable multi-agent version, you also can run:
python run_flow.py
代码在 app/agent/manus.py 中，核心类是 Manus,负责管理工具集，MCP连接，浏览器上下文，决策逻辑
关键的方法是：
Manus 继承自 ToolCallAgent， 代码在 app/agent/toolcall.py 中， ToolCallAgent 是一个通用的智能体框架，提供了一些基础的工具调用功能，如调用工具、管理工具集等。Manus 在 ToolCallAgent 的基础上，添加了更多的功能，如管理浏览器上下文、管理MCP连接等。
关键的方法有：
ToolCallAgent 继承自 ReActAgent, ReActAgent的代码在 中，ReActAgent是很简单的基础类，包含了三个方法，分别是 think，act，step，其中think 和 act 方法都是没有实现的，step中简单构建了整个框架，先进行think，根据返回内容看是否需要act，或者进行 act ，或者直接返回think的结果。
基础的Agent类，BaseAgent 类，代码是在 app/agent/base.py
BaseAgent类的有几个核心的属性：
浏览器上下文Agent
结合browser和Agent类，记录当前的Agent的state和Memory的信息，共同进行下一步的规划。
数据分析Agent
代码是 app/agent/data_analysis.py 中，仅仅包含了工具集合为：NormalPythonExecute, VisualizationPrepare, DataVisualization, Terminate四类工具
代码编程 SWEAgent
代码在 app/agent/swe.py 下，包含了自动化编码的prompt，和相关的工具集合，bash，StrReplaceEditor，Terminate三个工具。
流程的代码在 app.flow下，有三个文件，base.py flow_factory.py planning.py
base.py 中定义了 BaseFlow 类，是所有流程的基类，包含了简单地管理Agent的方法等。
如 属性primary_agent_key 是多个agent中的主agent，
还有 get_agent， add_agent， execute 方法，来管理agent。其中execute方法未定义。
flow_factory.py 中定义了 FlowFactory 类，是流程工厂类，负责创建流程。目前只支持一个流程，PlanningFlow。
planning.py 中定义了核心类 PlanningFlow 类，是流程的具体实现类，继承自 BaseFlow。
具有的属性有：
核心方法只有一个，就是 execute方法，该方法会根据当前状态，在多个Agent上执行不同的操作，如调用工具、更新状态等。
核心代码为
async def execute(self, input_text: str) -> str:
"""Execute the planning flow with agents."""
try:
if not self.primary_agent:
raise ValueError("No primary agent available")
# Create initial plan if input provided # 如果提供了输入，则说明是新的任务或者有新内容增加，则需要重新制定执行计划。
if input_text:
await self._create_initial_plan(input_text)
# Verify plan was created successfully
if self.active_plan_id not in self.planning_tool.plans:
logger.error(
f"Plan creation failed. Plan ID {self.active_plan_id} not found in planning tool."
)
return f"Failed to create plan for: {input_text}"
result = ""
while True:
# Get current step to execute
self.current_step_index, step_info = await self._get_current_step_info()
# Exit if no more steps or plan completed
if self.current_step_index is None:
result += await self._finalize_plan()
break
# Execute current step with appropriate agent
step_type = step_info.get("type") if step_info else None
executor = self.get_executor(step_type)
step_result = await self._execute_step(executor, step_info)
result += step_result + "\n"
# Check if agent wants to terminate
if hasattr(executor, "state") and executor.state == AgentState.FINISHED:
break
该方法的主要的做的内容是
方法 def _create_initial_plan 将要prompt转为Message，并带有工具调用去chat LLM，得到tool_calls后
PlanningFlow 在execute 中的最常调用的是 PlanningTool 这个类的 execute 函数。
核心函数代码为：
def execute(self, *, command: Literal["create", "update", "list", "get", "set_active", "mark_step", "delete"],
plan_id: Optional[str] = None,
title: Optional[str] = None,
steps: Optional[List[str]] = None,
step_index: Optional[int] = None,
step_status: Optional[Literal["not_started", "in_progress", "completed", "blocked"]] = None,
step_notes: Optional[str] = None,
**kwargs,
):
"""
Execute the planning tool with the given command and parameters.
Parameters:
- command: The operation to perform
- plan_id: Unique identifier for the plan
- title: Title for the plan (used with create command)
- steps: List of steps for the plan (used with create command)
- step_index: Index of the step to update (used with mark_step command)
- step_status: Status to set for a step (used with mark_step command)
- step_notes: Additional notes for a step (used with mark_step command)
"""
if command == "create":
return self._create_plan(plan_id, title, steps)
elif command == "update":
return self._update_plan(plan_id, title, steps)
elif command == "list":
return self._list_plans()
elif command == "get":
return self._get_plan(plan_id)
elif command == "set_active":
return self._set_active_plan(plan_id)
elif command == "mark_step":
return self._mark_step(plan_id, step_index, step_status, step_notes)
elif command == "delete":
return self._delete_plan(plan_id)
else:
raise ToolError()
这个类主要在管理 plans 格式为dict{dict_name: plan_info_dict}的形式。
代码在 app/llm.py 下，使用单例模式来实现，每一个配置名只会创建一次的实例。
LLM类会首先进行tokenier 的初始化
代码为：self.tokenizer = tiktoken.encoding_for_model(self.model)
根据指定的模型名称返回对应的预训练分词器。不在集合中的模型，则统一使用 cl100k_base 预训练模型。
LLM类的几个关键方法
def ask(self,
messages: List[Union[dict, Message]],
system_msgs: Optional[List[Union[dict, Message]]] = None,
stream: bool = True,
temperature: Optional[float] = None,
) -> str:
发送一段Prompt给LLM以返回 response。
先将List[Message]的列表格式中的转为 List[dict{'role':'', 'content':''}]的格式。
然后，计算总的token的长度，判断是否需要满足不超过最大token数量。
最后，调用LLM的chat，分streaming 和 not stream的方式分别处理。
方法
def ask_with_images(self,
messages: List[Union[dict, Message]],
images: List[Union[str, dict]],
system_msgs: Optional[List[Union[dict, Message]]] = None,
stream: bool = False,
temperature: Optional[float] = None,
) -> str
方法
def ask_tool(self,
messages: List[Union[dict, Message]],
system_msgs: Optional[List[Union[dict, Message]]] = None,
timeout: int = 300,
tools: Optional[List[dict]] = None,
tool_choice: TOOL_CHOICE_TYPE = ToolChoice.AUTO, # type: ignore
temperature: Optional[float] = None,
** kwargs,
) -> ChatCompletionMessage | None
这两个方法的实现都类似 chat，只是增加对image和工具调用的支持。
还有一个有意思的类是 TokenCounter 类，负责计算token的数量。基于上面创建的tokenizer对象，来计算每一个文本的token，TokenCounter实现封装，可以计算纯文本的，也可以计算带图片的，还支持本项目的Message格式的token数量的计算。
该模块的核心类是SandboxManus类。
代码是在 app/agent/sandbox_agent.py 上。一个多功能代理，可以使用多个沙盒工具（包括基于MCP的工具）解决各种任务
该类的核心方法有几个
create方法，根据输入的参数进行 SandboxManus的创建。只有三个步骤：
该模式是在 app/sandbox 文件夹下，核心类是 LocalSandboxClient ，代码在 app/sandbox/client.py 下
类LocalSandboxClient下有很多关键的方法：
DockerSandbox类的解析
初始化该类的时候传入两个参数，config和volume_bindings，其中config是Sandbox的配置，volume_bindings是格式为{host_path: container_path}的卷映射
asyncio.to_thread(self.container.get_archive, resolved_path)
然后从该对象中读取tar流，并使用临时文件保存下该tar流，然后一次性 tar.extractfile 得到文件内容。
await asyncio.to_thread(self.container.put_archive, parent_dir or "/", tar_stream)
AsyncDockerizedTerminal类的解析
主要是维护一个DockerSession，通过init函数初始化好Terminal的session，以传递命令到docker中。
负责在client和Docker之间传输命令
SandboxManager类
代码是在 app/sandbox/core/manager.py ，该类主要负责管理多个SandBox，
启动脚本是根目录下的 run_mcp.py，其中核心是 MCPRunner 类
MCPRunner类负责封装 MCPAgent，提供了三个核心方法：
MCPAgent 是核心类，继承于ToolCallAgent
代码在app/prompt文件夹。该文件夹下存放的是各个模块的prompt。
在browser.py, manus.py, mcp.py, planning.py, swe.py, toolcall.py, visualization.py 下，有 SYSTEM_PROMPT, NEXT_STEP_PROMPT 两个提示词。
整个项目的核心在于这些提示词的质量。
所有的工具集合都在这个文件夹 app/tool 下。
所有工具类的基础类是 BaseTool，用于组合BaseModel和工具功能的所有工具
AskHuman类，代码在 app/tool/ask_human.py 中，让LLM在信息不够详细的时候需求使用人类的介入
网络搜索工具，代码在 app/tool/search 文件夹下，含有多个不同的网络搜索，baidu,bing,google,duckduckgo
SandBox工具
基于daytona来实现SandBox的管理
bash工具，直接在项目本地进行bash操作，安全性较差的方式
通过 asyncio.create_subprocess_shell 的方式完成
浏览器使用工具 browser_use_tool.py ，基于Python的browser_use library库来实现
文件操作工具 file_operators.py
mcp服务封装 mcp.py
任务规划和计划管理模块 planning.py
Python代码执行模型，python_execute.py 直接开启线程调用 exec 来执行代码并返回
工具收集模块 tool_collection.py 收集和管理所有的工具
网络搜索模块 web_search.py 管理所有的搜索引擎，并设置不同的搜索方式。
OpenManus相比最近火热的OpenClaw来说，仅仅是和LLM交互的仍然为prompt，后续的Skills也就是prompt的另一种表现形式。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。