【AI Agent系列】【MetaGPT多智能体学习】6. 多智能体实战 - 基于MetaGPT实现游戏【你说我猜】（附完整代码）

2024-03-03 08:44
文章标签 实战 代码 实现 ai 学习 智能 系列 游戏 agent 完整 metagpt

本文主要是介绍【AI Agent系列】【MetaGPT多智能体学习】6. 多智能体实战 - 基于MetaGPT实现游戏【你说我猜】（附完整代码），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

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

本文为该课程的第四章（多智能体开发）的第四篇笔记。今天我们来完成第四章的作业：

基于 env 或 team 设计一个你的多智能体团队，尝试让他们完成 你画我猜文字版 ，要求其中含有两个agent，其中一个agent负责接收来自用户提供的物体描述并转告另一个agent，另一个agent将猜测用户给出的物体名称，两个agent将不断交互直到另一个给出正确的答案

系列笔记

  • 【AI Agent系列】【MetaGPT多智能体学习】0. 环境准备 - 升级MetaGPT 0.7.2版本及遇到的坑
  • 【AI Agent系列】【MetaGPT多智能体学习】1. 再理解 AI Agent - 经典案例和热门框架综述
  • 【AI Agent系列】【MetaGPT多智能体学习】2. 重温单智能体开发 - 深入源码，理解单智能体运行框架
  • 【AI Agent系列】【MetaGPT多智能体学习】3. 开发一个简单的多智能体系统，兼看MetaGPT多智能体运行机制
  • 【AI Agent系列】【MetaGPT多智能体学习】4. 基于MetaGPT的Team组件开发你的第一个智能体团队
  • 【AI Agent系列】【MetaGPT多智能体学习】5. 多智能体案例拆解 - 基于MetaGPT的智能体辩论（附完整代码）

文章目录

  • 系列笔记
  • 0. 需求分析
  • 1. 写代码 - 初版
    • 1.1 智能体1 - Describer实现
      • 1.1.1 Action定义 - DescribeWord
      • 1.1.2 Role定义 - Describer
    • 1.2 智能体2 - Guesser实现
      • 1.2.1 Action定义 - GuessWord
      • 1.2.2 Role定义 - Gusser
    • 1.3 定义Team，运行及结果
  • 2. 修改代码 - 效果优化
    • 2.1 存在的问题及分析
    • 2.2 Prompt优化
    • 2.3 回答正确后如何立刻停止游戏
    • 2.4 如何输出“游戏失败”的结果
  • 3. 完整代码
  • 4. 拓展 - 与人交互，人来猜词
  • 5. 总结

0. 需求分析

从上面的需求描述来看，你说我猜 游戏需要两个智能体：

  • 智能体1：Describer，用来接收用户提供的词语，并给出描述
  • 智能体2：Guesser，用来接收智能体1的描述，猜词

1. 写代码 - 初版

1.1 智能体1 - Describer实现

智能体1 Describer的任务是根据用户提供的词语，用自己的话描述出来。

1.1.1 Action定义 - DescribeWord

重点是 Prompt，这里我设置的Prompt接收两个参数，第一个参数word为用户输入的词语，也就是答案。第二个参数是Describer智能体的描述历史，因为在实际游戏过程中，描述是不会与前面的描述重复的。另外还设置了每次描述最多20个字，用来限制token的消耗。

class DescribeWord(Action):"""Action: Describe a word in your own language"""PROMPT_TMPL: str = """## 任务你现在在玩一个你画我猜的游戏，你需要用你自己的语言来描述"{word}"## 描述历史之前你的描述历史：{context}## 你必须遵守的限制1. 描述长度不超过20个字2. 描述中不能出现"{word}"中的字3. 描述不能与描述历史中的任何一条描述相同"""name: str = "DescribeWord"async def run(self, context: str, word: str):prompt = self.PROMPT_TMPL.format(context=context, word=word)logger.info(prompt)rsp = await self._aask(prompt)print(rsp)return rsp

1.1.2 Role定义 - Describer

（1）设置其 Action 为 DescribeWord
（2）设置其关注的消息来源为 UserRequirement 和 GuessWord
（3）重点重写了 _act 函数。

因为前面的Prompt中需要历史的描述信息，而描述是其自身发出的，因此历史描述信息的获取为：

if msg.sent_from == self.name:context = "\n".join(f"{msg.content}") # 自己的描述历史

另外，也在这里加了判断是否猜对了词语的逻辑：

elif msg.sent_from == "Gusser" and msg.content.find(self.word) != -1:print("回答正确！")return Message()

当回答对了之后，直接返回。

完整代码如下：

class Describer(Role):name: str = "Describer"profile: str = "Describer"word: str = ""def __init__(self, **data: Any):super().__init__(**data)self.set_actions([DescribeWord])self._watch([UserRequirement, GuessWord])async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo  # An instance of DescribeWordmemories = self.get_memories() # 获取全部的记忆context = ""for msg in memories:if msg.sent_from == self.name:context = "\n".join(f"{msg.content}") # 自己的描述历史elif msg.sent_from == "Gusser" and msg.content.find(self.word) != -1:print("回答正确！")return Message()print(context)rsp = await todo.run(context=context, word=self.word)msg = Message(content=rsp,role=self.profile,cause_by=type(todo),sent_from=self.name,)self.rc.memory.add(msg)return msg

1.2 智能体2 - Guesser实现

智能体2 - Guesser，用来接收智能体1的描述，猜词。

1.2.1 Action定义 - GuessWord

与 DescribeWord Action的Prompt类似，猜词的Prompt接收一个context来表示之前它的猜词历史，避免它老重复猜同一个词，陷入死循环。然后一个description来接收Describer的描述语句。

class GuessWord(Action):"""Action: Guess a word from the description"""PROMPT_TMPL: str = """## 背景你现在在玩一个你画我猜的游戏，你的任务是根据给定的描述，猜一个词语。## 猜测历史之前你的猜测历史：{context}## 轮到你了现在轮到你了，你需要根据描述{description}猜测一个词语，并遵循以下限制：### 限制1. 猜测词语不超过5个字2. 猜测词语不能与猜测历史重复3. 只输出猜测的词语，NO other texts"""name: str = "GuessWord"async def run(self, context: str, description: str):prompt = self.PROMPT_TMPL.format(context=context, description=description)logger.info(prompt)rsp = await self._aask(prompt)return rsp

1.2.2 Role定义 - Gusser

（1）设置其 Action 为 GuessWord
（2）设置其关注的消息来源为 DescribeWord
（3）重点重写了 _act 函数。

因为前面的Prompt中需要历史的猜词信息，而猜词是其自身发出的，因此猜词历史信息的获取为：

if msg.sent_from == self.name:context = "\n".join(f"{msg.content}")

Describer的描述信息获取为：

elif msg.sent_from == "Describer":description = "\n".join(f"{msg.content}")

完整代码如下：


class Gusser(Role):name: str = "Gusser"profile: str = "Gusser"def __init__(self, **data: Any):super().__init__(**data)self.set_actions([GuessWord])self._watch([DescribeWord])async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo  # An instance of DescribeWordmemories = self.get_memories() # 获取全部的记忆context= ""description = ""for msg in memories:if msg.sent_from == self.name:context = "\n".join(f"{msg.content}")elif msg.sent_from == "Describer":description = "\n".join(f"{msg.content}")print(context)rsp = await todo.run(context=context, description=description)msg = Message(content=rsp,role=self.profile,cause_by=type(todo),sent_from=self.name,)self.rc.memory.add(msg)print(rsp)return msg

1.3 定义Team，运行及结果

async def start_game(idea: str, investment: float = 3.0, n_round: int = 10):team = Team()team.hire([Describer(word=idea),Gusser(), ])team.invest(investment)team.run_project(idea)await team.run(n_round=n_round)def main(idea: str, investment: float = 3.0, n_round: int = 10):if platform.system() == "Windows":asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())asyncio.run(start_game(idea, investment, n_round))if __name__ == "__main__":fire.Fire(main("篮球"))

运行结果如下：

智能体产生描述：
在这里插入图片描述

猜词，检测结果：

在这里插入图片描述
可以看到，运行成功了，也能进行简单的交互。但是还是能看出不少问题的。

下面是进一步优化的过程。

2. 修改代码 - 效果优化

2.1 存在的问题及分析

（1）猜对答案后，它后面还是在循环运行，直到运行完刚开始设置的运行轮数：n_round: int = 10。如上面的运行结果，后面一直在输出“回答正确”。

（2）看下图的运行结果，回答了英文，导致一直认为不是正确答案。并且一直在重复这个词，所以，Prompt还需要优化：

在这里插入图片描述

（3）10轮后结束运行，如果这时候没有猜对答案，没有输出“你失败了”类似的文字。

总结下主要问题：

  • 回答正确后如何立刻停止游戏
  • Prompt需要优化
  • 如何输出“游戏失败”的结果

2.2 Prompt优化

Prompt优化的原则是，有啥问题堵啥问题…

（1）它既然输出了英文词语，那就限制它不让它输出英文单词，只输出中文。
（2）它重复输出了之前的猜词，说明猜词历史的限制没有生效，改变话术各种试（没有好的方法，只有各种试）。

修改之后的 Prompt：

class DescribeWord(Action):"""Action: Describe a word in your own language"""PROMPT_TMPL: str = """## 任务你现在在玩一个你画我猜的游戏，你需要用你自己的语言来描述"{word}"## 描述历史之前你的描述历史：{context}## 你必须遵守的限制1. 描述长度不超过20个字2. 描述中不能出现与"{word}"中的任何一个字相同的字，否则会有严重的惩罚。例如：描述的词为"雨伞"，那么生成的描述中不能出现"雨"，"伞","雨伞"3. 描述不能与描述历史中的任何一条描述相同, 例如：描述历史中已经出现过"一种工具"，那么生成的描述就不能再是"一种工具""""

class GuessWord(Action):"""Action: Guess a word from the description"""PROMPT_TMPL: str = """## 任务你现在在玩一个你画我猜的游戏，你需要根据描述"{description}"猜测出一个词语## 猜测历史之前你的猜测历史：{context}### 你必须遵守的限制1. 猜测词语不超过5个字，词语必须是中文2. 猜测词语不能与猜测历史重复3. 只输出猜测的词语，NO other texts"""

优化之后的运行效果，虽然还是有点小问题（描述中出现了重复和出现了答案中的字），但最终效果还行吧… ：

在这里插入图片描述

2.3 回答正确后如何立刻停止游戏

当 await team.run(n_round=n_round) 之后，不运行完 n_round 是不会返回的，而 Team 组件目前也没有接口来设置停止运行。因此想要立刻停止游戏，用Team组件几乎是不可能的（有方法的欢迎指教）。

所以我想了另一种办法：既然无法立刻停止游戏，那就停止两个智能体的行动，让他们一直等待n_round完就行了，就像等待游戏时间结束。

代码修改也很简单：

elif msg.sent_from == "Gusser" and msg.content.find(self.word) != -1:print("回答正确！")return ""

只要在回答正确后，直接return一个空字符串就行。为什么这样就可以？看源码：

def publish_message(self, msg):"""If the role belongs to env, then the role's messages will be broadcast to env"""if not msg:return

在运行完动作_act后，往环境中放结果消息，如果为空，就不忘环境中放消息了。这样Guesser也就接收不到 Describer 的消息，也就不动作了。剩下的 n_round 就是在那空转了。

看下运行效果：

在这里插入图片描述
可以看到，只输出了一次“回答正确”，之后就没有其余打印了，直到程序结束。

2.4 如何输出“游戏失败”的结果

如果 n_round 运行完之后，还没有猜对结果，就要宣告游戏失败了。怎么获取这个结果呢？

程序运行结束，只能是在这里返回：await team.run(n_round=n_round)

我们将它的返回值打出来看下是什么：

result = await team.run(n_round=n_round)
print(result)

打印结果如下：

在这里插入图片描述
可以看到它的返回结果就是所有的对话历史。那么判断游戏是否失败就好说了，有很多种方法，例如直接比较用户输入的词语是否与这个结果中的最后一行相同：

result = result.split(':')[-1].strip(' ')
if (result.find(idea) != -1):print("恭喜你，猜对了！")
else:print("很遗憾，你猜错了！")

运行效果：

在这里插入图片描述

3. 完整代码


import asyncio
from typing import Any
import platformimport 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 Teamclass DescribeWord(Action):"""Action: Describe a word in your own language"""PROMPT_TMPL: str = """## 任务你现在在玩一个你画我猜的游戏，你需要用你自己的语言来描述"{word}"## 描述历史之前你的描述历史：{context}## 你必须遵守的限制1. 描述长度不超过20个字2. 描述中不能出现与"{word}"中的任何一个字相同的字，否则会有严重的惩罚。例如：描述的词为"雨伞"，那么生成的描述中不能出现"雨"，"伞","雨伞"3. 描述不能与描述历史中的任何一条描述相同, 例如：描述历史中已经出现过"一种工具"，那么生成的描述就不能再是"一种工具""""name: str = "DescribeWord"async def run(self, context: str, word: str):prompt = self.PROMPT_TMPL.format(context=context, word=word)logger.info(prompt)rsp = await self._aask(prompt)# print(rsp)return rspclass GuessWord(Action):"""Action: Guess a word from the description"""PROMPT_TMPL: str = """## 任务你现在在玩一个你画我猜的游戏，你需要根据描述"{description}"猜测出一个词语## 猜测历史之前你的猜测历史：{context}### 你必须遵守的限制1. 猜测词语不超过5个字，词语必须是中文2. 猜测词语不能与猜测历史重复3. 只输出猜测的词语，NO other texts"""name: str = "GuessWord"async def run(self, context: str, description: str):prompt = self.PROMPT_TMPL.format(context=context, description=description)logger.info(prompt)rsp = await self._aask(prompt)return rspclass Describer(Role):name: str = "Describer"profile: str = "Describer"word: str = ""def __init__(self, **data: Any):super().__init__(**data)self.set_actions([DescribeWord])self._watch([UserRequirement, GuessWord])async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo  # An instance of DescribeWordmemories = self.get_memories() # 获取全部的记忆context = ""for msg in memories:if msg.sent_from == self.name:context += f"{msg.content}\n" # 自己的描述历史elif msg.sent_from == "Gusser" and msg.content.find(self.word) != -1:print("回答正确！")return ""# print(context)rsp = await todo.run(context=context, word=self.word)msg = Message(content=rsp,role=self.profile,cause_by=type(todo),sent_from=self.name,)self.rc.memory.add(msg)return msgclass Gusser(Role):name: str = "Gusser"profile: str = "Gusser"def __init__(self, **data: Any):super().__init__(**data)self.set_actions([GuessWord])self._watch([DescribeWord])async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo  # An instance of DescribeWordmemories = self.get_memories() # 获取全部的记忆context= ""description = ""for msg in memories:if msg.sent_from == self.name:context += f"{msg.content}\n"elif msg.sent_from == "Describer":description += f"{msg.content}\n"print(context)rsp = await todo.run(context=context, description=description)msg = Message(content=rsp,role=self.profile,cause_by=type(todo),sent_from=self.name,)self.rc.memory.add(msg)# print(rsp)return msgasync def start_game(idea: str, investment: float = 3.0, n_round: int = 10):team = Team()team.hire([Describer(word=idea),Gusser(), ])team.invest(investment)team.run_project(idea)result = await team.run(n_round=n_round)result = result.split(':')[-1].strip(' ')if (result.find(idea) != -1):print("恭喜你，猜对了！")else:print("很遗憾，你猜错了！")def main(idea: str, investment: float = 3.0, n_round: int = 3):if platform.system() == "Windows":asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())asyncio.run(start_game(idea, investment, n_round))if __name__ == "__main__":fire.Fire(main("打篮球运行"))

4. 拓展 - 与人交互，人来猜词

可以做下拓展，将猜词的Role换成你自己，你自己来猜词，与智能体进行交互。这实现起来比较简单。

代表人的智能体，只需要在实例化智能体时，将 Role 的 is_human 属性置为 true 即可：

team.hire([Describer(word=idea),Gusser(is_human=True),  # is_human=True 代表这个角色是人类，需要你的输入])

运行效果：

在这里插入图片描述
还可以引入另一个智能体来自动出词语。大家可以思考下应该怎么实现。

5. 总结

本文我们利用MetaGPT的Team组件实现了一个“你说我猜”的游戏。因为游戏比较简单，所以整体逻辑也比较简单。重点在于Prompt优化比较费劲，还有就是要注意何时结束游戏等细节。最后，也向大家展示了一下如何让人参与到游戏中。

站内文章一览

在这里插入图片描述

这篇关于【AI Agent系列】【MetaGPT多智能体学习】6. 多智能体实战 - 基于MetaGPT实现游戏【你说我猜】（附完整代码）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/769093。

相关文章

SpringBoot集成redisson实现延时队列教程

SpringBoot集成redisson实现延时队列教程

《SpringBoot集成redisson实现延时队列教程》文章介绍了使用Redisson实现延迟队列的完整步骤,包括依赖导入、Redis配置、工具类封装、业务枚举定义、执行器实现、Bean创建、消费... 目录1、先给项目导入Redisson依赖2、配置redis3、创建 RedissonConfig 配
阅读更多...
Python的Darts库实现时间序列预测

Python的Darts库实现时间序列预测

《Python的Darts库实现时间序列预测》Darts一个集统计、机器学习与深度学习模型于一体的Python时间序列预测库,本文主要介绍了Python的Darts库实现时间序列预测,感兴趣的可以了解... 目录目录一、什么是 Darts？二、安装与基本配置安装 Darts导入基础模块三、时间序列数据结构与
阅读更多...
Python使用FastAPI实现大文件分片上传与断点续传功能

Python使用FastAPI实现大文件分片上传与断点续传功能

《Python使用FastAPI实现大文件分片上传与断点续传功能》大文件直传常遇到超时、网络抖动失败、失败后只能重传的问题,分片上传+断点续传可以把大文件拆成若干小块逐个上传,并在中断后从已完成分片继... 目录一、接口设计二、服务端实现（FastAPI）2.1 运行环境2.2 目录结构建议2.3 serv
阅读更多...
C#实现千万数据秒级导入的代码

C#实现千万数据秒级导入的代码

《C#实现千万数据秒级导入的代码》在实际开发中excel导入很常见,现代社会中很容易遇到大数据处理业务,所以本文我就给大家分享一下千万数据秒级导入怎么实现,文中有详细的代码示例供大家参考,需要的朋友可... 目录前言一、数据存储二、处理逻辑优化前代码处理逻辑优化后的代码总结前言在实际开发中excel导入很
阅读更多...
MyBatis分页查询实战案例完整流程

MyBatis分页查询实战案例完整流程

《MyBatis分页查询实战案例完整流程》MyBatis是一个强大的Java持久层框架,支持自定义SQL和高级映射,本案例以员工工资信息管理为例,详细讲解如何在IDEA中使用MyBatis结合Page... 目录1. MyBATis框架简介2. 分页查询原理与应用场景2.1 分页查询的基本原理2.1.1 分
阅读更多...
SpringBoot+RustFS 实现文件切片极速上传的实例代码

SpringBoot+RustFS 实现文件切片极速上传的实例代码

《SpringBoot+RustFS实现文件切片极速上传的实例代码》本文介绍利用SpringBoot和RustFS构建高性能文件切片上传系统,实现大文件秒传、断点续传和分片上传等功能,具有一定的参考... 目录一、为什么选择 RustFS + SpringBoot？二、环境准备与部署2.1 安装 RustF
阅读更多...
Nginx部署HTTP/3的实现步骤

Nginx部署HTTP/3的实现步骤

《Nginx部署HTTP/3的实现步骤》本文介绍了在Nginx中部署HTTP/3的详细步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学... 目录前提条件第一步：安装必要的依赖库第二步：获取并构建 BoringSSL第三步：获取 Nginx
阅读更多...
MyBatis Plus实现时间字段自动填充的完整方案

MyBatis Plus实现时间字段自动填充的完整方案

《MyBatisPlus实现时间字段自动填充的完整方案》在日常开发中,我们经常需要记录数据的创建时间和更新时间,传统的做法是在每次插入或更新操作时手动设置这些时间字段,这种方式不仅繁琐,还容易遗漏,... 目录前言解决目标技术栈实现步骤1. 实体类注解配置2. 创建元数据处理器3. 服务层代码优化填充机制详
阅读更多...
Python实现Excel批量样式修改器(附完整代码)

Python实现Excel批量样式修改器(附完整代码)

《Python实现Excel批量样式修改器(附完整代码)》这篇文章主要为大家详细介绍了如何使用Python实现一个Excel批量样式修改器,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一... 目录前言功能特性核心功能界面特性系统要求安装说明使用指南基本操作流程高级功能技术实现核心技术栈关键函
阅读更多...
Java实现字节字符转bcd编码

Java实现字节字符转bcd编码

《Java实现字节字符转bcd编码》BCD是一种将十进制数字编码为二进制的表示方式,常用于数字显示和存储,本文将介绍如何在Java中实现字节字符转BCD码的过程,需要的小伙伴可以了解下... 目录前言BCD码是什么Java实现字节转bcd编码方法补充总结前言BCD码（Binary-Coded Decima
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2