跳到主要内容

Role:观察-思考-行动的智能体原子

30 秒导读: MetaGPT 把「一个智能体」抽象成 Role。它有自己的私有收件箱、自己的记忆、一串能做的动作(Action),每被驱动一次就走一遍 观察 → 思考 → 行动:看有没有该我处理的新消息,想这轮该做哪个动作,做完把结果作为一条消息发出去。整个 MetaGPT(产品经理、架构师、工程师协作写代码)都是许多这样的原子拼起来的。本章只讲单个原子怎么转。

1. 这是什么(零基础也能懂)​

一句话定义: Role 是 MetaGPT 里「一个会干活的智能体」的基类——一个自带收件箱、记忆和技能清单,靠「观察-思考-行动」循环推进工作的对象。

它在解决什么问题。 你想让多个 AI 分工协作(一个写需求、一个画架构、一个写代码),就得先回答一个更小的问题:单个 AI 到底怎样「自己动起来」? 它怎么知道现在轮到它了?轮到它时该做哪件事?做完给谁?Role 就是这个「单体智能体」的标准答案。

它是全框架的原子。 后面几章讲的 Environment 消息总线(第 3 章)、经典 SOP 流水线(第 4 章)、MGX 动态调度(第 5 章),本质都是把很多个 Role 摆在一起、让它们的消息互相喂。所以看懂一个 Role 怎么转,是看懂全框架的前提。

用起来什么样。 自定义一个角色,通常只需要:声明身份、给它几个动作、告诉它「关注谁产出的消息」。下面是一段最小示意:

# 示意,非源码:一个最小自定义 Role 的样子
class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "Engineer"          # 身份:会拼进 system prompt

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WriteCode])  # 技能清单:它能做的 Action
        self._watch([WriteDesign])     # 只关心「架构师产出的设计」这类消息

# 驱动一次:喂一条消息进它的收件箱,它自己走完 observe→think→act
msg = await SimpleCoder().run(with_message="给我实现登录功能")

一句话直觉。 把 Role 想成一个流水线上的工人:面前有个传送带(收件箱),他只捡「贴了我关心的标签」的零件(watch 过滤);捡到就动手(选一个动作做),做完把成品放回传送带(广播消息),供下一个工人捡。没有该我捡的零件时,他就闲着等(is_idle)。

本节到此不碰代码细节。记住一句话:Role = 一个「看消息 → 决定做什么 → 做 → 发消息」的自转小机器。

2. 顶层全景(它大概怎么转)​

2.1 一次驱动的主线​

外界(或环境)每调用一次 Role.run(),这个角色就完整走一遍循环。怎么读下面这张图:从上到下是一次 run 的时间顺序,中途任何一步没料就提前返回。

                 run(with_message)              role.py:529  @role_raise_decorator
                        │
                        ▼
        put_message()  把消息塞进私有收件箱        role.py:448
                        │
                        ▼
        ┌───────────────────────────────┐
        │  _observe  观察                │  role.py:399
        │  从收件箱捞新消息,按 watch/    │
        │  send_to 过滤,存进 memory      │
        └───────────────────────────────┘
                        │
              有该我处理的新消息?
              ├── 否 ─► 记日志「no news」直接 return(闲置等待)   role.py:543
              └── 是
                        ▼
        ┌───────────────────────────────┐
        │  react  反应(三种模式之一)    │  role.py:512
        │   ┌─────────────────────────┐ │
        │   │ _think 想:这轮做哪个动作 │ │  role.py:340
        │   │ _act   做:跑 Action     │ │  role.py:381
        │   └─────────────────────────┘ │
        └───────────────────────────────┘
                        │  产出一条 AIMessage
                        ▼
        publish_message()  广播给环境/自己          role.py:429
                        │
                        ▼
                     返回该消息

2.2 部件一句话职责​

一次循环牵动的部件(都定义在 role.py,数据容器 RoleContext 是核心):

部件干什么位置
Role.run单次驱动入口:收消息 → 观察 → 反应 → 广播role.py:529
Role._observe从私有收件箱过滤出「该我处理」的新消息进记忆role.py:399
Role.react按反应模式分派到 _react / _plan_and_actrole.py:512
Role._think状态机:决定这一轮的 state 和 todo(下一个动作)role.py:340
Role._act跑 todo 这个 Action,把结果包成 AIMessagerole.py:381
RoleContext角色的全部运行时状态:收件箱/记忆/watch/state/todorole.py:92
RoleReactMode三种反应模式的枚举:REACT / BY_ORDER / PLAN_AND_ACTrole.py:82

2.3 状态住在哪:RoleContext​

Role 本身存的是「静态设定」(名字、身份、动作清单);而「这轮跑到哪了」这类易变运行时状态全塞进一个叫 rc(RoleContext)的子对象。RoleContext 定义在 role.py:92,几个字段是本章的主角:

字段含义定义
msg_buffer私有收件箱:一个异步消息队列,别人发给我的消息先落这role.py:100
memory长记忆:我处理过、认为相关的消息都存这,供 think/act 读role.py:103
working_memory工作记忆:planner 模式下当前任务的临时草稿区role.py:105
watch关注的动作集合:一组 cause_by 字符串,决定我捡哪些消息role.py:108
state当前状态:动作清单里的下标;-1 表示初始/终止(无 todo)role.py:106
todo本轮待办:think 选出的那个 Action 实例role.py:107
react_mode反应模式(默认 REACT)role.py:110
max_react_loopREACT 模式下最多 think-act 几轮,防止无限反应role.py:113

理解这张表,后面每个 _ 方法都只是在读写这几个字段而已。

3. 核心原理(逐个机制,由浅入深)​

3.1 观察 _observe:从私有收件箱里挑「该我处理」的新消息​

它要解决的小问题。 环境会把消息广播给很多角色,一个角色的收件箱里会混进大量跟我无关、或我已经看过的消息。_observe 要从中挑出「既新、又归我管」的那几条。

两条挑选规则(核心暗线)。 一条消息被留下,必须同时满足:它是新的(不在我记忆里),并且归我管——归我管有两种:要么它的 cause_by(由哪个动作产生)在我的 watch 集合里,要么它的 send_to(收件人)点了我的名。

怎么读这张图:从左往右是过滤漏斗,一层层筛。

收件箱 msg_buffer.pop_all()        role.py:406
        │  (一批原始新消息 news)
        ▼
① 去重:不在 memory 里             role.py:408  old_messages
        │
        ▼
② 归我管:cause_by ∈ watch          role.py:411
   或   send_to 含 self.name
        │
        ▼
留下的存进 memory → rc.news         role.py:418
        │
        ▼
返回 len(rc.news):0 就表示「没料」   role.py:427

真实实现里,这两条规则就是一行列表推导 role.py:410-412(_observe):

self.rc.news = [
    n for n in news if (n.cause_by in self.rc.watch or self.name in n.send_to) and n not in old_messages
]

n.cause_by 是消息的「产生它的动作」标签(schema.py:239,Message.cause_by),self.rc.watch 是一组这样的标签字符串(由 _watch 填,见 3.5)。「关注某个动作」= 关注「由该动作产生的所有消息」,这是 MetaGPT 消息路由的基本口径。

新消息从哪来:两个来源。 正常情况是从私有收件箱 msg_buffer.pop_all() 一次性倒空(role.py:406)。但如果角色是从中断中恢复的(recovered),会先用 memory.find_news 从记忆里找出「上次观察点之后」还没处理的消息(role.py:403-404),避免恢复后重复或漏处理。find_news 的逻辑很朴素——把候选里已经在记忆中的剔掉(memory.py:84-92,find_news)。

一个副作用变量:latest_observed_msg。 每次观察完,它记下「这批里最后一条」(role.py:419)。这是给「中断-恢复」用的书签:下次恢复时从这条之后接着看。

返回值即闸门。 _observe 返回留下的新消息条数;run 拿它当闸门——返回 0 就打印「no news, waiting」直接 return(role.py:543),连想都不想。没有归我管的新消息,角色就不动。

3.2 反应 react:三种模式的总闸​

它要解决的小问题。 观察到有料之后,「该怎么把活干完」其实有不止一种范式:是每步都让 LLM 现想下一步(灵活),还是按固定顺序一个个走(确定),还是先规划出整张任务清单再逐个执行(适合复杂任务)?

三种模式,一个枚举。 RoleReactMode(role.py:82)列出三种,react()(role.py:512)据此分派:

模式枚举值怎么选动作分派到典型场景
REACT"react"每轮让 LLM 现选下一个动作(ReAct 范式)_react需要动态决策的通用角色
BY_ORDER"by_order"按动作清单顺序一个个走_react固定 SOP 流水线的角色
PLAN_AND_ACT"plan_and_act"先让 LLM 规划任务清单,再逐个执行_plan_and_act数据分析等需先拆解的任务

注意 REACT 与 BY_ORDER 都进 _react(区别只在 _think 里怎么选,见 3.3);只有 PLAN_AND_ACT 走另一条路 _plan_and_act。分派逻辑就在 react() 的 if/elif(role.py:514-519)。

收尾统一动作。 三条路无论走哪条,react() 最后都会 _set_state(-1)(role.py:520)——把状态复位成「无待办」,给下一次驱动一个干净起点;并给返回的 AIMessage 盖上「哪个角色产出的」标签(with_agent,role.py:521-522)。

_react:think-act 交替循环​

REACT/BY_ORDER 共用的循环体 _react(role.py:454)非常直白——想一下、做一下,循环,直到没得做或到上限:

# 真实结构,略去日志 —— role.py:459-470
actions_taken = 0
while actions_taken < self.rc.max_react_loop:
    has_todo = await self._think()   # 想:选出 state/todo
    if not has_todo:                 # think 说「没得做了」→ 停
        break
    rsp = await self._act()          # 做:跑那个 Action
    actions_taken += 1
return rsp                           # 返回最后一个动作的产物

max_react_loop 的语义(只从循环层面讲)。 它是这个 while 的上限闸——防止 REACT 模式下 LLM 无限地「再想再做」。默认值 1(role.py:113),即「想一次、做一次就结束」。REACT 模式下它由 _set_react_mode 按你传的值设定(见 3.5);而 BY_ORDER 模式下 _think 会把它强制改成动作清单的长度(role.py:354-355),好让每个动作正好各跑一遍。

_plan_and_act:先规划,再逐任务执行​

PLAN_AND_ACT 走 _plan_and_act(role.py:472),形态不同:先让 planner 生成一张任务计划,再 while 循环逐个 current_task 执行,每个任务交给 _act_on_task 处理、结果回喂给 planner 更新计划(role.py:480-488)。

_plan_and_act:
   planner.update_plan(goal)         生成/确认初始计划   role.py:477
        │
        ▼
   while planner.current_task:       逐个任务          role.py:480
        _act_on_task(task) ──► TaskResult
        planner.process_task_result(...)  评审/更新计划
        │
        ▼
   返回完成的计划作为响应             role.py:490

_act_on_task 在基类里是 raise NotImplementedError(role.py:498-510)——想用 planner 模式的角色必须自己实现它。Planner 的内部(计划怎么生成、任务怎么评审)属于第 5 章,本章到此为止。

3.3 思考 _think:一个选「下一步做哪个动作」的状态机​

它要解决的小问题。 一个角色可能会好几个动作(actions 清单)。_think 就是回答「这一轮该做清单里的哪一个?」——它不干活,只设定 state 和 todo。

「state」是什么。 state 就是动作清单 actions 里的下标(role.py:106)。_set_state(i) 做两件事:记下 state=i,并把 todo 指到 actions[i];若 i<0 则 todo=None(role.py:302-306,_set_state)。所以「选动作」= 「设 state」。

_think 的判断顺序(由简到繁的短路)。 它是一串 if,命中即返回:

顺序情形怎么选 state位置
①只有一个动作直接 _set_state(0),没得选role.py:342-346
②从中断恢复且有旧 state沿用旧 state 接着跑role.py:348-351
③BY_ORDER 模式_set_state(state+1),顺序推进role.py:353-357
④其余(REACT 多动作)让 LLM 选下一个 staterole.py:359-378

情形 ④ 才是「智能」所在:让 LLM 选状态。 多动作又要动态决策时,_think 把「对话历史 + 可选状态清单」填进一段提示词 STATE_TEMPLATE(role.py:54),问 LLM「下一步该进哪个阶段,只回一个数字」:

# 真实结构 —— role.py:359-378
prompt = self._get_prefix()
prompt += STATE_TEMPLATE.format(
    history=self.rc.history,          # 对话记录
    states="\n".join(self.states),   # 形如「0. WriteCode\n1. RunCode」
    n_states=len(self.states) - 1,
    previous_state=self.rc.state,
)
next_state = await self.llm.aask(prompt)
next_state = extract_state_value_from_output(next_state)   # 从回答里抠出数字
...
self._set_state(next_state)

STATE_TEMPLATE(role.py:54-69)明确要求 LLM「只回一个 0~n 的数字,认为目标已完成就回 -1」。回 -1 就意味着 _set_state(-1) → todo=None → _react 那轮循环因「无 todo」而停。extract_state_value_from_output 负责从 LLM 可能带杂质的回答里抠出那个整数;抠不出合法值就兜底成 -1(role.py:371-372)。

self.states 长什么样。 它在 set_actions 时同步生成:每加一个动作就 append 一行 "{下标}. {动作名}"(role.py:259)。所以 LLM 看到的是一份带编号的「技能菜单」,它只需报菜名对应的号。

3.4 行动 _act:跑动作,把结果封成 AIMessage​

它要解决的小问题。 _think 选好了 todo,_act 就把它跑掉,并把产物规整成一条标准消息——因为角色之间只靠消息通信,任何产物都得先变成 Message 才能流转。

三步。 读 _act(role.py:381-397):

  1. 跑动作: response = await self.rc.todo.run(self.rc.history)——把当前记忆(history)作为上下文喂给动作(role.py:383)。
  2. 规整成 AIMessage: 按 run 返回的类型分三种情况封装(role.py:384-394)——结构化输出(ActionOutput/ActionNode)会连 instruct_content 一起装进消息;已经是 Message 的直接用;纯字符串则包成 AIMessage。三种都盖上 cause_by=self.rc.todo(标明「我是这个动作产的」)和 sent_from=self。
  3. 进记忆: self.rc.memory.add(msg)——自己的产出也存进自己的记忆(role.py:395),下一轮 think/act 能看到。

关键细节:cause_by 是路由的钥匙。 _act 给消息盖的 cause_by(role.py:388),正是别的角色 _observe 时用来判断「归不归我管」的那把钥匙(回看 3.1)。于是一条隐形契约成立:A 角色关注 WriteDesign 动作,B 角色 _act 跑完 WriteDesign 产出的消息自带 cause_by=WriteDesign,A 就能在自己的 observe 里捞到它。 动作类型即消息的「频道」。

AIMessage 是 Message 的子类,专为承载「助手产出」而设,role="assistant",并提供 with_agent 打「哪个角色发的」标签(schema.py:439-450)。

3.5 配置动作与反应:set_actions / _watch / _set_react_mode​

前面几节讲「跑起来怎么转」,这节补齐「跑之前怎么配」——三个 setter,决定了循环的行为参数。

set_actions:装技能。 把动作清单塞进角色,顺便给每个动作注入 LLM、context、prefix,并同步生成 states 菜单(role.py:239-259)。传进来的可以是 Action 类(自动实例化)也可以是实例。

_watch:订阅频道。 把「关注哪些动作」转成一组字符串标签存进 rc.watch(role.py:284-288)。注意它存的是动作的字符串名(any_to_str(t)),好和消息的 cause_by 直接比对。默认角色关注 UserRequirement(在 _process_role_extra 里设,role.py:177)——即「用户提的需求」。

_set_react_mode:定循环模式。 设 react_mode,并按模式做相应初始化(role.py:261-282):REACT 模式会记下你给的 max_react_loop;PLAN_AND_ACT 模式会新建一个绑定了 goal 和 working_memory 的 Planner。这三种模式的语义前面 3.2 已讲。

4. 收发规约:publish_message / put_message / is_idle​

角色之间不共享内存、只靠消息,所以「消息怎么进、怎么出」有严格的两分法。这一节把收发规约收拢讲清。

4.1 一进一出,只有两个口​

MetaGPT 规定角色只有两个消息通道(见 role.py 顶部 RFC 116 注释):

方法方向干什么位置
put_message进把消息塞进自己的私有收件箱 msg_bufferrole.py:448
publish_message出把消息广播给环境,由环境转发给订阅者role.py:429

put_message 极简——就是 msg_buffer.push(message)(role.py:448-452)。收件箱是个异步队列 MessageQueue(schema.py:713),push/pop_all/empty 三个操作(schema.py:730-746)正好对应「投递 / 观察时倒空 / 判空」。

4.2 publish_message 的几条分流规则​

publish_message(role.py:429-446)不是无脑广播,它先看收件人 send_to 决定去向:

publish_message(msg)
   │
   ├─ send_to 含「发给自己」占位符? → 换成本角色地址   role.py:433-435
   ├─ sent_from 空?               → 填上本角色地址     role.py:436-437
   │
   ├─ send_to 全是「我自己」?       → 直接 put_message 回自己收件箱,不出环境  role.py:438-440
   ├─ 没有 env?                    → 不发(无处可发)   role.py:441-443
   └─ 否则                          → env.publish_message(msg) 交给环境总线  role.py:444-446

两个要点:一是**「发给自己」被短路成本地投递**(role.py:438-440),不经环境;二是没有环境时消息哪也不去(role.py:441-443)——单个孤立的 Role 广播出的消息会静默丢弃,这解释了为什么角色通常得先 set_env 挂到环境上才有意义(环境路由属第 3 章)。

4.3 is_idle:三个条件都空才算闲​

is_idle(role.py:556-559)判断角色「彻底没活了」——三个条件同时成立:没有待处理的新消息(rc.news 空)、没有待办动作(rc.todo 空)、收件箱也空(msg_buffer.empty())。团队用它判断「所有角色都闲了 → 本轮协作结束」(is_idle 也是 BaseRole 声明的抽象属性,base_role.py:12-14)。

5. 巧妙之处(可借鉴的技术)​

① 用「动作类型」当消息频道,实现零耦合路由。 角色不需要知道「谁会用我的产出」,只管给产出盖 cause_by=某动作;下游角色也不点名上游,只声明 _watch([某动作])。发布/订阅完全靠动作类型这个稳定标识对齐(role.py:388 盖章 vs role.py:411 过滤),角色之间彻底解耦。这是整个多智能体协作能松散拼装的地基。

② 状态即下标,思考即选号。 把「该做哪个动作」压缩成「选一个整数下标」(state),于是「让 LLM 做决策」退化成「让 LLM 回一个 0~n 的数字」(STATE_TEMPLATE,role.py:54)。决策空间越窄,LLM 越不容易跑偏、越好解析——比让它自由发挥「下一步做什么」稳健得多。

③ 一个 _react 循环容纳两种范式。 REACT(LLM 动态选)和 BY_ORDER(顺序走)复用同一段 think-act 循环(role.py:454),差异全部下沉到 _think 内部的分支(role.py:353-357 vs role.py:359-378)。加一种范式只需改 _think,循环骨架不动。

④ 中断-恢复的书签机制。 latest_observed_msg + recovered 让角色能在崩溃后从「上次观察点」精确续跑(role.py:403-404、_think 的恢复分支 role.py:348-351),而不是从头重放或漏处理消息。对长流程的健壮性很关键。

⑤ 记忆开关省内存。 enable_memory=False 时 _observe 不去比对旧消息、也不落库(role.py:408、role.py:136-138 注释)——无状态的原子角色(或用外部存储的角色)可以关掉记忆省开销。

6. 边界与局限​

单个 Role 不自带路由。 publish_message 在没有 env 时静默丢消息(role.py:441-443)。一个孤立 Role.run() 能自转,但产出无处可去——真正的多角色转发是 Environment 的职责(第 3 章),Role 只负责「进出两个口」。

_act_on_task 是抽象洞。 想用 PLAN_AND_ACT 模式的角色,必须自己实现 _act_on_task,否则 raise NotImplementedError(role.py:498-510)。基类只搭了 plan→act 的骨架。

LLM 选状态可能出错,靠兜底而非纠错。 _think 情形 ④ 依赖 LLM 回合法数字;回了非法值只是 warning 后强制置 -1(即「结束」)(role.py:371-372),并不会重试或纠正。极端情况下可能提前终止。

RoleContext.news 字段已弃用倾向。 源码注释标了 # TODO not used(role.py:109,指 news: list[Type[Message]] 那个字段声明),实际用的是 rc.news(在 _observe 里重新赋值)——读源码时别被同名字段迷惑。

_react 与 max_react_loop 的默认值偏保守。 默认 max_react_loop=1(role.py:113),即「想一次做一次就返回」;需要多轮反应的角色必须显式调大,否则循环只跑一轮就结束。

7. 横向对比​

同属 ai-agent-reference 货架、都在解「单体智能体怎么自转」的兄弟项目,取舍各异:

维度MetaGPT Role典型对照
循环范式observe→think→act,think 内含状态机选动作多数框架是 plan→tool-call→observe 的 ReAct 直循环
角色间通信消息发布/订阅,按 cause_by 动作类型路由常见做法是显式函数调用或共享黑板
决策粒度让 LLM「选一个状态号」,决策空间极窄常让 LLM 直接输出下一步工具+参数
状态载体全部收敛进 RoleContext 一个对象常散在 agent 实例的多个字段

MetaGPT 的差异点是把「协作」建模成「消息在角色间流动」,而单个 Role 是这条河里的一个泵站。要理解泵站怎么连成河,继续读同组其它章。

8. 代码地图(导航索引)​

主题文件路径符号名
单次驱动入口metagpt/roles/role.pyRole.run
观察:过滤新消息进记忆metagpt/roles/role.pyRole._observe
反应总闸:分派三种模式metagpt/roles/role.pyRole.react
think-act 交替循环metagpt/roles/role.pyRole._react
先规划后执行metagpt/roles/role.pyRole._plan_and_act
思考:状态机选动作metagpt/roles/role.pyRole._think
让 LLM 选状态的提示词metagpt/roles/role.pySTATE_TEMPLATE
行动:跑动作、封 AIMessagemetagpt/roles/role.pyRole._act
设 state / todometagpt/roles/role.pyRole._set_state
运行时状态容器metagpt/roles/role.pyRoleContext
三种反应模式枚举metagpt/roles/role.pyRoleReactMode
设反应模式metagpt/roles/role.pyRole._set_react_mode
订阅关注的动作metagpt/roles/role.pyRole._watch
装技能、生成 states 菜单metagpt/roles/role.pyRole.set_actions
出:广播消息metagpt/roles/role.pyRole.publish_message
进:塞进私有收件箱metagpt/roles/role.pyRole.put_message
是否闲置metagpt/roles/role.pyRole.is_idle
角色抽象基类metagpt/base/base_role.pyBaseRole
记忆:找新消息metagpt/memory/memory.pyMemory.find_news
记忆:按动作取消息metagpt/memory/memory.pyMemory.get_by_actions
消息与其路由字段metagpt/schema.pyMessage.cause_by / send_to
助手消息metagpt/schema.pyAIMessage.with_agent
异步收件箱队列metagpt/schema.pyMessageQueue

同组导航:index(MetaGPT 是什么 · 全景 · 阅读地图)· 02-action-actionnode(Action 与 ActionNode:本章的 _act 跑的就是它)· 03-environment-message-bus(Environment 总线:本章 publish_message 交出去之后的世界)· 04-classic-sop-pipeline(多个 Role 拼成 SOP 流水线)· 05-mgx-rolezero(RoleZero 与 TeamLeader 动态调度)。