使用MetaGPT 创建智能体（3）常用操作和晋级

1、使用记忆

这里的记忆，其实指的是消息的存储和获取

可以选择短期存储或长期存储，长期存储可以用于数据库进行存储

而短期存储指的是 MetaGPT 中叫做Memory的对象列表，

Role中 执行self.rc.memory.get(k=k)获得对应Memory对象，k为数值

使用添加

# content 消息内容
# role 消息角色
# cause_by 专门用于追踪消息的触发来源，建立消息与动作（Action）之间的关联
msg = Message(content=code_text, role=self.profile, cause_by=todo)self.rc.memory.add(msg)

2、创建和使用工具

这里的工具指的就是属于自己的类，然后在需要的地方使用，简单点讲就像一个公共的Role

1、对应函数上使用注解@register_tool()注册工具

2、调用DataInterpreter创建对象

role = DataInterpreter(tools=["函数名称"]) # 集成工具
await role.run(requirement)

像使用Role一样使用工具

3、人类介入

在具体业务中需要实际人员参加业务流程， 在关键决策中提供指导，或者在任务重扮演角色。简单点讲在执行任务的时候，有时候一些步骤需要人员进行输入处理

在定义角色对象的时候 ，给is_human传入true就好

team = Team()
team.hire([Role1(),Role2(),Role3(is_human=add_human),]
)

4、角色配置不同LLM

其实就是根据配置可以在Role对象里面配置对应的LLM，针对不同角色使用不同LLM，进行更优化的处理

第一步，定义配置：使用默认配置，或者从metagpt目录中加载自定义配置。

第二步，分配配置：将特定的LLM配置分配给Role和Action。配置的优先级：Action config > Role config > Global config（config in config2.yaml）。

ollama run qwen2.5:7b

c1 = Config.from_home("qwen.yaml") #加载指定文件
c2 = Config.default() # 默认的class DemoAction(Action):PROMPT_TEMPLATE: str = """编写一个可以{instruction}的java函数，并提供两个可运行的测试用例。代码和使用的说明使用中文返回“python your_code_here”，没有其他文本， 您的代码："""def __init__(self, **kwargs):super().__init__(**kwargs)self.config = c2name: str = "DemoAction"....

或者直接写入，

# 这段代码是官网上给的案例# 创建a1、a2和a3三个Action。并为a1指定`gpt4t`的配置。
a1 = Action(config=gpt4t, name="Say", instruction="Say your opinion with emotion and don't repeat it")
a2 = Action(name="Say", instruction="Say your opinion with emotion and don't repeat it")
a3 = Action(name="Vote", instruction="Vote for the candidate, and say why you vote for him/her")# 创建A，B，C三个角色，分别为“民主党候选人”、“共和党候选人”和“选民”。
# 虽然A设置了config为gpt4，但因为a1已经配置了Action config，所以A将使用model为gpt4的配置，而a1将使用model为gpt4t的配置。
A = Role(name="A", profile="Democratic candidate", goal="Win the election", actions=[a1], watch=[a2], config=gpt4)
# 因为B设置了config为gpt35，而为a2未设置Action config，所以B和a2将使用Role config，即model为gpt35的配置。
B = Role(name="B", profile="Republican candidate", goal="Win the election", actions=[a2], watch=[a1], config=gpt35)
# 因为C未设置config，而a3也未设置config，所以C和a3将使用Global config，即model为gpt4的配置。
C = Role(name="C", profile="Voter", goal="Vote for the candidate", actions=[a3], watch=[a1, a2])

5、进阶-智能体之间的通信

智能体之间的通信，通过Message处理。智能体之间的通讯有三方，首先是提供者，也就是发送消息的智能体；使用者，接收消息并进行相关处理的智能体；最后是环境Environment对象。

消息提供者，使用Message的sent_from、cause_by提供消息

消息使用者，使用_watch订阅相应消息

Environment对象负责将消息按各个智能体的订阅需求，广播给各个智能体

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# 写文档
class WriteDocument(Action):name: str = "WriteDocument"PROMPT_TEMPLATE: str = """以下是历史对话记录 : {msg} .你是一个需求分析员，以提供的需求写一个需求分析文档，只返回文档内容，如果程序员对文档内容的实现提出意见，请修改并返回文档内容："""async def run(self, msg: str):prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rsp# 读文档
class ReviewDocument(Action):name: str = "ReviewDocument"PROMPT_TEMPLATE: str = """以下是历史对话记录 : {msg}你是一个资深java程序员，请根据需求文档的内容，提出基于技术的意见，只返回意见您的意见:"""async def run(self, msg: str):prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rspclass Demander(Role):name: str = "张三"profile: str = "需求分析师"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteDocument])self._watch([UserRequirement, ReviewDocument]) # 监听用户输入和读文档的返回async def _act(self) -> Message:logger.info(f"{self._setting}: 准备好 {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有记忆poem_text = await WriteDocument().run(msg)logger.info(f'WriteDocument 的执行结果 : {poem_text}')msg = Message(content=poem_text, role=self.profile,cause_by=type(todo))return msgclass Programmer(Role):name: str = "李四"profile: str = "程序员"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewDocument])self._watch([WriteDocument]) # 监听需求分析师分析文档是否完成async def _act(self) -> Message:logger.info(f"{self._setting}: 准备好 {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 获取所有记忆poem_text = await ReviewDocument().run(msg)logger.info(f'ReviewDocument 的执行结果 : {poem_text}')msg = Message(content=poem_text, role=self.profile,cause_by=type(todo))return msgasync def main(topic: str, n_round=3):env = Environment()env.add_roles([Demander(), Programmer()])env.publish_message(Message(role="Human", content=topic, cause_by=UserRequirement,send_to='' or MESSAGE_ROUTE_TO_ALL),peekable=False,)while n_round > 0:n_round -= 1logger.debug(f"max {n_round=} left.")await env.run()return env.historyif __name__ == '__main__':asyncio.run(main(topic='写一个关于登录的需求分析文档'))

6、进阶-增量开发

这里的增量，我认为是针对已经生成的文档后再次迭代，优化修改已经生成过的文件，作为一个写代码的，感觉这块没有什么需要额外记录的，因为完全可以在代码层次上自己处理，这里只记一个官网地址，有兴趣的话自己看

https://docs.deepwisdom.ai/main/zh/guide/in_depth_guides/incremental_development.html

7、进阶-序列化和断点恢复

这里的断点指的是程序在执行的时候，有时候会遇到特殊情况停止运行，那这时候可以从中断处的结果继续运行，减少时间。

而这个时候为了支持这样的服务，就需要序列化以及反序列化，序列化指的是把当前内存中运行的数据存储成可操作的文件，而反序列化就是把对应文件读取成相关数据

当程序发生中断或结束后，在存储目录下的文件结构如下：

./workspacestorageteamteam.json          # 包含团队、环境、角色、动作等信息

主要存储的就是team.json

实现断点恢复，可以有两种方式

第一种是使用命令时处理，使用--recover_path声明序列化的文件夹

metagpt "xxx" --recover_path "./workspace/storage/team"

第二种是使用代码进行序列化，反序列化操作，需要先创建Team对象使用

team = Team()
...save_to = Path("./serial")
team.serialize(save_to)  # 序列化
team.deserialize(save_to, Context())  # 反序列化...