备案控制台
开发者社区 人工智能 文章 正文

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

2024-04-23 80
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 【AI Agent系列】【MetaGPT多智能体学习】6. 多智能体实战 - 基于MetaGPT实现游戏【你说我猜】（附完整代码）

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

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

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

系列笔记


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 DescribeWord
        memories = 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 DescribeWord
        memories = 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 platform
import fire
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
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
    
class 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 rsp
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 DescribeWord
        memories = 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 msg
    
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 DescribeWord
        memories = 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 msg
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)
    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
AI agent
通义星尘智能体
AI智能体
同学小张
目录
相关文章
汀丶人工智能
|
2天前
|
存储 人工智能 自然语言处理
AI Agent框架（LLM Agent）：LLM驱动的智能体如何引领行业变革，应用探索与未来展望
【7月更文挑战第2天】AI Agent框架（LLM Agent）：LLM驱动的智能体如何引领行业变革，应用探索与未来展望
汀丶人工智能
23 1
AI Agent框架（LLM Agent）：LLM驱动的智能体如何引领行业变革，应用探索与未来展望
程序员半支烟
|
8天前
|
人工智能 自然语言处理 前端开发
你要的AI Agent工具都在这里
只有让LLM（大模型）学会使用工具，才能做出一系列实用的AI Agent，才能发挥出LLM真正的实力。本篇，我们让AI Agent使用更多的工具，比如：外部搜索、分析CSV、文生图、执行代码等。
程序员半支烟
62 3
你要的AI Agent工具都在这里
向宇it
|
9天前
|
人工智能 图形学
【制作100个unity游戏之24】unity制作一个3D动物AI生态系统游戏2（附项目源码）
【制作100个unity游戏之24】unity制作一个3D动物AI生态系统游戏2（附项目源码）
向宇it
11 1
【制作100个unity游戏之24】unity制作一个3D动物AI生态系统游戏2（附项目源码）
楠竹11
|
4天前
|
机器学习/深度学习 人工智能 自然语言处理
能自动查找各种漏洞，专用于网络安全的AI Agent
【7月更文挑战第3天】研究表明，大型语言模型（LLM）可应用于网络安全，自动发现和利用漏洞，包括零日漏洞。多Agent系统中，HPTSA规划任务，其他Agent执行具体工作，提升发现和应对漏洞的效率。然而，LLM的复杂性可能导致可解释性、可控性问题，且系统鲁棒性、泛化能力和安全性仍需加强。[链接: arxiv.org/abs/2406.01637]
楠竹11
16 1
向宇it
|
9天前
|
人工智能 图形学
【制作100个unity游戏之24】unity制作一个3D动物AI生态系统游戏1（附项目源码）
【制作100个unity游戏之24】unity制作一个3D动物AI生态系统游戏1（附项目源码）
向宇it
12 2
雅纯
|
12天前
|
人工智能 自然语言处理 算法
哈啰集团全面接入通义灵码，AI 生成代码占比 20%，研发提效 12%
官宣！哈啰集团全面接入阿里云通义灵码专属版
雅纯
200 5
向宇it
|
9天前
|
人工智能 定位技术 图形学
【unity实战】制作敌人的AI，使用有限状态机、继承和抽象类多态 定义不同状态的敌人行为
【unity实战】制作敌人的AI，使用有限状态机、继承和抽象类多态 定义不同状态的敌人行为
向宇it
51 1
小森ai小小贾
|
15天前
|
机器学习/深度学习 人工智能 算法
AI - 集成学习
集成学习是一种机器学习策略，它通过组合多个模型（称为基学习器）来创建一个更强大、更稳健的预测模型。基学习器可以是不同类型或同类型的模型，如决策树、SVM、神经网络等。
小森ai小小贾
19 4
区块链WEB3技术团队
|
13天前
|
人工智能 搜索推荐 安全
AI技术如何个性化指导元宇宙学习
**元宇宙学习融合VR/AR、区块链、AI、5G通信、三维建模、云计算，确保沉浸式交互体验。AI个性化推荐，区块链保障数据安全，5G提供高速连接，而安全技术守护用户隐私。**
区块链WEB3技术团队
33 1
源码星辰
|
14天前
|
机器学习/深度学习 人工智能 自然语言处理
OpenCV与AI深度学习之常用AI名词解释学习
AGI：Artificial General Intelligence (通用人工智能)：是指具备与人类同等或超越人类的智能，能够表现出正常人类所具有的所有智能行为。又被称为强人工智能。
源码星辰
27 2

热门文章

最新文章

  • 1
    智能体-Agent能力升级！新增Assistant API & Tools API服务接口
  • 2
    揭秘Java Agent技术：解锁Java工具开发的新境界
  • 3
    用AI唤醒老照片里的记忆（GPU版）
  • 4
    智能文档检索和简便集成的开源 AI 问答工具：Danswer
  • 5
    IBM更新Watson为数据密集型AI应用提速
  • 6
    足球黑科技之AI与足球智能分析
  • 7
    人工智能、机器学习和深度学习
  • 8
    《阿里云AI产品必知必会系列电子书》——机器学习平台PAI——DesignerQuickStart使用教程（1）
  • 9
    从AI测温到安防机器人 智能安防会是新的“守门神”吗？
  • 10
    引发热议的AI翻译歧视，是理所应当的事么？
  • 1
    智能体-Agent能力升级！新增Assistant API & Tools API服务接口
    245
  • 2
    揭秘Java Agent技术：解锁Java工具开发的新境界
    241
  • 3
    Agent AI智能体的未来
    121
  • 4
    【AIGC】LangChain Agent 最新教程详解及示例学习
    439
  • 5
    【AGI】智能体简介及场景分析
    82
  • 6
    【AI智能体】SuperAGI-开源AI Agent 管理平台
    322
  • 7
    Llama3 中文通用Agent微调模型来啦！（附手把手微调实战教程）
    1350
  • 8
    android-agent-web中js-bridge案例
    34
  • 9
    关于多智能体应用，这届年轻人这么做
    85
  • 10
    变革来袭！多Agent框架MuAgent带你解锁代码开发新姿势
    234

    • 相关课程

    更多
  • 【科技少年】AI领航员探索教程（赛前训练）
  • 达摩院视觉AI精品课
  • AI开发者的Docker实践
  • AI社区开源学习实践
  • 趣味视觉AI应用入门与实战
  • AI数学基础

    • 相关电子书

    更多
  • Data+AI时代大数据平台应该如何建设
  • 大数据AI一体化的解读
  • 当搜索遇见AI

    • 相关实验场景

    更多
  • 云原生AI套件：五分钟微调宝可梦风格StableDiffusion
  • OpenSearch LLM智能问答版搭建对话式搜索服务
  • 函数计算X RDS PostgreSQL，基于LLM大语言模型构建AI知识库
  • 基于Higress构建AIGC ChatBot
  • 通过GPU云服务器生成AI视频
  • 使用计算巢AI应用，快速上手实践Stable Diffusion
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）