MetaGPT框架Agent开发流程详解

概述

智能体这一概念的起源可追溯至人工智能与强化学习领域，尤其在强化学习中得到了丰富的应用与拓展。在MetaGPT的视角下，智能体就如同生活在环境中的数字化个体。有一个简洁的公式清晰地阐明了这一点：

智能体 = 大语言模型（LLM） + 观察 + 思考 + 行动 + 记忆

这个公式概括了智能体的核心功能本质。若要深入理解每个组成部分的具体含义，不妨以人类行为作为类比，这样会更为直观：

1. 大语言模型（LLM）：它相当于智能体的“大脑”，赋予其处理信息、从交互中学习、做出决策并执行动作的能力。
2. 观察：如同智能体的感知器官，使其能够感知周围环境。例如，接收来自其他智能体的文字消息、监控摄像头传来的视觉数据、或客服录音中的音频信号。所有后续行动都基于这些观察结果。
3. 思考：这一阶段负责分析观察内容与记忆，并推导出可能采取的行动方案。这是智能体内部的决策过程，通常由LLM驱动。
4. 行动：智能体将思考与观察的结果以显式方式表现出来。行动可以是让LLM编写代码，也可以是手动预定义的操作（如读取本地文件）。此外，还能调用各类工具——比如查询天气、使用计算器进行数学运算等。
5. 记忆：智能体存储过去经验的能力。这一功能对学习至关重要，它使智能体能够回顾先前的结果，并据此调整后续行动。

多智能体

多智能体系统可以被想象成一个智能体社会，其运作同样有一个公式加以概括：

多智能体 = 智能体 + 环境 + 标准流程（SOP） + 通信 + 经济

每个组件都承担着不同的分工：

1. 智能体：在单智能体定义的基础上，多智能体系统中的智能体们协同工作，每个智能体都拥有独立的LLM、观察、思考、行动和记忆模块。
2. 环境：智能体生存与互动的公共空间。智能体从环境中获取关键信息，同时将自己的行动结果发布出去，供其他智能体使用。
3. 标准流程（SOP）：预先制定的程序与交互规则，确保系统运行有序、高效。例如，在汽车制造的SOP中，一个智能体负责焊接零件，另一个负责安装电缆，流水线作业有条不紊。
4. 通信：智能体之间的信息交换机制。协作、谈判、对抗等交互均依赖于通信。
5. 经济：多智能体环境中的价值交换体系，决定了资源如何分配、任务如何排序。

运行流程

在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框架，然后基于业务需求进行开发即可。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：MetaGPT框架Agent开发流程详解要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题