MetaGPT Agent 动态 Action 机制详解
这篇讲怎么在 MetaGPT 里写一个能在运行时动态换 Action 序列的 Agent。目标很具体：Agent 初始化时有三个动作 Print1/2/3，顺序跑完之后，动态生成 Print4/5/6 接着跑。核心是搞懂 MetaGPT 的 React 循环（run → react → think → act）和它的状态管理。
React 机制
run() 收到消息、存进 memory 后调用 react()；react() 是个 while 循环，每轮先 think() 决定下一个动作、再 act() 执行：
run(message)
└─ react():
while True:
think() → 决定下一个动作（设置 state / todo）
若 todo 为 None → 跳出
act() → 执行当前动作
状态就三个关键量：
self.actions # 动作列表 [Action1, Action2, ...]
self.rc.state # 当前状态索引 (int)
self.rc.todo # 当前要执行的动作 (Action 或 None)
_set_state() 把这两者绑在一起更新：
def _set_state(self, state: int):
self.rc.state = state
self.rc.todo = self.actions[state] if state >= 0 else None
约定：state = -1 表示无任务（todo=None），state = 0/1/2… 分别对应 actions[0/1/2…]。
实现
动作类继承 Action，run() 必须是 async（因为可能要调 LLM）：
class PrintAction(Action):
name: str = "PrintAction"
content: str = ""
async def run(self, *args, **kwargs) -> str:
logger.info(f"执行 {self.name}: {self.content}")
return self.content
Agent 里四个方法是关键。_think() 决定下一步：todo 为 None 时（刚开始或刚切阶段）从第 0 个动作起，否则往后挪一个，挪到头就置 None：
async def _think(self) -> None:
if self.rc.todo is None:
self._set_state(0)
return
if self.rc.state + 1 < len(self.states):
self._set_state(self.rc.state + 1)
else:
self.rc.todo = None
_act() 执行当前动作，并在第一阶段最后一个动作（state==2）跑完时触发切换：
async def _act(self) -> Message:
todo = self.rc.todo
result = await todo.run()
if self.phase == 1 and self.rc.state == 2:
self._switch_to_phase_2()
return Message(content=result, role=self.name)
切换阶段时用 _init_actions() 整体替换动作列表（不是 append，否则会变成 1–6 全在）。
一个容易踩的坑
切换后有个关键动作：必须把 self.rc.todo 重置为 None，而不是调 _set_state(0)。因为下一轮 _think() 里，如果 todo 已经是 actions[0]（非 None），就会走到“往后挪一个”的分支执行 _set_state(1)，直接跳过 Print4 从 Print5 开始。置 None 才能让 _think() 重新从第 0 个开始：
def _switch_to_phase_2(self) -> None:
self.phase = 2
self._init_actions([PrintAction(content=c) for c in ("4", "5", "6")])
self.rc.todo = None # 关键：让下一轮 _think() 从 actions[0] 重新开始
_react() 把这一切循环起来：
async def _react(self) -> Message:
msg = None
while True:
await self._think()
if self.rc.todo is None:
break
msg = await self._act()
return msg
完整代码
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message
import asyncio
class PrintAction(Action):
name: str = "PrintAction"
content: str = ""
async def run(self, *args, **kwargs) -> str:
logger.info(f"执行 {self.name}: {self.content}")
return self.content
class SimpleAgent(Role):
name: str = "SimpleAgent"
profile: str = "Simple Sequential Agent"
phase: int = 1
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._init_actions([PrintAction(content=c) for c in ("1", "2", "3")])
self._set_react_mode(react_mode=RoleReactMode.REACT.value)
async def _think(self) -> None:
if self.rc.todo is None:
self._set_state(0)
return
if self.rc.state + 1 < len(self.states):
self._set_state(self.rc.state + 1)
else:
self.rc.todo = None
async def _act(self) -> Message:
todo = self.rc.todo
result = await todo.run()
if self.phase == 1 and self.rc.state == 2:
self._switch_to_phase_2()
return Message(content=result, role=self.name)
def _switch_to_phase_2(self) -> None:
self.phase = 2
self._init_actions([PrintAction(content=c) for c in ("4", "5", "6")])
self.rc.todo = None
async def _react(self) -> Message:
msg = None
while True:
await self._think()
if self.rc.todo is None:
break
msg = await self._act()
return msg
async def main():
agent = SimpleAgent()
result = await agent.run("开始执行")
logger.info(f"最终结果: {result}")
if __name__ == "__main__":
asyncio.run(main())
跑起来输出依次是 1 2 3 4 5 6：前三个来自初始动作，后三个来自 state==2 时动态切换出来的第二阶段。
这套写法的价值在于动态性——第二阶段可以由第一阶段的结果决定，比如让 LLM 读完大纲再动态生成每一章的动作（MetaGPT 的 TutorialAssistant 就是这么干的），或者按结果走不同分支。要加第三阶段，在 _act() 里多一个 elif self.phase == 2 and self.rc.state == 2 的判断、再写个 _switch_to_phase_3() 即可。