多智能体协作

本篇将学习 MetaGPT中的 Environment 、 Team 组件。

1. Muti Agent 概念概述

• 多智能体系统 (Multi-Agent System, MAS) 是由一群具有一定自主性、协同性和学习能力的智能体组成的系统。
• 智能体在环境中相互协作，以达到某种目标或完成特定任务。

2. 多智能体组件介绍

2.1 Environment

• 概念: Environment 是多智能体系统中的一个核心概念，类似于强化学习中的环境。它为智能体提供了一个交互和通信的平台。
• 组成:
  • desc: 描述环境信息。
  • roles: 指定环境中的角色及其状态。
  • members: 表示环境中的角色及其对应的状态。
  • history: 记录环境中发生的消息。
• 功能:
  • 智能体可以向环境发布消息。
  • 智能体可以被其他角色观察。

环境的具体实现

• 在 MetaGPT 中，Environment 类负责管理智能体的活动和信息交流。
• 环境能够承载一批角色，角色可以发布消息，并可以被其他角色观察。

环境中的角色运行

• Environment 类中的 run 方法负责处理环境中所有角色的信息交互。
• 该方法按照角色声明的顺序依次执行每个角色的 run 方法。

角色行为

• 角色首先将接收到的信息存入其 msg_buffer。
• 根据观察结果进行思考和决策，然后执行相应的动作。
• 动作执行结果将被发送回环境，并由环境传递给其他订阅者。

理解

• Environment 作为多智能体系统的基础设施，为智能体之间的交互和通信提供了必要的平台。
• 智能体通过发布和订阅消息的方式在环境中进行协作，这有助于它们在完成各自任务的同时，保持整个系统的协调和效率。

点此下滑到Environment代码实践 👇

2.2 Team

Team 是基于 Environment 之上的二次封装，为多智能体活动提供了一个更为高级的抽象和管理工具。Team 不仅包含了 Environment 的所有功能，还增加了一些额外的组件和方法，以支持更为复杂的多智能体协作任务。

Team 类的主要组件包括：

1. env：一个 Environment 实例，用于智能体的即时消息交互。
2. investment：用于管理团队成本，即限制 token 花费。
3. idea：用于指定团队接下来要围绕的工作或目标。

Team 类提供的一些主要方法包括：

1. hire：向团队中添加员工角色。
2. invest：控制团队预算。
3. run_project：根据用户需求启动项目。
4. run：运行团队直到指定轮数或预算耗尽。

在 Team 运行时，首先将调用 run_project 方法给智能体们一个需求，接着在 n_round 的循环中，重复检查预算与运行 env，最后返回环境中角色的历史对话。

尽管 Team 类只是在 Env 上的简单封装，但它向我们展示了，我们该如何向多智能体系统发布启动消息以及引入可能的人类反馈。通过 Team，我们可以更方便地管理和协调多智能体团队，实现更为复杂和高效的任务协作。

3. 代码实现

在第一篇环境配置中，我们提供了Team封装的使用，这里我们仅基于Environment，不使用封装的Team进行多智能体的协作。

项目思路

为了实现一个文字版的你画我猜，我们需要 一个猜测者和一个描述者。

1. 开始时，用户给出一个物品词，这个物品词将被传递给描述者。
2. 描述者接收到用户给出的物品词后，将其转换为一个描述。然后，描述者使用 PassDescription 动作将描述传达给猜测者。
3. 猜测者接收到描述后，使用 GuessObject 动作进行猜测。而描述者监听猜测者的描述信息，提示是否正确。

动作定义

我们先进行基本动作定义，在此之前先导包

from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environment

动作比较单一，就是猜物品、描述物品。

# 定义新的动作类，用于猜测物体名称
class GuessObject(Action):name: str = "GuessObject"# 异步执行猜测物体名称的动作async def run(self, obj_desc: str):# 构造提示模板，要求模型根据提供的物体描述猜测物体名称prompt = f"根据以下描述猜测物体名称：\n{obj_desc}\n 你的猜测是？"# 使用_aask方法请求大模型进行猜测，并将结果返回rsp = await self._aask(prompt)return rsp# 定义新的动作类，用于向猜测者传达物体描述
class PassDescription(Action):name: str = "PassDescription"# 异步执行传达物体描述的动作async def run(self, obj_desc: str):# 构造提示模板，要求模型根据提供的物体描述构造传达信息prompt = f"请根据以下描述向猜测者传达信息：\n{obj_desc}\n 让猜测者猜猜是什么。你提供的信息是？"# 使用_aask方法请求大模型进行传达，并将结果返回rsp = await self._aask(prompt)return rsp

角色定义

描述者（Describer）

描述者负责接收用户给出的物品词，并将其转换为可供猜测者理解的描述。

class Describer(Role):name: str = "Describer"profile: str = "描述者"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([PassDescription])self._watch([GuessObject])async def _act(self) -> Message:logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todo# 获取所有记忆msg = self.get_memories()# 构造提示模板并请求大模型进行信息传达description = await PassDescription().run(msg)# 创建消息对象并返回msg = Message(content=description, role=self.profile,cause_by=type(todo))return msg

描述者首先获取所有记忆，然后构造一个提示模板，请求大模型进行信息传达。将得到的描述内容放入消息对象中，并返回给猜测者。

猜测者（Guesser）

猜测者基于描述者提供的描述，猜测用户心中所想的物品词。

class Guesser(Role):name: str = "Guesser"profile: str = "猜测者"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([GuessObject])self._watch([PassDescription])async def _act(self) -> Message:logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todo# 获取所有记忆msg = self.get_memories()print("猜测者拿到的msg : ", msg)# 构造提示模板并请求大模型进行猜测guess = await GuessObject().run(msg)# 创建消息对象并返回msg = Message(content=guess, role=self.profile,cause_by=type(todo))return msg

猜测者首先获取所有记忆，并打印出来以便调试。然后，构造一个提示模板，请求大模型进行猜测。将得到的猜测内容放入消息对象中，并返回给描述者。

环境配置

在运行的主函数中，我们主要按照如下流程执行：

主函数执行

环境初始化

首先，我们创建一个 Environment 实例，命名为 classroom。这个环境将管理游戏中所有角色的交互。

创建并配置角色

然后，我们创建两个角色：describer（描述者）和 guesser（猜测者）。这两个角色分别负责接收和传达物品描述，以及基于这些描述进行猜测。

发布游戏主题

接下来，我们使用 classroom.publish_message 方法发布游戏主题。这个方法允许我们向环境中的所有角色发送消息。在这个例子中，我们向描述者发送了一个包含物体“葡萄”的主题。

运行游戏

游戏运行在一个循环中，该循环会在指定轮数（n_round）内重复执行。在每次循环中，我们调用 classroom.run 方法来处理所有角色的动作。这个方法会按照顺序执行每个角色的 _act 方法，即他们的动作逻辑。

记录游戏历史

在每次循环后，我们打印出环境的历史记录，以便跟踪游戏的进展。

# 定义游戏的主要函数
async def main(topic: str, n_round=5):# 创建角色并添加到环境describer = Describer()guesser = Guesser()classroom.add_roles([describer, guesser])# 发布游戏主题classroom.publish_message(Message(role="Human", content=topic, cause_by=UserRequirement,send_to='Describer'),peekable=False,)# 运行游戏while n_round > 0:await classroom.run()print("history : ", classroom.history)n_round -= 1# 返回游戏历史记录return classroom.history# 运行游戏
asyncio.run(main(topic=' 物体 “葡萄” '))

至此，执行之后我们可以看到运行效果：

• 由于猜测者只观察watch描述者的动作，所以他的是仅有描述，没有人物给出的答案的。

• 如果我们输出classroom.history，则会看到第一条记录就是Human给出的答案。

• 如果刚刚直接跳过了team的部分，也可以点此上滑到Team 介绍 上 。在构造Team时，我们可以传入n_round进行直接执行，就不需要这种main函数了。