概述
智能体这一概念的起源可追溯至人工智能与强化学习领域，尤其在强化学习中得到了丰富的应用与拓展。在MetaGPT的视角下，智能体就如同生活在环境中的数字化个体。有一个简洁的公式清晰地阐明了这一点：
智能体 = 大语言模型（LLM） + 观察 + 思考 + 行动 + 记忆
这个公式概括了智能体的核心功能本质。若要深入理解每个组成部分的具体含义，不妨以人类行为作为类比，这样会更为直观：
- 大语言模型（LLM）：它相当于智能体的“大脑”，赋予其处理信息、从交互中学习、做出决策并执行动作的能力。
- 观察：如同智能体的感知器官，使其能够感知周围环境。例如，接收来自其他智能体的文字消息、监控摄像头传来的视觉数据、或客服录音中的音频信号。所有后续行动都基于这些观察结果。
- 思考：这一阶段负责分析观察内容与记忆，并推导出可能采取的行动方案。这是智能体内部的决策过程，通常由LLM驱动。
- 行动：智能体将思考与观察的结果以显式方式表现出来。行动可以是让LLM编写代码，也可以是手动预定义的操作（如读取本地文件）。此外，还能调用各类工具——比如查询天气、使用计算器进行数学运算等。
- 记忆：智能体存储过去经验的能力。这一功能对学习至关重要，它使智能体能够回顾先前的结果，并据此调整后续行动。
多智能体
多智能体系统可以被想象成一个智能体社会，其运作同样有一个公式加以概括：
多智能体 = 智能体 + 环境 + 标准流程（SOP） + 通信 + 经济
每个组件都承担着不同的分工：
- 智能体：在单智能体定义的基础上，多智能体系统中的智能体们协同工作，每个智能体都拥有独立的LLM、观察、思考、行动和记忆模块。
- 环境：智能体生存与互动的公共空间。智能体从环境中获取关键信息，同时将自己的行动结果发布出去，供其他智能体使用。
- 标准流程（SOP）：预先制定的程序与交互规则，确保系统运行有序、高效。例如，在汽车制造的SOP中，一个智能体负责焊接零件，另一个负责安装电缆，流水线作业有条不紊。
- 通信：智能体之间的信息交换机制。协作、谈判、对抗等交互均依赖于通信。
- 经济：多智能体环境中的价值交换体系，决定了资源如何分配、任务如何排序。
运行流程
在MetaGPT中，一个智能体的运行周期流程如下图所示：
对于智能体而言，思考与行动是其核心，这主要基于ReAct理论框架[1]。整个流程可概括为 observe -> think -> act 这样一个React循环。MetaGPT实现了三种模式的React，具体可参考“思考与行动 | MetaGPT”[2]。
关键对象
在MetaGPT中，有两个关键对象定义了智能体的基础：Action 和 Role。
Action类是对动作的逻辑抽象。用户只需调用self._aask函数，即可让LLM赋予该动作执行能力——该函数底层会调用LLM的API。Role类是对智能体的逻辑抽象。一个Role可以执行特定（或多个）Action，它具备记忆、思考能力，并能采用多种策略行动。简而言之，它是将所有组件凝聚在一起的实体。
对于 Action，通常只需继承该类并重载 run 方法即可。
class Action(SerializationMixin, ContextMixin, BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
# 省略其它....
async def _aask(self, prompt: str, system_msgs: Optional[list[str]] = None) -> str:
"""Append default prefix"""
return await self.llm.aask(prompt, system_msgs)
async def _run_action_node(self, *args, **kwargs):
"""Run action node"""
msgs = args[0]
context = "## History Messages\n"
context += "\n".join([f"{idx}: {i}" for idx, i in enumerate(reversed(msgs))])
return await self.node.fill(context=context, llm=self.llm)
async def run(self, *args, **kwargs):
"""Run action"""
if self.node:
return await self._run_action_node(*args, **kwargs)
raise NotImplementedError("The run method should be implemented in a subclass.")
而智能体的抽象类 Role 内部方法则更为复杂，其代码位于 Role.py[3]。关键方法包括：
set_actions：为智能体设置动作_set_react_mode：设置react模式_watch：observe（观测）action，并筛选由此action产生的msg_think：思考_act：行动_observe：观测_plan_and_act：规划与行动get_memories：获取内存数据run：开始运行
实现一个智能体时，主要取决于业务场景——需要重载不同方法，适配不同的react模式，以满足不同的业务需求。从源码来看，MetaGPT对扩展的支持非常完善。
智能体示例
MetaGPT提供了多个示例[4]，它们均基于数据解释器 DataInterpreter[5] 实现。数据解释器是一个通过编写代码来解决数据相关问题的智能体。它能够理解用户需求、制定计划、编写并执行代码，必要时还能调用工具——这些能力使其能够处理多种场景。
数据解释器agent的核心思路是：规划与行动（plan_and_act）。
# 提示词
REACT_THINK_PROMPT = """
# User Requirement
{user_requirement}
# Context
{context}
Output a json following the format:
.....
"""
class DataInterpreter(Role):
# 设置react模式为plan_and_act
@model_validator(mode="after")
def set_plan_and_tool(self) -> "Interpreter":
# code....
return self
# 设置持久化
@property
def working_memory(self):
return self.rc.working_memory
# 重写_think方法
async def _think(self) -> bool:
# code....
return need_action
# 重写_act方法
async def _act(self) -> Message:
# code....
return Message(content=code, role="assistant", cause_by=WriteAnalysisCode)
# 重写_plan_and_act方法
async def _plan_and_act(self) -> Message:
try:
# code....
return rsp
except Exception as e:
await self.execute_code.terminate()
raise e
# 重写_act_on_task方法
async def _act_on_task(self, current_task: Task) -> TaskResult:
# code....
return task_result
# 实际action的抽象
async def _write_and_exec_code(self, max_retry: int = 3):
# code....
return code, result, success
async def _write_code(
self,
counter: int,
plan_status: str = "",
tool_info: str = "",
):
# code....
return code, todo
async def _check_data(self):
# code....
如果想要自己动手开发智能体，数据解释器agent是一个极佳的参考范例，而且还有配套论文进行详细讲解。
需要注意的是，这是一个单智能体示例。与多智能体相比，除了react模式的选择和业务逻辑差异外，单智能体少了一个 watch 环节——因为无需在智能体之间进行通信。
辩论Agent
这是一个多agent的示例。不过严格来说，它并不算标准的agent示例——代码结构里只有一个agent，只是通过不同的属性初始化出了辩手智能体。但其中的思路仍然值得借鉴。相关描述可参考“辩论”[6]，完整代码位于 debate.py[7]。
# action
class SpeakAloud(Action):
"""Action: Speak out aloud in a debate (quarrel)"""
PROMPT_TEMPLATE: str = """
## BACKGROUND
Suppose you are {name}, you are in a debate with {opponent_name}.
## DEBATE HISTORY
Previous rounds:
{context}
## YOUR TURN
Now it's your turn, you should closely respond to your opponent's latest argument, state your position, defend your arguments, and attack your opponent's arguments,
craft a strong and emotional response in 80 words, in {name}'s rhetoric and viewpoints, your will argue:
"""
name: str = "SpeakAloud"
async def run(self, context: str, name: str, opponent_name: str):
prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)
rsp = await self._aask(prompt)
return rsp
# agent智能体
class Debator(Role):
name: str = ""
profile: str = ""
opponent_name: str = ""
def __init__(self, **data: Any):
super().__init__(**data)
self.set_actions([SpeakAloud])
self._watch([UserRequirement, SpeakAloud])
async def _observe(self) -> int:
await super()._observe()
self.rc.news = [msg for msg in self.rc.news if msg.send_to == {self.name}]
return len(self.rc.news)
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
todo = self.rc.todo
memories = self.get_memories()
context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)
msg = Message(
content=rsp,
role=self.profile,
cause_by=type(todo),
sent_from=self.name,
send_to=self.opponent_name,
)
self.rc.memory.add(msg)
return msg
# 根据不同的属性初始化两个agent实例使用
Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")
小结
整体阅读下来，基本上理清了MetaGPT框架中agent的开发流程。从实际所需的agent技能点来看，MetaGPT提供了良好的支持，扩展性也很出色，代码相对简洁——比autogen更为清爽。感觉值得进一步深入探索。此外，在开发过程中有一个特别明显的感受：这仿佛是在基于Spring之类的框架进行业务开发。当前AI的发展趋势，尤其是agent方向，确实在让AI开发变得越来越简化。对我们而言，只需要理解agent框架，然后基于业务需求进行开发即可。