# MetaGPT智能體學(xué)習(xí) ｜ 第四章：多智能體

作業(yè)：你畫我猜

目錄：

  • 初次嘗試版本
  • 改進(jìn)版本
    #解決消息傳遞機(jī)制
  • 實測案例

參考資料：

MetaGPT github：https://github.com/geekan/MetaGPT
MetaGPT 官方文檔：https://docs.deepwisdom.ai/main/zh/guide/get_started/introduction.html
MetaGPT 官方教程（飛書）：https://deepwisdom.feishu.cn/wiki/KhCcweQKmijXi6kDwnicM0qpnEf
Datawhale 合作課程：https://spvrm23ffj.feishu.cn/docx/RZNpd5uXfoebTPxMXCFcWHdMnZb

初次嘗試版本

設(shè)計邏輯：
Human: 出題
Drawer: 接收Human消息， 根據(jù)題目比劃
Guesser:接收Drawer消息， 根據(jù)比劃猜題
Drawer: 接收Guesser消息， 對比答案是否正確。若不正確，繼續(xù)比劃。
Guesser【重復(fù)】

加載需要模塊：


import asyncio
import platform
from typing import Any

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

參考辯論的寫法，Action設(shè)計：

class Draw(Action):
    """Action: By given an object, describe it without saying the name itself."""

    PROMPT_TEMPLATE: str = """
    ## BACKGROUND
    Suppose you are playing the game "You draw, I Guess". You are the actor to 'draw'. 
    ### The question
    {question} 
    ## Previous guess
    {context}
    ## YOUR TURN
    Now it's your turn, observe previous round. If the latest answer is EXACTLY the same as the question, you win and response "WIN". 
    Otherwise, describe the question without saying the word itself. Try new aspects other then previous rounds. 
    DO NOT mention the question itself in your response!
    If the question is a combination of multiple single words. Those words are alos not allowed present in your response.
    Your will respone in Chinese:
    """
    name: str = "draw"

    async def run(self, context: str, question:str, topic: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context, question=question, topic=topic)
        # logger.info(prompt)

        rsp = await self._aask(prompt)

        return rsp

class Guess(Action):
    """Action: By given a descriptin, guess the word or phase."""

    PROMPT_TEMPLATE: str = """
    ## BACKGROUND
    Suppose you are playing the game "You draw, I Guess". You are the actor to 'guess'. 
    ### The topic
    {topic}
    ## The description
    {context}
    ## YOUR TURN
    Now it's your turn, observe the description. Guess what it is, wihtout any other words. 
    Your respone:
    """
    name: str = "Guesser"

    async def run(self, context: str, topic: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context, topic=topic)
        # logger.info(prompt)

        rsp = await self._aask(prompt)

        return rsp

Roles 設(shè)計：

class Drawer(Role):
    name: str = "Drawer"
    profile: str = "Drawer"
    topic: str = ""

    def __init__(self, **data: Any):
        super().__init__(**data)
        self.set_actions([Draw])
        self._watch([UserRequirement, Guess])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo 

        memories = self.get_memories()
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
        question = ""
        for msg in memories:
            if msg.role == "Human":
                question = f"{msg.content}"
        rsp = await todo.run(context=context, topic=self.topic, question=question)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=type(todo),
            sent_from=self.name,
        )
        self.rc.memory.add(msg)

        return msg

class Guesser(Role):
    name: str = "Guesser"
    profile: str = "Guesser"
    topic: str = ""

    def __init__(self, **data: Any):
        super().__init__(**data)
        self.set_actions([Guess])
        self._watch([Draw])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo 

        memories = self.get_memories(k=1)
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
        # print(context)

        rsp = await todo.run(context=context, topic=self.topic)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=type(todo),
            sent_from=self.name,
        )
        self.rc.memory.add(msg)

        return msg

最后team封裝：

async def drawAguess(idea: str, topic: str, investment: float = 3.0, n_round: int = 5):
    """Run a team of presidents and watch they quarrel. :)"""
    team = Team()
    team.hire([
        Drawer(topic=topic) ,
        Guesser(topic=topic)])
    team.invest(investment)
    team.run_project(idea)
    await team.run(n_round=n_round)


idea= "冰球"
topic = "運(yùn)動"
investment = 3.0
n_round = 5

asyncio.run(drawAguess(idea, topic, investment, n_round))

問題：

  • 辯論例子中的_observe模塊困惑了我一整天，它的寫法會導(dǎo)致Drawer接收不到消息，導(dǎo)致程序直接結(jié)束。也沒有報錯，太難排查了，沒時間進(jìn)一步研究了。
  • Drawer沒法很好區(qū)分人類的題目和Guesser的答案，導(dǎo)致有概率出現(xiàn)一上來就說答對了的情況。后續(xù)有待研究改進(jìn)。

改進(jìn)版本

經(jīng)過研究，已經(jīng)能夠非常好的完成你畫我猜了。

  • 環(huán)境中的消息通過send_to能夠正確傳遞了
  • 增加了描述技巧和猜題技巧的提示詞
  • 增加了GPT和GLM分別扮演“畫”和“猜”角色的能力，看看誰比較厲害

有待改進(jìn)的地方：

  • 一次輸入多個詞，讓智能體自動完成多輪游戲
    • 策略1: 增加問題派發(fā)員
    • 策略2:在策略1的基礎(chǔ)上，讓派發(fā)員自己生成題目

完整測試代碼：


import asyncio
import platform
from typing import Any

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

from metagpt.config2 import Config
gpt4 = Config.default()
glm4 = Config.from_home("zhipu-glm4.yaml")




class Draw(Action):
    """Action: By given an object, describe it without saying the name itself."""

    PROMPT_TEMPLATE: str = """
    ## BACKGROUND
    Suppose you are playing the game "You draw, I Guess". You are the actor to 'draw'. 
    ### The question
    {question} 
    ## Previous guess
    {previous}
    ## YOUR TURN
    Now it's your turn, Describe the question without saying the word itself. 
    If "Previous guess" have content, try something new other then previous used. 
    DO NOT mention the question itself in your response!
    If the question is a combination of multiple words, ANY SINGLE word arr NOT allowed be present in your response.
    技巧：
    1. 如果意思正確但是字?jǐn)?shù)不同，請描述正確答案的文字長度，如“三個字”，“四個字”
    2. 如果是數(shù)字表示方式不同，請描述正確答案的數(shù)字類型，如“中文數(shù)字”，“阿拉伯?dāng)?shù)字“
    3. 如果給出的答案意思正確但是語言不匹配時，請描述正確答案的語言，如“改用中文”，“答案是英文”
    注意，不要強(qiáng)調(diào)規(guī)則，請直接說出描述本身。
    Your will respone in Chinese:
    """
    name: str = "draw"

    async def run(self, previous: str, question:str, topic: str):
        prompt = self.PROMPT_TEMPLATE.format(previous=previous, question=question, topic=topic)
        # logger.info(prompt)

        rsp = await self._aask(prompt)

        return rsp

class Guess(Action):
    """Action: By given a descriptin, guess the word or phase."""

    PROMPT_TEMPLATE: str = """
    ## BACKGROUND
    Suppose you are playing the game "You draw, I Guess". You are the actor to 'guess'. 
    ### The topic
    {topic}
    ## The description
    {context}
    ## YOUR TURN
    Now it's your turn, observe the description. Guess what it is, wihtout any other characters. 
    Use the same language as the description.
    再次強(qiáng)調(diào)，請直接說答案，不要帶有任何其他描述；也不要帶有任何標(biāo)點(diǎn)符號
    Your respone:
    """
    name: str = "Guesser"

    async def run(self, context: str, topic: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context, topic=topic)
        # logger.info(prompt)

        rsp = await self._aask(prompt)

        return rsp


class Drawer(Role):
    name: str = "Drawer"
    profile: str = "Drawer"
    topic: str = ""

    def __init__(self, **data: Any):
        super().__init__(**data)
        self.set_actions([Draw])
        self._watch([UserRequirement, Guess])

    async def _observe(self) -> int:
        await super()._observe()
        # accept messages sent (from opponent) to self, disregard own messages from the last round
        self.rc.news = [msg for msg in self.rc.news if msg.send_to == {self.name}]
        return len(self.rc.news)


    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo 

        memories = self.get_memories()
        mem0 = memories.pop(0)
        if mem0.role == "Human":
            question = f"{mem0.content}" 
        else:
            raise ValueError(f"Question not found: {mem0}")
        
        if len(memories) > 0:
            mem9 = memories[-1]
            latestAnswer = f"{mem9.content}"
            previousTry = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
        else:
            latestAnswer = ""
            previousTry = ""
        
        if latestAnswer == question:
            totalTries = len([msg for msg in memories if msg.role == "Guesser"])
            rsp = f"Win after {totalTries} tries!"
            logger.info(rsp)

            next_to = "Human"
        else:
            rsp = await todo.run(previous=previousTry, topic=self.topic, question=question)
            next_to = "Guesser"

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=type(todo),
            sent_from=self.name,
            send_to=next_to,
        )
        self.rc.memory.add(msg)

        return msg

class Guesser(Role):
    name: str = "Guesser"
    profile: str = "Guesser"
    topic: str = ""

    def __init__(self, **data: Any):
        super().__init__(**data)
        self.set_actions([Guess])
        self._watch([Draw])

    async def _observe(self) -> int:
        await super()._observe()
        # accept messages sent (from opponent) to self, disregard own messages from the last round
        self.rc.news = [msg for msg in self.rc.news if msg.send_to == {self.name}]
        return len(self.rc.news)
    
    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo 

        memories = self.get_memories(k=1)
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
        # print(context)

        rsp = await todo.run(context=context, topic=self.topic)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=type(todo),
            sent_from=self.name,
            send_to="Drawer"
        )
        self.rc.memory.add(msg)

        return msg

async def drawAguess(idea: str, topic: str, investment: float = 3.0, n_round: int = 5):
    """Run a team of presidents and watch they quarrel. :)"""
    team = Team()
    team.hire([
        Drawer(topic=topic,config=glm4) ,
        Guesser(topic=topic,config=gpt4)
        ])
    team.invest(investment)
    team.run_project(idea, send_to="Drawer")
    await team.run(n_round=n_round)

def main(q: str, topic:str, investment: float = 3.0, n_round: int = 10):
    """
    :param topic: Guess topic, such as "Sports/Movie/Star"
    :param q(uestion): A word, phase to play. Should be a subject belong to the topic.
    :param investment: contribute a certain dollar amount to watch the debate
    :param n_round: maximum rounds of the debate
    :return:
    """
    if platform.system() == "Windows":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(drawAguess(q, topic, investment, n_round))


if __name__ == "__main__":
    fire.Fire(main)

實測案例

python learn/drawAguess.py --q "深圳" --topic "地名"

INFO     | __main__:_act:98 - Drawer(Drawer): to do Draw(Draw)
這是中國一個著名的經(jīng)濟(jì)特區(qū)，位于廣東省，毗鄰香港，是一個年輕而充滿活力的城市。描述：這是一個由兩個漢字組成的城市名，具有改革開放的重要意義。
IINFO     | __main__:_act:154 - Guesser(Guesser): to do Guess(Guess)
深圳
INFO     | __main__:_act:98 - Drawer(Drawer): to do Draw(Draw)
INFO     | __main__:_act:119 - Win after 1 tries!

嘗試了很多，大多數(shù)都能像這樣一次答對，總體還是不錯的。

python learn/drawAguess.py --q "百元之戀" --topic "電影"

NFO     | __main__:_act:98 - Drawer(Drawer): to do Draw(Draw)
這是一部講述拳擊和人生挑戰(zhàn)的日本電影，描述了一個女性從自暴自棄到自我救贖的過程。這部電影的名字由三個字組成，其中包含了貨幣單位和一種情感。
INFO     | __main__:_act:154 - Guesser(Guesser): to do Guess(Guess)
百円の戀
INFO     | __main__:_act:98 - Drawer(Drawer): to do Draw(Draw)
這是一部以拳擊和自我成長為主題的日本電影，描繪了主角從失意沉淪到重燃斗志的心路歷程。電影的名字由三個字組成，反映了金錢與愛情的雙重含義，其中“元”字代表了貨幣單位，“戀”字則表達(dá)了情感主題。名字中的第一個字是中文數(shù)字。
INFO     | __main__:_act:154 - Guesser(Guesser): to do Guess(Guess)
千與千尋
INFO     | __main__:_act:98 - Drawer(Drawer): to do Draw(Draw)
這是一部日本電影，講述了主角通過拳擊運(yùn)動找回自我和愛情的故事。影片的名字由三個漢字組成，表達(dá)了金錢與情感的關(guān)系，開頭的字是中文數(shù)字“百”，而結(jié)尾的字代表了愛情。名字中既體現(xiàn)了影片中金錢的元素，也暗示了主角在經(jīng)歷挑戰(zhàn)后對生活的熱愛。
INFO     | __main__:_act:154 - Guesser(Guesser): to do Guess(Guess)
百萬愛情宣言
INFO     | __main__:_act:98 - Drawer(Drawer): to do Draw(Draw)
這是一部描繪女性通過拳擊找回生活意義和愛情的日本電影。片名由三個漢字組成，反映了影片中金錢與情感交織的主題，開頭的字使用了中文數(shù)字，最后一個字代表了愛情。
INFO     | __main__:_act:154 - Guesser(Guesser): to do Guess(Guess)
一億兆戀情
INFO     | __main__:_act:98 - Drawer(Drawer): to do Draw(Draw)
這是一部以拳擊為背景，講述女性從低谷中站起，重新找回生活熱情和愛情的日本電影。影片的名字由三個漢字組成，開頭的字是中文數(shù)字，代表著貨幣的價值，而最后一個字則是對愛的簡稱。這個名字既體現(xiàn)了電影中金錢的元素，也隱喻了主人公情感的轉(zhuǎn)變。
INFO     | __main__:_act:154 - Guesser(Guesser): to do Guess(Guess)
百萬愛

這個實例，其實第一輪已經(jīng)猜出來了，但是因為語言沒有匹配上，所以進(jìn)入了下一輪。但是下一輪中并沒有及時給予語言上的提示，導(dǎo)致越來越離譜。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容