基于MetaGPT构建LLM多智能体

bar

前言

你好，我是GISer Liu，在上一篇文章中，我们用了两万多字详细拆解了单个Agent的组成，并通过Github Trending订阅智能体理解MetaGPT框架的订阅模块如何解决应用问题，但是对于复杂，并行的任务，单个智能体是不能胜任；今天我们将进入多智能体开发的学习阶段；一起期待吧😀

一、介绍

在本文中，我们将分别详细介绍:

  • MetaGPT中Environment的设计思想；
  • 构建简单师生对话多Agent框架；
  • MetaGPT中Team的设计思想；
  • 构建 多Agent 开发团队；
  • 构建 多Agent 辩论团队;
  • 你画我猜多Agent框架实现；

二、Environment 环境设计思想

openai_gym

在MetaGPT框架中，Environment（环境）与Agent（智能体）这两个概念借鉴了强化学习的思想。而在强化学习中，Agent需要在环境中采取行动以最大化奖励。而在MetaGPT中，则提供了一个标准的环境组件Environment，用来管理Agent的活动与信息交流。

学习 agent 与环境进行交互的思想可以去OpenAI的GYM项目看看

1.环境设计原理

MetaGPT中的环境设计分为外部环境(ExtEnv)和内部环境，旨在帮助Agent代理与不同的外部应用场景（如游戏、手机应用等）以及内部开发和操作环境进行交互。

①外部环境（ExtEnv）

minecraft

定义：
外部环境是代理与外部世界交互的接口。它为代理提供了一种机制，使其能够与外部系统（例如游戏引擎、移动应用API）进行通信和交互。

继承和扩展：
ExtEnv类是所有外部环境的基础类，各种具体的外部环境（如Minecraft环境、狼人游戏环境等）会继承这个基础类，并在其上扩展实现特定的交互逻辑。

示例：

  1. 游戏环境：

    • 假设有一个在线游戏提供了API，允许查询玩家状态和执行游戏动作。
    • ExtEnv类封装了这些API，使代理能够调用这些API来查询游戏状态和执行动作。

    Agent执行某个Action，该Action中封装了执行API调用的逻辑

  2. 狼人sha游戏：

    • 在狼人游戏中，代理需要知道每晚和每天的游戏状态。
    • ExtEnv类定义了获取这些状态的方法，使代理能够在游戏中做出决策。
  • Minecraft开发API
  • Agent狼人sha实现案例
②内部环境

chatdev

(1)定义：
内部环境是代理及其团队直接使用的开发和操作环境。它类似于软件开发中的工作环境，包括开发工具、测试框架和配置文件等。

(2)继承和扩展：
内部环境类（XxxEnv）通常继承自一个基础环境类，并根据具体需求进行定制和扩展。这个基础环境类可以提供一些通用功能，比如日志记录、错误处理等。

(3)案例：

  • 开发环境：
    • 基础环境类可能提供一些通用的开发工具和测试框架。
    • 开发团队可以在这个基础上添加特定项目所需的工具和配置，例如数据库连接配置、CI/CD脚本等。

作者认为其思想和ChatDev的实现相似；

2.环境交互设计

MetaGPT还引入了两个重要的概念：observation_space和action_space。这些概念来自强化学习领域，用于描述代理从环境中获取的状态信息和可以采取的动作集合。

observation_space：

  • 表示代理可以从环境中获得的所有可能的状态。
    observation

  • 例如，在游戏环境中，observation_space可能包括玩家的位置、游戏时间、得分等。在上图Minecraft的案例中，观察空间就是周围的环境，角色的血量与护甲，拥有的工具与工具的数量；

action_space：

  • 表示代理在环境中可以执行的所有可能的动作。
  • 例如，在游戏环境中，action_space可能包括移动、跳跃、攻击等，同样在上面的案例中，action_space代表可选Action的集合，例如看到树以后选择砍树，看到怪物后选择逃离还是进攻；这需要Agent通过反思机制来判断进行；

通过定义这两个空间，MetaGPT能够更好地抽象不同环境中的具体细节，使得环境提供者可以专注于实现环境逻辑，而代理使用者可以专注于状态和动作的处理。

3.环境运行机制

agent&env

这里放这张图供大家思考

①Environment类的基本组成

以下是MetaGPT中Environment类的基本组成：

class Environment(ExtEnv):"""环境，承载一批角色，角色可以向环境发布消息，可以被其他角色观察到Environment, hosting a batch of roles, roles can publish messages to the environment, and can be observed by other roles"""model_config = ConfigDict(arbitrary_types_allowed=True)desc: str = Field(default="")  # 环境描述roles: dict[str, SerializeAsAny["Role"]] = Field(default_factory=dict, validate_default=True)member_addrs: Dict["Role", Set] = Field(default_factory=dict, exclude=True)history: str = ""  # For debugcontext: Context = Field(default_factory=Context, exclude=True)

参数说明如下：

  • model_config：配置模型的配置字典，允许任意类型作为字段。
  • desc：环境描述，默认值为空字符串。
  • roles：包含环境中所有角色的字典，键是角色名字，值是角色对象，默认值是一个空字典。
  • member_addrs：存储每个角色的地址集合的字典，键是角色对象，值是地址集合，默认值是一个空字典，不参与序列化。
  • history：记录环境历史信息的字符串，默认值为空字符串。
  • context：环境上下文对象，默认值是一个新的Context对象，不参与序列化。

知晓了环境的组成与Agent的交互方式以后，我们来理解一下多个Agent与环境的交互方式；

②Environment类的运行过程

试着想象一个大型圆桌会议，Environment提供了一个让Agent们统一上桌讨论的环境。接下来，我们来看看MetaGPT是如何实现这种机制的。
首先，当一个Environment运行时，会发生什么事情呢？来看一下Environment基类中定义的run方法：

async def run(self, k=1):"""处理一次所有信息的运行Process all Role runs at once"""for _ in range(k):futures = []for role in self.roles.values():future = role.run()futures.append(future)await asyncio.gather(*futures)logger.debug(f"is idle: {self.is_idle}")

当一个Environment运行时，其会遍历环境中的role（角色）列表，让它们逐个运行，即逐个做出各自的Actions，然后进行发言(将结果输出到环境)。

③单个Agent的运行机制

下面是每个Agent运行时所执行的事件：

@role_raise_decorator
async def run(self, with_message=None) -> Message | None:"""观察，并根据观察结果进行思考和行动"""if with_message:msg = Noneif isinstance(with_message, str):msg = Message(content=with_message)elif isinstance(with_message, Message):msg = with_messageelif isinstance(with_message, list):msg = Message(content="\n".join(with_message))if not msg.cause_by:msg.cause_by = UserRequirementself.put_message(msg)if not await self._observe():# 如果没有新的信息，则暂停并等待logger.debug(f"{self._setting}: 没有新的信息。正在等待...")returnrsp = await self.react()# 重置下一步要执行的动作self.set_todo(None)# 将响应消息发送到环境对象，以便将消息转发给订阅者self.publish_message(rsp)return rsp

run方法主要功能是观察环境，并根据观察结果进行思考和行动。如果有新的消息，它会将消息添加到队列中，并根据消息的内容进行处理。如果没有新的信息，它会暂停并等待。在处理完消息后，它会重置下一步要执行的动作，并将响应消息发送到环境对象。

def put_message(self, message):"""Place the message into the Role object's private message buffer."""if not message:returnself.rc.msg_buffer.push(message)

在Role的run方法中，Role首先会根据运行时是否传入信息（部分行动前可能需要前置知识消息），将信息存入RoleContext的msg_buffer中。

信息观察机制

在多智能体环境运行中，Role的每次行动将从Environment中先_observe（观察）消息。在observe的行动中，Role将从消息缓冲区和其他源准备新消息以进行处理，当未接受到指令时，Role将等待执行。

对于信息缓冲区中的信息，首先我们会根据self.recovered参数决定news是否来自于self.latest_observed_msg或者msg_buffer并读取。完成信息缓冲区中的读取后，如果设定好了ignore_memory则old_messages便不会再读取当前Role的memory。将news中的信息存入Role的memory后，我们将进一步从news中筛选，也就是我们设定的角色关注的信息（self.rc.watch），而self.rc.news将存储这些当前角色关注的消息，最近的一条将被赋给latest_observed_msg。最后，我们打印角色关注到的消息并返回。

这便是MetaGPT中环境的设计原理及其运行机制的详细解析。

run方法主要功能是观察环境，并根据观察结果进行思考和行动。如果有新的消息，它会将消息添加到队列中，并根据消息的内容进行处理。如果没有新的信息，它会暂停并等待。在处理完消息后，它会重置下一步要执行的动作，并将响应消息发送到环境对象，以便将消息转发。

def put_message(self, message):"""Place the message into the Role object's private message buffer."""if not message:returnself.rc.msg_buffer.push(message)

而在 role 的run方法中 role 首先将会根据运行时是否传入信息（部分行动前可能需要前置知识消息），将信息存入 rolecontext的 msg_buffer 中；
agent&env

最后，再看看，这张图，我想你会记忆更加深刻，当然，如果作者认知有偏颇，读者也可以在评论区指出，感谢支持

三、简单的师生交互多智能体系统

在上一节中，我们已经了解了environment环境的基本构成与它的运行逻辑，在这一节中，我们将学习如何利用environment来进行开发，进一步了解environment组件内部的活动，
现在设想一个多Agent交互的应用场景，我的想法是两人对话场景，如：

师生交互场景:

  • 首先用户输入一个主题;
  • 然后学生Agent负责根据用户的输入进行作文撰写
  • 当老师Agent发现学生Agent写作完毕以后，就会给学生提出学习意见；
  • 根据老师Agent给的意见，学生将修改自己的作品；
  • 如此循环直到设定的循环次数结束；这里环境则是教室；
    teacher&student

接下来我们用metagpt提供的API实现这一交互场景；

  • 首先，我们需要导入必要的包，并定义一个classroom环境，如下所示：
import asynciofrom metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environmentfrom metagpt.const import MESSAGE_ROUTE_TO_ALL
classroom = Environment()
  • 接着作者分别为老师和学生Agent撰写它们的行动WritingAction和ReviewAction，这里的思路基本就是简单的提示词工程，学生要求有写作格式和写作主题写作，老师有检查标准和检查功能；
    规范点说就是：
  1. 实现 WriteAction 方法：在这个方法中，学生Agent需要根据用户提供的主题撰写一篇作文。同时，当收到来自老师的修改建议后，也需要对作文进行相应的修改。
  2. 实现 ReviewAction 方法：在这个方法中，老师Agent需要读取学生撰写的作文，然后提出修改意见，以帮助学生进一步完善作文。

OK，开始编写：

class WriteAction(Action):"""学生Agent的撰写作文Action。"""name: str = "WriteEssay"PROMPT_TEMPLATE: str = """这里是历史对话记录：{msg}。请你根据用户提供的主题撰写一篇作文，只返回生成的作文内容，不包含其他文本。如果老师提供了关于作文的建议，请根据建议修改你的历史作文并返回。你的作文如下："""async def run(self, msg: str):"""根据用户提供的主题撰写一篇作文，并在收到老师的修改建议后进行修改。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rspclass ReviewAction(Action):"""老师Agent的审阅作文Action。"""name: str = "ReviewEssay"PROMPT_TEMPLATE: str = """这里是历史对话记录：{msg}。你是一名老师，现在请检查学生创作的关于用户提供的主题的作文，并给出你的修改建议。你更喜欢逻辑清晰的结构和有趣的口吻。只返回你的修改建议，不要包含其他文本。你的修改建议如下："""async def run(self, msg: str):"""审阅学生的作文，并给出修改建议。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rsp

接着，我们定义StudentAgent与TeacherAgent，与单智能体不同的是，我们需要声明每个Agent关注的动作(self._watch),只有当动作发生后，角色才开始行动，这样能保证整体的运行规律而不混乱；

class Student(Role):"""学生角色。"""name: str = "cheems"profile: str = "Student"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteAction])  # 设置学生的动作为撰写作文self._watch([UserRequirement, ReviewAction])  # 监听用户要求和老师的审阅动作async def _act(self) -> Message:"""学生动作：根据用户要求撰写作文或根据老师的修改建议修改作文。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆# logger.info(msg)essay_text = await WriteAction().run(msg)logger.info(f'student : {essay_text}')msg = Message(content=essay_text, role=self.profile, cause_by=type(todo))return msgclass Teacher(Role):"""老师角色。"""name: str = "laobai"profile: str = "Teacher"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewAction])  # 设置老师的动作为审阅作文self._watch([WriteAction])  # 监听学生的撰写作文动作async def _act(self) -> Message:"""老师动作：审阅学生的作文并给出修改建议。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆review_text = await ReviewAction().run(msg)logger.info(f'teacher : {review_text}')msg = Message(content=review_text, role=self.profile, cause_by=type(todo))return msg

要记得关注动作在init阶段；

设计完毕agent后,我们就可以开始撰写运行函数了，用户输入一个主题topic，并将topic发布在env中,以运行env，此时系统就开始工作了，我们可以通过修改对话轮数（n_round）来查看不同轮数checkPoint下的结果；

async def main(topic: str, n_round=5):"""运行函数，用户输入一个主题，并将主题发布在环境中，然后运行环境。"""classroom.add_roles([Student(), Teacher()])  # 向环境中添加学生和老师角色classroom.publish_message(Message(role="Human", content=topic, cause_by=UserRequirement,send_to='' or MESSAGE_ROUTE_TO_ALL),peekable=False,)# 发布一条消息，包含用户输入的主题，并将其发送给所有角色while n_round > 0:# self._save()n_round -= 1logger.debug(f"max {n_round=} left.")  # 输出剩余对话轮数await classroom.run()  # 运行环境return classroom.history  # 返回对话历史记录asyncio.run(main(topic='关于道德和法律的限制范围'))  # 运行主函数，输入主题为 "道德和法律的限制范围"

完整代码如下：

import asynciofrom metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environmentfrom metagpt.const import MESSAGE_ROUTE_TO_ALL
# 加载环境变量
from dotenv import load_dotenv 
load_dotenv()classroom = Environment()class WriteAction(Action):"""学生Agent的撰写作文Action。"""name: str = "WriteEssay"PROMPT_TEMPLATE: str = """这里是历史对话记录：{msg}。请你根据用户提供的主题撰写一篇作文，只返回生成的作文内容，不包含其他文本。如果老师提供了关于作文的建议，请根据建议修改你的历史作文并返回。你的作文如下："""async def run(self, msg: str):"""根据用户提供的主题撰写一篇作文，并在收到老师的修改建议后进行修改。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rspclass ReviewAction(Action):"""老师Agent的审阅作文Action。"""name: str = "ReviewEssay"PROMPT_TEMPLATE: str = """这里是历史对话记录：{msg}。你是一名老师，现在请检查学生创作的关于用户提供的主题的作文，并给出你的修改建议。你更喜欢逻辑清晰的结构和有趣的口吻。只返回你的修改建议，不要包含其他文本。你的修改建议如下："""async def run(self, msg: str):"""审阅学生的作文，并给出修改建议。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rspclass Student(Role):"""学生角色。"""name: str = "cheems"profile: str = "Student"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteAction])  # 设置学生的动作为撰写作文self._watch([UserRequirement, ReviewAction])  # 监听用户要求和老师的审阅动作async def _act(self) -> Message:"""学生动作：根据用户要求撰写作文或根据老师的修改建议修改作文。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆# logger.info(msg)essay_text = await WriteAction().run(msg)logger.info(f'student : {essay_text}')msg = Message(content=essay_text, role=self.profile, cause_by=type(todo))return msgclass Teacher(Role):"""老师角色。"""name: str = "laobai"profile: str = "Teacher"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewAction])  # 设置老师的动作为审阅作文self._watch([WriteAction])  # 监听学生的撰写作文动作async def _act(self) -> Message:"""老师动作：审阅学生的作文并给出修改建议。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆review_text = await ReviewAction().run(msg)logger.info(f'teacher : {review_text}')msg = Message(content=review_text, role=self.profile, cause_by=type(todo))return msgclass Student(Role):"""学生角色。"""name: str = "cheems"profile: str = "Student"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteAction])  # 设置学生的动作为撰写作文self._watch([UserRequirement, ReviewAction])  # 监听用户要求和老师的审阅动作async def _act(self) -> Message:"""学生动作：根据用户要求撰写作文或根据老师的修改建议修改作文。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆# logger.info(msg)essay_text = await WriteAction().run(msg)logger.info(f'student : {essay_text}')msg = Message(content=essay_text, role=self.profile, cause_by=type(todo))return msgclass Teacher(Role):"""老师角色。"""name: str = "laobai"profile: str = "Teacher"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewAction])  # 设置老师的动作为审阅作文self._watch([WriteAction])  # 监听学生的撰写作文动作async def _act(self) -> Message:"""老师动作：审阅学生的作文并给出修改建议。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆review_text = await ReviewAction().run(msg)logger.info(f'teacher : {review_text}')msg = Message(content=review_text, role=self.profile, cause_by=type(todo))return msgasync def main(topic: str, n_round=5):"""运行函数，用户输入一个主题，并将主题发布在环境中，然后运行环境。"""classroom.add_roles([Student(), Teacher()])  # 向环境中添加学生和老师角色classroom.publish_message(Message(role="Human", content=topic, cause_by=UserRequirement,send_to='' or MESSAGE_ROUTE_TO_ALL),peekable=False,)# 发布一条消息，包含用户输入的主题，并将其发送给所有角色while n_round > 0:# self._save()n_round -= 1logger.debug(f"max {n_round=} left.")  # 输出剩余对话轮数await classroom.run()  # 运行环境return classroom.history  # 返回对话历史记录asyncio.run(main(topic='关于道德和法律的限制范围'))  # 运行主函数，输入主题为 "道德和法律的限制范围"

运行结果如下：
result

很有趣，哈哈😂😂

四、MetaGPT中Team的设计思想

在上节中，我们通过师生交互的案例体验了多Agent开发的趣味性，现在让我们来了解一下Team。在官方介绍中，Team是一个重要的组件，它是基于Environment进行二次封装的结果。Team的代码如下：

class Team(BaseModel):"""Team: 由一个或多个角色（Agent）组成，具有SOP（标准运营程序）和一个用于即时消息传递的环境，专用于任意多Agent活动，如协同编写可执行代码。"""model_config = ConfigDict(arbitrary_types_allowed=True)env: Environment = Field(default_factory=Environment)  # Team的环境investment: float = Field(default=10.0)  # 团队投资idea: str = Field(default="")  # 团队想法

Team在Env的基础上增加了更多的组件。例如，Investment用于管理团队成本（限制Token花费），idea则用于告诉你的团队接下来应该围绕什么工作。Team有以下几个重要的方法：
①hire方法

  • 向团队中添加员工。
def hire(self, roles: list[Role]):"""招聘角色进行协作"""self.env.add_roles(roles)  # 在环境中添加角色

②invest方法

  • 计算Token,控制预算
def invest(self, investment: float):"""投资公司。当超过最大预算时，会引发NoMoneyException异常。"""self.investment = investmentCONFIG.max_budget = investmentlogger.info(f"Investment: ${investment}.")

③run_project方法

  • 发布需求
  • 初始化项目
def run_project(self, idea, send_to: str = ""):"""运行一个项目，从发布用户需求开始。"""self.idea = idea# 人类需求。self.env.publish_message(Message(role="Human", content=idea, cause_by=UserRequirement, send_to=send_to or MESSAGE_ROUTE_TO_ALL),peekable=False,)

在Team运行时，首先调用run_project方法给智能体提供一个需求，然后在n_round的循环过程中，重复检查预算和运行环境，最后返回环境中角色的历史对话。

@serialize_decorator
async def run(self, n_round=5, idea="", send_to="", auto_archive=True):"""运行公司，直到到达目标轮次或没有预算"""if idea:self.run_project(idea=idea, send_to=send_to)while n_round > 0:# self._save()n_round -= 1logger.debug(f"max {n_round=} left.")self._check_balance()await self.env.run()self.env.archive(auto_archive)return self.env.history

这里尽管Team类只是在Env上的简单封装，🤔但它向我们展示了如何向多智能体系统****发布启动消息以及引入可能的人类反馈。接下来，我们将使用Team，开发属于自己的第一个智能体团队。

五、基于Team的Agent开发团队

1.需求分析

学习完Team的设计思想后，我们就本系列课程3的思路进行研究，我们用Team将其实现一遍；还记得当初我们的需求吗？下面是当初是思路流程图：
requirement
本文中，我们需要构建一个包含需求分析，代码撰写，代码测试，代码评审的Team开发团队：
下面是作者是思路：

  1. 定义每个Agent执行的行动Action;
    • RequirementAnalysisAction：需求分析
    • CodeWriteAction：代码撰写
    • CodeTestAction：代码测试
    • CodeReviewAction：代码评审
  2. 基于SOP流程，确保每个Agent既可以观察到上个Agent的输出结果，也能保证****将自己的输出传递给下一个Agent;
  3. 初始化所有Agent，并将这些Agent添加进入Team实例，创建一个存在内部环境的智能体团队，使Agent之间能够进行交互。

现在我们开始撰写代码！😺😺

2.正式开发

先导入第三方库

import re
import fire # 新增了招募
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
import subprocess
# 加载环境变量
from dotenv import load_dotenv 
load_dotenv()

撰写每个Agent的Action,包括需求分析，代码撰写，代码测试，代码评审：

# 需求分析优化Action
class RequirementsOptAction(Action):PROMPT_TEMPLATE: str = """你要遵守的规范有：1.简要说明 (Brief Description)简要介绍该用例的作用和目的。2.事件流 (Flow of Event)包括基本流和备选流，事件流应该表示出所有的场景。3.用例场景 (Use-Case Scenario)包括成功场景和失败场景，场景主要是由基本流和备选流组合而成的。4.特殊需求 (Special Requirement)描述与该用例相关的非功能性需求（包括性能、可靠性、可用性和可扩展性等）和设计约束（所使用的操作系统、开发工具等）。5.前置条件 (Pre-Condition)执行用例之前系统必须所处的状态。6.后置条件 (Post-Condition)用例执行完毕后系统可能处于的一组状态。请优化以下需求，使其更加明确和全面：{requirements}"""name: str = "RequirementsOpt"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)return rsp.strip()  # 返回优化后的需求# 代码撰写Action
class CodeWriteAction(Action):PROMPT_TEMPLATE: str = """根据以下需求，编写一个能够实现{requirements}的Python函数，并提供两个可运行的测试用例。返回的格式为：```python\n你的代码\n```，请不要包含其他的文本。```python# your code here```"""name: str = "CodeWriter"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)code_text = CodeWriteAction.parse_code(rsp)return code_text@staticmethoddef parse_code(rsp): # 从模型生成中字符串匹配提取生成的代码pattern = r'```python(.*?)```'  # 使用非贪婪匹配match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_text# 代码测试Action
class CodeTestAction(Action):PROMPT_TEMPLATE: str = """上下文：{context}为给定的函数编写 {k} 个单元测试，并且假设你已经导入了该函数。返回 ```python 您的测试代码 ```，且不包含其他文本。your code:"""name: str = "CodeTest"async def run(self, code_text: str,k:int = 5):try:result = subprocess.run(['python', '-c', code_text],text=True,capture_output=True,check=True)return result.stdoutexcept subprocess.CalledProcessError as e:return e.stderrclass CodeReviewAction(Action):PROMPT_TEMPLATE: str = """context：{context}审查测试用例并提供一个关键性的review,在评论中，请包括对测试用例覆盖率的评估，以及对测试用例的可维护性和可读性的评估。同时，请提供具体的改进建议。"""name: str = "CodeReview"async def run(self, context: str):prompt = self.PROMPT_TEMPLATE.format(context=context)rsp = await self._aask(prompt)return rsp

在多智能体系统中，我们定义Agent有两个重点：

  1. 使用 set_actions方法 为Agent配备对应的 Action，这与单智能体思路相同;
  2. SOP流程中，每个Agent的输入都是上一个Agent的输出，因此每个Agent在初始化的时候都通过self._watch来监听上一个Agent的行动Action，以保证正确顺序执行；对于第一个Agent,我们监听用户的输入UserRequirement；

不知道大家有没有想过同时监听两个或多个Action的是什么结果呢？是两个Action都执行完，该Agent才执行自己的Action，还是任意一个执行完就执行自己的Action呢？大家可以试一试，作者996或许得在下一篇文章前会去试一试；

好了我们继续将Agent的设计一次完善,代码如下：作者这里直接使用官方案例，略有修改：

class RA(Role): #需求分析师缩写name: str = "yake"profile: str = "Requirement Analysis"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement])self.set_actions([RequirementsOptAction])class Coder(Role):name: str = "cheems"profile: str = "Coder"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([RequirementsOptAction])self.set_actions([CodeWriteAction])class Tester(Role):name: str = "Bob"profile: str = "Tester"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeTestAction])# self._watch([SimpleWriteCode])self._watch([CodeWriteAction,CodeReviewAction])  # 这里测试一下同时监听两个动作是什么效果async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo# context = self.get_memories(k=1)[0].content # use the most recent memory as contextcontext = self.get_memories()  # 获取所有记忆，避免重复检查code_text = await todo.run(context, k=5)  # specify argumentsmsg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msgclass Reviewer(Role):name: str = "Charlie"profile: str = "Reviewer"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeReviewAction])self._watch([CodeTestAction])

OK,当前Team中需要的Agent全部定义完毕，我们开始初始化Team,并通过用户输入运行；代码如下：

async def main(idea: str = "撰写一个python自动生成随机人物数据并保存到csv的tkinter程序,用户输入数量，则随机生成人物信息保存csv到当前文件夹下",investment: float = 3.0, # token限制3美金n_round: int = 5, # 循环5 轮add_human: bool = False, # 无需用户参与评审
):logger.info(idea)team = Team()team.hire([RA(),Coder(),Tester(),Reviewer(is_human=add_human),])team.invest(investment=investment) # 计算成本预算team.run_project(idea) # 初始化项目await team.run(n_round=n_round) # 开始循环if __name__ == "__main__":fire.Fire(main)

完整代码如下：

import re
import fire # 新增了招募
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
import subprocess
# 加载环境变量
from dotenv import load_dotenv 
load_dotenv()# 需求分析优化Action
class RequirementsOptAction(Action):PROMPT_TEMPLATE: str = """你要遵守的规范有：1.简要说明 (Brief Description)简要介绍该用例的作用和目的。2.事件流 (Flow of Event)包括基本流和备选流，事件流应该表示出所有的场景。3.用例场景 (Use-Case Scenario)包括成功场景和失败场景，场景主要是由基本流和备选流组合而成的。4.特殊需求 (Special Requirement)描述与该用例相关的非功能性需求（包括性能、可靠性、可用性和可扩展性等）和设计约束（所使用的操作系统、开发工具等）。5.前置条件 (Pre-Condition)执行用例之前系统必须所处的状态。6.后置条件 (Post-Condition)用例执行完毕后系统可能处于的一组状态。请优化以下需求，使其更加明确和全面：{requirements}"""name: str = "RequirementsOpt"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)return rsp.strip()  # 返回优化后的需求# 代码撰写Action
class CodeWriteAction(Action):PROMPT_TEMPLATE: str = """根据以下需求，编写一个能够实现{requirements}的Python函数，并提供两个可运行的测试用例。返回的格式为：```python\n你的代码\n```，请不要包含其他的文本。```python# your code here```"""name: str = "CodeWriter"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)code_text = CodeWriteAction.parse_code(rsp)return code_text@staticmethoddef parse_code(rsp): # 从模型生成中字符串匹配提取生成的代码pattern = r'```python(.*?)```'  # 使用非贪婪匹配match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_text# 代码测试Action
class CodeTestAction(Action):PROMPT_TEMPLATE: str = """上下文：{context}为给定的函数编写 {k} 个单元测试，并且假设你已经导入了该函数。返回 ```python 您的测试代码 ```，且不包含其他文本。your code:"""name: str = "CodeTest"async def run(self, context: str, k: int = 5):prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)rsp = await self._aask(prompt)code_text = CodeWriteAction.parse_code(rsp)return code_textclass CodeReviewAction(Action):PROMPT_TEMPLATE: str = """context：{context}审查测试用例并提供一个关键性的review,在评论中，请包括对测试用例覆盖率的评估，以及对测试用例的可维护性和可读性的评估。同时，请提供具体的改进建议。"""name: str = "CodeReview"async def run(self, context: str):prompt = self.PROMPT_TEMPLATE.format(context=context)rsp = await self._aask(prompt)return rsp
class RA(Role): #需求分析师缩写name: str = "yake"profile: str = "Requirement Analysis"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement])self.set_actions([RequirementsOptAction])class Coder(Role):name: str = "cheems"profile: str = "Coder"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([RequirementsOptAction])self.set_actions([CodeWriteAction])class Tester(Role):name: str = "Bob"profile: str = "Tester"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeTestAction])# self._watch([SimpleWriteCode])self._watch([CodeWriteAction,CodeReviewAction])  # 这里测试一下同时监听两个动作是什么效果async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo# context = self.get_memories(k=1)[0].content # use the most recent memory as contextcontext = self.get_memories()  # 获取所有记忆，避免重复检查code_text = await todo.run(context, k=5)  # specify argumentsmsg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msgclass Reviewer(Role):name: str = "Charlie"profile: str = "Reviewer"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeReviewAction])self._watch([CodeTestAction])async def main(idea: str = "撰写一个python自动生成随机人物数据并保存到csv的tkinter程序,用户输入数量，则随机生成人物信息保存csv到当前文件夹下",investment: float = 3.0, # token限制3美金n_round: int = 5, # 循环5 轮add_human: bool = False, # 无需用户参与评审
):logger.info(idea)team = Team()team.hire([RA(),Coder(),Tester(),Reviewer(is_human=add_human),])team.invest(investment=investment) # 计算成本预算team.run_project(idea) # 初始化项目await team.run(n_round=n_round) # 开始循环if __name__ == "__main__":fire.Fire(main)

运行效果如下：
result

嘿嘿😀，运行成功！可惜代码运行逻辑不稳定😣，容易报错，作者就删去了这部分代码 ；

总结

在本文中，各位读者和作者一起学习了MetaGPT多智能体开发中环境Environment的定义和Team的设计思想，并通过师生互动案例和开发小组案例，体验了其具体应用；虽然案例相对简单，但是也足以说明多Agent框架在复杂问题中的潜力了;
通过对任务的原子级分解，统筹成本和效率，作者认为Agent的开发一定逐渐会改变我们生活的方方面面；真令人激动！🫡
好了，不多说，感谢大家的支持。作者虽然已经熬夜一周了😣，但是这一周来对Agent的学习帮到了作者很多，希望作者的文章也能帮到你🎉🎉🎉😀；

课后作业

  • 你画我猜

基于 env 或 team 设计一个你的多智能体团队，尝试让他们完成 你画我猜文字版 ，要求其中含有两个agent，其中一个agent负责接收来自用户提供的物体描述并转告另一个agent，另一个agent将猜测用户给出的物体名称，两个agent将不断交互直到另一个给出正确的答案
（也可以在系统之上继续扩展，比如引入一个agent来生成词语，而人类参与你画我猜的过程中）
给出完整的代码和详细注释，并在后面补充实现效果：

下面是作者的思路和实现效果:

设计思路

1.Action方法设计
  • describe_item:接受用户提供的物体，对其进行描述并返回给猜测者，
  • guess_item:接受描述者的描述，猜测物体;
2.Agent设计

我们需要设计两个智能体（Agent）：描述者和猜测者：

  1. 描述者（DescriberAgent）：接收物体词汇并生成描述文本。
  2. 猜测者（GuesserAgent）：根据描述文本进行猜测。

游戏流程如下：

  • 用户将一个物体词汇发送给描述者。
  • 描述者生成描述文本，并将其发送给猜测者。
  • 猜测者根据描述文本进行猜测，并将猜测结果返回给描述者。
3.完整代码实现

以下是完整的代码实现：

import re
import fire
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
from dotenv import load_dotenv
from typing import ClassVarload_dotenv()# 描述Action
class DescribeItem(Action):PROMPT_TEMPLATE: str = """请根据以下物体词汇生成描述文本：可以对物体词汇侧面描写，但是不能直接说明其名称，你的生成内容是让别人猜测的；例如： "苹果": "这是一种红色或绿色的水果，圆形，味道甜或酸。""桌子": "这是一个家具，有四条腿，用来放置物品。",当前如下：词汇：{word}"""name: str = "DescribeItem"async def run(self, word):prompt = self.PROMPT_TEMPLATE.format(word=word)res = await self._aask(prompt)return res# 猜测Action
class GuessItem(Action):PROMPT_TEMPLATE: str = """根据以下描述文本进行猜测物体名称：描述：{description}例如：描述为："这是一种红色或绿色的水果，圆形，味道甜或酸。"，你需要猜测为: "苹果",你的输出格式如下，猜测结果用方括号扩住：[苹果]"""name: str = "Guess"async def run(self, description):prompt = self.PROMPT_TEMPLATE.format(description=description)result = await self._aask(prompt)return self.parse_item(result)@staticmethoddef parse_item(rsp):pattern = r'\[(.*?)\]'match = re.search(pattern, rsp, re.DOTALL)item = match.group(1) if match else rspreturn itemclass DescriberAgent(Role):name: str = "Describer"profile: str = "负责生成物体描述文本的描述者"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement,GuessItem])self.set_actions([DescribeItem])async def _act(self) -> Message:"""描述者动作：根据猜测者的回答修改描述。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆# logger.info(msg)prompt = "这是猜测者的返回：{msg},如果这不是正确答案，请修改描述"describe = await DescribeItem().run(prompt)logger.info(f'DescriberAgent : {describe}')msg = Message(content=describe, role=self.profile, cause_by=type(todo))return msgclass GuesserAgent(Role):name: str = "Guesser"profile: str = "负责猜测物体名称的猜测者"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([DescribeItem])self.set_actions([GuessItem])async def _act(self) -> Message:"""猜测者动作：根据描述者的描述修改猜测结果。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有对话记忆# logger.info(msg)prompt = "这是描述者的返回：{msg},如何这不是正确答案，请修改结果重新回答"guess = await GuessItem().run(msg)logger.info(f'GuesserAgent : {guess}')msg = Message(content=guess, role=self.profile, cause_by=type(todo))return msgasync def main(word: str = "猫", idea: str = "鸡你太美", investment: float = 3.0, add_human: bool = False, n_round=5):logger.info(idea)team = Team()team.hire([DescriberAgent(), GuesserAgent()])team.invest(investment=investment)team.run_project(idea) # 初始化项目await team.run(n_round=n_round) # 开始循环if __name__ == "__main__":fire.Fire(main)

实现效果如下：
result

本文已经足够长了，考虑到读者的用户体验，BabyAGI的内容将在下一篇中撰写实现；

项目地址

  • Github地址
  • 拓展阅读

如果觉得我的文章对您有帮助，三连+关注便是对我创作的最大鼓励！或者一个star🌟也可以😂.

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/bicheng/15844.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【vue】el-select选择器实现宽度自适应

【vue】el-select选择器实现宽度自适应

选择器的宽度根据内容长度进行变化 <div class"Space_content"><el-selectv-model"value":placeholder"$t(bot.roommessage)"class"select"size"small"style"margin-right: 10px"change"selectcha…
阅读更多...
JavaSE——集合框架二（1/6）-前置知识-可变参数、Collections工具类

JavaSE——集合框架二（1/6）-前置知识-可变参数、Collections工具类

目录 可变参数 Collections工具类 Collections的常用静态方法 实例演示 可变参数 可变参数 就是一种特殊形参&#xff0c;定义在方法、构造器的形参列表里&#xff0c;格式是&#xff1a;数据类型...参数名称 可变参数的特点和好处 特点&#xff1a;可以不传数据给它&am…
阅读更多...
SQL常用基础语句（一）-- ABCDE开头

SQL常用基础语句（一）-- ABCDE开头

AS 将列名从 count(*) 修改为 total select count(*) as total from users where status0 将列名 username 改为 uname&#xff0c; password 改为 upwd select username as uname, password as upwd from users BETWEEN AND 说明&#xff1a;BETWEEN 筛选的是 >value1且 &l…
阅读更多...
小程序主体变更是通过迁移吗?是需要2个小程序吗?

小程序主体变更是通过迁移吗?是需要2个小程序吗?

小程序迁移变更主体有什么作用&#xff1f;好多朋友都想做小程序迁移变更主体&#xff0c;但是又不太清楚具体有啥用&#xff0c;今天我就来详细说说。首先&#xff0c;小程序迁移变更主体最重要的作用就是可以修改主体。比如你的小程序原来是 A 公司的&#xff0c;现在 A 公司…
阅读更多...
操作系统实验四：多线程与信号量编程

操作系统实验四：多线程与信号量编程

操作系统实验上机 更多技术请访问&#xff1a;www.xuanworld.top 部分审核不通过的文章将发至个人博客&#xff1a;www.xuanworld.top 欢迎来52破解论坛阅读帖子&#xff1a;https://www.52pojie.cn/thread-1891208-1-1.html 实验名称实验序号实验日期实验人多线程与信号量…
阅读更多...
010-Linux磁盘介绍

010-Linux磁盘介绍

文章目录 1、名词 2、类型 3、尺寸 4、接口/协议/总线 5、命名 6、分区方式 MBR分区 GPT分区 1、名词 磁盘是计算机主要的存储介质&#xff0c;可以存储大量的二进制数据&#xff0c;并且断电后也能保持数据不丢失。早期计算机使用的磁盘是软磁盘&#xff08;Floppy D…
阅读更多...
基于yolov5和desnet的猫咪识别模型

基于yolov5和desnet的猫咪识别模型

前言 前段时间给学校的猫咪小程序搭建了识猫模型&#xff0c;可以通过猫咪的照片辨别出是那只猫猫&#xff0c;这里分享下具体的方案&#xff0c;先看效果图&#xff1a; 源代码在文末 模型训练 在训练服务器&#xff08;或你的个人PC&#xff09;上拉取本仓库代码。 图片数…
阅读更多...
10款免费黑科技软件，强烈推荐！

10款免费黑科技软件，强烈推荐！

1.AI视频生成——巨日禄 网页版https://aitools.jurilu.com/ "巨日禄 "是一款功能强大的文本视频生成器&#xff0c;可以快速将文本内容转换成极具吸引力的视频。操作简单&#xff0c;用户只需输入文字&#xff0c;选择喜欢的样式和模板&#xff0c; “巨日禄”就会…
阅读更多...
Day39贪心算法part06

Day39贪心算法part06

LC738单调递增的数字&#xff08;未掌握&#xff09; 思路分析&#xff1a;一旦出现strNum[i - 1] > strNum[i]的情况&#xff08;非单调递增&#xff09;&#xff0c;首先想让strNum[i - 1]–&#xff0c;然后strNum[i]给为9字符串是不可变的&#xff0c;不可以使用s.char…
阅读更多...
树莓派学习笔记——树莓派的三种GPIO编码方式

树莓派学习笔记——树莓派的三种GPIO编码方式

1、板载编码&#xff08;Board pin numbering&#xff09;: 板载编码是树莓派上的一种GPIO引脚编号方式&#xff0c;它指的是按照引脚在树莓派主板上的物理位置来编号。这种方式对于初学者来说可能比较直观&#xff0c;因为它允许你直接根据引脚在板上的位置来编程。 2、BCM编…
阅读更多...
Linux gurb2简介

Linux gurb2简介

文章目录 前言一、GRUB 2简介二、GRUB 2相关文件/文件夹2.1 /etc/default/grub文件2.2 /etc/grub.d/文件夹2.3 /boot/grub/grub.cfg文件 三、grubx64.efi参考资料 前言 简单来说&#xff0c;引导加载程序&#xff08;boot loader&#xff09;是计算机启动时运行的第一个软件程…
阅读更多...
文章解读与仿真程序复现思路——电力系统保护与控制EI\CSCD\北大核心《计及温控厌氧发酵和阶梯碳交易的农村综合能源低碳经济调度》

文章解读与仿真程序复现思路——电力系统保护与控制EI\CSCD\北大核心《计及温控厌氧发酵和阶梯碳交易的农村综合能源低碳经济调度》

本专栏栏目提供文章与程序复现思路&#xff0c;具体已有的论文与论文源程序可翻阅本博主免费的专栏栏目《论文与完整程序》 论文与完整源程序_电网论文源程序的博客-CSDN博客https://blog.csdn.net/liang674027206/category_12531414.html 电网论文源程序-CSDN博客电网论文源…
阅读更多...
网络域名是什么意思

网络域名是什么意思

网络域名&#xff0c;顾名思义&#xff0c;就是网络上的名字&#xff0c;类似于现实中的地址或姓名一样&#xff0c;用来标识网络上的一个或一组计算机或服务器的位置&#xff0c;以及它们的相应服务资源。网络域名是互联网上最基础的基础设施之一&#xff0c;是网络通信的“标…
阅读更多...
【mysql】更新操作是如何执行的

【mysql】更新操作是如何执行的

现有一张表&#xff0c;建表语句如下&#xff1a; mysql> create table T(ID int primary key, c int);如果要将 ID2 这一行的a字段值加 1&#xff0c;SQL语句会这么写&#xff1a; mysql> update T set c c 1 where ID 2;上面这条sql执行时&#xff0c;分析器会通过词…
阅读更多...
Nacos 微服务管理

Nacos 微服务管理

Nacos 本教程将为您提供Nacos的基本介绍&#xff0c;并带您完成Nacos的安装、服务注册与发现、配置管理等功能。在这个过程中&#xff0c;您将学到如何使用Nacos进行微服务管理。下方是官方文档&#xff1a; Nacos官方文档 1. Nacos 简介 Nacos&#xff08;Naming and Confi…
阅读更多...
操作符详解（上）（新手向）

操作符详解（上）（新手向）

操作符详解&#xff08;上&#xff09; 一&#xff0c;算术操作符&#xff08;双目操作符&#xff09;1:‘’,‘-’,‘*’2&#xff1a;‘/’&#xff0c;‘%’ 一&#xff0c;单目操作符1:‘’,‘-’2&#xff1a;‘!’3&#xff1a;‘&’4&#xff1a;‘*’5&#xff1a;…
阅读更多...
linux 排查java内存溢出(持续更新中)

linux 排查java内存溢出(持续更新中)

场景 tone.jar 启动后内存溢出,假设pid 为48044 排查 1.确定java程序的pid(进程id) ps 或 jps 都可以 ps -ef | grep tone jps -l 2.查看堆栈信息 jmap -heap 48044 3.查看对象的实例数量显示前30 jmap -histo:live 48044 | head -n 30 4.查看线程状态 jstack 48044
阅读更多...
Spring 事件监听

Spring 事件监听

参考&#xff1a;Spring事件监听流程分析【源码浅析】_private void processbean(final string beanname, fi-CSDN博客 一、简介 Spring早期通过实现ApplicationListener接口定义监听事件&#xff0c;Spring 4.2开始通过EventListener注解实现监听事件 FunctionalInterface p…
阅读更多...
Rustdesk客户端源码编译

Rustdesk客户端源码编译

1.安装VCPKG windows平台vcpkg安装-CSDN博客 2.使用VCPKG安装: windows平台vcpkg安装-CSDN博客 配置VCPKG_ROOT环境变量: 安装静态库: ./vcpkg install libvpx:x64-windows-static libyuv:x64-windows-static opus:x64-windows-static aom:x64-windows-static 静态库安装成…
阅读更多...
【C语言深度解剖】(15)：动态内存管理和柔性数组

【C语言深度解剖】(15)：动态内存管理和柔性数组

&#x1f921;博客主页&#xff1a;醉竺 &#x1f970;本文专栏&#xff1a;《C语言深度解剖》 &#x1f63b;欢迎关注&#xff1a;感谢大家的点赞评论关注&#xff0c;祝您学有所成&#xff01; ✨✨&#x1f49c;&#x1f49b;想要学习更多C语言深度解剖点击专栏链接查看&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023