OpenManus 源码解析

运行方式

官方文档中，给出了三种运行方式，分别是：

一行命令运行 OpenManus：

python main.py

如需使用 MCP 工具版本，可运行：

python run_mcp.py

如需体验不稳定的多智能体版本，可运行：

python run_flow.py

代码关系图

先看代码之间的继承关系，有助于理解后续的各种模式的实现。

其中，绿色的为未实现的部分，蓝色的为已经实现的部分，可以看出 OpenManus 是想使用 ReAct 和 COT 两种增强模型推理的能力，但 COT 目前暂未完善，红色部分为关键函数

主要的关键函数为 run，think，act。这几个函数的大致作用如下：

run: 执行整个流程，并且为了防止一直执行下去，判断是否达到最大的步数（step）

step: 执行 think 和 act 的过程称为一个 step

think: 调用大模型的过程，将任务和工具给到大模型，让其选择是否需要工具来完成本次任务

act: 一个工具的执行叫做一个 action

针对于每个 Agent 之间的 think，run，act 之间有什么细微区别，可以在后面运行模式中详细说明。

目前时间为 2025-04-01，后续代码可能会有变动

COT 和 ReAct 的区别：

特性	Chain-of-Thought (COT)	ReAct
推理方式	内部推理，仅依赖模型自身的知识和逻辑推导	内部推理 + 外部行动，允许与环境交互
外部依赖	无外部依赖	可以依赖外部工具或知识库
适用场景	需要多步逻辑推理的任务	需要动态决策和与环境交互的任务
输出形式	显式的推理链条	推理链条 + 行动记录
灵活性	固定在模型内部，无法动态获取新信息	动态性强，可根据外部反馈调整策略

COT 举例：

问题: 如果一个房间有 3 张桌子，每张桌子上有 4 把椅子，总共有多少把椅子？
推理: 每张桌子有 4 把椅子，一共有 3 张桌子，所以总数是 3 × 4 = 12。
答案: 12 把椅子。

ReAct 举例：

问题: 北京的天气如何？
推理: 我需要查询北京的实时天气信息。
行动: 查询天气 API。
结果: 北京当前温度为 20°C，晴天。
答案: 北京当前温度为 20°C，晴天。

Manus 模式

直接实例化 Manus 对象，运行了 BaseAgent 中的 run 方法，run 方法中，判断当前 Agent 的状态是否已经完成，并且步数是否已经达到最大步数(默认为 10)，如果二者满足其一，那么其实整个 Agent 就已经结束或者该结束了，不会执行下面的逻辑

async def run(self, request: Optional[str] = None) -> str:
       results: List[str] = []
       async with self.state_context(AgentState.RUNNING):
           # 判断当前 Agent 的状态是否已经完成，并且步数是否已经达到最大步数(默认为 10)
           while (self.current_step < self.max_steps and self.state != AgentState.FINISHED):
               # 步数+1
               self.current_step += 1
               # 执行一次 step
               step_result = await self.step()
               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})")

执行每一个 step（COT 和 ReAct 中都有 step 的实现，因为这里是Manus 模式，所以执行的是 ReActAgent 中的 step），step 中的逻辑也很简单，就是先进行一次模型的调用，然后判断是否调用工具

async def step(self) -> str:
    # 先执行 think
    should_act = await self.think()
    if not should_act:
        return "Thinking complete - no action needed"
    # 如果需要执行 act，那么再调用 act
    return await self.act()

think 执行的过程就是去调用大模型的过程

async def think(self) -> bool:
    try:
        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 = tool_calls = (
        response.tool_calls if response and response.tool_calls else []
    )

因为 Manus 是继承自 BrowserAgent 的，二者中都有 think 方法，但其中的 think 方法都在为生成一个叫 next_step_prompt 的变量做准备，这是将当前的浏览器窗口信息传递到模型中，然后让大模型做出后续决定的 prompt，prompt 翻译后如下所示：

为了实现目标，我接下来应该做什么？

当你看到[Current state starts here]时，请关注以下内容：

• 当前 URL 和页面标题{url_placeholder}
• 可用标签页{tabs_placeholder}
• 交互元素及其索引
• 视口上方{content_above_placeholder}或下方{content_below_placeholder}的内容（如已标注）
• 任何操作结果或错误{results_placeholder}

浏览器交互操作：

• 导航跳转：使用 browser_use 并设置 action=”go_to_url”, url=”…”
• 点击元素：使用 browser_use 并设置 action=”click_element”, index=N
• 输入文本：使用 browser_use 并设置 action=”input_text”, index=N, text=”…”
• 提取操作：使用浏览器功能，动作属性为“extract_content”，目标为“…”
• 滚动操作：使用浏览器功能，动作属性为“scroll_down”或“scroll_up”

同时考虑当前可见内容及视口之外可能存在的部分。
保持条理性——记住你的进度和目前已掌握的信息。

act 的执行过程，就是调用每个工具的过程了

async def act(self) -> str:
    results = []
    # 循环调用 think 出来的工具
    for command in self.tool_calls:
        result = await self.execute_tool(command)
        ......

MCP 模式

MCP 的运行流程和 Manus 的流程是一样的，只不过工具的来源是 MCP，并且 MCP 的 think 中没有了 next_step_prompt 这个 prompt

async def _initialize_and_list_tools(self) -> None:
    # 获取 MCP Server 的工具列表
    response = await self.session.list_tools()

    for tool in response.tools:
        server_tool = MCPClientTool(
            name=tool.name,
            description=tool.description,
            parameters=tool.inputSchema,
            session=self.session,
        )
        self.tool_map[tool.name] = server_tool
    self.tools = tuple(self.tool_map.values())

而 MCP Server 的 run 方法会注册支持的所有工具

class MCPServer:
def __init__(self, name: str = "openmanus"):
    self.tools["bash"] = Bash()
    self.tools["browser"] = BrowserUseTool()
    self.tools["editor"] = StrReplaceEditor()
    self.tools["terminate"] = Terminate()

def register_all_tools(self) -> None:
    # 注册所有工具
    for tool in self.tools.values():
        self.register_tool(tool)

def run(self, transport: str = "stdio") -> None:
    """Run the MCP server."""
    self.register_all_tools()
    ......

Flow 模式

Flow 模式也是未完成的状态，目前仅支持 PlanningFlow，原理是通过 prompt，将 planning_tool 作为工具描述给大模型，然后大模型做出相应的 plan（planning_tool 中提供了create，update 等操作）, System Prompt 如下：

system message:
你是一名规划助手。制定一个简洁、可操作的计划，包含清晰的步骤。
重点关注关键里程碑，而非详细的子步骤。
以清晰和高效为优化目标。

user message:
制定一个合理的计划，包含清晰的步骤以完成任务：{request}

得到一个 plan 之后，会去循环执行 plan 的每一步，其中每一步都会通过大模型然后选择出适当的工具来执行，prompt 如下，这里就和之前的逻辑一样了，会去执行每一个 step，然后执行 think 和 act。

当前计划状态：
{plan_status}

你的当前任务：
你现在正在处理第 {self.current_step_index} 步：“{step_text}”

请使用适当的工具执行此步骤。完成后，请提供你所完成内容的简要总结。

内置的工具列表

工具类别	具体名称	功能描述
WebSearch	GoogleSearchEngine，BaiduSearchEngine，DuckDuckGoSearchEngine，BingSearchEngine	执行网络搜索并返回相关链接的列表。这个函数尝试使用搜索引擎API来获取最新的结果。如果出现错误，它会返回到另一个搜索引擎。
FileSaver	FileSaver	将内容保存到指定路径的本地文件中。当您需要将文本、代码或生成的内容保存到本地文件系统上的文件中时，可以使用此工具。该工具接受内容和文件路径，并将内容保存到该位置。
StrReplaceEditor	StrReplaceEditor	支持沙盒功能的文件查看、创建和编辑工具。
CreateChatCompletion	CreateChatCompletion	创建一个具有指定输出格式的结构化内容
BrowserUseTool	BrowserUseTool	浏览器操作
PythonExecute	PythonExecute	安全的执行 Python 代码
Terminal	Terminal	执行系统命令
Terminate	Terminate	当无法继续运行或者完成任务时调用
PlanningTool	PlanningTool	一种规划工具，允许 Agent 创建和管理用于解决复杂任务的计划。该工具提供了创建计划、更新计划步骤以及跟踪进度的功能。
MCPClientTool	MCPClientTool	MCP 客户端工具

附件

代码关系图：

https://fastly.jsdelivr.net/gh/rexyan/warehouse@master/openmanus.drawio

https://fastly.jsdelivr.net/gh/rexyan/warehouse@master/openmanus-flow%20(4).drawio