多智能体协作
本篇将学习 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 类的主要组件包括:
- env:一个 Environment 实例,用于智能体的即时消息交互。
- investment:用于管理团队成本,即限制 token 花费。
- idea:用于指定团队接下来要围绕的工作或目标。
Team 类提供的一些主要方法包括:
- hire:向团队中添加员工角色。
- invest:控制团队预算。
- run_project:根据用户需求启动项目。
- run:运行团队直到指定轮数或预算耗尽。
在 Team 运行时,首先将调用 run_project 方法给智能体们一个需求,接着在 n_round 的循环中,重复检查预算与运行 env,最后返回环境中角色的历史对话。
尽管 Team 类只是在 Env 上的简单封装,但它向我们展示了,我们该如何向多智能体系统发布启动消息以及引入可能的人类反馈。通过 Team,我们可以更方便地管理和协调多智能体团队,实现更为复杂和高效的任务协作。
3. 代码实现
在第一篇环境配置中,我们提供了Team封装的使用,这里我们仅基于Environment,不使用封装的Team进行多智能体的协作。
项目思路
为了实现一个文字版的你画我猜,我们需要 一个猜测者和一个描述者。
- 开始时,用户给出一个物品词,这个物品词将被传递给描述者。
- 描述者接收到用户给出的物品词后,将其转换为一个描述。然后,描述者使用 PassDescription 动作将描述传达给猜测者。
- 猜测者接收到描述后,使用 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函数了。