聊聊MetaGPT框架的agent开发流程

概述

智能体的概念可以溯源到人工智能与强化学习，尤其是在强化学习中得到了延伸与实际的应用。在MetaGPT看来，可以将智能体想象成环境中的数字人，其中

❝

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

这个公式概括了智能体的功能本质。为了理解每个组成部分，可以与人类进行类比：

1. 大语言模型（LLM）：LLM作为智能体的“大脑”部分，使其能够处理信息，从交互中学习，做出决策并执行行动。
2. 观察：这是智能体的感知机制，使其能够感知其环境。智能体可能会接收来自另一个智能体的文本消息、来自监视摄像头的视觉数据或来自客户服务录音的音频等一系列信号。这些观察构成了所有后续行动的基础。
3. 思考：思考过程涉及分析观察结果和记忆内容并考虑可能的行动。这是智能体内部的决策过程，其可能由LLM进行驱动。
4. 行动：这些是智能体对其思考和观察的显式响应。行动可以是利用 LLM 生成代码，或是手动预定义的操作，如阅读本地文件。此外，智能体还可以执行使用工具的操作，包括在互联网上搜索天气，使用计算器进行数学计算等。
5. 记忆：智能体的记忆存储过去的经验。这对学习至关重要，因为它允许智能体参考先前的结果并据此调整未来的行动。

多智能体

多智能体系统可以视为一个智能体社会，其中

❝

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

这些组件各自发挥着重要的作用：

1. 智能体：在上面单独定义的基础上，在多智能体系统中的智能体协同工作，每个智能体都具备独特有的LLM、观察、思考、行动和记忆。
2. 环境：环境是智能体生存和互动的公共场所。智能体从环境中观察到重要信息，并发布行动的输出结果以供其他智能体使用。
3. 标准流程（SOP）：这些是管理智能体行动和交互的既定程序，确保系统内部的有序和高效运作。例如，在汽车制造的SOP中，一个智能体焊接汽车零件，而另一个安装电缆，保持装配线的有序运作。
4. 通信：通信是智能体之间信息交流的过程。它对于系统内的协作、谈判和竞争至关重要。
5. 经济：这指的是多智能体环境中的价值交换系统，决定资源分配和任务优先级。

运行流程

在MetaGPT中，一个智能体运行周期的流程图如下：

在MetaGPT中，对于智能体来说，思考与行动是必不可少的，主要是基于ReAct理论框架^[1]，其流程主要是observe -> think -> act即一个react的流程；MetaGPT实现了三种模式的react，具体请看：思考与行动 | MetaGPT^[2]。

关键对象

在MetaGPT中，有两个关键对象定义了智能体的基础：Action、Role。

• Action类是动作的逻辑抽象。用户可以通过简单地调用 <font style="color:rgb(60, 60, 67);">self._aask</font> 函数令 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)
        # logger.info(prompt)

        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()
        # accept messages sent (from opponent) to self, disregard own messages from the last round
        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  # An instance of SpeakAloud

        memories = self.get_memories()
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
        # print(context)

        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而言。感觉可以继续深挖MetaGPT，而且在开发的过程中，我个人一个最大的感受——好像是在基于Spring等框架——做业务开发，现在的AI发展趋势，尤其是在agent侧，的确是越来越简化AI的开发，对于我们来说，只需要了解agent框架，而后基于业务开发即可。