【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 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优化比较费劲，还有就是要注意何时结束游戏等细节。最后，也向大家展示了一下如何让人参与到游戏中。

站内文章一览

在这里插入图片描述