前言
在之前的文章中，我們詳細地描述了Agent的概念和組成，在代碼案例中體驗了Agent的記憶、工具、規劃決策模塊，并通過幾個Agent框架來加強讀者對Agent開發設計與應用的理解，接下來我們就要進入智能體Agent的實際開發中，請各位和我一起運行環境，開始Coding!!!😋😋 代碼已開源，文末有鏈接自取；
一、介紹
????本文中，我們將專注于單智能體開發，使用框架為MetaGPT。MetaGPT是一個基于Python的智能體開發框架，它提供了一系列的工具和方法，可以幫助開發者更加高效地進行智能體的開發。這里主要介紹如何使用MetaGPT框架進行智能體的開發，包括單動作Agent、多動作Agent和復雜Agent的開發。
二、MetaGPT中Agent的概念
????在MetaGPT的設計思想中，我們可以將Agent想象成為計算機環境中的虛擬人類，這個虛擬人類與環境交互過程可以抽象為以下組成：
Agent = LLM(大語言模型) + Observer(觀察) + Throught(思考)+Action(行動)+Memory(記憶)
這個公式抽象概括了MetaGPT Agent智能體的本質；
對于初學者，我認為將MetaGPT中的Agent與人類類比是一個很好的學習方式：
下面我們用類比的方式幫助各位讀者理解其概念：
- LLM大語言模型：LLM可以看做是智能體的“大腦”，其可以處理信息，并從交互中學習，做出決策并執行行動；
- 觀察：這是Agent的感知模塊，使其可以感知其環境；Agent可以接受來自另外一個Agent的文本信息，來自攝像頭的圖像視頻數據，來自用戶錄音的的音頻數據，還有來自API或函數的文本信息等；
OpenAI最新發布的GPT-4o則將視覺、聽覺、語言綜合到了一起，這使得Agent的反應更加迅速，并且可以從說話預語氣和語速中感受到額外的信息；以后以GPT-4o為LLM核心的Agent會更加強大；
- 思考：思考過程包括對觀察結果的分析和根據記憶選擇下一步行動，這不僅僅包括智能體Agent內部的決策機制，還可以通過某些函數進行執行，例如用四象限法分析某個人的財務狀況，用人格分析法來分析用戶的人格類型等，傳統的硬編碼方法都可以整理為Agent的某些思考插件；
- 行動：這是Agent對其觀察經過思考后的響應，行動既可以是用戶預設的硬編碼，也可以是由Agent使用包含代碼驗證機制的LLM代碼生成模塊生成的函數或API工具；例如通過聚合數據的API來獲取最新天氣信息和股市數據，利用Mathmatic或MatLab API輸入公式得到計算的結果等等；
- 記憶：智能體的記憶可以從非向量數據庫(MySQL等)和向量數據庫(Faiss)中存儲提取，這對Agent的決策和學習至關重要，其允許Agent參考歷史信息并據此調整未來的行動；用戶可以根據具體的業務需求選擇不同的數據庫或者混合使用；
下面是一個MetaGPT的一個Agent任務執行流程圖：
我們來理解一下這個圖：
- 1.Agent首先觀察當前環境，并將觀察得到結果存放到記憶中，
- 2.Agent根據觀察得到的信息進行思考，選擇下一步要執行的行動，也就是從Action1,Action2…中選擇要執行的行動；
- 3.Agent執行上一步選擇的行動，執行完畢后，我們可以得到執行的反饋，并將其存儲到記憶中，然后Agent根據反饋記憶和當前任務再次決策選擇下一步行動，如此循環，直到完成任務為止；
如果用唯物辯證法的角度看:
- Agent通過觀察學習就是首次(實踐)，從互聯網或API等渠道獲得大量信息(感性認識)，
- 然后根據當前任務(矛盾)和得到的信息(感性認識)進行思考，整理抽象出對解決任務的初步方案(理性認識)；然后用總結的經驗修改方案再次執行(再實踐)，從反饋中再次接受反饋思考(再認識)，
- 通過循環此過程(反復的認識和實踐過程)，Agent便可通過試錯，自我優化得到一個較為成功的解決方案（相對真理）；這個解決方案可以被永久存放在數據庫中，用于指導下一次的任務實踐；
- 臨時存儲每一步的記憶，可以避免Agent原地踏步，重復犯錯；
下面我介紹一下MetaGPT中，Agent的實現原理：
- 在MetaGPT中，Role類是智能體的邏輯抽象。一個Role能執行特定的Action，擁有記憶、思考并采用各種策略行動。
- 根據面向對象的思想，構建一個Agent就像構建一個類，其初始化的時候具有人設信息作為其類的屬性，也具有初始化方法，例如文件讀寫，網絡檢索等；
如果再其基礎上增加自我迭代的方法，例如可以修改自己的屬性(方法操作屬性)，如人設，增加或修改自己的方法，使其增加更多功能，結合強化學習的思想來構建一個自我迭代的Agent，作者覺得這是一個可以嘗試的有趣方向；
當然，回到正題，我們這里只關注一個執行動作的Agent,我將用python實現一個最簡單的Agent;
三、RoleContext
先簡要說說RoleContex的基本概念；
Agent在與環境上下文進行的交互，是通過內部的RoleContext對象來實現的，下面是RoleContext定義的源碼:
克隆MetaGPT項目到本地，在Vscode中搜索
RoleContext即可；
class RoleContext(BaseModel):
"""角色運行時上下文"""model_config = ConfigDict(arbitrary_types_allowed=True)# 環境變量，默認為None，排除序列化以避免遞歸錯誤env: "Environment" = Field(default=None, exclude=True)# 消息緩沖區，具有異步更新msg_buffer: MessageQueue = Field(default_factory=MessageQueue, exclude=True)# 記憶體memory: Memory = Field(default_factory=Memory)# 工作記憶體working_memory: Memory = Field(default_factory=Memory)# 狀態，默認為-1，表示初始或終止狀態，此時 todo 為 Nonestate: int = Field(default=-1)# 待處理的動作，默認為 Nonetodo: Action = Field(default=None, exclude=True)# 觀察列表watch: set[str] = Field(default_factory=set)# 新聞列表，默認為空列表，排除序列化，暫未使用news: list[Type[Message]] = Field(default=[], exclude=True)# 角色反應模式，默認為 RoleReactMode.REACTreact_mode: RoleReactMode = RoleReactMode.REACT# 最大反應循環次數，默認為 1max_react_loop: int = 1
我這里解釋一下上面的參數：
- env：Environment 對象，當 Role 被添加到 Environment 時，Role 會對 Environment 進行引用。
- msg_buffer：一個 MessageQueue 對象，用于 Role 與環境中其他 Role 進行信息交互。它是對 asyncio 的 Queue 進行了簡單的封裝，提供了非阻塞的 pop/push 方法。
- memory：記憶對象，用于存儲 Role 執行過程中產生的消息。當 Role 執行 _act 時，會將執行得到的響應轉換為 Message 對象存儲在 memory 中。當 Role 執行 _observe 時，會把 msg_buffer 中的所有消息轉移到 memory 中。
- state：記錄 Role 的執行狀態。初始狀態值為 -1，當所有 Action 執行完成后也會被重置為 -1。
- todo：下一個待執行的 Action。當 state >= 0 時，會指向最后一個 Action。
- watch：一個字符串列表，用于記錄當前 Role 觀察的 Action。在 _observe 方法中用于過濾消息。
- news：存儲在執行 _observe 時讀取到的與當前 Role 上下游相關的消息。
- react_mode：ReAct 循環的模式，支持 REACT、BY_ORDER、PLAN_AND_ACT 三種模式，默認為 REACT 模式。簡單來說，BY_ORDER 模式按照指定的 Action 順序執行，PLAN_AND_ACT 模式則是一次思考后執行多個動作，而 REACT 模式按照 ReAct 論文中的思考——行動循環來執行。
- max_react_loop：在 react_mode 為 REACT 模式時生效，用于設置最大的思考-行動循環次數，超過后會停止 _react 執行。這點我記得ChatDev中有同樣的配置；
當Agent啟動后，每一次的交互都要調用到RoleContext對象，以得到最新的環境理解，來確保下一步行為的準確性；
四、單動作Agent
1.思路
下面作者將帶領大家基于MetaGPT框架實現一個代碼生成Agent；先說一下需求，我們希望這個Agent可以根據我們的需求來生成代碼，因此我們需要重寫Role基類的_init_與_act_方法，在_init_方法中我們需要聲明Agent的基本配置屬性，例如名稱name和profile人設,然后我們使用self._init_action(v0.6.6)或self.set_actions(v0.8.1)函數為我們的Agent實現代碼編寫功能的綁定，這個Action需要接受用戶的輸入并且生成我們期望的代碼；
在Agent的_act方法中，我們需要編寫智能體具體的行動邏輯，智能體將從最新的記憶中獲取用戶輸入，并且運行我們配置的動作Action，而MetaGPT則將該動作作為待辦事項self.rc.to在幕后進行處理，最后返回一個完整的回復；
2.需求分析
想要實現這個AgentCoder Agent，我們需要進行需求分析，思考這個Agent它需要哪些能力；
- 1.記憶存儲:
- 使用Memory模塊存儲用戶輸入和Agent生成的代碼響應,方便后續的需求分析和代碼生成。
- Memory模塊應該提供添加、查詢、更新等接口,確保Agent能夠高效地訪問歷史記錄。
- 2.代碼編寫:
- 生成的代碼應該符合相關編碼規范,并且具有良好的可讀性和可維護性。
- 3.代碼提取：
- LLM生成的代碼中常帶有一些我們不需要的內容，因此我們需要使用正則表達式從LLM輸出的字符串中提取我們需要的代碼；
- 為了保證代碼正常提取，我們需要為Agent舉例說明，要求其將代碼輸入到指定的格式中方便提取；
下面是我的Agent執行流程圖；
- 首先，我們需要其接受用戶的輸入，并且調用記憶模塊存儲我們的輸入，
- 接著這個Agent根據已有信息和用戶輸入進行需求優化得到更加標準的需求報告，
- 最后，Agent根據優化的需求報告進行代碼編寫；
好了，需求分析完畢，開始準備Action編寫；
3.編寫CodeWrite動作
在MetaGPT中，Action類是動作的邏輯抽象，用戶可以通過調用self.aask函數來獲取LLM的回復。self_aask的函數定義如下：
async def _aask(self, prompt: str, system_msgs: Optional[list[str]] = None) -> str:"""Append default prefix"""return await self.llm.aask(prompt, system_msgs)
- prompt：用戶輸入
- system_msgs：LLM的預設定,可以看做是這個工具的功能，輸出要求等；
了解這個函數后我們就可以更好的構建Action;
在開始編碼之前，我們可以配置一下開發環境，作者使用環境如下:
- Github CodeSpace VSCode
- python 3.10
- metagpt 0.8.1
配置API Key 和API服務器可以在安裝完畢metagpt后執行：
metagpt --init-config
然后根目錄下就會生成config文件夾，在其中配置參數即可：
llm:api_type: 'openai' # or azure / ollama / groq etc. Check LLMType for more optionsmodel: 'deepseek-chat' # or gpt-3.5-turbobase_url: 'https://api.deepseek.com' # or forward url / other llm urlapi_key: 'sk-6cxxxxxxxxxxxxxxxxx74'# proxy: 'YOUR_LLM_PROXY_IF_NEEDED' # Optional. If you want to use a proxy, set it here.# pricing_plan: 'YOUR_PRICING_PLAN' # Optional. If your pricing plan uses a different name than the `model`.
作者這里使用是deepseek的api，如果有需要大家可以去嘗試一下：
deepseek 野生打廣告😀
好了我們言歸正傳，開始Coding!!!😎
下面是CodeWriteAction實現的具體代碼：
import re
import asyncio
from metagpt.actions import Actionclass CodeWrite(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(instruction=requirements)rsp = await self._aask(prompt)code_text = CodeWrite.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在這個代碼案例中：
- 1.我們定義了一個CodeWrite類，其繼承自Action類，
- 2.我們重寫了run方法，其表示這個方法會對用戶輸入做哪些處理
- 3.在name中定義了該行為的名稱
CodeWrite,這樣以來，我們就可以后續用name指代這個方法了； - 4.定義提示詞模版，這是屬于提示詞工程一部分，通過規范正式化的模版以及舉例說明，可以將用戶輸入格式化字符串打包；
- 5.我們通過調用
self.__aask方法，將打包后的prompt然后發給LLM，從而得到要求的輸出格式; - 6.利用正則表達式，我們提取其中生成的\Python代碼，然后返回提取的結果;這里我們的表達式用于提取以"···python", 開頭，以"···結尾"的字符串;
這里為了避免 ` 影響格式，用 · 進行替代；
運行效果如下：
4.設計CodeWriterAgent人設
Message
在開始設計Agent人設之前，我們首先需要了解一下Messge,在MetaGPT中，Message是最基本的信息類型，其基本構成如下：
可以看到，Message的組成包括以下幾個部分：
- content：用于存放消息的內容;
- instruct_content：與content功能相同，但存放的是結構化的數據;
- role：表示消息的角色，是調用LLM的參數的一部分，屬于meta信息的一部分;
- cause_by：用作分類標簽和路由標簽，表示消息的來源，即哪個動作導致產生的message;
- sent_from：用作展示時顯示的發言者信息，屬于meta信息的一部分;
- send_to：用作路由參數，用來篩選發給特定角色的消息;
- restricted_to：用作群發的路由參數，用來篩選發給特定角色的消息;
本文中，我們只需要使用到其中的content,role,cause_by這幾個參數，其中只有content是必選參數，其他都是可選參數；
之前，我們已經定義了的動作CodeWrite，這里我們如果想要調用這個動作，就需要將其配置綁定到具體的Agent初始化屬性上；
Agent設計
廢話不多數，上代碼：
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import loggerclass CodeWriter(Role):"""CodeWriter 角色類，繼承自 Role 基類"""name: str = "Cheems" # 角色昵稱profile: str = "CodeWriter" # 角色人設def __init__(self, **kwargs):"""初始化 CodeWriter 角色"""super().__init__(**kwargs) # 調用基類構造函數self.set_actions([CodeWrite]) # 為角色配備 CodeWrite 動作async def _act(self) -> Message:"""定義角色行動邏輯"""logger.info(f"{self._setting}: ready to {self.rc.todo}") # 記錄日志todo = self.rc.todo # 獲取待執行的動作 (CodeWriter)msg = self.get_memories(k=1)[0] # 獲取最近一條記憶 (用戶輸入)code_text = await todo.run(msg.content) # 執行 CodeWrite 動作，獲取生成的代碼msg = Message(content=code_text, role=self.profile, cause_by=type(todo)) # 構造 Message 對象return msg # 返回生成的 Message
- name: str = “cheems”;配置角色昵稱；
- profile: str = “CodeWriter”; 設定角色人設；
- self._init_actions(CodeWrite]); 為CodeWriter角色配備
CodeWrite動作；- self.rc.todo; 從代辦事項中獲取待執行的動作 (
CodeWrite)；- self.get_memories(k=1)[0]; 獲取最近一條記憶 (用戶輸入)；
- await todo.run(msg.content); 執行CodeWrite動作，獲取生成的代碼；
- Message(content=code_text, role=self.profile, cause_by=type(todo)); 構造 Message 對象并返回。
如此以來，我們便拿到了大模型給我們的輸出了，我們現在得到返回的一個基本通信Message對象；
運行CodeWrite
當前我們已經定義好了Action CodeWrite，也設計好了Agent CodeWriter,我們需要實例化我們的Agent類，并且用一個初始化消息content來激活運行CodeWriter，運行代碼如下：
import asyncioasync def main():msg = "如何用python獲取最新的股票統計數據？"role = CodeWriter() # 實例化CodeWritelogger.info(msg) # 記錄日志信息result = await role.run(msg) # 得到運行結果logger.info(result) # 記錄運行結果asyncio.run(main()) # 異步運行main函數
- 因為Jupyter已經存在循環的事件，我們這里需要將完整代碼存放在.py文件中運行;
完整代碼如下：
import re
import asyncio
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import loggerclass CodeWrite(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 = CodeWrite.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_textclass CodeWriter(Role):"""CodeWriter 角色類，繼承自 Role 基類"""name: str = "Cheems" # 角色昵稱profile: str = "CodeWriter" # 角色人設def __init__(self, **kwargs):"""初始化 CodeWriter 角色"""super().__init__(**kwargs) # 調用基類構造函數self.set_actions([CodeWrite]) # 為角色配備 CodeWrite 動作async def _act(self) -> Message:"""定義角色行動邏輯"""logger.info(f"{self._setting}: ready to {self.rc.todo}") # 記錄日志todo = self.rc.todo # 獲取待執行的動作 (CodeWriter)msg = self.get_memories(k=1)[0] # 獲取最近一條記憶 (用戶輸入)code_text = await todo.run(msg.content) # 執行 CodeWrite 動作，獲取生成的代碼msg = Message(content=code_text, role=self.profile, cause_by=type(todo)) # 構造 Message 對象return msg # 返回生成的 Messageasync def main():msg = "如何用python獲取最新的股票統計數據？"role = CodeWriter() # 實例化CodeWritelogger.info(msg) # 記錄日志信息result = await role.run(msg) # 得到運行結果logger.info(result) # 記錄運行結果asyncio.run(main()) # 異步運行main函數 # 正常python文件使用
運行命令:
python main.py
運行結果如下：
運行成功😀😀🏆🏆！！可以看到，我們的Agent生成了內容，并且按照我們的要求對內容進行了提取，得到我們規范格式的Python代碼；
五、多動作Agent
1.思考
我們現在已經學會構建一個單動作Agent去解決一些原子性問題和簡單問題，但是現實生活面臨的更多是復雜多樣的問題，因此單動作已經遠遠不能滿足現實需要，并且我們發現如果只是一個動作，那么我們直接運行動作函數即可，并不需要構建Agent，因此單動作并不能體現出Agent的作用；
Agent的潛力在于對多個動作模塊的組合（模塊化，原子化），規劃使用(線性工作流、并行工作流、優化迭代)，使其能完成類似于人類實踐的效果；通過鏈接這些Action，我們可以構建一個工作流，使智能體能夠完成更復雜的任務；
2.需求分析
我們已經在前面的案例中編寫了代碼編寫動作CodeWrite,現在我們在其基礎上繼續增加一個代碼執行動作CodeRun，用來將CodeWrite動作執行的返回代碼，并返回運行的結果,包括報錯或正常輸出，因此我們的需求分析需要進一步調整：
既然我們的Agent需要掌握代碼編寫和代碼執行動作，那我們稱這個新Agent名為Programmer,下面是我們的設計圖：
我們在之前的基礎上增加了兩個新的Action，原諒我在這里擅自新增了一個需求優化Action，作者覺得需求優化環節很有必要，即可以節省用戶的時間成本；又可以規范用戶需求，讓下一步動作CodeWrite生成的結果更加精準可靠；
4.編寫Requirements_opt動作
代碼如下：
from metagpt.actions import Actionclass RequirementsOpt(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將會對用戶輸入的需求按照要求的規范進行優化;得到更全面穩定的需求；
5.編寫CodeRun動作
CodeRun 動作用于執行生成的代碼并返回運行結果。
這里作者也學習到一個新的東西:一個動作可以利用LLM，也可以在沒有LLM的情況下運行，我們的動作中，只有需求優化和代碼編寫這部分需要與LLM進行交互，而代碼運行動作則不需要，因此，我們可以只啟動一個子進程來運行代碼并獲取結果；
- 在Python中，我們 通過標準庫中的
subprocess包來fork一個子進程，并運行一個外部的程序；subprocess包中定義有數個創建子進程的函數，這些函數分別以不同的方式創建子進程；- 第一個進程是你的Python程序本身，它執行了包含
Programmer類定義的代碼，第二個進程是由subprocess.run創建的，它執行了python -c命令，用于運行code_text中返回的python代碼，這兩個進程相互獨立，通過subprocess.run，我們的python程序可以啟動并與第二個進程進行交互，獲取其輸出結果；
import subprocess
from metagpt.actions import Actionclass CodeRun(Action):name: str = "CodeRun"async def run(self, code_text: str):try:result = subprocess.run(['python', '-c', code_text],text=True,capture_output=True)return result.stdoutexcept subprocess.CalledProcessError as e:return e.stderr
很好，我們已經成功構建了兩個動作，接下來我們繼續參考單動作階段的思路設計Programmer Agent；
6. 設計Programmer Agent人設
Programmer Agent 將整合三個動作：RequirementsOpt、CodeWrite 和 CodeRun。
代碼如下：
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import loggerclass Programmer(Role):"""Programmer 角色類，繼承自 Role 基類"""name: str = "cheems"profile: str = "Programmer"def __init__(self, **kwargs):"""初始化 Programmer 角色"""super().__init__(**kwargs) # 調用基類構造函數self.set_actions([RequirementsOpt, CodeWrite, CodeRun]) # 為角色配備三個動作self._set_react_mode(react_mode="by_order") # 順序執行async def _act(self) -> Message:"""定義角色行動邏輯"""logger.info(f"{self._setting}: 準備 {self.rc.todo}") # 記錄日志todo = self.rc.todo # 按照排列順序獲取待執行的動作msg = self.get_memories(k=1)[0] # 獲取最相似的一條記憶 (用戶輸入)# 優化需求》編寫代碼》運行代碼result = await todo.run(msg.content)# 構造 Message 對象msg = Message(content=run_result, role=self.profile, cause_by=type(todo))self.rc.memory.add(msg) # 將運行結果添加到記憶return msg # 返回最終的 Message
這里我們用Role類的_set_react_mode 方法來設定我們Action執行的先后順序，
7.運行Programmer Agent
這里我們也沒什么好講的，思路和之前一樣，我們編寫代碼，實例化Programmer Agent，并且執行器配置的動作；
import asyncioasync def main():msg = "如何用python爬取每日新聞數據？要求不使用API，安裝包的代碼要直接在python代碼中運行，而非手動輸入cmd;"role = Programmer() # 實例化 Programmerlogger.info(msg) # 記錄日志信息result = await role.run(msg) # 得到運行結果logger.info(result) # 記錄運行結果asyncio.run(main()) # 異步運行 main 函數
完整代碼如下：
import re
import asyncio
import subprocess
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger# 代碼撰寫Action
class CodeWrite(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 = CodeWrite.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 RequirementsOpt(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 CodeRun(Action):name: str = "CodeRun"async def run(self, code_text: str):try:result = subprocess.run(['python', '-c', code_text],text=True,capture_output=True,check=True)return result.stdoutexcept subprocess.CalledProcessError as e:return e.stderr# 設計Programmer Agent人設
class Programmer(Role):"""Programmer 角色類，繼承自 Role 基類"""name: str = "cheems"profile: str = "Programmer"def __init__(self, **kwargs):"""初始化 Programmer 角色"""super().__init__(**kwargs) # 調用基類構造函數self.set_actions([RequirementsOpt, CodeWrite, CodeRun]) # 為角色配備三個動作self._set_react_mode(react_mode="by_order") # 順序執行async def _act(self) -> Message:"""定義角色行動邏輯"""logger.info(f"{self._setting}: 準備 {self.rc.todo}") # 記錄日志todo = self.rc.todo # 按照排列順序獲取待執行的動作msg = self.get_memories(k=1)[0] # 獲取最相似的一條記憶 (用戶輸入)# 優化需求》編寫代碼》運行代碼result = await todo.run(msg.content)# 構造 Message 對象msg = Message(content=result, role=self.profile, cause_by=type(todo))self.rc.memory.add(msg) # 將運行結果添加到記憶return msg # 返回最終的 Message# 運行我們的Agent
async def main():msg = "如何用python爬取每日新聞數據？要求不使用API，安裝包的代碼要直接在python代碼中運行，而非手動輸入cmd;"role = Programmer() # 實例化 Programmerlogger.info(msg) # 記錄日志信息result = await role.run(msg) # 得到運行結果logger.info(result) # 記錄運行結果asyncio.run(main()) # 異步運行 main 函數
在命令行cmd執行命令python main.py,輸出如下：
@qianyouliang ? /workspaces/MetaGPT-Learn (main) $ python 3.單智能體-多Action.py
2024-05-16 12:46:11.434 | INFO | metagpt.const:get_metagpt_package_root:29 - Package root set to /workspaces/MetaGPT-Learn
2024-05-16 12:46:15.281 | INFO | __main__:main:120 - 如何用python設計一個ToDoList，使用Tkinter庫，盡量不要安裝額外包，安裝包的代碼要直接在python代碼中運行，而非手動輸入cmd;
2024-05-16 12:46:15.284 | INFO | __main__:_act:102 - cheems(Programmer): 準備 RequirementsOpt
根據您提供的規范，以下是對ToDoList應用程序的需求優化和設計：### 1. 簡要說明 (Brief Description)設計一個簡單的ToDoList應用程序，使用Python的Tkinter庫來創建圖形用戶界面（GUI）。該應用程序允許用戶添加、刪除和標記任務為已完成。應用程序應具有直觀的用戶界面，以便用戶可以輕松管理他們的任務列表。### 2. 事件流 (Flow of Event)#### 基本流
1. 用戶啟動應用程序。
2. 用戶看到一個空白的任務列表。
3. 用戶點擊“添加任務”按鈕。
4. 用戶在彈出的文本框中輸入任務描述。
5. 用戶點擊“確認”按鈕，任務被添加到列表中。
6. 用戶可以選擇一個任務并點擊“刪除”按鈕來移除任務。
7. 用戶可以選擇一個任務并點擊“標記為完成”按鈕來標記任務為已完成。#### 備選流
- 如果用戶嘗試添加一個空任務，系統應提示用戶輸入有效的任務描述。
- 如果用戶嘗試刪除或標記一個不存在的任務，系統應提示錯誤信息。### 3. 用例場景 (Use-Case Scenario)#### 成功場景
- 用戶成功添加一個新任務到列表中。
- 用戶成功刪除一個任務。
- 用戶成功標記一個任務為已完成。#### 失敗場景
- 用戶嘗試添加一個空任務，系統提示錯誤信息。
- 用戶嘗試刪除或標記一個不存在的任務，系統提示錯誤信息。### 4. 特殊需求 (Special Requirement)- 性能：應用程序應快速響應用戶的操作。
- 可靠性：應用程序應確保數據的完整性，即使在意外關閉的情況下也能恢復任務列表。
- 可用性：應用程序應提供清晰的指導和反饋，以便用戶了解如何操作。
- 可擴展性：應用程序應設計為易于添加新功能，如任務分類、優先級設置等。
- 設計約束：使用Python 3.x和Tkinter庫，不安裝額外包。### 5. 前置條件 (Pre-Condition)- Python 3.x已安裝。
- Tkinter庫已安裝。### 6. 后置條件 (Post-Condition)- 用戶可以查看、添加、刪除和標記任務。
- 任務列表的狀態在應用程序關閉后應保持不變。### Python代碼示例```python
import tkinter as tk
from tkinter import messageboxclass ToDoListApp:def __init__(self, root):self.root = rootself.tasks = []self.create_widgets()def create_widgets(self):self.task_listbox = tk.Listbox(self.root)self.task_listbox.pack(pady=10)self.add_button = tk.Button(self.root, text="添加任務", command=self.add_task)self.add_button.pack(pady=5)self.delete_button = tk.Button(self.root, text="刪除任務", command=self.delete_task)self.delete_button.pack(pady=5)self.complete_button = tk.Button(self.root, text="標記為完成", command=self.mark_complete)self.complete_button.pack(pady=5)def add_task(self):task_description = tk.simpledialog.askstring("添加任務", "請輸入任務描述:")if task_description:self.tasks.append(task_description)self.task_listbox.insert(tk.END, task_description)else:messagebox.showwarning("警告", "任務描述不能為空！")def delete_task(self):selected_task_index = self.task_listbox.curselection()if selected_task_index:del self.tasks[selected_task_index[0]]self.task_listbox.delete(selected_task_index)else:messagebox.showwarning("警告", "請選擇一個任務進行刪除！")def mark_complete(self):selected_task_index = self.task_listbox.curselection()if selected_task_index:self.task_listbox.itemconfig(selected_task_index, {'bg': 'gray'})else:messagebox.showwarning("警告", "請選擇一個任務進行標記！")if __name__ == "__main__":root = tk.Tk()app = ToDoListApp(root)root.mainloop()此代碼創建了一個基本的ToDoList應用程序，遵循了上述規范。用戶可以通過圖形界面添加、刪除和標記任務。注意，此代碼不包含持久化數據的功能，如果需要，可以添加文件操作來保存和加載任務列表。
2024-05-16 12:47:19.918 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model deepseek-chat not found in TOKEN_COSTS.
2024-05-16 12:47:19.919 | INFO | __main__:_act:102 - cheems(Programmer): 準備 CodeWrite
根據您提供的需求，以下是一個簡單的ToDoList應用程序的Python代碼示例，使用Tkinter庫來創建GUI。這個示例包含了添加、刪除和標記任務為已完成的功能，并且提供了兩個測試用例來驗證代碼的正確性。import tkinter as tk
from tkinter import messagebox, simpledialogclass ToDoListApp:def __init__(self, root):self.root = rootself.tasks = []self.create_widgets()def create_widgets(self):self.task_listbox = tk.Listbox(self.root, width=50)self.task_listbox.pack(pady=10)self.add_button = tk.Button(self.root, text="添加任務", command=self.add_task)self.add_button.pack(pady=5)self.delete_button = tk.Button(self.root, text="刪除任務", command=self.delete_task)self.delete_button.pack(pady=5)self.complete_button = tk.Button(self.root, text="標記為完成", command=self.mark_complete)self.complete_button.pack(pady=5)def add_task(self):task_description = simpledialog.askstring("添加任務", "請輸入任務描述:")if task_description:self.tasks.append(task_description)self.task_listbox.insert(tk.END, task_description)else:messagebox.showwarning("警告", "任務描述不能為空！")def delete_task(self):selected_task_index = self.task_listbox.curselection()if selected_task_index:del self.tasks[selected_task_index[0]]self.task_listbox.delete(selected_task_index)else:messagebox.showwarning("警告", "請選擇一個任務進行刪除！")def mark_complete(self):selected_task_index = self.task_listbox.curselection()if selected_task_index:self.task_listbox.itemconfig(selected_task_index, {'bg': 'gray'})else:messagebox.showwarning("警告", "請選擇一個任務進行標記！")# 測試用例1: 添加任務
def test_add_task():root = tk.Tk()app = ToDoListApp(root)app.add_task()assert app.tasks == [] # 因為輸入為空，所以任務列表應該為空app.add_task()assert len(app.tasks) == 1 # 添加了一個任務root.destroy()# 測試用例2: 刪除任務
def test_delete_task():root = tk.Tk()app = ToDoListApp(root)app.add_task()app.delete_task()assert app.tasks == [] # 刪除后任務列表應該為空root.destroy()if __name__ == "__main__":test_add_task()test_delete_task()root = tk.Tk()app = ToDoListApp(root)root.mainloop()請注意，這個示例代碼沒有包含持久化數據的功能，如果需要，可以添加文件操作來保存和加載任務列表。此外，測試用例使用了簡單的斷言來驗證功能是否按預期工作。在實際應用中，可能需要更復雜的測試框架和更多的測試用例來確保應用程序的穩定性和可靠性。
2024-05-16 12:48:08.914 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model deepseek-chat not found in TOKEN_COSTS.
2024-05-16 12:48:08.916 | INFO | __main__:_act:102 - cheems(Programmer): 準備 CodeRun
作者因為是在Linux服務器上運行的，沒有GUI，因此不能直接展示，現在我將其復制到Windows系統上，進行運行，下面是運行效果:
雖然運行后，有點問題，但是🤣🤣🤣哈哈，真的很有趣，不是嗎；各位讀者也可以復制我的代碼嘗試去更新迭代，增加更多的功能，使其能穩定運行；我想，用這個來生成需求分析和測試報告可能會比較好用😁😁；
六、思路整理
我們對今天的內容進行匯總，等各位將今天的只是消化理解后，我們再進行下一步；
1.需求分析
- 本階段，我們需要根據我們的需求，設計一個工作流，可以是線性的，也可以是并行的，可以對我們實際工作中的工作流進行抽象，然后繪制為流程圖，其中需要利用MetaGPT的設計思想，對每個環節的機制和信息傳遞進行構建，然后組合成為一個可用的工作流；
這一步非常重要，學習MetaGPT思想來分解現實中的復雜任務可以讓我們最高效快捷的后續開發，平時可以多聯系，總結；
2. 撰寫Action模塊
經過需求分析后，我們已經知道的我們的工作包含那些環節，哪些環節可以用那些工具或者代碼完成，例如我作為一個WebGIS開發者，我的日常工作中是前端開發，我嘗嘗需要構建一些重復性但不完全相同的項目文件，我需要使用到ArcGIS去處理一些數據，并且用一些插件轉化為geojson，我就可以將我的工作流拆分為ArcGIS處理數據[矢量，柵格]，ArcGIS[可視化,制圖]構建爬蟲Action進行文本和圖片視頻爬取，根據API獲得專業數據，例如遙感影像，地理坐標等，然后使用我定義好的工具(Python腳本)進行處理，如果使用過ArcGIS 模型構建器的讀者或許能更容易理解我的想法；
3. 設計Agent人設和工作流
在我多Action的案例中，我設定了AI的人設，這一點也很重要，這和ChatGPT的system 系統設定類似，可以限制LLM的輸出方向和結果，良好構建的prompt模板可以得到更加專業精準的回答，從而提高效率；我這里使用的工作流模式就是順序模式react_mode="by_order",當然MetaGPT中還包含著其他有趣的執行流程，感興趣可以去看看官方文檔；良好的工作流可以是我們之前測試成功率較高的解決方案，我們可以將其保存下來后續使用；
4. 運行工作流
當我們構建好Action，并且設計調試好Agent人設后，我們便可以運行我們的工作流，通過多次的測試，我們可以看到很多潛在的問題，這些問題反饋為新的代碼和提示詞繼續優化著我們的每個環節，經過多次調試后，就可以得到良好的Agent;
七、技術文檔生成Agent
這里作者會將給出完整技術文檔生成器的代碼，以及設計思路；讀者有疑問可以指出，至于為什么這個潦草描述，原因有二：
- 學習Agent思想，要理論結合實踐前進，基礎概念和理解在前期很重要(不是作者懶😂😂)；
- 之外，我們的文章長度已經足夠長，再閱讀下去容易造成大腦疲勞；
1.需求描述：
如何讓LLM寫一篇技術文檔？直接使用ChatGPT?存在問題：
- 內容太短；
- 質量不佳；
專業的事情交給專業的人去做，這里我們可以說交給專業的Agent去做,那么思考一下，博主是如何撰寫一篇技術博客的？哪些環節可以用Agent替代？
- ① 找主題（爬取熱點、主題活動）
- ② 列大綱（LLM根據熱點信息撰寫大綱）
- ③ 查資料（LLM通過WebSearch和API獲取信息，存入向量數據庫）
- ④ 寫內容 (基于熱點從向量數據庫中篩選信息，并根據寫作風格和文章格式開始撰寫內容)
- ⑤ 測試驗證 （構建一個Agent用于讀取每一段內容并于向量數據庫中內容做對比，判斷是否正確，對于代碼則調用代碼執行方法進行測試，根據反饋判斷是否正確）
- ⑥ 發布（調用平臺提供的API或者自動化發布工具實現自動發文）
下面是學習文檔給出的思路，和我給的大差不差，讀者自行參考：
這里就不分開描述了，完整代碼如下：
from datetime import datetime
from typing import Dict
import asyncio
from metagpt.actions.write_tutorial import WriteDirectory, WriteContent
from metagpt.const import TUTORIAL_PATH
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message
from metagpt.utils.file import Filefrom typing import Dictfrom metagpt.actions import Action
from metagpt.prompts.tutorial_assistant import DIRECTORY_PROMPT, CONTENT_PROMPT
from metagpt.utils.common import OutputParserclass WriteDirectory(Action):"""Action class for writing tutorial directories.Args:name: The name of the action.language: The language to output, default is "Chinese"."""name: str = "WriteDirectory"language: str = "Chinese"async def run(self, topic: str, *args, **kwargs) -> Dict:"""Execute the action to generate a tutorial directory according to the topic.Args:topic: The tutorial topic.Returns:the tutorial directory information, including {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}."""COMMON_PROMPT = """You are now a seasoned technical professional in the field of the internet. We need you to write a technical tutorial with the topic "{topic}"."""DIRECTORY_PROMPT = COMMON_PROMPT + """Please provide the specific table of contents for this tutorial, strictly following the following requirements:1. The output must be strictly in the specified language, {language}.2. Answer strictly in the dictionary format like {{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}.3. The directory should be as specific and sufficient as possible, with a primary and secondary directory.The secondary directory is in the array.4. Do not have extra spaces or line breaks.5. Each directory title has practical significance."""prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)resp = await self._aask(prompt=prompt)return OutputParser.extract_struct(resp, dict)class WriteContent(Action):"""Action class for writing tutorial content.Args:name: The name of the action.directory: The content to write.language: The language to output, default is "Chinese"."""name: str = "WriteContent"directory: dict = dict()language: str = "Chinese"async def run(self, topic: str, *args, **kwargs) -> str:"""Execute the action to write document content according to the directory and topic.Args:topic: The tutorial topic.Returns:The written tutorial content."""COMMON_PROMPT = """You are now a seasoned technical professional in the field of the internet. We need you to write a technical tutorial with the topic "{topic}"."""CONTENT_PROMPT = COMMON_PROMPT + """Now I will give you the module directory titles for the topic. Please output the detailed principle content of this title in detail. If there are code examples, please provide them according to standard code specifications. Without a code example, it is not necessary.The module directory titles for the topic is as follows:{directory}Strictly limit output according to the following requirements:1. Follow the Markdown syntax format for layout.2. If there are code examples, they must follow standard syntax specifications, have document annotations, and be displayed in code blocks.3. The output must be strictly in the specified language, {language}.4. Do not have redundant output, including concluding remarks.5. Strict requirement not to output the topic "{topic}"."""prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)return await self._aask(prompt=prompt)class TutorialAssistant(Role):"""Tutorial assistant, input one sentence to generate a tutorial document in markup format.Args:name: The name of the role.profile: The role profile description.goal: The goal of the role.constraints: Constraints or requirements for the role.language: The language in which the tutorial documents will be generated."""name: str = "Stitch"profile: str = "Tutorial Assistant"goal: str = "Generate tutorial documents"constraints: str = "Strictly follow Markdown's syntax, with neat and standardized layout"language: str = "Chinese"topic: str = ""main_title: str = ""total_content: str = ""def __init__(self, **kwargs):super().__init__(**kwargs)self.set_actions([WriteDirectory(language=self.language)])self._set_react_mode(react_mode=RoleReactMode.REACT.value)async def _think(self) -> None:"""Determine the next action to be taken by the role."""logger.info(self.rc.state)logger.info(self,)if self.rc.todo is None:self._set_state(0)returnif self.rc.state + 1 < len(self.states):self._set_state(self.rc.state + 1)else:self.rc.todo = Noneasync def _handle_directory(self, titles: Dict) -> Message:"""Handle the directories for the tutorial document.Args:titles: A dictionary containing the titles and directory structure,such as {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}Returns:A message containing information about the directory."""self.main_title = titles.get("title")directory = f"{self.main_title}\n"self.total_content += f"# {self.main_title}"actions = list()print(titles,"diandiandian")for first_dir in titles.get("目錄"):actions.append(WriteContent(language=self.language, directory=first_dir))key = list(first_dir.keys())[0]directory += f"- {key}\n"for second_dir in first_dir[key]:directory += f" - {second_dir}\n"self.set_actions(actions)self.rc.todo = Nonereturn Message(content=directory)async def _act(self) -> Message:"""Perform an action as determined by the role.Returns:A message containing the result of the action."""todo = self.rc.todoif type(todo) is WriteDirectory:msg = self.rc.memory.get(k=1)[0]self.topic = msg.contentresp = await todo.run(topic=self.topic)logger.info(resp)return await self._handle_directory(resp)resp = await todo.run(topic=self.topic)logger.info(resp)if self.total_content != "":self.total_content += "\n\n\n"self.total_content += respreturn Message(content=resp, role=self.profile)async def _react(self) -> Message:"""Execute the assistant's think and actions.Returns:A message containing the final result of the assistant's actions."""while True:await self._think()if self.rc.todo is None:breakmsg = await self._act()root_path = TUTORIAL_PATH / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")await File.write(root_path, f"{self.main_title}.md", self.total_content.encode('utf-8'))return msgasync def main():msg = "Git 教程"role = TutorialAssistant()logger.info(msg)result = await role.run(msg)logger.info(result)asyncio.run(main())
代碼運行效果如下：
填寫提示詞時需要求將“direction”字段改為“目錄”，我這里使用了“目錄”作為字段名
課后作業
最后，為各位讀者留一個課后作業，當然也是作者(中間商😂)的課后作業，幫助你我吸收理解MetaGPT這個框架的思路邏輯(只是為例方便大家理解Agent的Action迭代方式)；
題目:
編寫這樣一個 Agent:
- 這個 Agent 擁有三個動作 打印1 打印2 打印3（初始化時 init_action([print,print,print])）
- 重寫有關方法（請不要使用act_by_order，我希望你能獨立實現）使得 Agent 順序執行上面三個動作
- 當上述三個動作執行完畢后，為 Agent 生成新的動作 打印4 打印5 打印6 并順序執行，（之前我們初始化了三個 print 動作，執行完畢后，重新 init_action([…,…,…])，然后順序執行這個新生成的動作列表)
下面是作者的思路：
- 創建三個Action，分別用于打印輸出；
- 三個Action存放在一個類的自身任務列表(類的屬性)中，循環此列表，然后進行順序執行；每個元素為一個任務對象，，包含(ID,name,content,state),當任務執行完畢后，Action的狀態為2，執行失敗是1，未執行是0；每次執行則輸出Content的內容
- 創建Acton動態生成方法，當判斷到Action列表為空后，繼續生成3個新的Action，其中ID不能與之前重復；
- 為了保證體現出Agent的run->react->think->act流程，我們這里每次運行完畢后，將運行的結果進行存儲；并且基于記憶，調整下一步的輸出；
完整代碼如下，這里以Cheems與醫生的無限對話為例：
import asyncio
from metagpt.actions.write_tutorial import WriteDirectory, WriteContent
from metagpt.logs import logger
from metagpt.roles.role import Role
from metagpt.schema import Message
from metagpt.actions import Action
from metagpt.prompts.tutorial_assistant import DIRECTORY_PROMPT, CONTENT_PROMPT
from metagpt.utils.common import OutputParser# 定義 Action 類，繼承自 metagpt.actions.Action
class PrintAction(Action):def __init__(self, ID, name, content, day, period):super().__init__()self.ID = IDself.name = nameself.content = contentself.day = dayself.period = periodself.state = 0 # 初始化狀態為 0，未執行async def run(self):try:self.content = await self._aask(f"這是我們的聊天歷史:{self.content}，當前時間是第{self.day}天，{self.period}，請你繼續與我以一個心理醫生角度交談，內容要符合時間，而不是一個AI，,回復要推動事情發展前進，而不是只考慮一個話題；現在針對我的話進行回復即可")self.state = 2 # 執行成功，狀態更新為 2except Exception as e:print(e)self.state = 1 # 執行失敗，狀態更新為 1return self.content # 確保返回一個字符串class PrintAgent(Role):def __init__(self, **kwargs):super().__init__(**kwargs)self.actions = [] # 初始化動作列表self.action_id = 0 # 初始化動作 IDself.day = 1 # 從第1天開始self.periods = ["早晨", "下午", "晚上"]self.profile = "cheems"# 初始化前三個固定的 PrintActionself.add_initial_actions()def add_initial_actions(self):initial_contents = ['你叫什么名字？','Cheems,我感到很累？我整天996工作，但是我別無選擇，醫生，我該怎么辦','講講你的故事，Cheems，我愿意傾聽..']for index, content in enumerate(initial_contents):self.action_id += 1action = PrintAction(self.action_id, f"打印{self.action_id}", content, self.day, self.periods[index])self.add_action(action)def add_action(self, action):self.actions.append(action) # 添加動作到動作列表async def run_actions(self):for action in self.actions:todo = self.rc.todo # 按照排列順序獲取待執行的動作logger.info(f"{self._setting}: 準備動作{action.ID},今天是第{action.day}天, {action.period}") # 記錄日志msg = await action.run() # 順序執行動作msg = Message(content=msg or "", role=self.profile, cause_by=type(todo))self.rc.memory.add(msg) # 將執行結果添加到記憶self.day += 1self.actions = [] # 清空動作列表def generate_actions(self):# 生成三個新的動作，ID 遞增for i in range(3):self.action_id += 1msg = self.get_memories(k=1)[0] # 獲取最相似的1條記憶action = PrintAction(self.action_id, f"打印{self.action_id}", msg, self.day, self.periods[i])self.add_action(action)async def _act(self) -> Message:while True:if not self.actions: # 如果動作列表為空self.generate_actions() # 生成新的動作await self.run_actions() # 執行動作await asyncio.sleep(1) # 添加一個短暫的休眠，以防止無限循環導致過高的 CPU 占用return Message(content="動作執行完畢", role=self.profile)# 異步主函數
async def main():agent = PrintAgent() # 創建 醫生 實例await agent._act() # 執行動作# 運行異步主函數
asyncio.run(main())
運行效果如下：
各位可以運行，自行測試，時間有限，代碼只完成了基本功能，并不完善😥😥；
項目地址
- Github地址
- 拓展閱讀
如果覺得我的文章對您有幫助，三連+關注便是對我創作的最大鼓勵！或者一個star🌟也可以😂.
![[C++] 小游戲 能量 1.0.0 版本 zty出品](http://pic.xiahunao.cn/[C++] 小游戲 能量 1.0.0 版本 zty出品)
(比亞迪))
