本系列文章跟隨《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：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優化比較費勁，還有就是要注意何時結束游戲等細節。最后，也向大家展示了一下如何讓人參與到游戲中。

---

站內文章一覽