《MetaGPT智能体开发入门》学习笔记第五章多智能体

news/2024/9/23 14:29:59/

需求：两个智能体，智能体一和智能体二
智能体一可以写文章，然后更具智能体二的建议进行修改；
智能体二可以看智能体一的输出并且给出智能体一的内容修改建议；

解决：
MetaGPT有watch观察方法作为外部数据进行思考和action具体代码如下
智能体一的角色和action定义

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 = "WritePoem"PROMPT_TEMPLATE: str = """Here is the historical conversation record : {msg} .写一篇有human提供主题的100字发言稿，只返回生成的文稿不返回其他文本。如果老师提供了关于发言稿的修改建议，根据建议修改并返回。language: chineseyour poem:"""async def run(self, msg: str):prompt = self.PROMPT_TEMPLATE.format(msg = msg)rsp = await self._aask(prompt)return rsp# 定义学生角色 使用学生的action并且input老师的检查反馈
class Student(Role):name: str = "xiaoming"profile: str = "Student"def __init__(self, **kwargs):super().__init__(**kwargs)# self._init_actions([WritePoem])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(f'student : {poem_text}')msg = Message(content=poem_text, role=self.profile,cause_by=type(todo))return msg

智能体二的角色和action定义


# 老师检查学生写的诗词
class ReviewPoem(Action):name: str = "ReviewPoem"PROMPT_TEMPLATE: str = """Here is the historical conversation record : {msg} .检查学生创作关于human提供主题的文稿，并提出修改建议。您更喜欢表达优雅 引经据典 语句通顺连贯的内容。language: chineseyour comments:"""async def run(self, msg: str):prompt = self.PROMPT_TEMPLATE.format(msg = msg)rsp = await self._aask(prompt)return rsp# 定义老师角色并且观察也就是input学生写的诗词
class Teacher(Role):name: str = "laowang"profile: str = "Teacher"def __init__(self, **kwargs):super().__init__(**kwargs)# self._init_actions([ReviewPoem])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(f'teacher : {poem_text}')msg = Message(content=poem_text, role=self.profile,cause_by=type(todo))return msg

MAIN程序入口：


async def main(topic: str, n_round=3):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.historyasyncio.run(main(topic='写一个关于光伏性能源的200字文稿'))