【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。

相关文章

在React中引入Tailwind CSS的完整指南

在React中引入Tailwind CSS的完整指南

《在React中引入TailwindCSS的完整指南》在现代前端开发中,使用UI库可以显著提高开发效率,TailwindCSS是一个功能类优先的CSS框架,本文将详细介绍如何在Reac... 目录前言一、Tailwind css 简介二、创建 React 项目使用 Create React App 创建项目
阅读更多...
SpringBoot3实现Gzip压缩优化的技术指南

SpringBoot3实现Gzip压缩优化的技术指南

《SpringBoot3实现Gzip压缩优化的技术指南》随着Web应用的用户量和数据量增加,网络带宽和页面加载速度逐渐成为瓶颈,为了减少数据传输量,提高用户体验,我们可以使用Gzip压缩HTTP响应,... 目录1、简述2、配置2.1 添加依赖2.2 配置 Gzip 压缩3、服务端应用4、前端应用4.1 N
阅读更多...
SpringBoot实现数据库读写分离的3种方法小结

SpringBoot实现数据库读写分离的3种方法小结

《SpringBoot实现数据库读写分离的3种方法小结》为了提高系统的读写性能和可用性,读写分离是一种经典的数据库架构模式,在SpringBoot应用中,有多种方式可以实现数据库读写分离,本文将介绍三... 目录一、数据库读写分离概述二、方案一：基于AbstractRoutingDataSource实现动态
阅读更多...
Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

《PythonFastAPI+Celery+RabbitMQ实现分布式图片水印处理系统》这篇文章主要为大家详细介绍了PythonFastAPI如何结合Celery以及RabbitMQ实现简单的分布式... 实现思路FastAPI 服务器Celery 任务队列RabbitMQ 作为消息代理定时任务处理完整
阅读更多...
springboot循环依赖问题案例代码及解决办法

springboot循环依赖问题案例代码及解决办法

《springboot循环依赖问题案例代码及解决办法》在SpringBoot中,如果两个或多个Bean之间存在循环依赖（即BeanA依赖BeanB,而BeanB又依赖BeanA）,会导致Spring的... 目录1. 什么是循环依赖？2. 循环依赖的场景案例3. 解决循环依赖的常见方法方法 1：使用 @La
阅读更多...
Java枚举类实现Key-Value映射的多种实现方式

Java枚举类实现Key-Value映射的多种实现方式

《Java枚举类实现Key-Value映射的多种实现方式》在Java开发中,枚举(Enum)是一种特殊的类,本文将详细介绍Java枚举类实现key-value映射的多种方式,有需要的小伙伴可以根据需要... 目录前言一、基础实现方式1.1 为枚举添加属性和构造方法二、http://www.cppcns.co
阅读更多...
使用Python实现快速搭建本地HTTP服务器

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》：本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.
阅读更多...
使用C#代码在PDF文档中添加、删除和替换图片

使用C#代码在PDF文档中添加、删除和替换图片

《使用C#代码在PDF文档中添加、删除和替换图片》在当今数字化文档处理场景中,动态操作PDF文档中的图像已成为企业级应用开发的核心需求之一,本文将介绍如何在.NET平台使用C#代码在PDF文档中添加、... 目录引言用C#添加图片到PDF文档用C#删除PDF文档中的图片用C#替换PDF文档中的图片引言在当
阅读更多...
C#使用SQLite进行大数据量高效处理的代码示例

C#使用SQLite进行大数据量高效处理的代码示例

《C#使用SQLite进行大数据量高效处理的代码示例》在软件开发中,高效处理大数据量是一个常见且具有挑战性的任务,SQLite因其零配置、嵌入式、跨平台的特性,成为许多开发者的首选数据库,本文将深入探... 目录前言准备工作数据实体核心技术批量插入：从乌龟到猎豹的蜕变分页查询：加载百万数据异步处理：拒绝界面
阅读更多...
MySQL双主搭建+keepalived高可用的实现

MySQL双主搭建+keepalived高可用的实现

《MySQL双主搭建+keepalived高可用的实现》本文主要介绍了MySQL双主搭建+keepalived高可用的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,... 目录一、测试环境准备二、主从搭建1.创建复制用户2.创建复制关系3.开启复制，确认复制是否成功4.同
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2