1. 前言

本章任务：

了解MetaGPT中Agent的概念
实现一个单动作Agent
实现一个多动作Agent
实现一个复杂Agent：技术文档助手

2. MetaGPT中Agent的概念

MetaGPT中对于智能体的定义如下：

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

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

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

在MetaGPT中定义的一个agent运行示例如下：

一个agent在启动后他会观察自己能获取到的信息，加入自己的记忆中
下一步进行思考，决定下一步的行动，也就是从Action1，Action2，Action3中选择执行的Action
决定行动后，紧接着就执行对应行动，得到这个环节的结果

而在MetaGPT内 Role 类是智能体的逻辑抽象。一个 Role 能执行特定的 Action，拥有记忆、思考并采用各种策略行动。基本上，它充当一个将所有这些组件联系在一起的凝聚实体。

MetaGPT中对于多智能体的定义如下：

多智能体 = 智能体 + 环境 + SOP + 评审 + 路由 + 订阅 + 经济

3. 实现一个单动作Agent

3.1. 本节任务

利用MetaGPT框架实现一个生成代码的 Role，名为SimpleCoder。基于这个Role创建的Agent能够根据我们的需求来生成代码。

3.2. 需求分析

要实现一个 SimpleCoder ，我们需要分析这个 Role 它需要哪些能力

能够接受用户输入的需求
记忆用户的需求
根据自己已知的信息和需求来编写代码

3.3. 实现Role的思路

要自己实现一个最简单的Role，只需要重写Role基类的 _init_ 与 _act 方法

1、在 _init_ 方法中，声明 Role 的name（名称）profile（人设）
2、使用 self._init_action 函数为Role配备期望的动作
3、在_act方法中，我们需要编写智能体具体的行动逻辑，智能体将从最新的记忆中获取人类指令，运行配备的动作，最后返回一个完整的消息。

在 MetaGPT 中，类 Action 是动作的逻辑抽象。用户可以通过简单地调用 self._aask 函数来获取 LLM 的回复。
self._aask 函数如下：

async def _aask(self, prompt: str, system_msgs: Optional[list[str]] = None) -> str:
    """Append default prefix"""
    if not system_msgs:
        system_msgs = []
    system_msgs.append(self.prefix)
    return await self.llm.aask(prompt, system_msgs)

它将调用你预设好的 LLM 来根据输出的提示词生成回答。

3.4. 实现SimpleWriteCode动作

simple_write_code.py 文件中，实现SimpleWriteCode的动作

import re
from metagpt.actions import Action

class SimpleWriteCode(Action):

    PROMPT_TEMPLATE = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ``` with NO other texts,
    your code:
    """
    
    def __init__(self, name="SimpleWriteCode", context=None, llm=None):
        super().__init__(name, context, llm)

    async def run(self, instruction: str):

        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = SimpleWriteCode.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        pattern = r'```python(.*)```'
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text

SimpleWriteCode 继承自 Action，重写了__init__ 和 run 方法。
__init__ 方法用来初始化这个Action，而run方法决定了我们对传入的内容到底要做什么样的处理。
在__init__方法中，我们声明了这个这个动作的名称，行动前的一些前置知识context，要使用的llm。
在run方法中，我们需要声明当采取这个行动时，我们要对传入的内容做什么样的处理。run运行完成，返回代码内容。

3.5. 实现SimpleCoder角色

simple_coder.py 文件中，实现SimpleCoder角色

from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
from simple_write_code import SimpleWriteCode

class SimpleCoder(Role):
    def __init__(
        self,
        name: str = "Alice",
        profile: str = "SimpleCoder",
        **kwargs,
    ):
        super().__init__(name, profile, **kwargs)
        self._init_actions([SimpleWriteCode])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self._rc.todo}")
        todo = self._rc.todo # todo will be SimpleWriteCode()

        msg = self.get_memories(k=1)[0] # find the most recent messages

        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

SimpleCoder继承自 Role，重写了 __init__ 和 _act 方法。
__init__ 方法用来初始化这个Role，而_act方法决定了当这个角色行动时它的具体行动逻辑
在 __init__ 方法中声明了这个Role的name（名称），profile（人设），并且调用self._init_actions配备刚才写好的Action SimpleWriteCode
调用self._init_actions之后，SimpleWriteCode就会被加入到代办self._rc.todo中。
在_act方法中，智能体调用todo.run()方法，执行Action。

当调用Action时，我们需要获取用户的输入来作为instruction传递给Action，这里就涉及到我们该如何获取用户之前传递给agent的信息，在MetaGPT中，当用户与Agent交互时，所有的内容都会被存储在Agent自有的Memory中。
在MetaGPT中，Memory类是智能体的记忆的抽象。当初始化时，Role初始化一个Memory对象作为self._rc.memory属性，它将在之后的_observe中存储每个Message，以便后续的检索。简而言之，Role的记忆是一个含有Message的列表。

当需要获取记忆时（获取LLM输入的上下文），我们可以使用self.get_memories。函数定义如下：

1
2
3

def get_memories(self, k=0) -> list[Message]:
    """A wrapper to return the most recent k memories of this role, return all when k=0"""
    return self._rc.memory.get(k=k)

在SimpleCoder中，我们只需要获取最近的一条记忆，也就是用户下达的需求，将它传递给Action即可。

然后我们调用LLM，拿到LLM的输出，最后我们将拿到的信息封装为MetaGPT中通信的基本格式 Message 返回。

3.6. 运行单动作Agent

1、定义一个入口脚本 main.py

import asyncio
from metagpt.logs import logger
from simple_coder import SimpleCoder

async def main():
    msg = "write a function that calculates the sum of a list"
    role = SimpleCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

asyncio.run(main())

2、配置好LLM（第一章中已完成）
参考《基于MetaGPT实现一个订阅智能体：第一章前期准备》，给MetaGPT配置一个LLM。

3、运行Agent

1	python main.py

以上，我们就实现了一个简单的单动作Agent。

4. 实现一个多动作Agent

4.1. 为什么需要多动作Agent？

我们注意到一个智能体能够执行一个动作，但如果只有这些，实际上我们并不需要一个智能体。通过直接运行动作本身（输入提示词调用LLM），我们可以得到相同的结果。智能体的力量，或者说Role抽象的惊人之处，在于动作的组合（以及其他组件，比如记忆，但我们将把它们留到后面的部分）。通过连接动作，我们可以构建一个工作流程，使智能体能够完成更复杂的任务。

4.2. 本节任务

假设现在我们不仅希望用自然语言编写代码，而且还希望生成的代码立即执行，那么一个拥有多个动作的智能体可以满足我们的需求。
让我们称之为RunnableCoder，一个既写代码又立即运行的Role。我们需要两个Action：SimpleWriteCode 和 SimpleRunCode

4.3. 实现SimpleWriteCode动作

与上文中 simple_write_code.py 一致，不用变化。

4.4. 实现SimpleRunCode动作

一个动作可以调用LLM，也可以不调用LLM。
在SimpleRunCode的情况下，不涉及调用LLM。我们只需启动一个子进程来运行代码并获取结果。

在Python中，我们通过标准库中的subprocess包来fork一个子进程，并运行一个外部的程序。

simple_run_code.py 文件中，实现SimpleRunCode的动作

import subprocess
from metagpt.actions import Action
from metagpt.logs import logger

class SimpleRunCode(Action):
    def __init__(self, name="SimpleRunCode", context=None, llm=None):
        super().__init__(name, context, llm)

    async def run(self, code_text: str):
        result = subprocess.run(["python3", "-c", code_text], capture_output=True, text=True)
        code_result = result.stdout
        logger.info(f"{code_result=}")
        return code_result

4.5. 实现RunnableCoder角色

runnable_coder.py 文件中，实现RunnableCoder角色

from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
from simple_write_code import SimpleWriteCode
from simple_run_code import SimpleRunCode

class RunnableCoder(Role):
    def __init__(
        self,
        name: str = "Alice",
        profile: str = "RunnableCoder",
        **kwargs,
    ):
        super().__init__(name, profile, **kwargs)
        self._init_actions([SimpleWriteCode, SimpleRunCode])
        self._set_react_mode(react_mode="by_order")

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: 准备 {self._rc.todo}")
        todo = self._rc.todo

        msg = self.get_memories(k=1)[0] 
        result = await todo.run(msg.content)

        msg = Message(content=result, role=self.profile, cause_by=type(todo))
        self._rc.memory.add(msg)
        return msg

用 self._init_actions 初始化所有 Action
用 self._set_react_mode 方法来设定我们action执行的先后顺序。将 react_mode 设置为 “by_order”，这意味着 Role 将按照 self._init_actions 中指定的顺序执行 Action。
_act 定义每个action执行前后的处理逻辑。

此外，RunnableCoder继承自Role的run方法也很关键，它使得多个action都能被调用。

async def run(self, with_message=None):
    """Observe, and think and act based on the results of the observation"""
    if with_message:
        msg = None
        if isinstance(with_message, str):
            msg = Message(with_message)
        elif isinstance(with_message, Message):
            msg = with_message
        elif isinstance(with_message, list):
            msg = Message("\n".join(with_message))
        if not msg.cause_by:
            msg.cause_by = UserRequirement
        self.put_message(msg)

    if not await self._observe():
        # If there is no new information, suspend and wait
        logger.debug(f"{self._setting}: no news. waiting.")
        return

    rsp = await self.react()

    # Reset the next action to be taken.
    self._rc.todo = None
    # Send the response message to the Environment object to have it relay the message to the subscribers.
    self.publish_message(rsp)
    return rsp

4.6. 运行多动作Agent

1、定义一个入口脚本 main.py

import asyncio
from metagpt.logs import logger
from runnable_coder import RunnableCoder

async def main():
    msg = "write a function that calculates the sum of a list"
    role = RunnableCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)

asyncio.run(main())

2、配置好LLM（第一章中已完成）
参考《基于MetaGPT实现一个订阅智能体：第一章前期准备》，给MetaGPT配置一个LLM。

3、运行Agent

1	python main.py

以上，我们就实现了一个多动作Agent。
本节的完整代码，请参阅ai-agent-based-on-metagpt/muliti-action-agent。

5. 实现一个复杂Agent：技术文档助手

5.1. 本节任务

怎么让大模型为我们写一篇技术文档？
可能想到的是，我们告诉大模型：“请帮我生成关于Mysql的技术文档”，他可能很快地就能帮你完成这项任务，但是受限于大模型自身的token限制，我们无法实现让他一次性就输出我们希望的一个完整的技术文档。
当然我们可以将我们的技术文档拆解成一个一个很小的需求，然后一个一个的提问，但是这样来说不仅费时，而且还需要人工一直去跟他交互，比较麻烦。
而使用智能体，就可以解决这个问题。

5.2. 需求分析

因为token限制的原因，我们先通过 LLM 大模型生成教程的目录，再对目录按照二级标题进行分块，对于每块目录按照标题生成详细内容，最后再将标题和内容进行拼接，解决 LLM 大模型长文本的限制问题。

5.3. 需求实现

1、实现 WriteDirectory 动作，根据用户需求生成文章大纲
2、实现 WriteContent 动作，根据传入的子标题来生成内容
3、实现 TutorialAssistant 角色，重写_think、_act、_react方法，添加角色特有方法 _handle_directory

本节完整代码，请参阅ai-agent-based-on-metagpt/doc-agent。
关于代码的解释，请参阅《MetaGPT智能体开发入门》教程，写的非常详细了。

6. 本章作业

经过上面的学习，我想你已经对 MetaGPT 的框架有了基本了解，现在我希望你能够自己编写这样一个 agent

这个 Agent 拥有三个动作打印1 打印2 打印3（初始化时 init_action([print,print,print])）
重写有关方法（请不要使用act_by_order，我希望你能独立实现）使得 Agent 顺序执行上面三个动作
当上述三个动作执行完毕后，为 Agent 生成新的动作打印4 打印5 打印6 并顺序执行，（之前我们初始化了三个 print 动作，执行完毕后，重新 init_action([…,…,…])，然后顺序执行这个新生成的动作列表)

如果完成上面的任务，那这次作业已经可以算完成了，这个作业可以用多种思路去解决，比如我可以直接写死一整套的逻辑，甚至都不需要引入llm来完成这个工作，我希望大家通过这个作业来感受 MetaGPT 中 Agent 的行动逻辑，run->react->think->act 的这样一个过程，但你也可以试着在中间的某个环节中加入llm的交互，来尝试减少硬编码的工作，就像【实现一个更复杂的Agent：技术文档助手】中，我们实际上是让llm帮我们设计了 action 列表的内容。你也可以在此基础上做出更多的尝试，关于这个 Agent 我们还有更多可以思考的地方。

目前为止我们设计的所有思考模式都可以总结为是链式的思考（chain of thought），能否利用 MetaGPT 框架实现树结构的思考（tree of thought）图结构的思考（graph of thought）？试着实现让 ai 生成树结构的动作列表，并按照树的遍历方式执行他们，如果你实现，这将是加分项。

一个计算机技术爱好者与学习者

基于MetaGPT实现一个订阅智能体：第三章 MetaGPT框架组件介绍