利用MetaGPT 创建智能体（2）多智能体

先给上个文章利用MetaGPT 创建智能体（1）入门打个补丁：
补丁1： MeteGTP中Role和Action的关联和区别？这是这两天再利用MetaGPT时候心中的疑问，这里做个记录
Role（脚色）和 Action（动作）是MetaGPT中的两个紧张概念和对象，Role顾名思义也就是脚色，这Role子类中表示一个独立的功能模块，一个Role内里调用多个Action；Action（动作）详细使命中的最小操纵单位。
举个例子的话就是，一个讲师（Role），如果要授课的话就需要备课（Action），开始授课（Action），上课期间还有提问（Action），末了由多个讲师（Role）来完成整个学生上课的这个功能。

---

补丁2： 安装volcengine-python-sdk[ark]~=1.0.94报错

1. Could not build wheels for volcengine-python-sdk, which is required to install pyproject.toml-based projects

复制代码

由于 Windows 系统有最长路径限制，导致安装失败，按照以下方式设置：

---

• 按下 Win+R ，输入 regedit 打开注册表编辑器。
  
• 设置 \HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\FileSystem 路径下的变量 LongPathsEnabled 为 1
  

---

---

再蹭个热度：
deepseek前段时间爆火，这里写个怎么在MetaGPT中利用deepseek（当地启动版），利用的话也很方便，第一启动ollama内里的deepseek模子，打开控制台，输入

1. ollama run deepseek-r1

复制代码

等候模子下载运行就好，修改config2.yaml配置文件

1. llm:
2.   api_type: "ollama"  # or azure / ollama / groq etc.
3.   model: "deepseek-r1"  # or gpt-3.5-turbo
4.   base_url: "http://localhost:11434/api"  # or forward url / other llm url
5.   api_key: "ollama"

复制代码

然后就可以利用了，有一说一确实不错。

---

---

正文

---

多智能体指的就是创建多个智能体去进行执行使命，这也是实际业务中的常用处理。
多个智能体的利用大致3步，
1、定义多个Role，Action
2、Role中定义_watch监听相关Action
3、利用Team对象到场多个Role启动

---

在 MetaGPT 多智能体框架中，Team 对象是和谐和管理多个智能体（Agent）协作的焦点组件。它的作用雷同于一个“假造团队”，负责组织智能体之间的交互、使命分配、通信和谐，以及团体协作流程的控制。
Team 的焦点作用

• 多智能体协作的容器
  Team 作为多个智能体的容器，会合管理智能体的生命周期（创建、烧毁）、状态监控和资源分配。
  
• 使命分配与调度
  将复杂使命拆解为子使命，并根据智能体的脚色（Role）和能力（Skill）动态分配使命，确保高效协作。
  
• 通信与和谐
  管理智能体之间的通信（如消息通报、事故触发），避免辩说并确保信息同步。比方，在主动化客服系统中，和谐“用户意图辨认”和“工单生成”两个智能体的协作。
  
• 环境管理
  提供共享的上下文环境（如全局数据、知识库），使智能体可以访问统一的信息源，避免重复计算或数据不一致。
  
• 容错与恢复
  监控智能体运行状态，处理非常（如使命超时、逻辑错误），并触发恢复机制（如重试使命、切换备用智能体）。
  

代码

创建Role1和对应的Action1，第一个Role需要吸收用户传来的指令，以是_watch监听用户UserRequirement消息，因为需要详细操纵以是也不用重写_act函数（全部代码在最下面）

1. def parse_code(rsp):
2.     # r 定义原始字符串，忽略其中的所有转义字符
3.     pattern = r"```python(.*)```"
4.     match = re.search(pattern, rsp, re.DOTALL)
5.     code_text = match.group(1) if match else rsp
6.     return code_text
8. class Action1(Action):
9.     name:str = "Action1"
10.     PROMPT_TEMPLATE:str = """ 编写一个python函数，可以 {instruction} """
12.     async def run(self, instruction: str):
13.         prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
15.         # 令 LLM 赋予这个动作能力
16.         rsp = await self._aask(prompt)
17.         # 解析返回内容
18.         code_text = parse_code(rsp)
20.         return code_text
22. # 角色
23. class Role1(Role):
24.     name:str = "Role1"
25.     profile:str = "Role1"
27.     def __init__(self, **kwargs):
28.         super().__init__(**kwargs)
29.         self._watch([UserRequirement])  # 监听来自用户或其他智能体的重要上游消息
30.         self.set_actions([Action1])

复制代码

编写Role2和Action2，因为需要等候Action的执行内容，以是_watch监听Action1

1. class Action2(Action):
2.     name:str = "Action2"
3.     PROMPT_TEMPLATE:str = """
4.         上下文：｛context｝
5.         假设您已导入给定函数，则使用pytest为其编写{k}个单元测试。
6.     """
8.     async def run(self, context: str, k:int=3):
9.         prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)
11.         rsp = await self._aask(prompt)
12.         code_text = parse_code(rsp)
14.         return code_text
16. class Role2(Role):
17.     name:str = "Role2"
18.     profile:str = "Role2"
20.     def __init__(self, **kwargs):
21.         super().__init__(**kwargs)
22.         self.set_actions([Action2])
23.         self._watch([Action1])  # 监听来自用户或其他智能体的重要上游消息
25.     async def _act(self) -> Message:
26.         todo = self.rc.todo
27.         context = self.get_memories() # 查找消息
29.         code_text = await todo.run(context, k=5) # 这里开始执行Action2的处理
30.         # role 消息角色
31.         # cause_by 专门用于追踪消息的触发来源，建立消息与动作（Action）之间的关联
32.         msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
34.         return msg

复制代码

末了Role3，Action3，监听Action2，监听是可以监听多个Action的

1. class Action3(Action):
2.     name:str = "Action3"
3.     PROMPT_TEMPLATE:str = """
4.         上下文：｛context｝
5.         审查测试用例并提供一条关键意见：
6.     """
8.     async def run(self, context: str):
9.         prompt = self.PROMPT_TEMPLATE.format(context=context)
10.         rsp = await self._aask(prompt)
11.         return rsp
13. class Role3(Role):
14.     name:str = "Role3"
15.     profile:str = "Role3"
17.     def __init__(self, **kwargs):
18.         super().__init__(**kwargs)
19.         self.set_actions([Action3])
20.         self._watch([Action2])

复制代码

末了是运行，运行时创建Team对象，添加脚色，然后运行

1. # 运行函数
2. async def main(idea:str,
3.                investment:float,
4.                n_round:int):
5.     # MetaGPT中的团队
6.     team = Team()
7.     # 添加角色
8.     team.hire([
9.         Role1(),
10.         Role2(),
11.         Role3()
12.     ])
13.     # 注入资源 invest 属性用于设定或表示团队对某个任务或目标的资源投入程度
14.     team.invest(investment=investment)
15.     team.run_project(idea) # 启动团队协作流程
16.     await team.run(n_round=n_round)
18. if __name__ == '__main__':
19.     asyncio.run(main('计算列表乘积的函数', 3.0, 3))

复制代码

三个智能体
Role1，执行Action1，监听用户输入；
Role2，监听Action1，执行Action2；
Role3，监听Action2，执行Action3；
这就是多个智能体的合作利用，重点总结的话就是，监听_watch其他人，Team运行。
末了是全部代码

1. import asyncioimport refrom metagpt.actions import Action, UserRequirementfrom metagpt.roles import Rolefrom metagpt.schema import Messagefrom metagpt.team import Team# 多个智能体案例def parse_code(rsp):
2.     # r 定义原始字符串，忽略其中的所有转义字符
3.     pattern = r"```python(.*)```"
4.     match = re.search(pattern, rsp, re.DOTALL)
5.     code_text = match.group(1) if match else rsp
6.     return code_text
8. class Action1(Action):
9.     name:str = "Action1"
10.     PROMPT_TEMPLATE:str = """ 编写一个python函数，可以 {instruction} """
12.     async def run(self, instruction: str):
13.         prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
15.         # 令 LLM 赋予这个动作能力
16.         rsp = await self._aask(prompt)
17.         # 解析返回内容
18.         code_text = parse_code(rsp)
20.         return code_text
22. # 角色
23. class Role1(Role):
24.     name:str = "Role1"
25.     profile:str = "Role1"
27.     def __init__(self, **kwargs):
28.         super().__init__(**kwargs)
29.         self._watch([UserRequirement])  # 监听来自用户或其他智能体的重要上游消息
30.         self.set_actions([Action1])
31. class Action2(Action):
32.     name:str = "Action2"
33.     PROMPT_TEMPLATE:str = """
34.         上下文：｛context｝
35.         假设您已导入给定函数，则使用pytest为其编写{k}个单元测试。
36.     """
38.     async def run(self, context: str, k:int=3):
39.         prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)
41.         rsp = await self._aask(prompt)
42.         code_text = parse_code(rsp)
44.         return code_text
46. class Role2(Role):
47.     name:str = "Role2"
48.     profile:str = "Role2"
50.     def __init__(self, **kwargs):
51.         super().__init__(**kwargs)
52.         self.set_actions([Action2])
53.         self._watch([Action1])  # 监听来自用户或其他智能体的重要上游消息
55.     async def _act(self) -> Message:
56.         todo = self.rc.todo
57.         context = self.get_memories() # 查找消息
59.         code_text = await todo.run(context, k=5) # 这里开始执行Action2的处理
60.         # role 消息角色
61.         # cause_by 专门用于追踪消息的触发来源，建立消息与动作（Action）之间的关联
62.         msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
64.         return msg
65. class Action3(Action):
66.     name:str = "Action3"
67.     PROMPT_TEMPLATE:str = """
68.         上下文：｛context｝
69.         审查测试用例并提供一条关键意见：
70.     """
72.     async def run(self, context: str):
73.         prompt = self.PROMPT_TEMPLATE.format(context=context)
74.         rsp = await self._aask(prompt)
75.         return rsp
77. class Role3(Role):
78.     name:str = "Role3"
79.     profile:str = "Role3"
81.     def __init__(self, **kwargs):
82.         super().__init__(**kwargs)
83.         self.set_actions([Action3])
84.         self._watch([Action2])
85. # 运行函数async def main(idea:str,               investment:float,               n_round:int):    # MetaGPT中的团队    team = Team()    # 添加脚色    team.hire([        Role1(),        Role2(),        Role3()    ])    # 注入资源，invest 属性用于设定或表示团队对某个使命或目标的资源投入程度    team.invest(investment=investment)    team.run_project(idea) # 启动团队协作流程    await team.run(n_round=n_round)if __name__ == '__main__':    # fire.Fire(main) 快速将 Python 对象（如函数、类、模块）转换为命令行接口（CLI）    # main('编写一个计算列表乘积的函数', 3.0, 5)    asyncio.run(main('计算列表乘积的函数', 2.0, 3))

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。