推荐语
深入解析OpenManus的内部结构和功能实现，带你一探智能代理的奥秘。
核心内容：
1. OpenManus概览与目录结构解析
2. 核心Agent类的设计与功能
3. 代码调用逻辑梳理与任务执行流程
杨芳贤
53AI创始人/腾讯云(TVP)最具价值专家
我之前演示过几个OpenManus的demo，其实也就是demo，包括manus，现在也就是demo阶段，复杂的plan和flow，现在的代码支撑和LLM的能力都有待改善，但是我们这期不是吐槽文章，是来把OpenManus给打开看看它的实现是怎么样的，其实Manus也都差不多，甚至OWL也都差不多，我们看一个就够了2- flow 就是来做multiagent的planning和管理任务框架的都是这种形式，来定义系统提示词和agent的role纯纯干活的了
先看base：base.py 模块定义了抽象基类 BaseAgent，用于管理代理的状态、内存、执行循环（包括运行、步骤执行、卡住处理）和消息，并提供初始化和配置功能，为构建具有特定行为的代理提供基础框架。在看planning：planning.py模块定义了PlanningAgent类，该代理通过PlanningTool和Terminate等工具创建、管理和执行任务计划。它具有初始化、计划创建(create_initial_plan)、思考(think)、行动(act)、计划状态更新(update_plan_status)、步骤跟踪(step_execution_tracker)等功能，并能根据工具执行结果动态调整计划，处理初始请求(run)并检索当前计划状态(get_plan)。然后是 react.py:react.py模块定义了继承自BaseAgent的抽象类ReActAgent，它通过think(思考，决定下一步行动) 和act(执行行动) 两个抽象方法（需子类实现）以及step方法（整合think和act）来处理和执行任务，并提供基础的任务处理框架。manus.py:manus.py模块定义了Manus类，一个继承自ToolCallAgent(实际上你前面提到了继承自 PlanningAgent, 请确认是哪个) 的通用智能代理。Manus具有预定义的名称、描述、系统提示、步骤限制(max_observe,max_steps)，并利用包含PythonExecute、WebSearch、BrowserUseTool、FileSaver和Terminate等工具的available_tools集合来执行各种任务，并通过_handle_special_tool方法处理（如清理）BrowserUseTool的结果。
async run(request): if state != IDLE: raise RuntimeError
if request: update_memory(USER, request)
async with state_context(RUNNING): while current_step < max_steps and state != FINISHED: current_step += 1 step_result = await step() # 调用子类的 step() 方法 if is_stuck(): handle_stuck_state()
results.append(step_result)
if current_step >= max_steps: state = IDLE results.append("Terminated: Reached max steps")
return "\n".join(results)
async step(): should_act = await think() # 调用子类的 think() if should_act: return await act() # 调用子类的 act() else: return "Thinking complete - no action needed"
async run(request): if request: await create_initial_plan(request) return await super().run()
async create_initial_plan(request): messages = [...] response = await llm.ask_tool(..., tool_choice=ToolChoice.AUTO)
for tool_call in response.tool_calls: if tool_call.function.name == "planning": result = await execute_tool(tool_call) update_memory(TOOL, result, tool_call_id=tool_call.id)
async think(): prompt = f"CURRENT PLAN STATUS:\n{await self.get_plan()}\n\n{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: latest_tool_call = self.tool_calls[0] if latest_tool_call 不是 planning tool 且 不是 special tool: self.step_execution_tracker[latest_tool_call.id] = { "step_index": self.current_step_index, "tool_name": latest_tool_call.function.name, "status": "pending" } return result
async act(): result = await super().act() if self.tool_calls: latest_tool_call = self.tool_calls[0] if latest_tool_call.id in self.step_execution_tracker: self.step_execution_tracker[latest_tool_call.id]["status"] = "completed" self.step_execution_tracker[latest_tool_call.id]["result"] = result
if latest_tool_call 不是 planning tool 且 不是 special tool: await self.update_plan_status(latest_tool_call.id) return result
async update_plan_status(tool_call_id):
async _get_current_step_index():
async think(): self.working_dir = await self.bash.execute("pwd") # 获取当前工作目录 self.next_step_prompt = self.next_step_prompt.format(current_dir=self.working_dir) # 更新提示 return await super().think() # 调用 ToolCallAgent 的 think()
好了今天这节课就解释到这，大家可以结合我的解释自己去run一下代码，甚至自己按着这个逻辑来新写一套multi-agents的框架也不是特别难的事
粤ICP备14082021号
免费POC，
零成本试错
