基于MetaGPT构建单智能体

BAR

前言

在之前的文章中，我们详细地描述了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任务执行流程图：
agent-frame
我们来理解一下这个图：

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执行流程图；
uml

首先，我们需要其接受用户的输入，并且调用记忆模块存储我们的输入，
接着这个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", 开头，以"···结尾"的字符串;

这里为了避免 ` 影响格式，用 · 进行替代；

运行效果如下：
output_createAction

4.设计CodeWriterAgent人设

Message

在开始设计Agent人设之前，我们首先需要了解一下Messge,在MetaGPT中，Message是最基本的信息类型，其基本构成如下：
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

运行结果如下：
execute_result
运行成功😀😀🏆🏆！！可以看到，我们的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系统上，进行运行，下面是运行效果:
code_result
虽然运行后，有点问题，但是🤣🤣🤣哈哈，真的很有趣，不是吗；各位读者也可以复制我的代码尝试去更新迭代，增加更多的功能，使其能稳定运行；我想，用这个来生成需求分析和测试报告可能会比较好用😁😁；

六、思路整理

我们对今天的内容进行汇总，等各位将今天的只是消化理解后，我们再进行下一步；

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())

代码运行效果如下：
cmd

填写提示词时需要求将“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())