[AI Agent学习] MetaGPT源码浅析

前言

工作上,需要使用AI Agent,所以需要深入学习一下AI Agent,光阅读各种文章,总觉无法深入细节,所以开看各类AI Agent相关的开源项目,此为第一篇,学习一下MetaGPT的源码。

基本目标

MetaGPT是一个多智能体框架,他抽象了一个软件公司中的主要角色,用不同的AI Agent去扮演,这些AI Agent包括产品经理、软件架构师、项目经理、工程师,这些AI Agent会按照开发团队设计好的SOP去交互并最终产出一个项目。

老习惯:不为读而读,为解决某些问题或理清某些概念而读,那么面对MetaGPT,我有以下目标:

  • MetaGPT是怎么抽象出的软件公司开发流程的?SOP具体在代码上是怎么实现的?

  • MetaGPT中不同AI Agent是怎么交互的?

  • 从效果上看,MetaGPT输出的内容是比较格式化的,要做到这样的效果,prompt是怎么写的?

  • MetaGPT是怎么抽象具体的职业的?比如产品经理是怎么抽象的。

本文主要来从源码中,找到上面问题的答案。

入口

虽然README.md中说,很多同学运行时会有点问题,但我感觉,运行起来还是很轻松的,相比于早期的sd-webui,那是友好太多了,所以这里不多提,直接按README中的内容run一下就好了。

直接看到入口方法

# startup.py async def startup(idea: str,investment: float = 3.0,n_round: int = 5,code_review: bool = False,run_tests: bool = False,implement: bool = True
):# 1.开公司company = SoftwareCompany()# 2.雇员工company.hire([ProductManager(),  # 产品经理Architect(), # 架构师ProjectManager(), # 项目经理])if implement or code_review:# 3.雇开发company.hire([Engineer(n_borg=5, use_code_review=code_review)])if run_tests:company.hire([QaEngineer()])# 4.设置金额(这次运行最多能消耗多少美金的GPT4)company.invest(investment)# 5.老板的需求company.start_project(idea)# 6.跑几轮await company.run(n_round=n_round)

从startup看,整个流程很清晰,可读性很高!

先看看SoftwareCompany类,部分代码如下:

# metagpt/software_company.pyclass SoftwareCompany(BaseModel):# 环境environment: Environment = Field(default_factory=Environment)investment: float = Field(default=10.0)idea: str = Field(default="")async def run(self, n_round=3):"""Run company until target round or no money"""while n_round > 0:# self._save()n_round -= 1logger.debug(f"{n_round=}")self._check_balance()await self.environment.run()return self.environment.history

通过SoftwareCompany类抽象一个软件公司,阅读代码后,你会发现SoftwareCompany类中的environment对象很重要,公司里的不同AI Agent都会与environment对象交互,这个抽象也很巧妙,就是在公司里,同事间的交流确实在公司这个“环境”里传播。

然后就是run方法,看到startup方法的最后,就是调用SoftwareCompany类的run方法,该方法会检测一下余额以及运行的轮数,如果余额不够或轮数超了,就会停止运行。

因为MetaGPT底层是使用openai api的,每次请求都需要花钱,你设置一个预算,默认是3美元,跑满3美元,就算任务没有完成,也会强行结束。

SoftwareCompany类的run方法会调用environment对象的run方法,代码如下:

# metagpt/environment.py/Environmentasync def run(self, k=1):"""Process all Role runs at once"""for _ in range(k):futures = []for role in self.roles.values():future = role.run()futures.append(future)# 当调用 asyncio.gather(*futures) 时,它会同时运行传递进来的协程,这些协程会在后台并发执行,而不会阻塞主线程。await asyncio.gather(*futures)

从上面代码可知,这才是真正的入口,循环所有的roles,然后调用每个role的run方法,整个过程通过python协程异步并发的运行,从而提高程序的运行效率。

这里的role就是不同身份的员工,其本质就是使用不同Prompt的请求openai的代码逻辑。

我们需要阅读一下role相关的代码,来尝试理解顺序性的问题。

role

在startup方法中,我们首先雇佣了3个不同的role:

company.hire([ProductManager(),  # 产品经理Architect(), # 架构师ProjectManager(),])

雇佣的顺序是有讲究的。一开始雇佣的是ProductManager实例,也就是,这里已经执行了实例ProductManager类的实例化代码了:

# metagpt/roles/product_manager.pyclass ProductManager(Role):def __init__(self, name: str = "Alice", profile: str = "Product Manager", goal: str = "Efficiently create a successful product",constraints: str = "") -> None:super().__init__(name, profile, goal, constraints)# 1.写产品需求文档(Product Requirement Document)self._init_actions([WritePRD])# 2.观察老板的需求self._watch([BossRequirement])

上面代码中,_init_actions与_watch都是Role类中的方法,也就是当前产品经理这个role自己的方法。它初始化了自己的动作:写需求文档,以及定了自己要观察的东西:老板的需求。

BossRequirement怎么来的?在startup方法中,我们调用了SoftwareCompany的start_project方法,该方法会就我们的需要以BOSS的身份发送到SoftwareCompany的environment中,代码如下:

# metagpt/software_company.py/SoftwareCompanydef start_project(self, idea):"""Start a project from publishing boss requirement."""self.idea = idea# Role初始观察到的messageself.environment.publish_message(Message(role="BOSS", content=idea, cause_by=BossRequirement))

这个过程类似于,公司里,老板走到产研区域,大喊一声,我要做个xxxx,然后大家就知道了,metagpt就是抽象了这个过程,其他的role可以通过SoftwareCompany的environment的读到老板和其他role的信息。

当environment对象的run方法被运行时,role的run方法就会被执行,

async def run(self, message=None):if message:# 1.如果有人给你发消息,则将消息存入role的短期记忆中if isinstance(message, str):message = Message(message)if isinstance(message, Message):self.recv(message)if isinstance(message, list):self.recv(Message("\n".join(message)))# 2.观察环境中的信息,看看有没有需要处理的elif not await self._observe():# If there is no new information, suspend and waitlogger.debug(f"{self._setting}: no news. waiting.")return# 3.如果有需要处理的,则通过 _react 处理,并获得处理结果rsp = await self._react()# 4.将处理的结果发到 environment 中self._publish_message(rsp)return rsp

run方法的逻辑也很清晰,先观察environment中的信息,如果有需要处理的,则_react进行处理并将结果发送回environment中。

那具体怎么观察呢?看到_observe方法的代码:

# metagpt/roles/role.py/Roleasync def _observe(self) -> int:if not self._rc.env:return 0# 从env的短期记忆中,获取信息env_msgs = self._rc.env.memory.get()# 从env中观察要观察的对象,获得对应的messageobserved = self._rc.env.memory.get_by_actions(self._rc.watch)# 记下来(role自己的memory)self._rc.news = self._rc.memory.remember(observed)  # remember recent exact or similar memoriesfor i in env_msgs:# 将环境中的信息记到role memory中self.recv(i)news_text = [f"{i.role}: {i.content[:20]}..." for i in self._rc.news]if news_text:logger.debug(f'{self._setting} observed: {news_text}')return len(self._rc.news)def recv(self, message: Message) -> None:if message in self._rc.memory.get():return# 记一下信息self._rc.memory.add(message)

上面代码中,self._rc.env 对象就是当前role所在的环境,这里用了常见的双向关联技巧。

当其他role通过publish_message方法向environment发消息时,消息其实存在environment的memory中,在_observe方法中,首先environment的memory中读取消息,然后在通过get_by_actions方法去获得需要某个action产生的message,以ProductManager为里,ProductManager通过_watch方法将BossRequirement设置为需要观察的对象。

当role调用get_by_actions时,会去找BossRequirement对应的message(其实就是需要message,然后交给GPT4去处理),get_by_actions代码如下:

# metagpt/memory.py/Memorydef get_by_action(self, action: Type[Action]) -> list[Message]:"""Return all messages triggered by a specified Action"""return self.index[action]def get_by_actions(self, actions: Iterable[Type[Action]]) -> list[Message]:"""Return all messages triggered by specified Actions"""rsp = []for action in actions:if action not in self.index:continuersp += self.index[action]return rspdef add(self, message: Message):"""Add a new message to storage, while updating the index"""if message in self.storage:returnself.storage.append(message)if message.cause_by:self.index[message.cause_by].append(message)

因为get_by_actions其实就是从self.index中找到action这个key对应的value,所以self.index是什么就很关键,而这就需要看到add方法,以BossRequirement为例,就是将key设置为BossRequirement,然后存message。

简单而言,get_by_actions方法会获取由role的_watch方法指定的要观察的对象其对应的message,作为self._rc.news,然后role将这些news存到memory中。

如果role通过_observe方法观察到了news,那么就需要执行_react方法:

# metagpt/roles/role.pyasync def _react(self) -> Message:"""Think first, then act"""await self._think()logger.debug(f"{self._setting}: {self._rc.state=}, will do {self._rc.todo}")return await self._act()

先看到_think方法:

# metagpt/roles/role.py/Roleasync def _think(self) -> None:# 1.如果只有一个动作,那么就直接让_act执行这一个动作则可if len(self._actions) == 1:self._set_state(0)returnprompt = self._get_prefix()# 2.整合role的memory和state到prompt中,让GPT4处理prompt += STATE_TEMPLATE.format(history=self._rc.history, states="\n".join(self._states),n_states=len(self._states) - 1)next_state = await self._llm.aask(prompt)logger.debug(f"{prompt=}")if not next_state.isdigit() or int(next_state) not in range(len(self._states)):logger.warning(f'Invalid answer of state, {next_state=}')next_state = "0"self._set_state(int(next_state))

_think方法的作用是设置state,然后_act才会根据设置好的state决定要执行的action,如果当前role只有一个state,那么state直接设置成0则可,如果有多个action,那么就交由GPT4去判断:将role的memory中的信息全部取出作为history,然后将role的所有state也整合起来,一起放到prompt中,交由GPT4去选择当前history下要做什么action是最好的,让其返回state。

然后看_act方法:

async def _act(self) -> Message:logger.info(f"{self._setting}: ready to {self._rc.todo}")# 让role执行相应的actionresponse = await self._rc.todo.run(self._rc.important_memory)if isinstance(response, ActionOutput):msg = Message(content=response.content, instruct_content=response.instruct_content,role=self.profile, cause_by=type(self._rc.todo))else:msg = Message(content=response, role=self.profile, cause_by=type(self._rc.todo))# 记下当前action返回的msgself._rc.memory.add(msg)# logger.debug(f"{response}")return msg

_act方法的逻辑就是执行self._rc.todo对应的action,这个action是_think方法通过_set_state方法设置的,比如产品经理只有一个action:WritePRD,那么就会执行WritePRD的run方法,需要注意的是,为了让不同的role可以交互,_act还会产生message,每个message都会设置cause_by参数,以表明当前的message是由哪个动作产生的,其他role就可以通过_watch方法看到需要的message了。

随后,我们看到WritePRD的细节,代码如下:

#  metagpt/actions/write_prd.py
class WritePRD(Action):def __init__(self, name="", context=None, llm=None):super().__init__(name, context, llm)async def run(self, requirements, *args, **kwargs) -> ActionOutput:sas = SearchAndSummarize()rsp = ""info = f"### Search Results\n{sas.result}\n\n### Search Summary\n{rsp}"if sas.result:logger.info(sas.result)logger.info(rsp)prompt = PROMPT_TEMPLATE.format(requirements=requirements, search_information=info,format_example=FORMAT_EXAMPLE)logger.debug(prompt)prd = await self._aask_v1(prompt, "prd", OUTPUT_MAPPING)return prd

WritePRD类的本质也是一个prompt,它会接受requirements参数,从metagpt整个流程看,requirements参数就是老板的需求message,也就是产品经理要基于老板需求message写出PRD(产品需求文档),获得结果后,自己再记到role到memory中。

role间的交互

一开始时,用户以BOSS role输入一个需求message,这个message被丢到environment上,代码如下,主要publish_message方法的cause_by使用了BossRequirement,为了方便解释,我将需求定为:开发一个Crypto的量化交易系统。

# metagpt/software_company.py/SoftwareCompanydef start_project(self, idea):self.idea = idea# Role初始观察到的messageself.environment.publish_message(Message(role="BOSS", content=idea, cause_by=BossRequirement))

ProductManager通过_watch方法关注BossRequirement,从而在运行ProductManager时,_observe方法会收到【开发一个Crypto的量化交易系统】的message作为self._rc.news,随后调用_think和_act方法,因为只有WritePRD这一个action,所以就调用WritePRD的run方法了,并将role memory中记忆的【开发一个Crypto的量化交易系统】的message作为requirements,传入了prompt中。

从startup方法相关代码可知,ProductManager后是Architect(架构师),Architect通过_watch方法关注WritePRD。

在ProductManager的_act中,将Message添加到environment中,cause_by为WritePRD,此时就可以被Architect通过_watch方法关注,并通过_observe方法获得WritePRD产出的结果,存入Architect的记忆中,然后作为WriteDesign这个action的输入,WriteDesign会将WritePRD的内容放到它的prompt中。

至此,role的交互流程就比较清晰了。

回答问题

Metagpt还有很多代码细节,但读到这,已经可以回答开头的问题了:

1.MetaGPT是怎么抽象出的软件公司开发流程的?SOP具体在代码上是怎么实现的?SOP就是不同role之间交互的顺序,比如产品经理需要BOSS的message作为他prompt的约束,而产品经理的产出是架构师的输入,不同角色间的输入输出,就是SOP。

2.MetaGPT中不同AI Agent是怎么交互的?role通过_watch确定要从哪个role的哪个action中获得这个action的输出,具体获取的过程在_observer方法中,获得其他role的message后,如果当前的role有多个action可执行,则通过_think去选一个action(使用env的memory,即当前环境中发生的内容),再通过_act去具体的执行action,执行action时,会从role的memory中获取需要的message。

3.从效果上看,MetaGPT输出的内容是比较格式化的,要做到这样的效果,prompt是怎么写的?MetaGPT的prompt的设计形式值得学习,它主要使用MarkDown格式来设计prompt,多数prompt中都会有context、example,从而让GPT4更好的发挥zero-shot能力,想知道具体的,建议直接拉代码下来看。

4.MetaGPT是怎么抽象具体的职业的?从职业这个角度讲,主要通过action和承接的message来抽象,比如产品经理,就抽象成,接收老板需求,产出产品需求文档,将需求文档给到架构师的对象,然后每个role生成的结果都会放到env中,其他人也可以看到(很多角色只有一个action,就不会用到env中的message)。

结尾

我自己跑了几遍MetaGPT,还是有明显的局限性的。如果我按example的形式,让他写python小游戏,过程是丝滑的,但我换成让MetaGPT帮我设计一个今日头条的广告推荐系统,他就给了我下面这样的东西,很明显,不太可用。

322697900ee16441f4327cea4f2926c0.png

此外,在阅读源码的过程,也发现MetaGPT团队自己提出的一些问题,比如大段的代码受限GPT4的tokens限制还做不到,切开生成,可能又需要提供额外的code作为context,效果也没那么理想。

当然MetaGPT的team我还是很respect的,代码设计清晰,然后让MetaGPT可以实现自举的这个目标也很酷。

以上。

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

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

相关文章

博客系统的自动化测试

本次自动化实战内容:本次测试根据博客管理系统这个项目,首先设计UI自动化测试用例,然后使用Selenium4自动化测试工具和JUnit5单元测试框架,实现web端的自动化测试。 本次项目总体实现思路:目录下有一个common包&#…

机器学习(19)---神经网络详解

神经网络 一、神经网络概述1.1 神经元模型1.2 激活函数 二、感知机2.1 概述2.2 实现逻辑运算2.3 多层感知机 三、神经网络3.1 工作原理3.2 前向传播3.3 Tensorflow实战演示3.3.1 导入数据集查看3.3.2 数据预处理3.3.3 建立模型3.3.4 评估模型 四、反向传播五、例题5.1 题15.2 题…

Day1-DeepWalk

论文《DeepWalk: Online Learning of Social Representations》 2014年发表在数据挖掘顶会ACM SIGKDD(KDD)上的论文 目的:学习节点表示 推动:将自然语言处理里面的无监督学习方法迁移至此 思路:将图结构序列化&#x…

ajax method to retrieve images as a blob

go 服务端: 就是先把这个图片读出来 然后返回二进制的数据 byteFile, err : ioutil.ReadFile("." "/processed/" uuidStr"processed.png")if err ! nil {fmt.Println(err)}c.Header("Content-Disposition", "att…

MYSQL不常用但好用写法

ORDER BY FIELD() 自定义排序逻辑 MySql 中的排序 ORDER BY 除了可以用 ASC 和 DESC,还可以通过 「ORDER BY FIELD(str,str1,…)」 自定义字符串/数字来实现排序。这里用 order_diy 表举例,结构以及表数据展示: ORDER BY FIELD(str,str1,…) …

Android 滑动事件消费监控,Debug 环境下通用思路

Android Debug 环境下滑动事件消费监控通用思路 背景 Android 开发中,经常会遇到滑动事件冲突。在一些简单的场景下,我们如果能够知道是那个 View 拦截了事件,那我们能够很容易得解决。解决方法通常就是内部拦截法或者外部拦截法。ViewPage…

SWC 流程

一个arxml 存储SWC (可以存多个,也可以一个arxml存一个SWC)一个arxml 存储 composition (只能存一个)一个arxml 存储 system description (通过import dbc自动生成system) 存储SWC和composition的arxml文件分开&#…

注入之SQLMAP(工具注入)

i sqlmap是一个自动化的SQL注入工具,其主要功能是扫描,发现并利用给定的URL和SQL注入漏洞,其广泛的功能和选项包括数据库指纹,枚举,数据库提权,访问目标文件系统,并在获取操作权限时执行任…

Java学习星球,十月集训,五大赛道(文末送书)

目录 什么是知识星球?我的知识星球能为你提供什么?专属专栏《Java基础教程系列》内容概览:《Java高并发编程实战》、《MySQL 基础教程系列》内容概览:《微服务》、《Redis中间件》、《Dubbo高手之路》、《华为OD机试》内容概览&am…

北工大汇编——综合题(1)

题目要求 统计字符数。从键盘输入一行字符,统计字母、空格、数字、其他宇符的个数,并显示。要求:提示输入一行宇符串;键盘输入宇符串,Enter 键结束输入,并换行显示结果。 题目代码 DATAS SEGMENT;此处输…

JSP ssm 零配件管理系统myeclipse开发mysql数据库springMVC模式java编程计算机网页设计

一、源码特点 java ssm 零配件管理系统是一套完善的web设计系统(系统采用SSM框架进行设计开发,springspringMVCmybatis),对理解JSP java编程开发语言有帮助,系统具有完整的源代码和数据库,系统主要采用…

一文了解什么SEO

搜索引擎优化 (SEO) 是一门让页面在 Google 等搜索引擎中排名更高的艺术和科学。 一、搜索引擎优化的好处 搜索引擎优化是在线营销的关键部分,因为搜索是用户浏览网络的主要方式之一。 搜索结果以有序列表的形式呈现,网站在该列表中的排名越高&#x…

Android开发MVP架构记录

Android开发MVP架构记录 安卓的MVP(Model-View-Presenter)架构是一种常见的软件设计模式,用于帮助开发者组织和分离应用程序的不同组成部分。MVP架构的目标是将应用程序的业务逻辑(Presenter)、用户界面(V…

Selenium自动化测试框架常见异常分析及解决方法

01 pycharm中导入selenium报错 现象: pycharm中输入from selenium import webdriver, selenium标红 原因1: pycharm使用的虚拟环境中没有安装selenium, 解决方法: 在pycharm中通过设置或terminal面板重新安装selenium 原因2: 当前项目下有selenium.py,和系统包名冲突导致, …

Linux学习第19天:Linux并发与竞争实例: 没有规矩不成方圆

Linux版本号4.1.15 芯片I.MX6ULL 大叔学Linux 品人间百味 思文短情长 先说点题外话,上周参加行业年会,停更了一周。接下来的周五就要开启国庆中秋双节模式,所以有的时候,尤其是工作以后…

JAVA学习-全网最详细

🌈write in front🌈 🧸大家好,我是Aileen🧸.希望你看完之后,能对你有所帮助,不足请指正!共同学习交流. 🆔本文由Aileen_0v0🧸 原创 CSDN首发🐒 如…

表格内日期比较计算

需求&#xff1a;在表格中新增数据&#xff0c;计算开始日期中最早的和结束日期中最晚的&#xff0c;回显到下方。 <el-formref"formRef":model"ruleForm":rules"rules"style"margin-top: 20px;"label-position"top">…

Golang 的 GMP:并发编程的艺术

前言 在 Golang 的并发编程中&#xff0c;GMP 是一个重要的概念&#xff0c;它代表了 Goroutine、M&#xff08;线程&#xff09;和 P&#xff08;调度器&#xff09;。这个强大的三位一体的并发模型使得 Golang 在处理并发任务时非常高效和灵活。通过 GMP 的组合&#xff0c;…

美轮美奂,尽在眼前——Aerial for Mac 高清鸟瞰屏保程序

想要让您的 Mac 屏幕焕发别样风采&#xff1f;那么&#xff0c;Aerial for Mac 高清鸟瞰屏保程序一定不容错过。这款应用程序将为您带来最优质的高清鸟瞰视频壁纸&#xff0c;让您的屏幕焕发无限活力和美感。 Aerial for Mac 高清鸟瞰屏保程序是一款专为 Mac 设计的屏幕保护程…

2023-9-25 货仓选址

题目链接&#xff1a;货仓选址 #include <iostream> #include <algorithm>using namespace std;const int N 100010;int n; int a[N];int main() {cin >> n;for(int i 0; i < n; i ) cin >> a[i];sort(a, a n);int res 0;for(int i 0; i < …
推荐文章