MetaGPT认知架构实现：感知、思考与行动循环的底层原理与实战定制

MetaGPT认知架构实现：感知、思考与行动循环

>摘要：MetaGPT的Role类不仅是简单的Agent封装，而是一个完整的认知架构实现。本文深入源码级别解析_observe、_think、_act三大核心方法的协作机制，揭秘ReAct模式的底层实现原理，并提供三种React模式的实战对比与自定义认知流的完整开发指南。

一、从ReAct到认知架构：Agent的智能跃迁

大型语言模型（LLM）驱动的Agent正在从"问答机器"向"自主智能体"演进。MetaGPT框架通过标准化认知架构（Standardized Cognitive Architecture），将人类的感知-思考-行动循环（Observe-Think-Act Loop）编码为可复用的软件范式[³]。

与简单的Prompt链不同，MetaGPT的Role类实现了状态ful的持续交互：

• 感知（Observe）：主动筛选环境信息，建立情境意识
• 思考（Think）：基于记忆与目标进行决策，选择行动策略
• 行动（Act）：执行具体操作，改变环境状态并生成新记忆

+-------------+ +-------------------+ +-------------------+ | Environment | | Role | | Memory | | (环境) | | (认知中枢) | | (记忆存储) | +------+------+ +---------+---------+ +---------+---------+ | | | | ① Publish Message | | |------------------------->| | | | ② _observe() | | | - 消息过滤(watch) | | | - 记忆检索 | | | | | | ③ _think() | | | - 决策逻辑 | | | - Action选择 | | | | | | ④ _act() | | | - 执行todo | | | - 结果封装 | | ⑤ 新消息广播 | | |<-------------------------| | | | ⑥ 记忆更新 | | |-------------------------->|

二、Role内部机制解剖：三大核心阶段

MetaGPT的Role类通过run()方法驱动认知循环，其核心调用链为：run()→react()→_observe()→_think()→_act()→publish_message()[³][¹⁰]。

2.1 _observe()：感知过滤与注意力机制

_observe()方法负责从环境中获取并筛选信息，相当于智能体的感官系统。

源码级工作流程：

asyncdef_observe(self)->bool:"""观察环境，返回是否有重要消息需要处理"""# 1. 从环境获取最新新闻（所有广播的消息）news=self.rc.env.get_news()# 2. 基于_watch列表进行过滤（注意力机制）filtered_news=[msgformsginnewsifmsg.cause_byinself.rc.watch# 只关注特定Action产生的消息]# 3. 存储至短期记忆（Memory）formsginfiltered_news:self.rc.memory.add(msg)# 4. 返回是否触发后续思考returnlen(filtered_news)>0

关键设计：

• 订阅制消息过滤：通过_watch()方法声明关注的Action类型，避免信息过载
• 记忆即时持久化：观察到的消息立即写入RoleContext的Memory，供后续阶段检索

实战示例：自定义观察逻辑实现敏感信息拦截

frommetagpt.rolesimportRolefrommetagpt.schemaimportMessagefromtypingimportListclassSecureObserver(Role):def__init__(self,**kwargs):super().__init__(**kwargs)self._watch([WriteCode,ReviewCode])# 标准关注列表asyncdef_observe(self)->bool:# 调用父类标准观察逻辑news=self.rc.env.get_news()# 扩展：敏感信息过滤（如API密钥）sensitive_keywords=["api_key","password","secret"]filtered=[]formsginnews:ifmsg.cause_byinself.rc.watch:# 安全检查ifany(kwinmsg.contentforkwinsensitive_keywords):msg.content="[FILTERED-SENSITIVE-DATA]"filtered.append(msg)self.rc.memory.add(msg)returnlen(filtered)>0

2.2 _think()：决策引擎与Action选择

_think()是认知架构的决策中枢，负责从候选Action中选择下一步执行项（设置为rc.todo）。
标准ReAct模式的实现：

asyncdef_think(self)->None:"""思考阶段：动态选择Action"""# 1. 构建决策提示词（包含可用Action列表与历史记忆）prompt=self._construct_decision_prompt()# 2. 调用LLM进行推理response=awaitself.llm.aask(prompt)# 3. 解析选择的Actionselected_action=self._parse_action(response)# 4. 设置待办事项（驱动_act阶段）self.rc.todo=selected_action self.rc.state=self._get_action_index(selected_action)

决策Prompt模板解析：

你是一名{profile}，目标是{goal}。 可选动作： {action_descriptions} 历史上下文： {memory_context} 当前任务：{current_task} 请从可选动作中选择一个最合适的，并说明理由。格式： Action: [动作名称] Reason: [选择理由]

2.3 _act()：执行与反馈封装

_act()是认知循环的执行终端，负责运行选定的Action并处理输出。
标准实现与自定义扩展：

asyncdef_act(self)->Message:"""执行todo（Action）并封装结果"""# 获取当前待办Actiontodo=self.rc.todo# 准备输入：从历史记忆中获取上下文context=self.get_memories(k=1)[0].content# 取最近1条记忆# 执行Actionresult=awaittodo.run(context)# 封装为Message（标准化输出格式）msg=Message(content=result,role=self.profile,cause_by=type(todo),# 标记由哪个Action产生sent_from=self.name,send_to=self.rc.envifself.is_publicelse{self.name})# 写入自身记忆（供后续循环使用）self.rc.memory.add(msg)returnmsg

三、React模式深度解析：三种认知策略

MetaGPT提供三种react_mode，决定_think与_act的编排方式：

3.1 标准ReAct模式（react）

循环机制：

第1轮: _think(选择Action1) → _act(执行Action1) → 产生Observation1 ↑ ↓ 第2轮: _think(基于Observation1选择Action2) ← _observe(获取Observation1) ↓ _act(执行Action2) → ... → 达到max_react_loop或LLM决定停止

代码配置：

classReActAgent(Role):def__init__(self,**kwargs):super().__init__(**kwargs)self.set_actions([Analyze,Plan,Execute,Verify])# 启用ReAct模式，最多循环6次（3轮Think-Act）self._set_react_mode(react_mode="react",max_react_loop=6)asyncdef_think(self)->None:"""可重写以实现更复杂的决策逻辑，如工具选择"""# 调用父类标准ReAct决策awaitsuper()._think()# 扩展：记录决策路径用于后续分析self.decision_log.append({"step":self.rc.react_loop,"action":self.rc.todo,"memory_count":len(self.rc.memory.get())})

3.2 顺序执行模式（by_order）

适用于标准作业程序（SOP）明确的场景，如代码生成→测试→修复的固定流程。

+------------+ +------------+ +------------+ +------------+ | WriteCode | --> | RunCode | --> | DebugCode | --> | SubmitCode | | (Action1)| | (Action2) | | (Action3) | | (Action4) | +------------+ +------------+ +------------+ +------------+

实现代码：

classCodePipeline(Role):def__init__(self,**kwargs):super().__init__(**kwargs)# 严格按列表顺序执行self.set_actions([WriteCode,RunCode,DebugCode,SubmitCode])self._set_react_mode(react_mode="by_order")asyncdef_act(self)->Message:"""保留标准执行逻辑，但可添加步骤间数据传递"""todo=self.rc.todo step_idx=self.rc.state# 当前执行到第几步# 获取前一步的结果作为输入（除第一步外）ifstep_idx>0:prev_result=self.get_memories(k=1)[0].content context=f"前一步结果：{prev_result}\n当前任务："else:context=self.get_memories(k=1)[0].content result=awaittodo.run(context)# 步骤特定处理：如RunCode失败时跳过SubmitCodeiftodo.name=="RunCode"and"Error"inresult:self.rc.state=2# 跳转到DebugCodemsg=Message(content=result,role=self.profile,cause_by=type(todo))self.rc.memory.add(msg)returnmsg

3.3 三种模式性能对比

在"编写并测试Python函数"任务上的测试数据：

四、自定义认知流：重写三大方法实现业务逻辑

通过继承Role并重写认知方法，可实现特定领域的认知策略。

4.1 场景一：条件触发观察（Conditional Observation）

仅在满足特定条件时才进入思考阶段：

classTriggeredAnalyst(Role):def__init__(self,**kwargs):super().__init__(**kwargs)self._watch([DataReport])self.trigger_keywords=["error","failure","critical"]asyncdef_observe(self)->bool:"""仅当检测到异常关键词时才激活"""news=self.rc.env.get_news()critical_msgs=[]formsginnews:ifmsg.cause_byinself.rc.watch:self.rc.memory.add(msg)# 全部存入记忆# 触发条件检查ifany(kwinmsg.content.lower()forkwinself.trigger_keywords):critical_msgs.append(msg)msg.urgent=True# 标记紧急# 只有紧急消息才触发_thinkreturnlen(critical_msgs)>0

4.2 场景二：多路径决策树（Custom Think Strategy）

在_think中实现复杂的业务规则：

classApprovalManager(Role):asyncdef_think(self)->None:"""基于金额和角色的审批路由逻辑"""last_msg=self.get_memories(k=1)[0]request=self.parse_request(last_msg)# 业务规则引擎ifrequest.amount<1000:self.rc.todo=SelfApprove()# 小额自审批elifrequest.amount<10000:self.rc.todo=ManagerApprove()# 经理审批else:self.rc.todo=BoardApprove()# 董事会审批self.rc.state=self._get_action_index(self.rc.todo)# 记录决策审计日志logger.info(f"审批路由决策：金额{request.amount}->{self.rc.todo.name}")

4.3 场景三：批量执行与结果聚合（Custom Act Pattern）

在_act中实现批处理逻辑：

classBatchProcessor(Role):def__init__(self,**kwargs):super().__init__(**kwargs)self.set_actions([ProcessSingleItem])self.batch_size=5asyncdef_act(self)->Message:"""批量处理多条消息，而非单条"""# 获取待处理的批次pending_items=self.get_memories(k=self.batch_size)# 并行执行tasks=[self.rc.todo.run(item.content)foriteminpending_items]results=awaitasyncio.gather(*tasks,return_exceptions=True)# 聚合结果success_count=sum(1forrinresultsifnotisinstance(r,Exception))summary=f"批次处理完成：{success_count}/{len(results)}成功"# 生成聚合消息msg=Message(content=summary,role=self.profile,cause_by=type(self.rc.todo),meta={"batch_results":results}# 详细结果存入meta)self.rc.memory.add(msg)returnmsg

4.4 场景四：认知循环干预（React Hook）

在标准react流程中插入自定义钩子：

classMonitoredAgent(Role):asyncdefreact(self)->Message:"""包裹标准react流程，添加监控与干预"""logger.info(f"=== 开始认知循环 #{self.rc.react_loop}===")# Pre-observe: 检查资源限制ifself.rc.react_loop>=self.max_react_loop-1:logger.warning("接近最大循环次数，准备强制终止")# 调用标准react流程msg=awaitsuper().react()# Post-act: 结果质量检查ifself._is_low_quality(msg):logger.warning("输出质量过低，触发自我修正")self.rc.memory.add(Message(content="之前的结果质量不佳，请重新思考",role="system"))# 强制再执行一轮修正msg=awaitself.react()returnmsgdef_is_low_quality(self,msg:Message)->bool:"""质量评估启发式规则"""returnlen(msg.content)<50or"error"inmsg.content.lower()

五、认知架构的调试与优化

5.1 认知路径可视化

通过重写方法添加调试输出：

asyncdef_think(self)->None:logger.debug(f"[Think] 当前记忆数量:{len(self.rc.memory.get())}")logger.debug(f"[Think] 可用Actions:{[a.nameforainself.actions]}")awaitsuper()._think()logger.info(f"[Think] 决策结果:{self.rc.todo.nameifself.rc.todoelse'None'}")

5.2 避免常见陷阱

六、总结与展望

MetaGPT的认知架构通过_observe、_think、_act的标准化抽象，为Multi-Agent系统提供了可扩展的智能体开发框架：

• 感知层的过滤机制确保Agent专注于相关信息
• 决策层的ReAct模式支持从简单顺序执行到复杂动态规划
• 执行层的标准化Message封装保证多Agent协作的互操作性
  通过重写这三个核心方法，开发者可以实现：
• 领域特定的感知策略（如异常检测、敏感信息过滤）
• 业务规则驱动的决策逻辑（如审批流、路由策略）
• 批处理与异步执行（提升吞吐量的自定义_act）

本文章基于metaGPT 官方文档。仅供学习参考，请勿用于商业用途。