OpenManus 源码解读

OpenManus采用了清晰的分层架构设计，通过继承关系构建了一个灵活可扩展的智能体框架，本文将自底向上分层解读。

分层解读

BaseAgent

BaseAgent 该类通过继承 ABC，来定义了一个抽象类，提供了智能体的基础框架，包括状态管理、内存管理和执行循环控制。通过 state_context 上下文管理器实现了安全的状态转换，使用 Memory 类管理对话历史。

首先看一下这个类的成员变量：

Name：智能体的名称
Description：描述
System_prompt：系统级 Prompt
Next_step_prompt：用于确定下一个动作的提示词
llm：指定所选用的大模型，openManus 通过 LLM 类封装了各种大模型
Memory：内存管理对象
State：当前的状态，包括：
1. IDLE，空闲
2. RUNNING，运行中
3. FINISHED，结束
4. ERROR，错误
Max_steps：设置最大运行步骤
Current_step：当前执行的步骤

类的一些主要方法：

state_context，用于更新状态
update_memory，更新内存，其实就是向一个 Message List 中 Append 新 Message
Run，执行方法
Step，抽象方法，由子类实现
is_stuck、handle_stuck_state，根据检测重复内容判断是否出现循环阻塞，如果出现的话给大模型新的 prompt：Observed duplicate responses. Consider new strategies and avoid repeating ineffective paths already attempted. 让其解决循环问题。

ReActAgent

OpenManus 实现了基于 ReAct 模式的执行流程，这种实现式使得OpenManus能够像人类一样思考和行动，通过不断迭代来解决复杂问题。特别是在处理多步骤任务时，这种方法显示出了强大的优势。主要包含以下步骤：

初始化，初始化实例和系统 prompt 及可用工具信息
用户输入
思考阶段，思考当前状态，决定下一步行动
行动行动，根据思考选择合适工具执行，获取结果
循环迭代 3-4，直到任务完成或者达到最大运行步骤

ReActAgent 继承了 BaseAgent，也是一个抽象类，主要是提供了两个抽象方法：

Think，对应 ReAct 的思考过程
Act，对用 ReAct 的行动过程
此外还给出了 Step 方法的一个具体实现，其实也非常简单就是对 think 和 act 的调用。

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

ToolCallAgent 赋予了 OpenManus 工具调用的功能，可以解析大模型的工具调用意图，并执行响应的工具。

这一层增加了一些成员变量：

available_tools，可用的工具集合，默认是两个工具，一个用于格式化解析，一个用于结束输出。
tool_choices，工具选择选项：
- Auto，自动选择
- Required，需要指定？
special_tool_names，特殊工具名称集合？
tool_calls，工具调用集合，可以看成函数集合
current_base64_image，用来处理图片
max_observe，最大观察次数

TollCallAgent 还对 ReActAgent 类中定义的抽象方法进行了实现：

async def think(self) -> bool:  
    """Process current state and decide next actions using tools"""  
    # 处理下一步的提示词
    if self.next_step_prompt:  
        user_msg = Message.user_message(self.next_step_prompt)  
        self.messages += [user_msg]  

    try:  
        # Get response with tool options  
        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,  
        )  
    except ValueError:  
        raise  
    except Exception as e:  
        # Check if this is a RetryError containing TokenLimitExceeded  
        if hasattr(e, "__cause__") and isinstance(e.__cause__, TokenLimitExceeded):  
            token_limit_error = e.__cause__  
            logger.error(  
                f"🚨 Token limit error (from RetryError): {token_limit_error}"  
            )  
            self.memory.add_message(  
                Message.assistant_message(  
                    f"Maximum token limit reached, cannot continue execution: {str(token_limit_error)}"  
                )  
            )  
            self.state = AgentState.FINISHED  
            return False  
        raise  
    self.tool_calls = tool_calls = (  
        response.tool_calls if response and response.tool_calls else []  
    )  
    content = response.content if response and response.content else ""  
  
    # Log response info  
    logger.info(f"✨ {self.name}'s thoughts: {content}")  
    logger.info(  
        f"🛠️ {self.name} selected {len(tool_calls) if tool_calls else 0} tools to use"    )  
    if tool_calls:  
        logger.info(  
            f"🧰 Tools being prepared: {[call.function.name for call in tool_calls]}"  
        )  
        logger.info(f"🔧 Tool arguments: {tool_calls[0].function.arguments}")  
  
    try:  
        if response is None:  
            raise RuntimeError("No response received from the LLM")  
  
        # Handle different tool_choices modes  
        if self.tool_choices == ToolChoice.NONE:  
            if tool_calls:  
                logger.warning(  
                    f"🤔 Hmm, {self.name} tried to use tools when they weren't available!"                )  
            if content:  
                self.memory.add_message(Message.assistant_message(content))  
                return True  
            return False  
        # Create and add assistant message  
        assistant_msg = (  
            Message.from_tool_calls(content=content, tool_calls=self.tool_calls)  
            if self.tool_calls  
            else Message.assistant_message(content)  
        )  
        self.memory.add_message(assistant_msg)  
  
        if self.tool_choices == ToolChoice.REQUIRED and not self.tool_calls:  
            return True  # Will be handled in act()  
  
        # For 'auto' mode, continue with content if no commands but content exists        if self.tool_choices == ToolChoice.AUTO and not self.tool_calls:  
            return bool(content)  
  
        return bool(self.tool_calls)  
    except Exception as e:  
        logger.error(f"🚨 Oops! The {self.name}'s thinking process hit a snag: {e}")  
        self.memory.add_message(  
            Message.assistant_message(  
                f"Error encountered while processing: {str(e)}"  
            )  
        )  
        return False

其他模块解读

参考资料

OpenManus源码深度剖析：3小时打造的智能体框架

ReAct