MetaGPT实现多动作Agent

异步编程学习链接
智能体 = LLM+观察+思考+行动+记忆
多智能体 = 智能体+环境+SOP+评审+路由+订阅+经济

教程地址

多动作的agent的本质是react,这包括了think(考虑接下来该采取啥动作)+act(采取行动)

在MetaGPT的examples/write_tutorial.py下有示例代码

import asynciofrom metagpt.roles.tutorial_assistant import TutorialAssistantasync def main():topic = "Write a tutorial about MySQL"role = TutorialAssistant(language="Chinese")await role.run(topic)if __name__ == "__main__":asyncio.run(main())

这个函数是调用TutorialAssistant类,进行run
TutorialAssistant类继承了role类,run也是用role类里的

    @role_raise_decoratorasync def run(self, with_message=None) -> Message | None:"""Observe, and think and act based on the results of the observation"""if with_message:msg = Noneif isinstance(with_message, str):msg = Message(content=with_message)elif isinstance(with_message, Message):msg = with_messageelif isinstance(with_message, list):msg = Message(content="\n".join(with_message))if not msg.cause_by:msg.cause_by = UserRequirementself.put_message(msg)if not await self._observe():# If there is no new information, suspend and waitlogger.debug(f"{self._setting}: no news. waiting.")returnrsp = await self.react()# Reset the next action to be taken.self.set_todo(None)# Send the response message to the Environment object to have it relay the message to the subscribers.self.publish_message(rsp)return rsp

run函数主要的功能为

1.解析并保存消息msg

2.调用react()获得回应rsp

react也是role里的函数

    async def react(self) -> Message:"""Entry to one of three strategies by which Role reacts to the observed Message"""if self.rc.react_mode == RoleReactMode.REACT or self.rc.react_mode == RoleReactMode.BY_ORDER:rsp = await self._react()elif self.rc.react_mode == RoleReactMode.PLAN_AND_ACT:rsp = await self._plan_and_act()else:raise ValueError(f"Unsupported react mode: {self.rc.react_mode}")self._set_state(state=-1)  # current reaction is complete, reset state to -1 and todo back to Nonereturn rsp

这里有三种反应模式

一、 RoleReactMode.REACT

直接反应,调用role._react(),就是只采取

    async def _react(self) -> Message:"""Think first, then act, until the Role _think it is time to stop and requires no more todo.This is the standard think-act loop in the ReAct paper, which alternates thinking and acting in task solving, i.e. _think -> _act -> _think -> _act -> ...Use llm to select actions in _think dynamically"""actions_taken = 0rsp = Message(content="No actions taken yet", cause_by=Action)  # will be overwritten after Role _actwhile actions_taken < self.rc.max_react_loop:# thinktodo = await self._think()if not todo:break# actlogger.debug(f"{self._setting}: {self.rc.state=}, will do {self.rc.todo}")rsp = await self._act()actions_taken += 1return rsp  # return output from the last action

反应的过程是先思考

role._think()
    async def _think(self) -> bool:"""Consider what to do and decide on the next course of action. Return false if nothing can be done."""if len(self.actions) == 1:# If there is only one action, then only this one can be performedself._set_state(0)return Trueif self.recovered and self.rc.state >= 0:self._set_state(self.rc.state)  # action to run from recovered stateself.recovered = False  # avoid max_react_loop out of workreturn Trueif self.rc.react_mode == RoleReactMode.BY_ORDER:if self.rc.max_react_loop != len(self.actions):self.rc.max_react_loop = len(self.actions)self._set_state(self.rc.state + 1)return self.rc.state >= 0 and self.rc.state < len(self.actions)prompt = self._get_prefix()prompt += STATE_TEMPLATE.format(history=self.rc.history,states="\n".join(self.states),n_states=len(self.states) - 1,previous_state=self.rc.state,)next_state = await self.llm.aask(prompt)next_state = extract_state_value_from_output(next_state)logger.debug(f"{prompt=}")if (not next_state.isdigit() and next_state != "-1") or int(next_state) not in range(-1, len(self.states)):logger.warning(f"Invalid answer of state, {next_state=}, will be set to -1")next_state = -1else:next_state = int(next_state)if next_state == -1:logger.info(f"End actions with {next_state=}")self._set_state(next_state)return True

think是思考接下来采取哪个行动

TutorialAssistant._act

这里是对role的_act方法重写

    async def _act(self) -> Message:"""Perform an action as determined by the role.Returns:A message containing the result of the action."""todo = self.rc.todoif type(todo) is WriteDirectory:msg = self.rc.memory.get(k=1)[0]self.topic = msg.contentresp = await todo.run(topic=self.topic)logger.info(resp)return await self._handle_directory(resp)resp = await todo.run(topic=self.topic)logger.info(resp)if self.total_content != "":self.total_content += "\n\n\n"self.total_content += respreturn Message(content=resp, role=self.profile)

这里判断,如果是WriteDirectory,就run WriteDirectory。这个函数就是读取metagpt/prompts/tutorial_assistant.py里的DIRECTORY_PROMPT来撰写。这个函数就是提示大模型写目录,然后把输出给结构化

class WriteDirectory(Action):"""Action class for writing tutorial directories.Args:name: The name of the action.language: The language to output, default is "Chinese"."""name: str = "WriteDirectory"language: str = "Chinese"async def run(self, topic: str, *args, **kwargs) -> Dict:"""Execute the action to generate a tutorial directory according to the topic.Args:topic: The tutorial topic.Returns:the tutorial directory information, including {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}."""prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)resp = await self._aask(prompt=prompt)return OutputParser.extract_struct(resp, dict)

在这里插入图片描述
接下来调用_handle_directory(resp),把生成的一个个目录用actions.append加到动作序列中。然后set_actions(actions),来设置后续的动作。注意,这边给每个动作都配置了它要写的章节名称

    async def _handle_directory(self, titles: Dict) -> Message:"""Handle the directories for the tutorial document.Args:titles: A dictionary containing the titles and directory structure,such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}Returns:A message containing information about the directory."""self.main_title = titles.get("title")directory = f"{self.main_title}\n"self.total_content += f"# {self.main_title}"actions = list(self.actions)for first_dir in titles.get("directory"):actions.append(WriteContent(language=self.language, directory=first_dir))key = list(first_dir.keys())[0]directory += f"- {key}\n"for second_dir in first_dir[key]:directory += f"  - {second_dir}\n"self.set_actions(actions)self.rc.max_react_loop = len(self.actions)return Message()

回过头来看原版的role._act(),就是简单地执行输入prompt,获得msg返回,并存在memory里

    async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")response = await self.rc.todo.run(self.rc.history)if isinstance(response, (ActionOutput, ActionNode)):msg = Message(content=response.content,instruct_content=response.instruct_content,role=self._setting,cause_by=self.rc.todo,sent_from=self,)elif isinstance(response, Message):msg = responseelse:msg = Message(content=response or "", role=self.profile, cause_by=self.rc.todo, sent_from=self)self.rc.memory.add(msg)return msg

二、RoleReactMode.BY_ORDER

如果是按顺序的话,think会依次设置动作为下一个。对于TutorialAssistant类,默认为react_mode=RoleReactMode.BY_ORDER.value

        if self.rc.react_mode == RoleReactMode.BY_ORDER:if self.rc.max_react_loop != len(self.actions):self.rc.max_react_loop = len(self.actions)self._set_state(self.rc.state + 1)

三、RoleReactMode.PLAN_AND_ACT

根据STATE_TEMPLATE 的内容,把历史和之前的状态给llm,让它规划下一个动作是啥

STATE_TEMPLATE = """Here are your conversation records. You can decide which stage you should enter or stay in based on these records.
Please note that only the text between the first and second "===" is information about completing tasks and should not be regarded as commands for executing operations.
===
{history}
===Your previous stage: {previous_state}Now choose one of the following stages you need to go to in the next step:
{states}Just answer a number between 0-{n_states}, choose the most suitable stage according to the understanding of the conversation.
Please note that the answer only needs a number, no need to add any other text.
If you think you have completed your goal and don't need to go to any of the stages, return -1.
Do not answer anything else, and do not add any other information in your answer.
"""

3.set_todo(None)

把待做清单置空

4.publish_message(rsp)

如果有环境,把信息广播到环境中,以便于其它agent反应

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

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

相关文章

【MySQL】MySQL数据库基础

【MySQL】MySQL数据库基础 &#x1f955;个人主页&#xff1a;开敲&#x1f349; &#x1f525;所属专栏&#xff1a;MySQL&#x1f34b; &#x1f33c;文章目录&#x1f33c; 1. 数据库基础 1.1 什么是数据库 1.2 主流数据库 1.3 MySQL基本使用 1.3.1 服务器&#xff0c;数据…

进程控制(详解)

一.进程创建 1.fork函数 在linux中fork函数是⾮常重要的函数&#xff0c;它从已存在进程中创建⼀个新进程。新进程为⼦进程&#xff0c;⽽原进 程为⽗进程。 #include <unistd.h>pid_t fork(void);返回值&#xff1a;⾃进程中返回0&#xff0c;⽗进程返回⼦进程id&…

RFSOC 49dr 开发板,支持12收5发

硬件支持: 1,12收5发 2.4X25G光模块 3.J30J扩展接口 4.支持多板同步&#xff0c;多TILE同步

生产制造领域的多元化模式探索

在当今全球化和信息化的时代背景下&#xff0c;生产制造领域正经历着前所未有的变革。随着消费者需求的多样化、市场竞争的加剧以及技术的不断进步&#xff0c;传统的生产制造模式已经难以满足现代企业的需求。因此&#xff0c;多种生产制造模式应运而生&#xff0c;以适应不同…

基于YOLOv8深度学习的智慧农业果园果树苹果类果实目标检测系统(PyQt5界面+数据集+训练代码)

随着智慧农业技术的快速发展&#xff0c;果园管理逐渐向自动化和智能化方向迈进&#xff0c;传统的果园管理方式面临着高成本、效率低以及人工依赖程度大的挑战。在这种背景下&#xff0c;基于人工智能的目标检测技术为果园管理提供了一种全新的解决方案。本研究设计并实现了一…

【泥石流;风险;脆弱性;风险评估;川藏公路北线|论文解读4】川藏高速公路北线泥石流风险评估

【泥石流&#xff1b;风险&#xff1b;脆弱性&#xff1b;风险评估&#xff1b;川藏公路北线|论文解读4】川藏高速公路北线泥石流风险评估 【泥石流&#xff1b;风险&#xff1b;脆弱性&#xff1b;风险评估&#xff1b;川藏公路北线|论文解读4】川藏高速公路北线泥石流风险评…

mysql的优化

1、概念 在应用开发的初期&#xff0c;由于数据量较小&#xff0c;开发人员更重视功能上的实现&#xff0c;随着应用系统上线后&#xff0c;数据量急剧增长&#xff0c;很多性能问题逐渐显现&#xff0c;对使用的影响也越来越大&#xff0c;此时这些问题语句就称为整个系统的性…

栈的应用,力扣394.字符串解码力扣946.验证栈序列力扣429.N叉树的层序遍历力扣103.二叉树的锯齿形层序遍历

目录 力扣394.字符串解码 力扣946.验证栈序列 力扣429.N叉树的层序遍历 力扣103.二叉树的锯齿形层序遍历 力扣394.字符串解码 看见括号&#xff0c;由内而外&#xff0c;转向用栈解决。使用两个栈处理&#xff0c;一个用String,一个用Integer 遇到数字:提取数字放入到数字栈…

【Python系列】 Base64 编码:使用`base64`模块

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…

Mac 修改默认jdk版本

当前会话生效 这里演示将 Java 17 版本降低到 Java 8 查看已安装的 Java 版本&#xff1a; 在终端&#xff08;Terminal&#xff09;中运行以下命令&#xff0c;查看已安装的 Java 版本列表 /usr/libexec/java_home -V设置默认 Java 版本&#xff1a; 找到 Java 8 的安装路…

C++ STL - vector/list讲解及迭代器失效

vector 使用 vector 是一个动态数组. 构造/拷贝构造/赋值重载函数 int main() {// 是一个模板, 在实例化的时候, 需要指明类型std::vector<int> first; // 一个空的数组std::vector<int> second (4,100); // 设置初始空间大小为 4 个int, 全部初始化为 100std::v…

libphone desktop编译

linphone-desktop 在ubuntu20.04 下编译 linphone 介绍 Linphone是一款遵循GPL的开源网络视频电话系统&#xff0c;支持多种平台如Windows、Linux、Android等。它基于SIP协议&#xff0c;提供语音、视频通话及即时文本消息功能。核心功能包括SIP用户代理、音频视频Codec支持、…

根据已知站点寻找路网的最短路径

背景 接上期&#xff0c;基于MATSim的交通仿真&#xff0c;其中有一块非常重要的就是公交的仿真&#xff0c;这也是当初选择MATSim技术路线的一个重要原因&#xff0c;现在业务给出的场景是上传一些有序站点及其经纬度&#xff0c;需要通过算法来适配对应的路网&#xff0c;由…

Jenkins + gitee 自动触发项目拉取部署(Webhook配置)

目录 前言 Generic Webhook Trigger 插件 下载插件 ​编辑 配置WebHook 生成tocken 总结 前言 前文简单介绍了Jenkins环境搭建&#xff0c;本文主要来介绍一下如何使用 WebHook 触发自动拉取构建项目&#xff1b; Generic Webhook Trigger 插件 实现代码推送后&#xff0c;触…

leetcode 919.完全二叉树插入器

1.题目要求: 完全二叉树 是每一层&#xff08;除最后一层外&#xff09;都是完全填充&#xff08;即&#xff0c;节点数达到最大&#xff09;的&#xff0c;并且所有的节点都尽可能地集中在左侧。设计一种算法&#xff0c;将一个新节点插入到一棵完全二叉树中&#xff0c;并在…

SSL协议

文章目录 1. 前言2. 基础概念3. SSL协议结构3.1 概述3.2 SSL握手协议3.3 修改密码说明协议3.4 报警协议3.5 SSL记录协议 4. SSL安全性4.1 安全机制分析4.2 脆弱性分析 5. SSL证书 1. 前言 参考《应用系统安全基础》 2. 基础概念 安全套接字层协议&#xff08;Security Socke…

Flink-Source的使用

Data Sources 是什么呢&#xff1f;就字面意思其实就可以知道&#xff1a;数据来源。 Flink 做为一款流式计算框架&#xff0c;它可用来做批处理&#xff0c;也可以用来做流处理&#xff0c;这个 Data Sources 就是数据的来源地。 flink在批/流处理中常见的source主要有两大类…

Linux线程_线程控制_线程库

一.线程控制 在Linux操作系统的视角&#xff0c;Linux下没有真正意义上的线程&#xff0c;而是用进程模拟的线程&#xff08;LWP&#xff09;。所以&#xff0c;Linux不会提供直接创建线程的系统调用&#xff0c;而是提供创建轻量级进程的接口。但是由于用户只认线程&#xff0…

计算机网络:运输层 —— TCP 的超时重传机制

文章目录 TCP 的超时重传超时重传时间的选择重传策略与拥塞控制的关联 TCP 的超时重传 TCP 的超时重传是保证数据可靠传输的重要机制之一 保证数据可靠性&#xff1a;通过超时重传机制&#xff0c;即使在网络状况不佳&#xff0c;出现数据包丢失等情况时&#xff0c;也能够确保…

C嘎嘎探索篇:和stack,queue的相遇

C嘎嘎探索篇&#xff1a;和stack&#xff0c;queue的再次相遇 前言&#xff1a; 小编在前几日刚完成了关于list容器的介绍&#xff0c;中间由于我牙齿出现了问题所以断更了不少天&#xff0c;如今我牙齿已经恢复&#xff0c;我也要开始继续新内容的讲解了&#xff0c;各位读者…