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

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


站内文章一览

相关文章
|
2天前
|
人工智能 算法 自动驾驶
用AI自动设计智能体,数学提分25.9%,远超手工设计
【9月更文挑战第18天】《智能体自动设计(ADAS)》是由不列颠哥伦比亚大学等机构的研究者们发布的一篇关于自动化设计智能体系统的最新论文。研究中提出了一种创新算法——“Meta Agent Search”,此算法通过迭代生成并优化智能体设计,从而实现更高效的智能体系统构建。实验表明,相比人工设计的智能体,Meta Agent Search生成的智能体在多个领域均有显著的性能提升。然而,该方法也面临着实际应用中的有效性与鲁棒性等挑战。论文详细内容及实验结果可于以下链接查阅:https://arxiv.org/pdf/2408.08435。
27 12
|
17天前
|
监控 Linux
Zabbix 5.0 LTS的agent服务部署实战篇
文章介绍了如何在CentOS 7.6操作系统上部署Zabbix 5.0 LTS版本的agent服务,包括配置软件源、安装agent、修改配置文件、启动服务,并在Zabbix web界面添加监控。
45 4
Zabbix 5.0 LTS的agent服务部署实战篇
|
18天前
|
人工智能 自然语言处理 物联网
智能体进化发展了一年,现在的RPA Agent迭代到什么程度了?
智能体技术在过去一年迅速发展,RPA Agent已成为连接多种应用系统的关键工具。实在智能推出的实在Agent 7.0,通过自然语言处理和屏幕识别技术,实现了从需求输入到任务执行的全流程自动化,大幅降低了智能体构建门槛。该平台不仅能在企业级应用中提供专业服务,还能满足个人用户的多样化需求,真正实现了端到端的自动化解决方案。
44 5
智能体进化发展了一年,现在的RPA Agent迭代到什么程度了?
|
17天前
|
监控 关系型数据库 MySQL
zabbix agent集成percona监控MySQL的插件实战案例
这篇文章是关于如何使用Percona监控插件集成Zabbix agent来监控MySQL的实战案例。
28 2
zabbix agent集成percona监控MySQL的插件实战案例
|
16天前
|
人工智能 JSON 自然语言处理
你的Agent稳定吗?——基于大模型的AI工程实践思考
本文总结了作者在盒马智能客服的落地场景下的一些思考,从工程的角度阐述对Agent应用重要的稳定性因素和一些解法。
|
10天前
|
存储 机器学习/深度学习 人工智能
深入浅出 AI 智能体(AI Agent)|技术干货
随着人工智能技术的发展,智能体(AI Agents)逐渐成为人与大模型交互的主要方式。智能体能执行任务、解决问题,并提供个性化服务。其关键组成部分包括规划、记忆和工具使用,使交互更加高效、自然。智能体的应用涵盖专业领域问答、资讯整理、角色扮演等场景,极大地提升了用户体验与工作效率。借助智能体开发平台,用户可以轻松打造定制化AI应用,推动AI技术在各领域的广泛应用与深度融合。
130 0
|
21天前
|
存储 弹性计算 前端开发
阿里云服务领域Agent智能体:从概念到落地的思考、设计与实践
本文讲述了作者团队在阿里云的服务领域Agent是如何设计与实践的,以及到目前为止的一些阶段性成果,作者做出了总结和整理。
|
机器学习/深度学习 人工智能 开发者
资源 | AI领域最优论文+代码查找神器:966个ML任务、8500+论文任君挑选!
查找论文及对应源码的神器 Papers With Code 刚刚推出了最新版本,可以用图形界面查找你想要的 SOTA 实现,从应用领域到具体任务再到实现代码一步到位。
|
5天前
|
机器学习/深度学习 人工智能 搜索推荐
AI与未来医疗:革命性的技术,触手可及的健康
本文探讨了人工智能(AI)在医疗领域的应用及其潜力。从诊断辅助、个性化治疗到药物研发,AI正在改变我们理解健康和疾病的方式。同时,本文也讨论了AI面临的伦理挑战和数据隐私问题,呼吁建立相应的法律法规来保障患者权益。最后,本文对AI在未来医疗中的角色进行了展望,认为AI将继续推动医疗领域的创新和发展。
27 8
|
2天前
|
机器学习/深度学习 人工智能 自然语言处理
AI技术在自然语言处理中的应用
【9月更文挑战第17天】本文主要介绍了AI技术在自然语言处理(NLP)领域的应用,包括文本分类、情感分析、机器翻译和语音识别等方面。通过实例展示了AI技术如何帮助解决NLP中的挑战性问题,并讨论了未来发展趋势。

热门文章

最新文章