基于MetaGPT構建LLM多智能體

bar

前言

你好，我是GISer Liu，在上一篇文章中，我們用了兩萬多字詳細拆解了單個Agent的組成，并通過Github Trending訂閱智能體理解MetaGPT框架的訂閱模塊如何解決應用問題，但是對于復雜，并行的任務，單個智能體是不能勝任；今天我們將進入多智能體開發的學習階段；一起期待吧😀

一、介紹

在本文中，我們將分別詳細介紹:

MetaGPT中Environment的設計思想；
構建簡單師生對話多Agent框架；
MetaGPT中Team的設計思想；
構建多Agent 開發團隊；
構建多Agent 辯論團隊;
你畫我猜多Agent框架實現；

二、Environment 環境設計思想

openai_gym

在MetaGPT框架中，Environment（環境）與Agent（智能體）這兩個概念借鑒了強化學習的思想。而在強化學習中，Agent需要在環境中采取行動以最大化獎勵。而在MetaGPT中，則提供了一個標準的環境組件Environment，用來管理Agent的活動與信息交流。

學習 agent 與環境進行交互的思想可以去OpenAI的GYM項目看看

1.環境設計原理

MetaGPT中的環境設計分為外部環境(ExtEnv)和內部環境，旨在幫助Agent代理與不同的外部應用場景（如游戲、手機應用等）以及內部開發和操作環境進行交互。

①外部環境（ExtEnv）

minecraft

定義：
外部環境是代理與外部世界交互的接口。它為代理提供了一種機制，使其能夠與外部系統（例如游戲引擎、移動應用API）進行通信和交互。

繼承和擴展：
ExtEnv類是所有外部環境的基礎類，各種具體的外部環境（如Minecraft環境、狼人游戲環境等）會繼承這個基礎類，并在其上擴展實現特定的交互邏輯。

示例：

游戲環境：
- 假設有一個在線游戲提供了API，允許查詢玩家狀態和執行游戲動作。
- ExtEnv類封裝了這些API，使代理能夠調用這些API來查詢游戲狀態和執行動作。
Agent執行某個Action，該Action中封裝了執行API調用的邏輯
狼人sha游戲：
- 在狼人游戲中，代理需要知道每晚和每天的游戲狀態。
- ExtEnv類定義了獲取這些狀態的方法，使代理能夠在游戲中做出決策。

Minecraft開發API
Agent狼人sha實現案例

②內部環境

chatdev

(1)定義：
內部環境是代理及其團隊直接使用的開發和操作環境。它類似于軟件開發中的工作環境，包括開發工具、測試框架和配置文件等。

(2)繼承和擴展：
內部環境類（XxxEnv）通常繼承自一個基礎環境類，并根據具體需求進行定制和擴展。這個基礎環境類可以提供一些通用功能，比如日志記錄、錯誤處理等。

(3)案例：

開發環境：
- 基礎環境類可能提供一些通用的開發工具和測試框架。
- 開發團隊可以在這個基礎上添加特定項目所需的工具和配置，例如數據庫連接配置、CI/CD腳本等。

作者認為其思想和ChatDev的實現相似；

2.環境交互設計

MetaGPT還引入了兩個重要的概念：observation_space和action_space。這些概念來自強化學習領域，用于描述代理從環境中獲取的狀態信息和可以采取的動作集合。

observation_space：

表示代理可以從環境中獲得的所有可能的狀態。
例如，在游戲環境中，observation_space可能包括玩家的位置、游戲時間、得分等。在上圖Minecraft的案例中，觀察空間就是周圍的環境，角色的血量與護甲，擁有的工具與工具的數量；

action_space：

表示代理在環境中可以執行的所有可能的動作。
例如，在游戲環境中，action_space可能包括移動、跳躍、攻擊等，同樣在上面的案例中，action_space代表可選Action的集合，例如看到樹以后選擇砍樹，看到怪物后選擇逃離還是進攻；這需要Agent通過反思機制來判斷進行；

通過定義這兩個空間，MetaGPT能夠更好地抽象不同環境中的具體細節，使得環境提供者可以專注于實現環境邏輯，而代理使用者可以專注于狀態和動作的處理。

3.環境運行機制

agent&env

這里放這張圖供大家思考

①Environment類的基本組成

以下是MetaGPT中Environment類的基本組成：

class Environment(ExtEnv):"""環境，承載一批角色，角色可以向環境發布消息，可以被其他角色觀察到Environment, hosting a batch of roles, roles can publish messages to the environment, and can be observed by other roles"""model_config = ConfigDict(arbitrary_types_allowed=True)desc: str = Field(default="")  # 環境描述roles: dict[str, SerializeAsAny["Role"]] = Field(default_factory=dict, validate_default=True)member_addrs: Dict["Role", Set] = Field(default_factory=dict, exclude=True)history: str = ""  # For debugcontext: Context = Field(default_factory=Context, exclude=True)

參數說明如下：

model_config：配置模型的配置字典，允許任意類型作為字段。
desc：環境描述，默認值為空字符串。
roles：包含環境中所有角色的字典，鍵是角色名字，值是角色對象，默認值是一個空字典。
member_addrs：存儲每個角色的地址集合的字典，鍵是角色對象，值是地址集合，默認值是一個空字典，不參與序列化。
history：記錄環境歷史信息的字符串，默認值為空字符串。
context：環境上下文對象，默認值是一個新的Context對象，不參與序列化。

知曉了環境的組成與Agent的交互方式以后，我們來理解一下多個Agent與環境的交互方式；

②Environment類的運行過程

試著想象一個大型圓桌會議，Environment提供了一個讓Agent們統一上桌討論的環境。接下來，我們來看看MetaGPT是如何實現這種機制的。
首先，當一個Environment運行時，會發生什么事情呢？來看一下Environment基類中定義的run方法：

async def run(self, k=1):"""處理一次所有信息的運行Process all Role runs at once"""for _ in range(k):futures = []for role in self.roles.values():future = role.run()futures.append(future)await asyncio.gather(*futures)logger.debug(f"is idle: {self.is_idle}")

當一個Environment運行時，其會遍歷環境中的role（角色）列表，讓它們逐個運行，即逐個做出各自的Actions，然后進行發言(將結果輸出到環境)。

③單個Agent的運行機制

下面是每個Agent運行時所執行的事件：

@role_raise_decorator
async def run(self, with_message=None) -> Message | None:"""觀察，并根據觀察結果進行思考和行動"""if with_message:msg = Noneif isinstance(with_message, str):msg = Message(content=with_message)elif isinstance(with_message, Message):msg = with_messageelif isinstance(with_message, list):msg = Message(content="\n".join(with_message))if not msg.cause_by:msg.cause_by = UserRequirementself.put_message(msg)if not await self._observe():# 如果沒有新的信息，則暫停并等待logger.debug(f"{self._setting}: 沒有新的信息。正在等待...")returnrsp = await self.react()# 重置下一步要執行的動作self.set_todo(None)# 將響應消息發送到環境對象，以便將消息轉發給訂閱者self.publish_message(rsp)return rsp

run方法主要功能是觀察環境，并根據觀察結果進行思考和行動。如果有新的消息，它會將消息添加到隊列中，并根據消息的內容進行處理。如果沒有新的信息，它會暫停并等待。在處理完消息后，它會重置下一步要執行的動作，并將響應消息發送到環境對象。

def put_message(self, message):"""Place the message into the Role object's private message buffer."""if not message:returnself.rc.msg_buffer.push(message)

在Role的run方法中，Role首先會根據運行時是否傳入信息（部分行動前可能需要前置知識消息），將信息存入RoleContext的msg_buffer中。

信息觀察機制

在多智能體環境運行中，Role的每次行動將從Environment中先_observe（觀察）消息。在observe的行動中，Role將從消息緩沖區和其他源準備新消息以進行處理，當未接受到指令時，Role將等待執行。

對于信息緩沖區中的信息，首先我們會根據self.recovered參數決定news是否來自于self.latest_observed_msg或者msg_buffer并讀取。完成信息緩沖區中的讀取后，如果設定好了ignore_memory則old_messages便不會再讀取當前Role的memory。將news中的信息存入Role的memory后，我們將進一步從news中篩選，也就是我們設定的角色關注的信息（self.rc.watch），而self.rc.news將存儲這些當前角色關注的消息，最近的一條將被賦給latest_observed_msg。最后，我們打印角色關注到的消息并返回。

這便是MetaGPT中環境的設計原理及其運行機制的詳細解析。

run方法主要功能是觀察環境，并根據觀察結果進行思考和行動。如果有新的消息，它會將消息添加到隊列中，并根據消息的內容進行處理。如果沒有新的信息，它會暫停并等待。在處理完消息后，它會重置下一步要執行的動作，并將響應消息發送到環境對象，以便將消息轉發。

def put_message(self, message):"""Place the message into the Role object's private message buffer."""if not message:returnself.rc.msg_buffer.push(message)

而在 role 的run方法中 role 首先將會根據運行時是否傳入信息（部分行動前可能需要前置知識消息），將信息存入 rolecontext的 msg_buffer 中；
agent&env

最后，再看看，這張圖，我想你會記憶更加深刻，當然，如果作者認知有偏頗，讀者也可以在評論區指出，感謝支持

三、簡單的師生交互多智能體系統

在上一節中，我們已經了解了environment環境的基本構成與它的運行邏輯，在這一節中，我們將學習如何利用environment來進行開發，進一步了解environment組件內部的活動，
現在設想一個多Agent交互的應用場景，我的想法是兩人對話場景，如：

師生交互場景:

首先用戶輸入一個主題;
然后學生Agent負責根據用戶的輸入進行作文撰寫
當老師Agent發現學生Agent寫作完畢以后，就會給學生提出學習意見；
根據老師Agent給的意見，學生將修改自己的作品；
如此循環直到設定的循環次數結束；這里環境則是教室；

接下來我們用metagpt提供的API實現這一交互場景；

首先，我們需要導入必要的包，并定義一個classroom環境，如下所示：

import asynciofrom metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environmentfrom metagpt.const import MESSAGE_ROUTE_TO_ALL
classroom = Environment()

接著作者分別為老師和學生Agent撰寫它們的行動WritingAction和ReviewAction，這里的思路基本就是簡單的提示詞工程，學生要求有寫作格式和寫作主題寫作，老師有檢查標準和檢查功能；
規范點說就是：

實現 WriteAction 方法：在這個方法中，學生Agent需要根據用戶提供的主題撰寫一篇作文。同時，當收到來自老師的修改建議后，也需要對作文進行相應的修改。
實現 ReviewAction 方法：在這個方法中，老師Agent需要讀取學生撰寫的作文，然后提出修改意見，以幫助學生進一步完善作文。

OK，開始編寫：

class WriteAction(Action):"""學生Agent的撰寫作文Action。"""name: str = "WriteEssay"PROMPT_TEMPLATE: str = """這里是歷史對話記錄：{msg}。請你根據用戶提供的主題撰寫一篇作文，只返回生成的作文內容，不包含其他文本。如果老師提供了關于作文的建議，請根據建議修改你的歷史作文并返回。你的作文如下："""async def run(self, msg: str):"""根據用戶提供的主題撰寫一篇作文，并在收到老師的修改建議后進行修改。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rspclass ReviewAction(Action):"""老師Agent的審閱作文Action。"""name: str = "ReviewEssay"PROMPT_TEMPLATE: str = """這里是歷史對話記錄：{msg}。你是一名老師，現在請檢查學生創作的關于用戶提供的主題的作文，并給出你的修改建議。你更喜歡邏輯清晰的結構和有趣的口吻。只返回你的修改建議，不要包含其他文本。你的修改建議如下："""async def run(self, msg: str):"""審閱學生的作文，并給出修改建議。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rsp

接著，我們定義StudentAgent與TeacherAgent，與單智能體不同的是，我們需要聲明每個Agent關注的動作(self._watch),只有當動作發生后，角色才開始行動，這樣能保證整體的運行規律而不混亂；

class Student(Role):"""學生角色。"""name: str = "cheems"profile: str = "Student"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteAction])  # 設置學生的動作為撰寫作文self._watch([UserRequirement, ReviewAction])  # 監聽用戶要求和老師的審閱動作async def _act(self) -> Message:"""學生動作：根據用戶要求撰寫作文或根據老師的修改建議修改作文。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶# logger.info(msg)essay_text = await WriteAction().run(msg)logger.info(f'student : {essay_text}')msg = Message(content=essay_text, role=self.profile, cause_by=type(todo))return msgclass Teacher(Role):"""老師角色。"""name: str = "laobai"profile: str = "Teacher"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewAction])  # 設置老師的動作為審閱作文self._watch([WriteAction])  # 監聽學生的撰寫作文動作async def _act(self) -> Message:"""老師動作：審閱學生的作文并給出修改建議。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶review_text = await ReviewAction().run(msg)logger.info(f'teacher : {review_text}')msg = Message(content=review_text, role=self.profile, cause_by=type(todo))return msg

要記得關注動作在init階段；

設計完畢agent后,我們就可以開始撰寫運行函數了，用戶輸入一個主題topic，并將topic發布在env中,以運行env，此時系統就開始工作了，我們可以通過修改對話輪數（n_round）來查看不同輪數checkPoint下的結果；

async def main(topic: str, n_round=5):"""運行函數，用戶輸入一個主題，并將主題發布在環境中，然后運行環境。"""classroom.add_roles([Student(), Teacher()])  # 向環境中添加學生和老師角色classroom.publish_message(Message(role="Human", content=topic, cause_by=UserRequirement,send_to='' or MESSAGE_ROUTE_TO_ALL),peekable=False,)# 發布一條消息，包含用戶輸入的主題，并將其發送給所有角色while n_round > 0:# self._save()n_round -= 1logger.debug(f"max {n_round=} left.")  # 輸出剩余對話輪數await classroom.run()  # 運行環境return classroom.history  # 返回對話歷史記錄asyncio.run(main(topic='關于道德和法律的限制范圍'))  # 運行主函數，輸入主題為 "道德和法律的限制范圍"

完整代碼如下：

import asynciofrom metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environmentfrom metagpt.const import MESSAGE_ROUTE_TO_ALL
# 加載環境變量
from dotenv import load_dotenv 
load_dotenv()classroom = Environment()class WriteAction(Action):"""學生Agent的撰寫作文Action。"""name: str = "WriteEssay"PROMPT_TEMPLATE: str = """這里是歷史對話記錄：{msg}。請你根據用戶提供的主題撰寫一篇作文，只返回生成的作文內容，不包含其他文本。如果老師提供了關于作文的建議，請根據建議修改你的歷史作文并返回。你的作文如下："""async def run(self, msg: str):"""根據用戶提供的主題撰寫一篇作文，并在收到老師的修改建議后進行修改。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rspclass ReviewAction(Action):"""老師Agent的審閱作文Action。"""name: str = "ReviewEssay"PROMPT_TEMPLATE: str = """這里是歷史對話記錄：{msg}。你是一名老師，現在請檢查學生創作的關于用戶提供的主題的作文，并給出你的修改建議。你更喜歡邏輯清晰的結構和有趣的口吻。只返回你的修改建議，不要包含其他文本。你的修改建議如下："""async def run(self, msg: str):"""審閱學生的作文，并給出修改建議。"""prompt = self.PROMPT_TEMPLATE.format(msg=msg)rsp = await self._aask(prompt)return rspclass Student(Role):"""學生角色。"""name: str = "cheems"profile: str = "Student"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteAction])  # 設置學生的動作為撰寫作文self._watch([UserRequirement, ReviewAction])  # 監聽用戶要求和老師的審閱動作async def _act(self) -> Message:"""學生動作：根據用戶要求撰寫作文或根據老師的修改建議修改作文。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶# logger.info(msg)essay_text = await WriteAction().run(msg)logger.info(f'student : {essay_text}')msg = Message(content=essay_text, role=self.profile, cause_by=type(todo))return msgclass Teacher(Role):"""老師角色。"""name: str = "laobai"profile: str = "Teacher"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewAction])  # 設置老師的動作為審閱作文self._watch([WriteAction])  # 監聽學生的撰寫作文動作async def _act(self) -> Message:"""老師動作：審閱學生的作文并給出修改建議。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶review_text = await ReviewAction().run(msg)logger.info(f'teacher : {review_text}')msg = Message(content=review_text, role=self.profile, cause_by=type(todo))return msgclass Student(Role):"""學生角色。"""name: str = "cheems"profile: str = "Student"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteAction])  # 設置學生的動作為撰寫作文self._watch([UserRequirement, ReviewAction])  # 監聽用戶要求和老師的審閱動作async def _act(self) -> Message:"""學生動作：根據用戶要求撰寫作文或根據老師的修改建議修改作文。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶# logger.info(msg)essay_text = await WriteAction().run(msg)logger.info(f'student : {essay_text}')msg = Message(content=essay_text, role=self.profile, cause_by=type(todo))return msgclass Teacher(Role):"""老師角色。"""name: str = "laobai"profile: str = "Teacher"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([ReviewAction])  # 設置老師的動作為審閱作文self._watch([WriteAction])  # 監聽學生的撰寫作文動作async def _act(self) -> Message:"""老師動作：審閱學生的作文并給出修改建議。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶review_text = await ReviewAction().run(msg)logger.info(f'teacher : {review_text}')msg = Message(content=review_text, role=self.profile, cause_by=type(todo))return msgasync def main(topic: str, n_round=5):"""運行函數，用戶輸入一個主題，并將主題發布在環境中，然后運行環境。"""classroom.add_roles([Student(), Teacher()])  # 向環境中添加學生和老師角色classroom.publish_message(Message(role="Human", content=topic, cause_by=UserRequirement,send_to='' or MESSAGE_ROUTE_TO_ALL),peekable=False,)# 發布一條消息，包含用戶輸入的主題，并將其發送給所有角色while n_round > 0:# self._save()n_round -= 1logger.debug(f"max {n_round=} left.")  # 輸出剩余對話輪數await classroom.run()  # 運行環境return classroom.history  # 返回對話歷史記錄asyncio.run(main(topic='關于道德和法律的限制范圍'))  # 運行主函數，輸入主題為 "道德和法律的限制范圍"

運行結果如下：
result

很有趣，哈哈😂😂

四、MetaGPT中Team的設計思想

在上節中，我們通過師生交互的案例體驗了多Agent開發的趣味性，現在讓我們來了解一下Team。在官方介紹中，Team是一個重要的組件，它是基于Environment進行二次封裝的結果。Team的代碼如下：

class Team(BaseModel):"""Team: 由一個或多個角色（Agent）組成，具有SOP（標準運營程序）和一個用于即時消息傳遞的環境，專用于任意多Agent活動，如協同編寫可執行代碼。"""model_config = ConfigDict(arbitrary_types_allowed=True)env: Environment = Field(default_factory=Environment)  # Team的環境investment: float = Field(default=10.0)  # 團隊投資idea: str = Field(default="")  # 團隊想法

Team在Env的基礎上增加了更多的組件。例如，Investment用于管理團隊成本（限制Token花費），idea則用于告訴你的團隊接下來應該圍繞什么工作。Team有以下幾個重要的方法：
①hire方法

向團隊中添加員工。

def hire(self, roles: list[Role]):"""招聘角色進行協作"""self.env.add_roles(roles)  # 在環境中添加角色

②invest方法

計算Token,控制預算

def invest(self, investment: float):"""投資公司。當超過最大預算時，會引發NoMoneyException異常。"""self.investment = investmentCONFIG.max_budget = investmentlogger.info(f"Investment: ${investment}.")

③run_project方法

發布需求
初始化項目

def run_project(self, idea, send_to: str = ""):"""運行一個項目，從發布用戶需求開始。"""self.idea = idea# 人類需求。self.env.publish_message(Message(role="Human", content=idea, cause_by=UserRequirement, send_to=send_to or MESSAGE_ROUTE_TO_ALL),peekable=False,)

在Team運行時，首先調用run_project方法給智能體提供一個需求，然后在n_round的循環過程中，重復檢查預算和運行環境，最后返回環境中角色的歷史對話。

@serialize_decorator
async def run(self, n_round=5, idea="", send_to="", auto_archive=True):"""運行公司，直到到達目標輪次或沒有預算"""if idea:self.run_project(idea=idea, send_to=send_to)while n_round > 0:# self._save()n_round -= 1logger.debug(f"max {n_round=} left.")self._check_balance()await self.env.run()self.env.archive(auto_archive)return self.env.history

這里盡管Team類只是在Env上的簡單封裝，🤔但它向我們展示了如何向多智能體系統****發布啟動消息以及引入可能的人類反饋。接下來，我們將使用Team，開發屬于自己的第一個智能體團隊。

五、基于Team的Agent開發團隊

1.需求分析

學習完Team的設計思想后，我們就本系列課程3的思路進行研究，我們用Team將其實現一遍；還記得當初我們的需求嗎？下面是當初是思路流程圖：
requirement
本文中，我們需要構建一個包含需求分析，代碼撰寫，代碼測試，代碼評審的Team開發團隊：
下面是作者是思路：

定義每個Agent執行的行動Action;
- RequirementAnalysisAction：需求分析
- CodeWriteAction：代碼撰寫
- CodeTestAction：代碼測試
- CodeReviewAction：代碼評審
基于SOP流程，確保每個Agent既可以觀察到上個Agent的輸出結果，也能保證****將自己的輸出傳遞給下一個Agent;
初始化所有Agent，并將這些Agent添加進入Team實例，創建一個存在內部環境的智能體團隊，使Agent之間能夠進行交互。

現在我們開始撰寫代碼！😺😺

2.正式開發

先導入第三方庫

import re
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
import subprocess
# 加載環境變量
from dotenv import load_dotenv 
load_dotenv()

撰寫每個Agent的Action,包括需求分析，代碼撰寫，代碼測試，代碼評審：

# 需求分析優化Action
class RequirementsOptAction(Action):PROMPT_TEMPLATE: str = """你要遵守的規范有：1.簡要說明 (Brief Description)簡要介紹該用例的作用和目的。2.事件流 (Flow of Event)包括基本流和備選流，事件流應該表示出所有的場景。3.用例場景 (Use-Case Scenario)包括成功場景和失敗場景，場景主要是由基本流和備選流組合而成的。4.特殊需求 (Special Requirement)描述與該用例相關的非功能性需求（包括性能、可靠性、可用性和可擴展性等）和設計約束（所使用的操作系統、開發工具等）。5.前置條件 (Pre-Condition)執行用例之前系統必須所處的狀態。6.后置條件 (Post-Condition)用例執行完畢后系統可能處于的一組狀態。請優化以下需求，使其更加明確和全面：{requirements}"""name: str = "RequirementsOpt"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)return rsp.strip()  # 返回優化后的需求# 代碼撰寫Action
class CodeWriteAction(Action):PROMPT_TEMPLATE: str = """根據以下需求，編寫一個能夠實現{requirements}的Python函數，并提供兩個可運行的測試用例。返回的格式為：```python\n你的代碼\n```，請不要包含其他的文本。```python# your code here```"""name: str = "CodeWriter"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)code_text = CodeWriteAction.parse_code(rsp)return code_text@staticmethoddef parse_code(rsp): # 從模型生成中字符串匹配提取生成的代碼pattern = r'```python(.*?)```'  # 使用非貪婪匹配match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_text# 代碼測試Action
class CodeTestAction(Action):PROMPT_TEMPLATE: str = """上下文：{context}為給定的函數編寫 {k} 個單元測試，并且假設你已經導入了該函數。返回 ```python 您的測試代碼 ```，且不包含其他文本。your code:"""name: str = "CodeTest"async def run(self, code_text: str,k:int = 5):try:result = subprocess.run(['python', '-c', code_text],text=True,capture_output=True,check=True)return result.stdoutexcept subprocess.CalledProcessError as e:return e.stderrclass CodeReviewAction(Action):PROMPT_TEMPLATE: str = """context：{context}審查測試用例并提供一個關鍵性的review,在評論中，請包括對測試用例覆蓋率的評估，以及對測試用例的可維護性和可讀性的評估。同時，請提供具體的改進建議。"""name: str = "CodeReview"async def run(self, context: str):prompt = self.PROMPT_TEMPLATE.format(context=context)rsp = await self._aask(prompt)return rsp

在多智能體系統中，我們定義Agent有兩個重點：

使用 set_actions方法為Agent配備對應的 Action，這與單智能體思路相同;
SOP流程中，每個Agent的輸入都是上一個Agent的輸出，因此每個Agent在初始化的時候都通過self._watch來監聽上一個Agent的行動Action，以保證正確順序執行；對于第一個Agent,我們監聽用戶的輸入UserRequirement；

不知道大家有沒有想過同時監聽兩個或多個Action的是什么結果呢？是兩個Action都執行完，該Agent才執行自己的Action，還是任意一個執行完就執行自己的Action呢？大家可以試一試，作者996或許得在下一篇文章前會去試一試；

好了我們繼續將Agent的設計一次完善,代碼如下：作者這里直接使用官方案例，略有修改：

class RA(Role): #需求分析師縮寫name: str = "yake"profile: str = "Requirement Analysis"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement])self.set_actions([RequirementsOptAction])class Coder(Role):name: str = "cheems"profile: str = "Coder"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([RequirementsOptAction])self.set_actions([CodeWriteAction])class Tester(Role):name: str = "Bob"profile: str = "Tester"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeTestAction])# self._watch([SimpleWriteCode])self._watch([CodeWriteAction,CodeReviewAction])  # 這里測試一下同時監聽兩個動作是什么效果async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo# context = self.get_memories(k=1)[0].content # use the most recent memory as contextcontext = self.get_memories()  # 獲取所有記憶，避免重復檢查code_text = await todo.run(context, k=5)  # specify argumentsmsg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msgclass Reviewer(Role):name: str = "Charlie"profile: str = "Reviewer"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeReviewAction])self._watch([CodeTestAction])

OK,當前Team中需要的Agent全部定義完畢，我們開始初始化Team,并通過用戶輸入運行；代碼如下：

async def main(idea: str = "撰寫一個python自動生成隨機人物數據并保存到csv的tkinter程序,用戶輸入數量，則隨機生成人物信息保存csv到當前文件夾下",investment: float = 3.0, # token限制3美金n_round: int = 5, # 循環5 輪add_human: bool = False, # 無需用戶參與評審
):logger.info(idea)team = Team()team.hire([RA(),Coder(),Tester(),Reviewer(is_human=add_human),])team.invest(investment=investment) # 計算成本預算team.run_project(idea) # 初始化項目await team.run(n_round=n_round) # 開始循環if __name__ == "__main__":fire.Fire(main)

完整代碼如下：

import re
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
import subprocess
# 加載環境變量
from dotenv import load_dotenv 
load_dotenv()# 需求分析優化Action
class RequirementsOptAction(Action):PROMPT_TEMPLATE: str = """你要遵守的規范有：1.簡要說明 (Brief Description)簡要介紹該用例的作用和目的。2.事件流 (Flow of Event)包括基本流和備選流，事件流應該表示出所有的場景。3.用例場景 (Use-Case Scenario)包括成功場景和失敗場景，場景主要是由基本流和備選流組合而成的。4.特殊需求 (Special Requirement)描述與該用例相關的非功能性需求（包括性能、可靠性、可用性和可擴展性等）和設計約束（所使用的操作系統、開發工具等）。5.前置條件 (Pre-Condition)執行用例之前系統必須所處的狀態。6.后置條件 (Post-Condition)用例執行完畢后系統可能處于的一組狀態。請優化以下需求，使其更加明確和全面：{requirements}"""name: str = "RequirementsOpt"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)return rsp.strip()  # 返回優化后的需求# 代碼撰寫Action
class CodeWriteAction(Action):PROMPT_TEMPLATE: str = """根據以下需求，編寫一個能夠實現{requirements}的Python函數，并提供兩個可運行的測試用例。返回的格式為：```python\n你的代碼\n```，請不要包含其他的文本。```python# your code here```"""name: str = "CodeWriter"async def run(self, requirements: str):prompt = self.PROMPT_TEMPLATE.format(requirements=requirements)rsp = await self._aask(prompt)code_text = CodeWriteAction.parse_code(rsp)return code_text@staticmethoddef parse_code(rsp): # 從模型生成中字符串匹配提取生成的代碼pattern = r'```python(.*?)```'  # 使用非貪婪匹配match = re.search(pattern, rsp, re.DOTALL)code_text = match.group(1) if match else rspreturn code_text# 代碼測試Action
class CodeTestAction(Action):PROMPT_TEMPLATE: str = """上下文：{context}為給定的函數編寫 {k} 個單元測試，并且假設你已經導入了該函數。返回 ```python 您的測試代碼 ```，且不包含其他文本。your code:"""name: str = "CodeTest"async def run(self, context: str, k: int = 5):prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)rsp = await self._aask(prompt)code_text = CodeWriteAction.parse_code(rsp)return code_textclass CodeReviewAction(Action):PROMPT_TEMPLATE: str = """context：{context}審查測試用例并提供一個關鍵性的review,在評論中，請包括對測試用例覆蓋率的評估，以及對測試用例的可維護性和可讀性的評估。同時，請提供具體的改進建議。"""name: str = "CodeReview"async def run(self, context: str):prompt = self.PROMPT_TEMPLATE.format(context=context)rsp = await self._aask(prompt)return rsp
class RA(Role): #需求分析師縮寫name: str = "yake"profile: str = "Requirement Analysis"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement])self.set_actions([RequirementsOptAction])class Coder(Role):name: str = "cheems"profile: str = "Coder"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([RequirementsOptAction])self.set_actions([CodeWriteAction])class Tester(Role):name: str = "Bob"profile: str = "Tester"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeTestAction])# self._watch([SimpleWriteCode])self._watch([CodeWriteAction,CodeReviewAction])  # 這里測試一下同時監聽兩個動作是什么效果async def _act(self) -> Message:logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")todo = self.rc.todo# context = self.get_memories(k=1)[0].content # use the most recent memory as contextcontext = self.get_memories()  # 獲取所有記憶，避免重復檢查code_text = await todo.run(context, k=5)  # specify argumentsmsg = Message(content=code_text, role=self.profile, cause_by=type(todo))return msgclass Reviewer(Role):name: str = "Charlie"profile: str = "Reviewer"def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([CodeReviewAction])self._watch([CodeTestAction])async def main(idea: str = "撰寫一個python自動生成隨機人物數據并保存到csv的tkinter程序,用戶輸入數量，則隨機生成人物信息保存csv到當前文件夾下",investment: float = 3.0, # token限制3美金n_round: int = 5, # 循環5 輪add_human: bool = False, # 無需用戶參與評審
):logger.info(idea)team = Team()team.hire([RA(),Coder(),Tester(),Reviewer(is_human=add_human),])team.invest(investment=investment) # 計算成本預算team.run_project(idea) # 初始化項目await team.run(n_round=n_round) # 開始循環if __name__ == "__main__":fire.Fire(main)

運行效果如下：
result

嘿嘿😀，運行成功！可惜代碼運行邏輯不穩定😣，容易報錯，~~作者就刪去了這部分代碼~~ ；

總結

在本文中，各位讀者和作者一起學習了MetaGPT多智能體開發中環境Environment的定義和Team的設計思想，并通過師生互動案例和開發小組案例，體驗了其具體應用；雖然案例相對簡單，但是也足以說明多Agent框架在復雜問題中的潛力了;
通過對任務的原子級分解，統籌成本和效率，作者認為Agent的開發一定逐漸會改變我們生活的方方面面；真令人激動！🫡
好了，不多說，感謝大家的支持。作者雖然已經熬夜一周了😣，但是這一周來對Agent的學習幫到了作者很多，希望作者的文章也能幫到你🎉🎉🎉😀；

課后作業

你畫我猜

基于 env 或 team 設計一個你的多智能體團隊，嘗試讓他們完成你畫我猜文字版，要求其中含有兩個agent，其中一個agent負責接收來自用戶提供的物體描述并轉告另一個agent，另一個agent將猜測用戶給出的物體名稱，兩個agent將不斷交互直到另一個給出正確的答案
（也可以在系統之上繼續擴展，比如引入一個agent來生成詞語，而人類參與你畫我猜的過程中）
給出完整的代碼和詳細注釋，并在后面補充實現效果：

下面是作者的思路和實現效果:

設計思路

1.Action方法設計

describe_item:接受用戶提供的物體，對其進行描述并返回給猜測者，
guess_item:接受描述者的描述，猜測物體;

2.Agent設計

我們需要設計兩個智能體（Agent）：描述者和猜測者：

描述者（DescriberAgent）：接收物體詞匯并生成描述文本。
猜測者（GuesserAgent）：根據描述文本進行猜測。

游戲流程如下：

用戶將一個物體詞匯發送給描述者。
描述者生成描述文本，并將其發送給猜測者。
猜測者根據描述文本進行猜測，并將猜測結果返回給描述者。

3.完整代碼實現

以下是完整的代碼實現：

import re
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 dotenv import load_dotenv
from typing import ClassVarload_dotenv()# 描述Action
class DescribeItem(Action):PROMPT_TEMPLATE: str = """請根據以下物體詞匯生成描述文本：可以對物體詞匯側面描寫，但是不能直接說明其名稱，你的生成內容是讓別人猜測的；例如： "蘋果": "這是一種紅色或綠色的水果，圓形，味道甜或酸。""桌子": "這是一個家具，有四條腿，用來放置物品。",當前如下：詞匯：{word}"""name: str = "DescribeItem"async def run(self, word):prompt = self.PROMPT_TEMPLATE.format(word=word)res = await self._aask(prompt)return res# 猜測Action
class GuessItem(Action):PROMPT_TEMPLATE: str = """根據以下描述文本進行猜測物體名稱：描述：{description}例如：描述為："這是一種紅色或綠色的水果，圓形，味道甜或酸。"，你需要猜測為: "蘋果",你的輸出格式如下，猜測結果用方括號擴住：[蘋果]"""name: str = "Guess"async def run(self, description):prompt = self.PROMPT_TEMPLATE.format(description=description)result = await self._aask(prompt)return self.parse_item(result)@staticmethoddef parse_item(rsp):pattern = r'\[(.*?)\]'match = re.search(pattern, rsp, re.DOTALL)item = match.group(1) if match else rspreturn itemclass DescriberAgent(Role):name: str = "Describer"profile: str = "負責生成物體描述文本的描述者"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([UserRequirement,GuessItem])self.set_actions([DescribeItem])async def _act(self) -> Message:"""描述者動作：根據猜測者的回答修改描述。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶# logger.info(msg)prompt = "這是猜測者的返回：{msg},如果這不是正確答案，請修改描述"describe = await DescribeItem().run(prompt)logger.info(f'DescriberAgent : {describe}')msg = Message(content=describe, role=self.profile, cause_by=type(todo))return msgclass GuesserAgent(Role):name: str = "Guesser"profile: str = "負責猜測物體名稱的猜測者"def __init__(self, **kwargs):super().__init__(**kwargs)self._watch([DescribeItem])self.set_actions([GuessItem])async def _act(self) -> Message:"""猜測者動作：根據描述者的描述修改猜測結果。"""logger.info(f"{self._setting}: ready to {self.rc.todo}")todo = self.rc.todomsg = self.get_memories()  # 獲取所有對話記憶# logger.info(msg)prompt = "這是描述者的返回：{msg},如何這不是正確答案，請修改結果重新回答"guess = await GuessItem().run(msg)logger.info(f'GuesserAgent : {guess}')msg = Message(content=guess, role=self.profile, cause_by=type(todo))return msgasync def main(word: str = "貓", idea: str = "雞你太美", investment: float = 3.0, add_human: bool = False, n_round=5):logger.info(idea)team = Team()team.hire([DescriberAgent(), GuesserAgent()])team.invest(investment=investment)team.run_project(idea) # 初始化項目await team.run(n_round=n_round) # 開始循環if __name__ == "__main__":fire.Fire(main)