【AI Agent系列】【MetaGPT多智能体学习】4. 基于MetaGPT的Team组件开发你的第一个智能体团队

本系列文章跟随《MetaGPT多智能体课程》(https://github.com/datawhalechina/hugging-multi-agent),深入理解并实践多智能体系统的开发。

本文为该课程的第四章(多智能体开发)的第二篇笔记。主要是对MetaGPT中Team组件的学习和实践。

文章目录

  • 0. Team组件介绍
    • 0.1 基本参数
    • 0.2 重要函数
      • 0.2.1 hire - 雇佣员工,往Team中添加Role
      • 0.2.2 invest - 投资,设置程序总预算
      • 0.2.3 run_project
      • 0.2.4 run - Team开始运行的入口
    • 0.3 总结
  • 1. 基于Team开发你的第一个智能体团队
    • 1.1 demo需求描述
    • 1.2 写代码
      • 1.2.1 SimpleCoder
      • 1.2.2 SimpleTester
      • 1.2.3 SimpleReviewer
      • 1.2.4 组成Team并运行
      • 1.2.5 完整代码
      • 1.2.6 运行过程及结果展示
  • 2. 总结

0. Team组件介绍

我们在刚开始搭建环境的时候,跑的第一个例子就使用了Team组件。当时只是复制粘贴,用它将程序跑起来了,但其背后的机制和原理是什么还没有学习过。下面从部分源码中,看下Team组件的运行机制。

0.1 基本参数

class Team(BaseModel):"""Team: Possesses one or more roles (agents), SOP (Standard Operating Procedures), and a env for instant messaging,dedicated to env any multi-agent activity, such as collaboratively writing executable code.团队:拥有一个或多个角色(代理人)、标准操作流程(SOP)和一个用于即时通讯的环境,致力于开展任何多代理活动,如协作编写可执行代码。"""model_config = ConfigDict(arbitrary_types_allowed=True)env: Optional[Environment] = Noneinvestment: float = Field(default=10.0)idea: str = Field(default="")

其中主要三个参数:

  • env:多智能体运行的环境
  • investment:投资,用来设置整个程序运行的预算,控制token消耗,当程序运行超过这个预设值后,会强制停止
  • idea:用户的输入、需求

0.2 重要函数

0.2.1 hire - 雇佣员工,往Team中添加Role

这个函数实现的功能其实就是往自身的环境中添加Role。

def hire(self, roles: list[Role]):"""Hire roles to cooperate"""self.env.add_roles(roles)

0.2.2 invest - 投资,设置程序总预算

用来设置整个程序运行的预算,控制token消耗,当程序运行超过这个预设值后,会强制停止。

def invest(self, investment: float):"""Invest company. raise NoMoneyException when exceed max_budget."""self.investment = investmentself.cost_manager.max_budget = investmentlogger.info(f"Investment: ${investment}.")

0.2.3 run_project

这个函数的名有点欺骗性,你可能以为这是开始运行整个Team的入口,其实不是。它只是往Team的环境中放入第一条用户消息而已。

idea 为用户的输入或需求。这个函数的主要功能是调用了 Environment 的 publish_message 往环境中送入了一个用户消息。

def run_project(self, idea, send_to: str = ""):"""Run a project from publishing user requirement."""self.idea = idea# Human requirement.self.env.publish_message(Message(role="Human", content=idea, cause_by=UserRequirement, send_to=send_to or MESSAGE_ROUTE_TO_ALL),peekable=False,)

0.2.4 run - Team开始运行的入口

这个才是Team运行的入口函数,当输入了idea时,会转到 run_project 去往自身的环境中放置用户消息。然后在 while循环中,循环运行各个Role。

n_round指定循环的次数,这里默认为3,执行三次 self.env.run()env.run我们上篇文章已经知道了,就是顺序执行环境中所有Role的run函数。

_check_balance函数的功能是检查当前程序消耗的token或钱数是否超过了预算。如果超过了预算,直接弹窗警告 raise NoMoneyException

 @serialize_decorator
async def run(self, n_round=3, idea="", send_to="", auto_archive=True):"""Run company until target round or no money"""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
def _check_balance(self):if self.cost_manager.total_cost >= self.cost_manager.max_budget:raise NoMoneyException(self.cost_manager.total_cost, f"Insufficient funds: {self.cost_manager.max_budget}")

0.3 总结

看了上面的几个重要函数,是否觉得有点眼熟?这不就是将上篇文章中我们在运行多智能体系统时的main函数拆分成了 hire / run_project / run 函数嘛。

async def main(topic: str, n_round=3):## 类比 Team 的 hire 函数添加 Rolesclassroom.add_roles([Student(), Teacher()])## 类比 Team 的 run_project 函数往环境中写入用户消息classroom.publish_message(Message(role="Human", content=topic, cause_by=UserRequirement,send_to='' or MESSAGE_ROUTE_TO_ALL),peekable=False,)## 类比 Team 的 run 函数控制循环次数while n_round > 0:# self._save()n_round -= 1logger.debug(f"max {n_round=} left.")await classroom.run()return classroom.history

所以,Team 组件的本质,就是对 Environment 接口的封装,同时在此基础上增加了 invest 的预算控制而已。

1. 基于Team开发你的第一个智能体团队

1.1 demo需求描述

总的需求,简单的软件开发流程:一个写代码,一个测试代码,一个review代码。

所以需要三个智能体Role:

  • SimpleCoder,Action是 SimpleWriteCode,写代码
  • SimpleTester,Action是 SimpleWriteTest,接收 SimpleCoder 的代码进行测试。也接收 SimpleReviewer 的修改意见进行测试用例改写。
  • SimpleReviewer,Action是 SimpleWriteReview,接收 SimpleTester 的测试用例,检查其覆盖范围和质量,给出测试用例的修改意见。

1.2 写代码

1.2.1 SimpleCoder

SimpleCoder主要用来写代码。

  • 它的Action是SimpleWriteCode,通过 self.set_actions([SimpleWriteCode]) 将该Action设置给SimpleCoder
  • 它的行动指令来源是 UserRequirement,当环境中出现 UserRequirement 来源的消息时,它开始执行Action。通过 self._watch([UserRequirement]) 设置其关注的消息来源。
def parse_code(rsp):pattern = r"```python(.*)```"match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_textclass SimpleWriteCode(Action):PROMPT_TEMPLATE: str = """Write a python function that can {instruction}.Return ```python your_code_here ```with NO other texts,your code:"""name: str = "SimpleWriteCode"async def run(self, instruction: str):prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)rsp = await self._aask(prompt)code_text = parse_code(rsp)return code_textclass SimpleCoder(Role):name: str = "Alice"profile: str = "SimpleCoder"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement])self.set_actions([SimpleWriteCode])

1.2.2 SimpleTester

SimpleTester 用来写测试用例代码。

  • 其Action为SimpleWriteTest,通过 self.set_actions([SimpleWriteTest]) 指定。
  • 其行动指令来源,一个是 SimpleWriteCode,接收主代码,根据主代码写单测的测试用例。第二个来源是 SimpleWriteReview,接收测试用例修改意见,根据修改意见完善测试用例。通过 self._watch([SimpleWriteCode, SimpleWriteReview]) 来指定关注的消息来源。
class SimpleWriteTest(Action):PROMPT_TEMPLATE: str = """Context: {context}Write {k} unit tests using pytest for the given function, assuming you have imported it.Return ```python your_code_here ```with NO other texts,your code:"""name: str = "SimpleWriteTest"async def run(self, context: str, k: int = 3):prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)rsp = await self._aask(prompt)code_text = parse_code(rsp)return code_textclass SimpleTester(Role):name: str = "Bob"profile: str = "SimpleTester"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([SimpleWriteTest])# self._watch([SimpleWriteCode])self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this tooasync 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()  # use all memories as contextcode_text = await todo.run(context, k=5)  # specify argumentsmsg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msg

1.2.3 SimpleReviewer

SimpleReviewer 用来对测试用例代码进行Review,给出修改意见。

  • 其Action为SimpleWriteReview,通过 self.set_actions([SimpleWriteReview]) 指定。
  • 其行动指令来源为 SimpleWriteTest,接收测试用例代码,根据测试用例代码给出修改意见。通过 self._watch([SimpleWriteTest]) 来指定关注的消息来源。
class SimpleWriteReview(Action):PROMPT_TEMPLATE: str = """Context: {context}Review the test cases and provide one critical comments:"""name: str = "SimpleWriteReview"async def run(self, context: str):prompt = self.PROMPT_TEMPLATE.format(context=context)rsp = await self._aask(prompt)return rspclass SimpleReviewer(Role):name: str = "Charlie"profile: str = "SimpleReviewer"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([SimpleWriteReview])self._watch([SimpleWriteTest])

1.2.4 组成Team并运行

下面就是将上面的三个 Role 放到一个 Team 中。

  • hire 函数添加上面的三个 Role 到 Team 中
  • invest 函数设置总预算
  • run_project 函数将 idea 任务放到环境中
  • run 函数让整个 Team 运行起来
async def main(idea: str = "write a function that calculates the product of a list",investment: float = 3.0,n_round: int = 5,add_human: bool = False,
):logger.info(idea)team = Team()team.hire([SimpleCoder(),SimpleTester(),SimpleReviewer(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)

1.2.5 完整代码

"""
Filename: MetaGPT/examples/build_customized_multi_agents.py
Created Date: Wednesday, November 15th 2023, 7:12:39 pm
Author: garylin2099
"""
import reimport firefrom metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Teamdef parse_code(rsp):pattern = r"```python(.*)```"match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_textclass SimpleWriteCode(Action):PROMPT_TEMPLATE: str = """Write a python function that can {instruction}.Return ```python your_code_here ```with NO other texts,your code:"""name: str = "SimpleWriteCode"async def run(self, instruction: str):prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)rsp = await self._aask(prompt)code_text = parse_code(rsp)return code_textclass SimpleCoder(Role):name: str = "Alice"profile: str = "SimpleCoder"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement])self.set_actions([SimpleWriteCode])class SimpleWriteTest(Action):PROMPT_TEMPLATE: str = """Context: {context}Write {k} unit tests using pytest for the given function, assuming you have imported it.Return ```python your_code_here ```with NO other texts,your code:"""name: str = "SimpleWriteTest"async def run(self, context: str, k: int = 3):prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)rsp = await self._aask(prompt)code_text = parse_code(rsp)return code_textclass SimpleTester(Role):name: str = "Bob"profile: str = "SimpleTester"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([SimpleWriteTest])# self._watch([SimpleWriteCode])self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this tooasync 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()  # use all memories as contextcode_text = await todo.run(context, k=5)  # specify argumentsmsg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msgclass SimpleWriteReview(Action):PROMPT_TEMPLATE: str = """Context: {context}Review the test cases and provide one critical comments:"""name: str = "SimpleWriteReview"async def run(self, context: str):prompt = self.PROMPT_TEMPLATE.format(context=context)rsp = await self._aask(prompt)return rspclass SimpleReviewer(Role):name: str = "Charlie"profile: str = "SimpleReviewer"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([SimpleWriteReview])self._watch([SimpleWriteTest])async def main(idea: str = "write a function that calculates the product of a list",investment: float = 3.0,n_round: int = 5,add_human: bool = False,
):logger.info(idea)team = Team()team.hire([SimpleCoder(),SimpleTester(),SimpleReviewer(is_human=add_human),])team.invest(investment=investment)team.run_project(idea)await team.run(n_round=n_round)# 最后这两句可以合成一句:await team.run(n_round=n_round, idea=idea)if __name__ == "__main__":fire.Fire(main)

1.2.6 运行过程及结果展示

(1)用户消息输入,SimpleCoder开始动作,写出代码

在这里插入图片描述

(2)SimpleTester 接收到 SimpleCoder 写完的代码,开始写测试用例。

在这里插入图片描述

(3)SimpleReviewer 接收到 SimpleTester 写的测试用例,开始审核并给出修改意见

在这里插入图片描述

(4)SimpleTester 接收到 SimpleReviewer 的修改意见,开始优化测试用例。
在这里插入图片描述
(5)SimpleReviewer 接收到 SimpleTester 优化后的测试用例,进行审核并再次给出修改意见
在这里插入图片描述
(6)SimpleTester 和 SimpleReviewer 之间循环交互 n 次

2. 总结

通过本节内容,学习了MetaGPT中Team组件的原理与使用方法。

Team 组件就是在原来 Environment 组件的基础上进行封装,增加了一个invest来控制整体成本。其主要函数为 hireinvestrun

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/710557.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【Django】执行查询—检索对象

检索对象 检索全部对象 >>> all_entries Entry.objects.all()通过过滤器检索指定对象 通过添加过滤条件精炼原始 QuerySet。两种最常见的精炼 QuerySet 的方式是: filter(**kwargs) 返回一个新的 QuerySet,包含的对象满足给定查询参数。 ex…

【Tomcat】在 linux 上实现 Catlina.log 自动分割,防止文件过大

背景描述 catalina.out即标准输出和标准出错,所有输出到这两个位置的都会进入catalina.out,这里包含tomcat运行自己输出的日志以及应用里向console输出的日志。默认这个日志文件是不会进行自动切割的,所以我们需要借助其他工具进行切割&…

MySQL:快照读和当前读

mysql读取数据实际上有两种读取模式:当前读和快照读 快照读:快照读的执行方式是生成 ReadView,直接利用 MVCC 机制来进行读取,并不会对记录进行加锁。当前读:每次读取的都是当前最新的数据,但是读的时候不…

剑指offer面试题22 栈的压入弹出序列

考察点 辅助栈知识点 题目 分析 这道题目要求输入俩个序列,第一个序列表示栈的压入顺序,要求判断第二个序列是否是该栈的弹出顺序。遇到这类题目思维一定要往辅助栈上靠,因为关于栈的考点其实就是这个。这种题目的解题思路就是归纳&#x…

「Python系列」Python pyecharts模块

文章目录 一、pyecharts安装二、pyecharts应用三、pyecharts图表类型四、pyecharts特点与功能特点:功能: 五、相关链接 一、pyecharts安装 要安装 Python 的 pyecharts 模块,你可以使用 pip,这是 Python 的包管理工具。请按照以下…

Payment Without Change

题目链接&#xff1a;Problem - 1256A - Codeforces 解题思路&#xff1a; 题目的大致意思就是手中的硬币数拿出若干枚正好等于s&#xff0c;分三种情况 .如果n > s && b < s,输出no .如果b > s,输出yes .如果n * (a < (s / n) ? a : (s / n)) b >…

数据结构(C语言版)02---链表

链表&#xff1a; 声明&#xff1a;LNode* LinkList在链表里面这两个是等价的; #include<stdio.h> #include<stdlib.h> typedef int Elemtype; typedef struct LNode{Elemtype data;struct LNode* next; }LNode,*LinkList; 链表打印函数&#xff1a; //打印v…

【iOS ARKit】RealityKit 同步机制

协作 Session 可以很方便地实现多用户之间的AR体验实时共享&#xff0c;但开发者需要自行负责并确保AR场景的完整性&#xff0c;自行负责虚拟物体的创建与销毁。为简化同步操作&#xff0c;RealityKit 内建了同步机制&#xff0c;RealityKit 同步机制基于 Multipeer Connectivi…

Python标准库sys常用函数、方法及代码实战解析【第108篇—标准库sys常用函数】

Python标准库sys常用函数、方法及代码实战解析 在Python的标准库中&#xff0c;sys 模块是一个常用而强大的工具&#xff0c;它提供了与Python解释器交互的函数和变量。本文将介绍sys模块的一些常用函数和方法&#xff0c;并通过实际的代码实例来解析它们的用法。 1. sys.argv…

C语言统计文件夹下所有文件数量/C语言删除文件夹下所有文件

代码解法不唯一&#xff0c;欢迎在评论区留下你的实现方式和想法&#xff0c;我会将好的解法更新到文章中&#xff01;&#xff01; 在C语言中&#xff0c;可以使用标准库中的函数来完成对文件夹中文件数量的判断和删除操作。下面是使用POSIX标准的代码示例&#xff0c;这段代…

Kafka是如何防止消息丢失的

Kafka通过一系列机制来防止消息丢失&#xff0c;主要包括以下几个方面&#xff1a; 生产者端&#xff08;Producer&#xff09;保证&#xff1a; 同步发送&#xff1a;生产者默认是异步发送消息的&#xff0c;但如果希望保证消息不丢失&#xff0c;可以选择将异步发送改为同步…

2024.2.19

1.TCP模型 服务器端 #include <myhead.h> #define SER_IP "192.168.199.129" #define SER_PORT 8899int main(int argc, const char *argv[]) {//1.创建用于连接的套接字文件int sfdsocket(AF_INET,SOCK_STREAM,0);if(sfd-1){perror("socket error"…

P1331 海战

难度&#xff1a;普及- 题目背景 在峰会期间&#xff0c;武装部队得处于高度戒备。警察将监视每一条大街&#xff0c;军队将保卫建筑物&#xff0c;领空将布满了 F-2003 飞机。 此外&#xff0c;巡洋船只和舰队将被派去保护海岸线。不幸的是&#xff0c;因为种种原因&#x…

react 原理揭秘

1.目标 A. 能够知道setState()更新数据是异步的 B. 能够知道JSX语法的转化过程 C. 能够说出React组件的更新机制 D. 能够对组件进行性能优化 E. 能够说出虚拟DOM和Diff算法 2.目录 A. setState()的说明 B. JSX语法的转化过程 C. 组件更新机制 D. 组件性能优化 E. 虚拟DOM和D…

[Vulnhub]靶场 Web Machine(N7)

kali:192.168.56.104 主机探测: arp-scan -l 靶机ip:192.168.56.104 端口扫描 nmap -p- 192.168.56.106 看一下web 目录扫描 gobuster dir -u http://192.168.56.106 -x html,txt,php,bak,zip --wordlist/usr/share/wordlists/dirbuster/directory-list-2.3-medium.txt exp…

【QT 5 +Linux下软件qt软件打包+qt生成软件创建可以安装压缩包+学习他人文章+第三篇:学习打包】

【QT 5 Linux下软件qt软件打包qt生成软件创建可以安装压缩包学习他人文章第三篇&#xff1a;学习打包】 1、前言2、实验环境3、自我学习总结-本篇总结&#xff08;1&#xff09;了解安装包的目录结构&#xff08;2&#xff09;了解要编写文件与编写脚本1. control文件2. postin…

NVMFS5113PLWFT1G汽车级功率MOSFET 60V 10A/64A满足AEC-Q101标准

AEC-Q101认证标准详细解读&#xff1a; AEC-Q101是一种汽车电子元件可靠性标准&#xff0c;由汽车电子委员会&#xff08;Automotive Electronics Council&#xff0c;简称AEC&#xff09;制定。该标准旨在确保在汽车环境中使用的电子元件具有足够的可靠性和耐久性。 AEC-Q10…

springboot/ssm工作量统计系统Java工作记录可视化管理系统web

springboot/ssm工作量统计系统Java工作记录可视化管理系统web 基于springboot(可改ssm)vue项目 开发语言&#xff1a;Java 框架&#xff1a;springboot/可改ssm vue JDK版本&#xff1a;JDK1.8&#xff08;或11&#xff09; 服务器&#xff1a;tomcat 数据库&#xff1a;…

探索JavaScript中的构造函数,巩固你的JavaScript基础

&#x1f90d; 前端开发工程师、技术日更博主、已过CET6 &#x1f368; 阿珊和她的猫_CSDN博客专家、23年度博客之星前端领域TOP1 &#x1f560; 牛客高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 &#x1f35a; 蓝桥云课签约作者、上架课程《Vue.js 和 E…

element-ui附件上传及在线查看详细总结,后续赋源码

一、附件上传 1、在element-ui上面复制相应代码 a、accept"image/*,.pdf,.docx,.xlsx,.doc,.xls" 是规定上传文件的类型&#xff0c;若是不限制&#xff0c;可以直接将accept‘all即可&#xff1b; b、:action"action" 这个属性就是你的上传附件的地址&am…