前言
你好,我是GISer Liu,在上一篇文章中,我们用了两万多字详细拆解了单个Agent的组成,并通过Github Trending订阅智能体理解MetaGPT框架的订阅模块如何解决应用问题,但是对于复杂,并行的任务,单个智能体是不能胜任;今天我们将进入多智能体开发的学习阶段;一起期待吧😀
一、介绍
在本文中,我们将分别详细介绍:
- MetaGPT中Environment的设计思想;
- 构建简单师生对话多Agent框架;
- MetaGPT中Team的设计思想;
- 构建 多Agent 开发团队;
- 构建 多Agent 辩论团队;
- 你画我猜多Agent框架实现;
二、Environment 环境设计思想
在MetaGPT框架中,Environment
(环境)与Agent
(智能体)这两个概念借鉴了强化学习的思想。而在强化学习中,Agent
需要在环境中采取行动以最大化奖励。而在MetaGPT
中,则提供了一个标准的环境组件Environment
,用来管理Agent的活动与信息交流
。
学习 agent 与环境进行交互的思想可以去OpenAI的GYM项目看看
1.环境设计原理
MetaGPT中的环境设计分为外部环境(ExtEnv)
和内部环境
,旨在帮助Agent代理
与不同的外部应用场景(如游戏、手机应用等)以及内部开发和操作环境进行交互。
①外部环境(ExtEnv)
定义:
外部环境是代理与外部世界交互的接口。它为代理提供了一种机制,使其能够与外部系统(例如游戏引擎、移动应用API)进行通信和交互。
继承和扩展:
ExtEnv
类是所有外部环境的基础类,各种具体的外部环境(如Minecraft环境、狼人游戏环境等)会继承这个基础类,并在其上扩展实现特定的交互逻辑。
示例:
-
游戏环境:
- 假设有一个在线游戏提供了API,允许查询玩家状态和执行游戏动作。
ExtEnv
类封装了这些API,使代理能够调用这些API来查询游戏状态和执行动作。
Agent执行某个Action,该Action中封装了执行API调用的逻辑
-
狼人sha游戏:
- 在狼人游戏中,代理需要知道每晚和每天的游戏状态。
- ExtEnv类定义了获取这些状态的方法,使代理能够在游戏中做出决策。
- Minecraft开发API
- Agent狼人sha实现案例
②内部环境
(1)定义:
内部环境是代理及其团队直接使用的开发和操作环境。它类似于软件开发中的工作环境,包括开发工具、测试框架和配置文件等。
(2)继承和扩展:
内部环境类(XxxEnv)通常继承自一个基础环境类,并根据具体需求进行定制和扩展。这个基础环境类可以提供一些通用功能,比如日志记录、错误处理等。
(3)案例:
- 开发环境:
- 基础环境类可能提供一些通用的开发工具和测试框架。
- 开发团队可以在这个基础上添加特定项目所需的工具和配置,例如数据库连接配置、CI/CD脚本等。
作者认为其思想和ChatDev的实现相似;
2.环境交互设计
MetaGPT还引入了两个重要的概念:observation_space
和action_space
。这些概念来自强化学习领域,用于描述代理从环境中获取的状态信息和可以采取的动作集合。
observation_space:
-
表示代理可以从环境中获得的所有可能的状态。
-
例如,在游戏环境中,
observation_space
可能包括玩家的位置、游戏时间、得分等。在上图Minecraft
的案例中,观察空间就是周围的环境,角色的血量与护甲,拥有的工具与工具的数量;
action_space:
- 表示代理在环境中可以执行的所有可能的动作。
- 例如,在游戏环境中,
action_space
可能包括移动、跳跃、攻击等,同样在上面的案例中,action_space
代表可选Action的集合,例如看到树以后选择砍树,看到怪物后选择逃离还是进攻;这需要Agent
通过反思机制来判断进行;
通过定义这两个空间,MetaGPT能够更好地抽象不同环境中的具体细节,使得环境提供者可以专注于实现环境逻辑,而代理使用者可以专注于状态和动作的处理。
3.环境运行机制
这里放这张图供大家思考
①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 中;
最后,再看看,这张图,我想你会记忆更加深刻,当然,如果作者认知有偏颇,读者也可以在评论区指出,感谢支持
三、简单的师生交互多智能体系统
在上一节中,我们已经了解了environment
环境的基本构成与它的运行逻辑,在这一节中,我们将学习如何利用environment
来进行开发,进一步了解environment
组件内部的活动,
现在设想一个多Agent交互的应用场景,我的想法是两人对话场景,如:
师生交互场景:
- 首先用户输入一个主题;
- 然后学生Agent负责根据用户的输入进行作文撰写
- 当老师Agent发现学生Agent写作完毕以后,就会给学生提出学习意见;
- 根据老师Agent给的意见,学生将修改自己的作品;
- 如此循环直到设定的循环次数结束;这里环境则是教室;
接下来我们用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
,这里的思路基本就是简单的提示词工程,学生要求有写作格式和写作主题写作,老师有检查标准和检查功能;
规范点说就是:
- 实现
WriteAction
方法:在这个方法中,学生Agent需要根据用户提供的主题撰写一篇作文。同时,当收到来自老师的修改建议后,也需要对作文进行相应的修改。 - 实现
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='关于道德和法律的限制范围')) # 运行主函数,输入主题为 "道德和法律的限制范围"
运行结果如下:
很有趣,哈哈😂😂
四、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将其实现一遍;还记得当初我们的需求吗?下面是当初是思路流程图:
本文中,我们需要构建一个包含需求分析,代码撰写,代码测试,代码评审的Team开发团队:
下面是作者是思路:
- 定义每个Agent执行的行动Action;
RequirementAnalysisAction
:需求分析CodeWriteAction
:代码撰写CodeTestAction
:代码测试CodeReviewAction
:代码评审
- 基于SOP流程,确保每个
Agent
既可以观察到上个Agent
的输出结果,也能保证****将自己的输出传递给下一个Agent
; - 初始化所有
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有两个重点:
- 使用
set_actions
方法 为Agent
配备对应的Action
,这与单智能体思路相同; - 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)
运行效果如下:
嘿嘿😀,运行成功!可惜代码运行逻辑不稳定😣,容易报错,
作者就删去了这部分代码;
总结
在本文中,各位读者和作者一起学习了MetaGPT多智能体开发中环境Environment
的定义和Team
的设计思想,并通过师生互动案例和开发小组案例,体验了其具体应用;虽然案例相对简单,但是也足以说明多Agent框架在复杂问题中的潜力了;
通过对任务的原子级分解,统筹成本和效率,作者认为Agent的开发一定逐渐会改变我们生活的方方面面;真令人激动!🫡
好了,不多说,感谢大家的支持。作者虽然已经熬夜一周了😣,但是这一周来对Agent
的学习帮到了作者很多,希望作者的文章也能帮到你🎉🎉🎉😀;
课后作业
- 你画我猜
基于 env 或 team 设计一个你的多智能体团队,尝试让他们完成 你画我猜文字版 ,要求其中含有两个agent,其中一个agent负责接收来自用户提供的物体描述并转告另一个agent,另一个agent将猜测用户给出的物体名称,两个agent将不断交互直到另一个给出正确的答案
(也可以在系统之上继续扩展,比如引入一个agent来生成词语,而人类参与你画我猜的过程中)
给出完整的代码和详细注释,并在后面补充实现效果:
下面是作者的思路和实现效果:
设计思路
1.Action方法设计
- describe_item:接受用户提供的物体,对其进行描述并返回给猜测者,
- guess_item:接受描述者的描述,猜测物体;
2.Agent设计
我们需要设计两个智能体(Agent):描述者和猜测者:
- 描述者(DescriberAgent):接收物体词汇并生成描述文本。
- 猜测者(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)
实现效果如下:
本文已经足够长了,考虑到读者的用户体验,BabyAGI的内容将在下一篇中撰写实现;
项目地址
- Github地址
- 拓展阅读
如果觉得我的文章对您有帮助,三连+关注便是对我创作的最大鼓励!或者一个star🌟也可以😂.