网站展示型推广有哪些,西安广告公司排名,施工企业岗位证书有哪些,网站内怎样做关键词有效果本系列文章跟随《MetaGPT多智能体课程》#xff08;https://github.com/datawhalechina/hugging-multi-agent#xff09;#xff0c;深入理解并实践多智能体系统的开发。
本文为该课程的第四章#xff08;多智能体开发#xff09;的第一篇笔记。主要记录下多智能体的运行…本系列文章跟随《MetaGPT多智能体课程》https://github.com/datawhalechina/hugging-multi-agent深入理解并实践多智能体系统的开发。
本文为该课程的第四章多智能体开发的第一篇笔记。主要记录下多智能体的运行机制及跟着教程实现一个简单的多智能体系统。
系列笔记
【AI Agent系列】【MetaGPT多智能体学习】0. 环境准备 - 升级MetaGPT 0.7.2版本及遇到的坑【AI Agent系列】【MetaGPT多智能体学习】1. 再理解 AI Agent - 经典案例和热门框架综述【AI Agent系列】【MetaGPT多智能体学习】2. 重温单智能体开发 - 深入源码理解单智能体运行框架 文章目录 系列笔记0. 多智能体间互通的方式 - Enviroment组件0.1 多智能体间协作方式简介0.2 Environment组件 - 深入源码0.2.1 参数介绍0.2.2 add_roles函数 - 承载角色0.2.3 publish_message函数 - 接收角色发布的消息 / 环境中的消息被角色得到0.2.4 run函数 - 运行入口 1. 开发一个简单的多智能体系统1.1 多智能体需求描述1.2 代码实现1.2.1 学生智能体1.2.2 老师智能体1.2.3 创建多智能体交流的环境1.2.4 运行1.2.5 完整代码 2. 总结 0. 多智能体间互通的方式 - Enviroment组件
0.1 多智能体间协作方式简介
在上次课中我也曾写过 多智能体运行机制 的笔记这篇文章【AI Agent系列】【MetaGPT】【深入源码】智能体的运行周期以及多智能体间如何协作。其中我自己通过看源码总结出了MetaGPT多智能体间协作的方式
1每一个Role都在不断观察环境中的信息_observe函数 2当观察到自己想要的信息后就会触发后续相应的动作 3如果没有观察到想要的信息则会一直循环观察 4执行完动作后会将产生的msg放到环境中publish_message供其它Role智能体来使用。
现在再结合《MetaGPT多智能体》课程的教案发现还是理解地有些浅了我只关注了智能体在关注自己想要的信息观察到了之后就开始行动而忽略了其背后的基础组件 - Enviroment。
0.2 Environment组件 - 深入源码
MetaGPT中对Environment的定义环境承载一批角色角色可以向环境发布消息可以被其他角色观察到。短短一句话总结三个功能
承载角色接收角色发布的消息环境中的消息被角色得到
下面我们分开来细说。
0.2.1 参数介绍
首先来看下 Environment 的参数
class Environment(ExtEnv):环境承载一批角色角色可以向环境发布消息可以被其他角色观察到Environment, hosting a batch of roles, roles can publish messages to the environment, and can be observed by other rolesmodel_config ConfigDict(arbitrary_types_allowedTrue)desc: str Field(default) # 环境描述roles: dict[str, SerializeAsAny[Role]] Field(default_factorydict, validate_defaultTrue)member_addrs: Dict[Role, Set] Field(default_factorydict, excludeTrue)history: str # For debugcontext: Context Field(default_factoryContext, excludeTrue)desc环境的描述roles字典类型指定当前环境中的角色member_addrs字典类型表示当前环境中的角色以及他们对应的状态history记录环境中发生的消息记录context当前环境的一些上下文信息
0.2.2 add_roles函数 - 承载角色
def add_roles(self, roles: Iterable[Role]):增加一批在当前环境的角色Add a batch of characters in the current environmentfor role in roles:self.roles[role.profile] rolefor role in roles: # setup system message with rolesrole.set_env(self)role.context self.context从源码中看这个函数的功能有两个
1给 Environment 的 self.roles 参数赋值上面我们已经知道它是一个字典类型现在看来它的 key 为 role 的 profile值为 role 本身。 疑问 role 的 profile 默认是空profile: str 可以没有值那如果使用者懒得写profile这里会不会最终只有一个 Role导致无法实现多智能体欢迎讨论交流。 2给添加进来的 role 设置环境信息
Role的 set_env 函数源码如下它又给env设置了member_addrs。有点绕
def set_env(self, env: Environment):Set the environment in which the role works. The role can talk to the environment and can also receivemessages by observing.self.rc.env envif env:env.set_addresses(self, self.addresses)self.llm.system_prompt self._get_prefix()self.set_actions(self.actions) # reset actions to update llm and prefix通过 add_roles 函数将 role 和 Environment 关联起来。
0.2.3 publish_message函数 - 接收角色发布的消息 / 环境中的消息被角色得到
Environment 中的 publish_message 函数是 Role 在执行完动作之后将自身产生的消息发布到环境中调用的接口。Role的调用方式如下
def publish_message(self, msg):If the role belongs to env, then the roles messages will be broadcast to envif not msg:returnif not self.rc.env:# If env does not exist, do not publish the messagereturnself.rc.env.publish_message(msg)通过上面的代码来看一个Role只能存在于一个环境中 然后看下 Environment 中源码
def publish_message(self, message: Message, peekable: bool True) - bool:Distribute the message to the recipients.In accordance with the Message routing structure design in Chapter 2.2.1 of RFC 116, as already plannedin RFC 113 for the entire system, the routing information in the Message is only responsible forspecifying the message recipient, without concern for where the message recipient is located. How toroute the message to the message recipient is a problem addressed by the transport framework designedin RFC 113.logger.debug(fpublish_message: {message.dump()})found False# According to the routing feature plan in Chapter 2.2.3.2 of RFC 113for role, addrs in self.member_addrs.items():if is_send_to(message, addrs):role.put_message(message)found Trueif not found:logger.warning(fMessage no recipients: {message.dump()})self.history f\n{message} # For debugreturn True其中重点是这三行代码检查环境中的所有Role是否订阅了该消息或该消息是否应该发送给环境中的某个Role如果订阅了则调用Role的put_message函数Role的 put_message函数的作用是将message放到本身的msg_buffer中
for role, addrs in self.member_addrs.items():if is_send_to(message, addrs):role.put_message(message)这样就实现了将环境中的某个Role的Message放到环境中并通知给环境中其它的Role的机制。
0.2.4 run函数 - 运行入口
run函数的源码如下
async def run(self, k1):处理一次所有信息的运行Process all Role runs at oncefor _ in range(k):futures []for role in self.roles.values():future role.run()futures.append(future)await asyncio.gather(*futures)logger.debug(fis idle: {self.is_idle})k 1, 表示处理一次消息为什么要有这个值难道还能处理多次意义是什么 该函数其实就是遍历一遍当前环境中的所有Role然后运行Role的run函数运行单智能体。
Role的run里面就是用该环境观察信息行动发布信息到环境。这样多个Role的运行就通过 Environment 串起来了这些之前已经写过了不再赘述见【AI Agent系列】【MetaGPT】【深入源码】智能体的运行周期以及多智能体间如何协作。
1. 开发一个简单的多智能体系统
复现教程中的demo。
1.1 多智能体需求描述
两个智能体学生 和 老师任务及预期流程学生写诗 — 老师给改进意见 — 学生根据意见改进 — 老师给改进意见 — … n轮 … — 结束
1.2 代码实现
1.2.1 学生智能体
学生智能体的主要内容就是根据指定的内容写诗。
因此首先定义一个写诗的ActionWritePoem。
然后定义学生智能体指定它的动作就是写诗self.set_actions([WritePoem])
那么它什么时候开始动作呢通过 self._watch([UserRequirement, ReviewPoem]) 设置其观察的信息。当它观察到环境中有了 UserRequirement 或者 ReviewPoem 产生的信息之后开始动作。UserRequirement为用户的输入信息类型。
class WritePoem(Action):name: str WritePoemPROMPT_TEMPLATE: str Here is the historical conversation record : {msg} .Write a poem about the subject provided by human, Return only the content of the generated poem with NO other texts.If the teacher provides suggestions about the poem, revise the students poem based on the suggestions and return.your poem:async def run(self, msg: str):prompt self.PROMPT_TEMPLATE.format(msg msg)rsp await self._aask(prompt)return rspclass Student(Role):name: str xiaomingprofile: str Studentdef __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WritePoem])self._watch([UserRequirement, ReviewPoem])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)poem_text await WritePoem().run(msg)logger.info(fstudent : {poem_text})msg Message(contentpoem_text, roleself.profile,cause_bytype(todo))return msg1.2.2 老师智能体
老师的智能体的任务是根据学生写的诗给出修改意见。
因此创建一个Action为ReviewPoem。
然后创建老师的智能体指定它的动作为Reviewself.set_actions([ReviewPoem])
它的触发实际应该是学生写完诗以后因此加入self._watch([WritePoem])
class ReviewPoem(Action):name: str ReviewPoemPROMPT_TEMPLATE: str Here is the historical conversation record : {msg} .Check student-created poems about the subject provided by human and give your suggestions for revisions. You prefer poems with elegant sentences and retro style.Return only your comments with NO other texts.your comments:async def run(self, msg: str):prompt self.PROMPT_TEMPLATE.format(msg msg)rsp await self._aask(prompt)return rspclass Teacher(Role):name: str laowangprofile: str Teacherdef __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewPoem])self._watch([WritePoem])async def _act(self) - Message:logger.info(f{self._setting}: ready to {self.rc.todo})todo self.rc.todomsg self.get_memories() # 获取所有记忆poem_text await ReviewPoem().run(msg)logger.info(fteacher : {poem_text})msg Message(contentpoem_text, roleself.profile,cause_bytype(todo))return msg1.2.3 创建多智能体交流的环境
前面我们已经知道了多智能体之间的交互需要一个非常重要的组件 - Environment。
因此创建一个供多智能体交流的环境通过 add_roles 加入智能体。
然后当用户输入内容时将该内容添加到环境中publish_message从而触发相应智能体开始运行。这里cause_byUserRequirement代表消息是由用户产生的学生智能体关心这类消息观察到之后就开始行动了。
classroom Environment()classroom.add_roles([Student(), Teacher()])classroom.publish_message(Message(roleHuman, contenttopic, cause_byUserRequirement,send_to or MESSAGE_ROUTE_TO_ALL),peekableFalse,
)
1.2.4 运行
加入相应头文件指定交互次数就可以开始跑了。注意交互次数直接决定了你的程序的效果和你的钱包
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_ALLasync def main(topic: str, n_round3):while n_round 0:# self._save()n_round - 1logger.debug(fmax {n_round} left.)await classroom.run()return classroom.historyasyncio.run(main(topicwirte a poem about moon))1.2.5 完整代码
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_ALLclassroom Environment()class WritePoem(Action):name: str WritePoemPROMPT_TEMPLATE: str Here is the historical conversation record : {msg} .Write a poem about the subject provided by human, Return only the content of the generated poem with NO other texts.If the teacher provides suggestions about the poem, revise the students poem based on the suggestions and return.your poem:async def run(self, msg: str):prompt self.PROMPT_TEMPLATE.format(msg msg)rsp await self._aask(prompt)return rspclass ReviewPoem(Action):name: str ReviewPoemPROMPT_TEMPLATE: str Here is the historical conversation record : {msg} .Check student-created poems about the subject provided by human and give your suggestions for revisions. You prefer poems with elegant sentences and retro style.Return only your comments with NO other texts.your comments:async def run(self, msg: str):prompt self.PROMPT_TEMPLATE.format(msg msg)rsp await self._aask(prompt)return rspclass Student(Role):name: str xiaomingprofile: str Studentdef __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WritePoem])self._watch([UserRequirement, ReviewPoem])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)poem_text await WritePoem().run(msg)logger.info(fstudent : {poem_text})msg Message(contentpoem_text, roleself.profile,cause_bytype(todo))return msgclass Teacher(Role):name: str laowangprofile: str Teacherdef __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewPoem])self._watch([WritePoem])async def _act(self) - Message:logger.info(f{self._setting}: ready to {self.rc.todo})todo self.rc.todomsg self.get_memories() # 获取所有记忆poem_text await ReviewPoem().run(msg)logger.info(fteacher : {poem_text})msg Message(contentpoem_text, roleself.profile,cause_bytype(todo))return msgasync def main(topic: str, n_round3):classroom.add_roles([Student(), Teacher()])classroom.publish_message(Message(roleHuman, contenttopic, cause_byUserRequirement,send_to or MESSAGE_ROUTE_TO_ALL),peekableFalse,)while n_round 0:# self._save()n_round - 1logger.debug(fmax {n_round} left.)await classroom.run()return classroom.historyasyncio.run(main(topicwirte a poem about moon))运行结果 2. 总结
总结一下整个流程吧画了个图。 从最外围环境 classroom开始用户输入的信息通过 publish_message 发送到所有Role中通过Role的 put_message 放到自身的 msg_buffer中。
当Role运行run时_observe从msg_buffer中提取信息然后只过滤自己关心的消息例如Teacher只过滤出来自WritePoem的其它消息虽然在msg_buffer中但不处理。同时msg_buffer中的消息会存入memory中。
run完即执行完相应动作后通过publish_message将结果消息发布到环境中环境的publish_message又将这个消息发送个全部的Role这时候所有的Role的msg_buffer中都有了这个消息。完成了智能体间信息的一个交互闭环。 站内文章一览
