CH12‐24 - SunXiaoXiang/learn_wowagent GitHub Wiki
环境配置
使用 wowAgent 作为虚拟环境
conda create -n wowAgent python=3.10 -y
conda activate wowAgent
pip install openai python-dotenv
pip install metagpt==0.8.0
metagpt 真的装了不少东西，需要点时间。
使用MetaGPT需要配置模型API。
- 在当前工作目录中创建一个名为config的文件夹，并在其中添加一个名为config2.yaml的新文件。
- 将示例config2.yaml文件的内容复制到您的新文件中。
- 将您自己的值填入文件中： 智谱 API
国内的大模型，智谱的效果是非常好的。 config2.yaml
llm:
api_type: 'zhipuai'
api_key: 'YOUR_API_KEY'
model: 'glm-4'
科大讯飞的大模型 Spark API： 科大讯飞的API无法支持异步，所以回答一两个简单的问题还可以，如果要做步骤多于两个的任务，目前效果还不太可观。
config2.yaml
llm:
api_type: 'spark'
app_id: 'YOUR_APPID'
api_key: 'YOUR_API_KEY'
api_secret: 'YOUR_API_SECRET'
domain: 'generalv3.5'
base_url: 'wss://spark-api.xf-yun.com/v3.5/chat'
百度 千帆 API
千帆的TPM比较低，适合当作回答问题来用。面对比较复杂的任务就会报错使用超限制。
config2.yaml
llm:
api_type: 'qianfan'
api_key: 'YOUR_API_KEY'
secret_key: 'YOUR_SECRET_KEY'
model: 'ERNIE-Bot-4'
Supported models: {'ERNIE-3.5-8K-0205', 'ERNIE-Bot-turbo-AI', 'ChatLaw', 'Qianfan-Chinese-Llama-2-13B', 'Yi-34B-Chat', 'ERNIE-Bot-4', 'Llama-2-70b-chat', 'ChatGLM2-6B-32K', 'Llama-2-7b-chat', 'Llama-2-13b-chat', 'ERNIE-Bot-8k', 'ERNIE-Speed', 'ERNIE-3.5-4K-0205', 'ERNIE-Bot', 'Mixtral-8x7B-Instruct', 'EB-turbo-AppBuilder', 'ERNIE-Bot-turbo', 'BLOOMZ-7B', 'XuanYuan-70B-Chat-4bit', 'Qianfan-BLOOMZ-7B-compressed', 'Qianfan-Chinese-Llama-2-7B', 'AquilaChat-7B'}
月之暗面 Moonshot API
月之暗面的TPM也比较低，只能当作回答问题来用。面对复杂的任务会报错使用超限制。若要使用建议充值去提升TPM。
config2.yaml
llm:
api_type: 'moonshot'
base_url: 'https://api.moonshot.cn/v1'
api_key: 'YOUR_API_KEY'
model: 'moonshot-v1-8k'
本地ollama API
config2.yaml
llm:
api_type: 'ollama'
base_url: 'http://192.168.0.70:11434/api'
model: 'qwen2:7b'
repair_llm_output: true
代码中192.168.0.70就是部署了大模型的电脑的IP，
请根据实际情况进行替换
有一个小细节需要注意，冒号后面需要有个空格，否则会报错。
如何检验自己是否配置成功呢？
from metagpt.config2 import Config
def print_llm_config():
# 加载默认配置
config = Config.default()
# 获取LLM配置
llm_config = config.llm
# 打印LLM配置的详细信息
if llm_config:
print(f"API类型: {llm_config.api_type}")
print(f"API密钥: {llm_config.api_key}")
print(f"模型: {llm_config.model}")
else:
print("没有配置LLM")
if __name__ == "__main__":
print_llm_config()
执行上面的代码，如果输出的llm类型、密钥都没问题，就说明配置成功。
安装ollama qwen2:1.5b模型
- 安装 ollama
- ollama run qwen2:1.5b 执行命令安装模型
- 按照流式配置，并验证安装 验证的代码
# 我们先用requets库来测试一下大模型
import json
import requests
# 192.168.0.70就是部署了大模型的电脑的IP，
# 请根据实际情况进行替换
BASE_URL = "http://192.168.0.70:11434/api/chat"
payload = {
"model": "qwen2:1.5b",
"messages": [
{
"role": "user",
"content": "请写一篇1000字左右的文章，论述法学专业的就业前景。"
}
]
}
response = requests.post(BASE_URL, json=payload)
print(response.text)
想要流式输出的话，就这么验证
# 我们先用requets库来测试一下大模型
import json
import requests
# 192.168.0.70就是部署了大模型的电脑的IP，
# 请根据实际情况进行替换
BASE_URL = "http://192.168.0.70:11434/api/chat"
payload = {
"model": "qwen2:1.5b",
"messages": [
{
"role": "user",
"content": "请写一篇1000字左右的文章，论述法学专业的就业前景。"
}
],
"stream": True
}
response = requests.post(BASE_URL, json=payload, stream=True) # 在这里设置stream=True告诉requests不要立即下载响应内容
# 检查响应状态码
if response.status_code == 200:
# 使用iter_content()迭代响应体
for chunk in response.iter_content(chunk_size=1024): # 你可以设置chunk_size为你想要的大小
if chunk:
# 在这里处理chunk（例如，打印、写入文件等）
rtn = json.loads(chunk.decode('utf-8')) # 假设响应是文本，并且使用UTF-8编码
print(rtn["message"]["content"], end="")
else:
print(f"Error: {response.status_code}")
# 不要忘记关闭响应
response.close()
mac 上，执行可以，并验证如下：
智能体
在MetaGPT看来，可以将智能体想象成环境中的数字人，其中
智能体 = 大语言模型（LLM） + 观察 + 思考 + 行动 + 记忆
这个公式概括了智能体的功能本质。为了理解每个组成部分，让我们将其与人类进行类比：
- 大语言模型（LLM）：LLM作为智能体的“大脑”部分，使其能够处理信息，从交互中学习，做出决策并执行行动。
- 观察：这是智能体的感知机制，使其能够感知其环境。智能体可能会接收来自另一个智能体的文本消息、来自监视摄像头的视觉数据或来自客户服务录音的音频等一系列信号。这些观察构成了所有后续行动的基础。
- 思考：思考过程涉及分析观察结果和记忆内容并考虑可能的行动。这是智能体内部的决策过程，其可能由LLM进行驱动。
- 行动：这些是智能体对其思考和观察的显式响应。行动可以是利用 LLM 生成代码，或是手动预定义的操作，如阅读本地文件。此外，智能体还可以执行使用工具的操作，包括在互联网上搜索天气，使用计算器进行数学计算等。
- 记忆：智能体的记忆存储过去的经验。这对学习至关重要，因为它允许智能体参考先前的结果并据此调整未来的行动。
多智能体
多智能体系统可以视为一个智能体社会，其中
多智能体 = 智能体 + 环境 + 标准流程（SOP） + 通信 + 经济
这些组件各自发挥着重要的作用：
- 智能体：在上面单独定义的基础上，在多智能体系统中的智能体协同工作，每个智能体都具备独特有的LLM、观察、思考、行动和记忆。
- 环境：环境是智能体生存和互动的公共场所。智能体从环境中观察到重要信息，并发布行动的输出结果以供其他智能体使用。
- 标准流程（SOP）：这些是管理智能体行动和交互的既定程序，确保系统内部的有序和高效运作。例如，在汽车制造的SOP中，一个智能体焊接汽车零件，而另一个安装电缆，保持装配线的有序运作。
- 通信：通信是智能体之间信息交流的过程。它对于系统内的协作、谈判和竞争至关重要。
- 经济：这指的是多智能体环境中的价值交换系统，决定资源分配和任务优先级。
任务
对于每一个任务，至少要明确两点：目标和期望。目标和期望都可以用自然语言去描述。
其他需要明确的是 上下文、回调、输出、使用的工具。
回调可以是一个python函数。使用的工具可以是一个python列表。
你可以用pydantic去约束输出的合适。把大模型的模糊输出变为强制结构化输出。
工具
一个常用的工具就是搜索引擎。例如谷歌的serper。国内的替代品是什么？
还有爬虫工具
通过 py 代码，做 meteGPT 的第一个代码的快速验证
import asyncio
from metagpt.roles import (
Architect,
Engineer,
ProductManager,
ProjectManager,
)
from metagpt.team import Team
async def startup(idea: str):
# 创建团队
company = Team()
# 招聘角色
company.hire(
[
ProductManager(), # 产品经理
Architect(), # 架构师
ProjectManager(), # 项目经理
Engineer(), # 工程师
]
)
# 投资
company.invest(investment=3.0)
# 运行项目
company.run_project(idea=idea)
# 运行团队，进行多轮迭代
await company.run(n_round=5)
if __name__ == "__main__":
# 定义项目想法
idea = "开发一个基于 AI 的任务管理系统"
# 运行异步函数
asyncio.run(startup(idea))
在 macOS 上，无法使用 ollama 来跑通这个示范例子，实际代码使用的是 zhipu api 完成的。
单个动作的单智能体
使用现有的智能体完成单个动作
import asyncio
from metagpt.roles.product_manager import ProductManager
# 定义提示信息
prompt = """
# Role: The software development team
## Background :
I am a software development team.
Now we need to develop a brushing program with html, js, vue3, and element-plus.
Brushing questions can give people a deeper grasp of the knowledge points involved in the questions.
## Profile:
- author: Li Wei
- version: 0.1
- Language: Chinese
- description: I'm a software development team.
## Goals:
- Development requirements document for developing a problem brushing program using HTML, JS, VUE3, and element-plus.
## Constrains:
1. The final delivery program is an HTML single file, and no other files.
2. The question type includes at least 2 true/false questions, 2 multiple-choice questions, and 2 fill-in-the-blank questions.
3. The content of the question is related to the basic theory of artificial intelligence agent.
4. At least 10 sample questions will be given in the brushing process.
5. Write the question in the form of a list in the script section of the HTML file.
6. Vue3 and element-plus are introduced in the header part of HTML in the form of CDN.
## Skills:
1. Strong JS language development ability
2. Familiar with the use of VUE3 and element-plus
3. Have a good understanding of the basic theory of artificial intelligence
4. Have a typographic aesthetic, and use serial numbers, indentations, dividers, and line breaks to beautify the typography of information
Please combine the above requirements to improve the development requirements document of the brushing program.
"""
async def main():
# 初始化角色
role = ProductManager()
# 运行角色并获取结果
result = await role.run(prompt)
# 打印结果
print(result)
if __name__ == "__main__":
# 运行异步函数
asyncio.run(main())
这个方法最简单，只需要按照实际要求，修改 prompt 部分，就可以完成单动作单智能体。
定制智能体
如果一个智能体能够执行某些动作（无论是由LLM驱动还是其他方式），它就具有一定的用途。简单来说，我们定义智能体应该具备哪些行为，为智能体配备这些能力，我们就拥有了一个简单可用的智能体！
假设我们想用自然语言编写代码，并想让一个智能体为我们做这件事。让我们称这个智能体为 SimpleCoder，我们需要两个步骤来让它工作：
- 定义一个编写代码的动作
- 为智能体配备这个动作
定义动作
在 MetaGPT 中，类 Action 是动作的逻辑抽象。用户可以通过简单地调用 self._aask 函数令 LLM 赋予这个动作能力，即这个函数将在底层调用 LLM api。
from metagpt.actions import Action
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction} and provide two runnnable test cases.
Return ```python your_code_here ```with NO other texts,
your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
rsp = await self._aask(prompt)
code_text = SimpleWriteCode.parse_code(rsp)
return code_text
@staticmethod
def parse_code(rsp):
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1) if match else rsp
return code_text
定义角色
在 MetaGPT 中，Role 类是智能体的逻辑抽象。一个 Role 能执行特定的 Action，拥有记忆、思考并采用各种策略行动。基本上，它充当一个将所有这些组件联系在一起的凝聚实体。目前，让我们只关注一个执行动作的智能体，并看看如何定义一个最简单的 Role。
在这个示例中，我们创建了一个 SimpleCoder，它能够根据人类的自然语言描述编写代码。步骤如下：
- 我们为其指定一个名称和配置文件。
- 我们使用 self._init_action 函数为其配备期望的动作 SimpleWriteCode。
- 我们覆盖 _act 函数，其中包含智能体具体行动逻辑。我们写入，我们的智能体将从最新的记忆中获取人类指令，运行配备的动作，MetaGPT将其作为待办事项 (self.rc.todo) 在幕后处理，最后返回一个完整的消息。
import re
import os
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
class SimpleCoder(Role):
name: str = "Alice"
profile: str = "SimpleCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteCode])
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
todo = self.rc.todo # todo will be SimpleWriteCode()
msg = self.get_memories(k=1)[0] # find the most recent messages
code_text = await todo.run(msg.content)
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
运行这个角色
现在我们可以让我们的智能体开始工作，只需初始化它并使用一个起始消息运行它。
async def main():
msg = "write a function that calculates the sum of a list"
role = SimpleCoder()
logger.info(msg)
result = await role.run(msg)
logger.info(result)
return result
rtn = await main()
完整代码
import asyncio
import re
import logging
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 定义动作
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction} and provide two runnable test cases.
Return ```python your_code_here ``` with NO other texts.
Your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
# 格式化提示
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
# 调用 LLM 生成代码
rsp = await self._aask(prompt)
# 解析代码
code_text = SimpleWriteCode.parse_code(rsp)
return code_text
@staticmethod
def parse_code(rsp: str) -> str:
# 使用正则表达式提取代码块
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1).strip() if match else rsp
return code_text
# 定义角色
class SimpleCoder(Role):
name: str = "Alice"
profile: str = "SimpleCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteCode]) # 设置动作
async def _act(self) -> Message:
# 记录日志
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
# 获取待执行的动作
todo = self.rc.todo # todo 是 SimpleWriteCode 的实例
# 从记忆中获取最新的人类指令
msg = self.get_memories(k=1)[0]
# 执行动作
code_text = await todo.run(msg.content)
# 返回消息
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
# 主程序
async def main():
# 定义任务
msg = "write a function that calculates the sum of a list"
# 初始化角色
role = SimpleCoder()
# 记录任务
logger.info(f"Task: {msg}")
# 运行角色并获取结果
result = await role.run(msg)
# 记录结果
logger.info(f"Generated Code:\n{result.content}")
return result
# 运行异步函数
if __name__ == "__main__":
asyncio.run(main())
执行结果
智能体的力量，或者说Role抽象的惊人之处，在于动作的组合（以及其他组件，比如记忆，但我们将把它们留到后面的部分）。通过连接动作，我们可以构建一个工作流程，使智能体能够完成更复杂的任务。
假设现在我们不仅希望用自然语言编写代码，而且还希望生成的代码立即执行。一个拥有多个动作的智能体可以满足我们的需求。让我们称之为RunnableCoder，一个既写代码又立即运行的Role。我们需要两个Action：SimpleWriteCode 和 SimpleRunCode。
定义动作
首先，定义 SimpleWriteCode。我们将重用上面创建的那个。
接下来，定义 SimpleRunCode。如前所述，从概念上讲，一个动作可以利用LLM，也可以在没有LLM的情况下运行。在SimpleRunCode的情况下，LLM不涉及其中。我们只需启动一个子进程来运行代码并获取结果。我们希望展示的是，对于动作逻辑的结构，我们没有设定任何限制，用户可以根据需要完全灵活地设计逻辑。
# SimpleWriteCode 这个类与上一节一模一样
from metagpt.actions import Action
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction} and provide two runnnable test cases.
Return ```python your_code_here ```with NO other texts,
your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
rsp = await self._aask(prompt)
code_text = SimpleWriteCode.parse_code(rsp)
return code_text
@staticmethod
def parse_code(rsp):
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1) if match else rsp
return code_text
# 本节新增了SimpleRunCode这个类
class SimpleRunCode(Action):
name: str = "SimpleRunCode"
async def run(self, code_text: str):
result = subprocess.run(["python", "-c", code_text], capture_output=True, text=True)
code_result = result.stdout
logger.info(f"{code_result=}")
return code_result
定义角色
与定义单一动作的智能体没有太大不同！让我们来映射一下：
- 用 self.set_actions 初始化所有 Action
- 指定每次 Role 会选择哪个 Action。我们将 react_mode 设置为 "by_order"，这意味着 Role 将按照 self.set_actions 中指定的顺序执行其能够执行的 Action。在这种情况下，当 Role 执行 _act 时，self.rc.todo 将首先是 SimpleWriteCode，然后是 SimpleRunCode。
- 覆盖 _act 函数。Role 从上一轮的人类输入或动作输出中检索消息，用适当的 Message 内容提供当前的 Action (self.rc.todo)，最后返回由当前 Action 输出组成的 Message。
import re
import os
import subprocess
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
class RunnableCoder(Role):
name: str = "Alice"
profile: str = "RunnableCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteCode, SimpleRunCode])
self._set_react_mode(react_mode="by_order")
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
# By choosing the Action by order under the hood
# todo will be first SimpleWriteCode() then SimpleRunCode()
todo = self.rc.todo
msg = self.get_memories(k=1)[0] # find the most k recent messages
result = await todo.run(msg.content)
msg = Message(content=result, role=self.profile, cause_by=type(todo))
self.rc.memory.add(msg)
return msg
async def main():
msg = "write a function that calculates the sum of a list"
role = RunnableCoder()
logger.info(msg)
result = await role.run(msg)
logger.info(result)
return result
rtn = await main()
print(rtn)
完整代码
import asyncio
import re
import subprocess
import logging
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 定义动作：SimpleWriteCode
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction} and provide two runnable test cases.
Return ```python your_code_here ``` with NO other texts.
Your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
# 格式化提示
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
# 调用 LLM 生成代码
rsp = await self._aask(prompt)
# 解析代码
code_text = SimpleWriteCode.parse_code(rsp)
return code_text
@staticmethod
def parse_code(rsp: str) -> str:
# 使用正则表达式提取代码块
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1).strip() if match else rsp
return code_text
# 定义动作：SimpleRunCode
class SimpleRunCode(Action):
name: str = "SimpleRunCode"
async def run(self, code_text: str):
# 使用子进程运行代码
result = subprocess.run(["python", "-c", code_text], capture_output=True, text=True)
code_result = result.stdout
logger.info(f"{code_result=}")
return code_result
# 定义角色：RunnableCoder
class RunnableCoder(Role):
name: str = "Alice"
profile: str = "RunnableCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteCode, SimpleRunCode]) # 设置动作
self._set_react_mode(react_mode="by_order") # 设置执行顺序
async def _act(self) -> Message:
# 记录日志
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
# 获取待执行的动作
todo = self.rc.todo # todo 是 SimpleWriteCode 或 SimpleRunCode 的实例
# 从记忆中获取最新的人类指令
msg = self.get_memories(k=1)[0]
# 执行动作
result = await todo.run(msg.content)
# 返回消息
msg = Message(content=result, role=self.profile, cause_by=type(todo))
self.rc.memory.add(msg)
return msg
# 主程序
async def main():
# 定义任务
msg = "write a function that calculates the sum of a list"
# 初始化角色
role = RunnableCoder()
# 记录任务
logger.info(f"Task: {msg}")
# 运行角色并获取结果
result = await role.run(msg)
# 记录结果
logger.info(f"Final Result:\n{result.content}")
return result
# 运行异步函数
if __name__ == "__main__":
asyncio.run(main())
执行结果
角色介绍
功能说明 输入一句话，生成一篇偏技术类教程文档，支持自定义语言。 设计思路 先通过 LLM 大模型生成教程的目录，再对目录按照二级标题进行分块，对于每块目录按照标题生成详细内容，最后再将标题和内容进行拼接。分块的设计解决了 LLM 大模型长文本的限制问题。 原文链接： https://docs.deepwisdom.ai/v0.8/zh/guide/use_cases/agent/tutorial_assistant.html 编写 WriteDirectory 动作 我们先来实现根据用户需求生成文章大纲的代码
from metagpt.actions import Action
from typing import Dict, Union
import ast
def extract_struct(text: str, data_type: Union[type(list), type(dict)]) -> Union[list, dict]:
"""Extracts and parses a specified type of structure (dictionary or list) from the given text.
The text only contains a list or dictionary, which may have nested structures.
Args:
text: The text containing the structure (dictionary or list).
data_type: The data type to extract, can be "list" or "dict".
Returns:
- If extraction and parsing are successful, it returns the corresponding data structure (list or dictionary).
- If extraction fails or parsing encounters an error, it throw an exception.
返回：
- 如果提取和解析成功，它将返回相应的数据结构（列表或字典）。
- 如果提取失败或解析遇到错误，则抛出异常。
Examples:
>>> text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
>>> result_list = OutputParser.extract_struct(text, "list")
>>> print(result_list)
>>> # Output: [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}]
>>> text = 'xxx {"x": 1, "y": {"a": 2, "b": {"c": 3}}} xxx'
>>> result_dict = OutputParser.extract_struct(text, "dict")
>>> print(result_dict)
>>> # Output: {"x": 1, "y": {"a": 2, "b": {"c": 3}}}
"""
# Find the first "[" or "{" and the last "]" or "}"
start_index = text.find("[" if data_type is list else "{")
end_index = text.rfind("]" if data_type is list else "}")
if start_index != -1 and end_index != -1:
# Extract the structure part
structure_text = text[start_index : end_index + 1]
try:
# Attempt to convert the text to a Python data type using ast.literal_eval
result = ast.literal_eval(structure_text)
# Ensure the result matches the specified data type
if isinstance(result, list) or isinstance(result, dict):
return result
raise ValueError(f"The extracted structure is not a {data_type}.")
except (ValueError, SyntaxError) as e:
raise Exception(f"Error while extracting and parsing the {data_type}: {e}")
else:
logger.error(f"No {data_type} found in the text.")
return [] if data_type is list else {}
class WriteDirectory(Action):
"""Action class for writing tutorial directories.
Args:
name: The name of the action.
language: The language to output, default is "Chinese".
用于编写教程目录的动作类。
参数：
name：动作的名称。
language：输出的语言，默认为"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"]}]}.
根据主题执行生成教程目录的操作。
参数：
topic：教程主题。
返回：
教程目录信息，包括{"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}".
您现在是互联网领域的经验丰富的技术专业人员。
我们需要您撰写一个关于"{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.
请按照以下要求提供本教程的具体目录：
1. 输出必须严格符合指定语言，{language}。
2. 回答必须严格按照字典格式，如{{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}。
3. 目录应尽可能具体和充分，包括一级和二级目录。二级目录在数组中。
4. 不要有额外的空格或换行符。
5. 每个目录标题都具有实际意义。
"""
prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
resp = await self._aask(prompt=prompt)
return extract_struct(resp, dict)
完整代码如下：
import asyncio
import ast
import logging
from datetime import datetime
from pathlib import Path
from typing import Dict, Union
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
from metagpt.utils.file import File
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 定义工具函数：从文本中提取数据结构
def extract_struct(text: str, data_type: Union[type(list), type(dict)]) -> Union[list, dict]:
"""
从给定文本中提取并解析指定类型的结构（字典或列表）。
文本仅包含一个列表或字典，可能具有嵌套结构。
Args:
text: 包含结构（字典或列表）的文本。
data_type: 要提取的数据类型，可以是 "list" 或 "dict"。
Returns:
- 如果提取和解析成功，返回相应的数据结构（列表或字典）。
- 如果提取失败或解析出错，抛出异常。
Examples:
>>> text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
>>> result_list = extract_struct(text, list)
>>> print(result_list)
[1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}]
>>> text = 'xxx {"x": 1, "y": {"a": 2, "b": {"c": 3}}} xxx'
>>> result_dict = extract_struct(text, dict)
>>> print(result_dict)
{"x": 1, "y": {"a": 2, "b": {"c": 3}}}
"""
# 找到第一个 "[" 或 "{" 和最后一个 "]" 或 "}"
start_index = text.find("[" if data_type is list else "{")
end_index = text.rfind("]" if data_type is list else "}")
if start_index != -1 and end_index != -1:
# 提取结构部分
structure_text = text[start_index : end_index + 1]
try:
# 使用 ast.literal_eval 将文本转换为 Python 数据类型
result = ast.literal_eval(structure_text)
# 确保结果与指定的数据类型匹配
if isinstance(result, list) or isinstance(result, dict):
return result
raise ValueError(f"提取的结构不是 {data_type}。")
except (ValueError, SyntaxError) as e:
raise Exception(f"提取和解析 {data_type} 时出错: {e}")
else:
logger.error(f"文本中未找到 {data_type}。")
return [] if data_type is list else {}
# 定义动作：WriteDirectory
class WriteDirectory(Action):
"""用于编写教程目录的动作类。
Args:
name: 动作的名称。
language: 输出的语言，默认为 "Chinese"。
"""
name: str = "WriteDirectory"
language: str = "Chinese"
async def run(self, topic: str, *args, **kwargs) -> Dict:
"""根据主题生成教程目录。
Args:
topic: 教程主题。
Returns:
教程目录信息，包括 {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}。
"""
COMMON_PROMPT = """
您现在是互联网领域的经验丰富的技术专业人员。
我们需要您撰写一个关于 "{topic}" 的技术教程。
"""
DIRECTORY_PROMPT = COMMON_PROMPT + """
请按照以下要求提供本教程的具体目录：
1. 输出必须严格符合指定语言，{language}。
2. 回答必须严格按照字典格式，如 {{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}。
3. 目录应尽可能具体和充分，包括一级和二级目录。二级目录在数组中。
4. 不要有额外的空格或换行符。
5. 每个目录标题都具有实际意义。
"""
prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
resp = await self._aask(prompt=prompt)
return extract_struct(resp, dict)
# 定义动作：WriteContent
class WriteContent(Action):
"""用于编写教程内容的动作类。
Args:
name: 动作的名称。
directory: 要编写的内容。
language: 输出的语言，默认为 "Chinese"。
"""
name: str = "WriteContent"
directory: dict = dict()
language: str = "Chinese"
async def run(self, topic: str, *args, **kwargs) -> str:
"""根据目录和主题编写教程内容。
Args:
topic: 教程主题。
Returns:
编写的教程内容。
"""
COMMON_PROMPT = """
您现在是互联网领域的经验丰富的技术专业人员。
我们需要您撰写一个关于 "{topic}" 的技术教程。
"""
CONTENT_PROMPT = COMMON_PROMPT + """
现在我将为您提供该主题的模块目录标题。
请详细输出该标题的详细原理内容。
如果有代码示例，请按照标准代码规范提供。
如果没有代码示例，则不需要提供。
该主题的模块目录标题如下：
{directory}
严格按照以下要求限制输出：
1. 遵循 Markdown 语法格式进行布局。
2. 如果有代码示例，必须遵循标准语法规范，有文档注释，并以代码块形式显示。
3. 输出必须严格符合指定语言，{language}。
4. 不要有冗余输出，包括总结性语句。
5. 严格要求不要输出主题 "{topic}"。
"""
prompt = CONTENT_PROMPT.format(
topic=topic, language=self.language, directory=self.directory
)
return await self._aask(prompt=prompt)
# 定义角色：TutorialAssistant
class TutorialAssistant(Role):
"""教程助手，输入一句话生成 Markdown 格式的教程文档。
Args:
name: 角色名称。
profile: 角色描述。
goal: 角色的目标。
constraints: 角色的约束或要求。
language: 生成教程文档的语言。
"""
name: str = "Stitch"
profile: str = "Tutorial Assistant"
goal: str = "生成教程文档"
constraints: str = "严格遵循 Markdown 语法，布局整洁规范"
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="by_order")
async def _handle_directory(self, titles: Dict) -> Message:
"""处理教程文档的目录。
Args:
titles: 包含标题和目录结构的字典，
例如 {"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}
Returns:
包含目录信息的消息。
"""
self.main_title = titles.get("title")
directory = f"{self.main_title}\n"
self.total_content += f"# {self.main_title}"
actions = list()
for first_dir in titles.get("directory"):
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 = None
return Message(content=directory)
async def _act(self) -> Message:
"""执行角色确定的动作。
Returns:
包含动作结果的消息。
"""
todo = self.rc.todo
if isinstance(todo, WriteDirectory):
msg = self.rc.memory.get(k=1)[0]
self.topic = msg.content
resp = 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 += resp
return Message(content=resp, role=self.profile)
async def _react(self) -> Message:
"""执行助手的思考和动作。
Returns:
包含助手动作最终结果的消息。
"""
while True:
await self._think()
if self.rc.todo is None:
break
msg = await self._act()
root_path = Path("tutorials") / 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 msg
# 主程序
async def main():
msg = "AI Agent开发教程"
role = TutorialAssistant()
logger.info(msg)
result = await role.run(msg)
logger.info(result)
return role.total_content
# 运行异步函数
if __name__ == "__main__":
result = asyncio.run(main())
print(result)
import asyncio
import logging
from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional
import aiohttp
from aiocron import crontab
from bs4 import BeautifulSoup
from pydantic import BaseModel, Field
from pytz import BaseTzInfo
from metagpt.actions import Action
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environment as _ # noqa: F401
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Define SubscriptionRunner
class SubscriptionRunner(BaseModel):
"""A simple wrapper to manage subscription tasks for different roles using asyncio."""
tasks: Dict[Role, asyncio.Task] = Field(default_factory=dict)
class Config:
arbitrary_types_allowed = True
async def subscribe(
self,
role: Role,
trigger: AsyncGenerator[Message, None],
callback: Callable[Message], Awaitable[None](/SunXiaoXiang/learn_wowagent/wiki/Message],-Awaitable[None),
):
"""Subscribes a role to a trigger and sets up a callback to be called with the role's response."""
loop = asyncio.get_running_loop()
async def _start_role():
async for msg in trigger:
resp = await role.run(msg)
await callback(resp)
self.tasks[role] = loop.create_task(_start_role(), name=f"Subscription-{role}")
async def unsubscribe(self, role: Role):
"""Unsubscribes a role from its trigger and cancels the associated task."""
task = self.tasks.pop(role)
task.cancel()
async def run(self, raise_exception: bool = True):
"""Runs all subscribed tasks and handles their completion or exception."""
try:
while True:
for role, task in self.tasks.items():
if task.done():
if task.exception():
if raise_exception:
raise task.exception()
logger.opt(exception=task.exception()).error(
f"Task {task.get_name()} run error"
)
else:
logger.warning(
f"Task {task.get_name()} has completed. "
"If this is unexpected behavior, please check the trigger function."
)
self.tasks.pop(role)
break
else:
await asyncio.sleep(1)
except asyncio.CancelledError:
# Cleanup tasks on cancellation
for task in self.tasks.values():
task.cancel()
await asyncio.gather(*self.tasks.values(), return_exceptions=True)
logger.info("All tasks have been cancelled.")
# Define Action: CrawlOSSTrending
class CrawlOSSTrending(Action):
"""Action to crawl GitHub Trending repositories."""
async def run(self, url: str = "https://github.com/trending"):
try:
async with aiohttp.ClientSession() as client:
async with client.get(url) as response:
response.raise_for_status()
html = await response.text()
soup = BeautifulSoup(html, "html.parser")
repositories = []
for article in soup.select("article.Box-row"):
repo_info = {}
repo_info["name"] = (
article.select_one("h2 a")
.text.strip()
.replace("\n", "")
.replace(" ", "")
)
repo_info["url"] = (
"https://github.com" + article.select_one("h2 a")["href"].strip()
)
# Description
description_element = article.select_one("p")
repo_info["description"] = (
description_element.text.strip() if description_element else None
)
# Language
language_element = article.select_one(
'span[itemprop="programmingLanguage"]'
)
repo_info["language"] = (
language_element.text.strip() if language_element else None
)
# Stars and Forks
stars_element = article.select("a.Link--muted")[0]
forks_element = article.select("a.Link--muted")[1]
repo_info["stars"] = stars_element.text.strip()
repo_info["forks"] = forks_element.text.strip()
# Today's Stars
today_stars_element = article.select_one(
"span.d-inline-block.float-sm-right"
)
repo_info["today_stars"] = (
today_stars_element.text.strip() if today_stars_element else None
)
repositories.append(repo_info)
return repositories
except Exception as e:
logger.error(f"Failed to crawl GitHub Trending: {e}")
return []
# Define Action: AnalysisOSSTrending
TRENDING_ANALYSIS_PROMPT = """# Requirements
You are a GitHub Trending Analyst, aiming to provide users with insightful and personalized recommendations based on the latest
GitHub Trends. Based on the context, fill in the following missing information, generate engaging and informative titles,
ensuring users discover repositories aligned with their interests.
# The title about Today's GitHub Trending
## Today's Trends: Uncover the Hottest GitHub Projects Today! Explore the trending programming languages and discover key domains capturing developers' attention. From ** to **, witness the top projects like never before.
## The Trends Categories: Dive into Today's GitHub Trending Domains! Explore featured projects in domains such as ** and **. Get a quick overview of each project, including programming languages, stars, and more.
## Highlights of the List: Spotlight noteworthy projects on GitHub Trending, including new tools, innovative projects, and rapidly gaining popularity, focusing on delivering distinctive and attention-grabbing content for users.
---
# Format Example
# [Title]
## Today's Trends
Today, ** and ** continue to dominate as the most popular programming languages. Key areas of interest include **, ** and **.
The top popular projects are Project1 and Project2.
## The Trends Categories
1. Generative AI
- [Project1](https://github/xx/project1): [detail of the project, such as star total and today, language, ...]
- [Project2](https://github/xx/project2): ...
...
## Highlights of the List
1. [Project1](https://github/xx/project1): [provide specific reasons why this project is recommended].
...
---
# Github Trending
{trending}
"""
class AnalysisOSSTrending(Action):
"""Action to analyze GitHub Trending repositories."""
async def run(self, trending: Any):
return await self._aask(TRENDING_ANALYSIS_PROMPT.format(trending=trending))
# Define Role: OssWatcher
class OssWatcher(Role):
"""Role to watch and analyze GitHub Trending repositories."""
def __init__(
self,
name="Codey",
profile="OssWatcher",
goal="Generate an insightful GitHub Trending analysis report.",
constraints="Only analyze based on the provided GitHub Trending data.",
):
super().__init__(name=name, profile=profile, goal=goal, constraints=constraints)
self.set_actions([CrawlOSSTrending, AnalysisOSSTrending])
self._set_react_mode(react_mode="by_order")
async def _act(self) -> Message:
logger.info(f"{self._setting}: ready to {self.rc.todo}")
todo = self.rc.todo
memories = self.get_memories(k=1)
if not memories:
return Message(content="No data in memory", role=self.profile, cause_by=type(todo))
msg = memories[0] # Find the most recent message
result = await todo.run(msg.content)
msg = Message(content=str(result), role=self.profile, cause_by=type(todo))
self.rc.memory.add(msg)
return msg
# Define Callback Function
async def wxpusher_callback(msg: Message):
print(msg.content)
# Define Trigger
async def trigger():
for i in range(5): # Trigger 5 times
yield Message("https://github.com/trending") # Valid URL
await asyncio.sleep(5) # Wait 5 seconds between triggers
# Run Entry
async def main():
callbacks = [wxpusher_callback] # Use wxpusher_callback by default
# If no callbacks are provided, use a default print callback
if not callbacks:
async def _print(msg: Message):
print(msg.content)
callbacks.append(_print)
# Callback function
async def callback(msg):
await asyncio.gather(*(call(msg) for call in callbacks))
runner = SubscriptionRunner()
await runner.subscribe(OssWatcher(), trigger(), callback)
await runner.run()
# Run Async Function
if __name__ == "__main__":
asyncio.run(main())
MetaGPT的核心优势也在于轻松灵活地开发一个智能体团队。
我们需要三个步骤来建立团队并使其运作：
定义每个角色能够执行的预期动作
基于标准作业程序（SOP）确保每个角色遵守它。通过使每个角色观察上游的相应输出结果，并为下游发布自己的输出结果，可以实现这一点。
初始化所有角色，创建一个带有环境的智能体团队，并使它们之间能够进行交互。
内容来自于： https://docs.deepwisdom.ai/v0.8/zh/guide/tutorials/multi_agent_101.html
定义动作
我们可以定义三个具有各自动作的Role：
SimpleCoder 具有 SimpleWriteCode 动作，接收用户的指令并编写主要代码
SimpleTester 具有 SimpleWriteTest 动作，从 SimpleWriteCode 的输出中获取主代码并为其提供测试套件
SimpleReviewer 具有 SimpleWriteReview 动作，审查来自 SimpleWriteTest 输出的测试用例，并检查其覆盖范围和质量
import re
from metagpt.actions import Action, UserRequirement
# 构造写代码的动作
def parse_code(rsp):
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1) if match else rsp
return code_text
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction}.
Return ```python your_code_here ```with NO other texts,
your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
rsp = await self._aask(prompt)
code_text = parse_code(rsp)
return code_text
# 构造写测试样例的动作
class SimpleWriteTest(Action):
PROMPT_TEMPLATE: str = """
Context: {context}
Write {k} unit tests using pytest for the given function, assuming you have imported it.
Return ```python your_code_here ```with NO other texts,
your code:
"""
name: str = "SimpleWriteTest"
async def run(self, context: str, k: int = 3):
prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)
rsp = await self._aask(prompt)
code_text = parse_code(rsp)
return code_text
# 构造审查代码的动作
class SimpleWriteReview(Action):
PROMPT_TEMPLATE: str = """
Context: {context}
Review the test cases and provide one critical comments:
"""
name: str = "SimpleWriteReview"
async def run(self, context: str):
prompt = self.PROMPT_TEMPLATE.format(context=context)
rsp = await self._aask(prompt)
return rsp
定义角色
在许多多智能体场景中，定义Role可能只需几行代码。对于SimpleCoder，我们做了两件事：
-
使用 set_actions 为Role配备适当的 Action，这与设置单智能体相同
-
多智能体操作逻辑：我们使Role _watch 来自用户或其他智能体的重要上游消息。回想我们的 SOP，SimpleCoder接收用户指令，这是由 MetaGPT 中的UserRequirement引起的Message。因此，我们添加了 self._watch([UserRequirement])。
# 构造写代码的角色
from metagpt.roles import Role
class SimpleCoder(Role):
name: str = "Alice"
profile: str = "SimpleCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([UserRequirement])
self.set_actions([SimpleWriteCode])
与上述相似，对于 SimpleTester，我们：
-
使用 set_actions 为SimpleTester配备 SimpleWriteTest 动作
-
使Role _watch 来自其他智能体的重要上游消息。回想我们的 SOP，SimpleTester从 SimpleCoder 中获取主代码，这是由 SimpleWriteCode 引起的 Message。因此，我们添加了 self._watch([SimpleWriteCode])。
-
重写 _act 函数，就像我们在智能体入门中的单智能体设置中所做的那样。在这里，我们希望SimpleTester将所有记忆用作编写测试用例的上下文，并希望有 5 个测试用例。
from metagpt.logs import logger
from metagpt.schema import Message
class SimpleTester(Role):
name: str = "Bob"
profile: str = "SimpleTester"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteTest])
self._watch([SimpleWriteCode])
# self._watch([SimpleWriteCode, SimpleWriteReview]) # feel free to try this too
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 context
context = self.get_memories() # use all memories as context
code_text = await todo.run(context, k=5) # specify arguments
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
# 按照相同的过程定义 SimpleReviewer：
class SimpleReviewer(Role):
name: str = "Charlie"
profile: str = "SimpleReviewer"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteReview])
self._watch([SimpleWriteTest])
创建一个团队并添加角色
现在我们已经定义了三个 Role，是时候将它们放在一起了。我们初始化所有角色，设置一个 Team，并hire 它们。
运行 Team，我们应该会看到它们之间的协作！
import asyncio
from metagpt.team import Team
async def main(
idea: str = "write a function that calculates the product of a list",
investment: float = 3.0,
n_round: int = 5,
):
logger.info(idea)
team = Team()
team.hire(
[
SimpleCoder(),
SimpleTester(),
SimpleReviewer(),
]
)
team.invest(investment=investment)
team.run_project(idea)
await team.run(n_round=n_round)
await main()
def calculate_product(numbers):
product = 1
for number in numbers:
product *= number
return product
This function takes a list of numbers as input and returns the product of all the numbers in the list. The product is calculated by multiplying each number in the list together, starting with 1 (since that's the default value when no other numbers are multiplied).
def calculate_product(numbers):
product = 1
for number in numbers:
product *= number
return product
import asyncio
import platform
from typing import Any
import fire
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
class SpeakAloud(Action):
"""Action: Speak out aloud in a debate (quarrel)"""
PROMPT_TEMPLATE: str = """
## BACKGROUND
Suppose you are {name}, you are in a debate with {opponent_name}.
## DEBATE HISTORY
Previous rounds:
{context}
## YOUR TURN
Now it's your turn, you should closely respond to your opponent's latest argument, state your position, defend your arguments, and attack your opponent's arguments,
craft a strong and emotional response in 80 words, in {name}'s rhetoric and viewpoints, your will argue:
"""
name: str = "SpeakAloud"
async def run(self, context: str, name: str, opponent_name: str):
prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)
# logger.info(prompt)
rsp = await self._aask(prompt)
return rsp
class Debator(Role):
name: str = ""
profile: str = ""
opponent_name: str = ""
def __init__(self, **data: Any):
super().__init__(**data)
self.set_actions([SpeakAloud])
self._watch([UserRequirement, SpeakAloud])
async def _observe(self) -> int:
await super()._observe()
# accept messages sent (from opponent) to self, disregard own messages from the last round
self.rc.news = [msg for msg in self.rc.news if msg.send_to == {self.name}]
return len(self.rc.news)
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
todo = self.rc.todo # An instance of SpeakAloud
memories = self.get_memories()
context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
# print(context)
rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)
msg = Message(
content=rsp,
role=self.profile,
cause_by=type(todo),
sent_from=self.name,
send_to=self.opponent_name,
)
self.rc.memory.add(msg)
return msg
async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
"""Run a team of presidents and watch they quarrel. :)"""
Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")
team = Team()
team.hire([Biden, Trump])
team.invest(investment)
team.run_project(idea, send_to="Biden") # send debate topic to Biden and let him speak first
await team.run(n_round=n_round)
def main(idea: str, investment: float = 3.0, n_round: int = 10):
"""
:param idea: Debate topic, such as "Topic: The U.S. should commit more in climate change fighting"
or "Trump: Climate change is a hoax"
:param investment: contribute a certain dollar amount to watch the debate
:param n_round: maximum rounds of the debate
:return:
"""
if platform.system() == "Windows":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
asyncio.run(debate(idea, investment, n_round))
if __name__ == "__main__":
fire.Fire(main) # run as python debate.py --idea="TOPIC" --investment=3.0 --n_round=5
现在我们设定，需要多智能体系统为我们根据我们给定的主题提供一篇优美的英文诗，除了完成写作的 agent 外，我们再设定一名精通诗句的老师来查看并修改学生的作品。 流程 系统首先接收用户的需求（写关于XX主题的诗），在系统中，当学生关注到布置的题目后就会开始创作，当老师发现学生写作完成后就会给学生提出意见，根据老师给出的意见，学生将修改自己的作品，直到设定循环结束。 插入模块
import asyncio
from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.environment import Environment
from metagpt.const import MESSAGE_ROUTE_TO_ALL
声明一个名为 classroom 的 env，我们将所有的 role 都放在其中
classroom = Environment()
定义角色 定义 Student 角色与 Teacher 角色，与单智能体不同的部分是，我们需要声明每个角色关注的动作（self._watch），只有当关注的动作发生后，角色才会开始行动。
class Student(Role):
name: str = "xiaoming"
profile: str = "Student"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._init_actions([WritePoem])
self._watch([UserRequirement, ReviewPoem])
async def _act(self) -> Message:
logger.info(f"{self._setting}: ready to {self.rc.todo}")
todo = self.rc.todo
msg = self.get_memories() # 获取所有记忆
# logger.info(msg)
poem_text = await WritePoem().run(msg)
logger.info(f'student : {poem_text}')
msg = Message(content=poem_text, role=self.profile,
cause_by=type(todo))
return msg
class Teacher(Role):
name: str = "laowang"
profile: str = "Teacher"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._init_actions([ReviewPoem])
self._watch([WritePoem])
async def _act(self) -> Message:
logger.info(f"{self._setting}: ready to {self.rc.todo}")
todo = self.rc.todo
msg = self.get_memories() # 获取所有记忆
poem_text = await ReviewPoem().run(msg)
logger.info(f'teacher : {poem_text}')
msg = Message(content=poem_text, role=self.profile,
cause_by=type(todo))
return msg
定义动作
编写 WritePoem 与 ReviewPoem 方法，在 WritePoem 方法中我们需要实现根据用户提供的主题来编写诗句，并且根据 teacher 的建议修改诗句，在 ReviewPoem 方法中，我们需要读取 student 的诗歌作品，并且给出修改意见。
class WritePoem(Action):
name: str = "WritePoem"
PROMPT_TEMPLATE: str = """
Here is the historical conversation record : {msg} .
Write a poem about the subject provided by human, Return only the content of the generated poem with NO other texts.
If the teacher provides suggestions about the poem, revise the student's poem based on the suggestions and return.
your poem:
"""
async def run(self, msg: str):
prompt = self.PROMPT_TEMPLATE.format(msg = msg)
rsp = await self._aask(prompt)
return rsp
class ReviewPoem(Action):
name: str = "ReviewPoem"
PROMPT_TEMPLATE: str = """
Here is the historical conversation record : {msg} .
Check student-created poems about the subject provided by human and give your suggestions for revisions. You prefer poems with elegant sentences and retro style.
Return only your comments with NO other texts.
your comments:
"""
async def run(self, msg: str):
prompt = self.PROMPT_TEMPLATE.format(msg = msg)
rsp = await self._aask(prompt)
return rsp
运行 提供一个主题，将topic发布在env中运行env，系统就将开始工作了，你可以修改对话轮数（n_round）来达到你希望的效果
async def main(topic: str, n_round=3):
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 -= 1 #如果n_round = 1 ，就只有学生写诗、然后老师没办法进行review
logger.debug(f"max {n_round=} left.")
await classroom.run()
return classroom.history
asyncio.run(main(topic='wirte a poem about moon'))
完成本节，你将能够：
- 理解智能体之间如何进行交互
- 开发你的第一个智能体团队
运行“软件公司”示例
metagpt "write a function that calculates the product of a list"
开发你的第一个智能体团队
希望你会发现软件创业示例很有启发。也许现在你已经有了灵感，想要开发一个根据你的独特需求而定制的智能体团队。在本节中，我们将继续在智能体入门中的简单代码示例中添加更多角色，并引入智能体之间的交互协作。
让我们还雇佣一名测试人员和一名审阅人员携手与编码人员一起工作。这开始看起来像一个开发团队了，不是吗？总的来说，我们需要三个步骤来建立团队并使其运作：
- 定义每个角色能够执行的预期动作
- 基于标准作业程序（SOP）确保每个角色遵守它。通过使每个角色观察上游的相应输出结果，并为下游发布自己的输出结果，可以实现这一点。
- 初始化所有角色，创建一个带有环境的智能体团队，并使它们之间能够进行交互。
完整的代码在本教程的末尾可用
定义动作和角色
与前面课程相同的过程，我们可以定义三个具有各自动作的Role：
- SimpleCoder具有 SimpleWriteCode 动作，接收用户的指令并编写主要代码
- SimpleTester 具有 SimpleWriteTest 动作，从 SimpleWriteCode 的输出中获取主代码并为其提供测试套件
- SimpleReviewer 具有 SimpleWriteReview 动作，审查来自 SimpleWriteTest 输出的测试用例，并检查其覆盖范围和质量
通过上述概述，我们使得 SOP（标准作业程序）变得更加清晰明了。接下来，我们将详细讨论如何根据 SOP 来定义Role。
首先导入模块
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
def parse_code(rsp):
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1) if match else rsp
return code_text
定义动作
我们列举了三个 Action：
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction}.
Return ```python your_code_here ```with NO other texts,
your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
rsp = await self._aask(prompt)
code_text = parse_code(rsp)
return code_text
class SimpleWriteTest(Action):
PROMPT_TEMPLATE: str = """
Context: {context}
Write {k} unit tests using pytest for the given function, assuming you have imported it.
Return ```python your_code_here ```with NO other texts,
your code:
"""
name: str = "SimpleWriteTest"
async def run(self, context: str, k: int = 3):
prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)
rsp = await self._aask(prompt)
code_text = parse_code(rsp)
return code_text
class SimpleWriteReview(Action):
PROMPT_TEMPLATE: str = """
Context: {context}
Review the test cases and provide one critical comments:
"""
name: str = "SimpleWriteReview"
async def run(self, context: str):
prompt = self.PROMPT_TEMPLATE.format(context=context)
rsp = await self._aask(prompt)
return rsp
定义角色
在许多多智能体场景中，定义Role可能只需几行代码。对于SimpleCoder，我们做了两件事：
- 使用 set_actions 为Role配备适当的 Action，这与设置单智能体相同
- 多智能体操作逻辑：我们使Role _watch 来自用户或其他智能体的重要上游消息。回想我们的SOP，SimpleCoder接收用户指令，这是由MetaGPT中的UserRequirement引起的Message。因此，我们添加了 self._watch([UserRequirement])。
class SimpleCoder(Role):
name: str = "Alice"
profile: str = "SimpleCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._watch([UserRequirement])
self.set_actions([SimpleWriteCode])
与上述相似，对于 SimpleTester，我们：
- 使用 set_actions 为SimpleTester配备 SimpleWriteTest 动作
- 使Role _watch 来自其他智能体的重要上游消息。回想我们的SOP，SimpleTester从 SimpleCoder 中获取主代码，这是由 SimpleWriteCode 引起的 Message。因此，我们添加了 self._watch([SimpleWriteCode])。
一个扩展的问题：想一想如果我们使用self._watch([SimpleWriteCode,SimpleWriteReview]) 会意味着什么，可以尝试这样做
此外，你可以为智能体定义自己的操作逻辑。这适用于Action需要多个输入的情况，你希望修改输入，使用特定记忆，或进行任何其他更改以反映特定逻辑的情况。因此，我们：
- 重写 _act 函数，就像我们在前面教程中的单智能体设置中所做的那样。在这里，我们希望SimpleTester将所有记忆用作编写测试用例的上下文，并希望有5个测试用例。
class SimpleTester(Role):
name: str = "Bob"
profile: str = "SimpleTester"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteTest])
self._watch([SimpleWriteCode])
# self._watch([SimpleWriteCode, SimpleWriteReview]) # feel free to try this too
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 context
context = self.get_memories() # use all memories as context
code_text = await todo.run(context, k=5) # specify arguments
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
按照相同的过程定义 SimpleReviewer：
class SimpleReviewer(Role):
name: str = "Charlie"
profile: str = "SimpleReviewer"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteReview])
self._watch([SimpleWriteTest])
创建一个团队并添加角色
现在我们已经定义了三个 Role，是时候将它们放在一起了。我们初始化所有角色，设置一个 Team，并hire 它们。
运行 Team，我们应该会看到它们之间的协作！
import fire
import typer
from metagpt.logs import logger
from metagpt.team import Team
app = typer.Typer()
@app.command()
def main(
idea: str = typer.Argument(..., help="write a function that calculates the product of a list"),
investment: float = typer.Option(default=3.0, help="Dollar amount to invest in the AI company."),
n_round: int = typer.Option(default=5, help="Number of rounds for the simulation."),
):
logger.info(idea)
team = Team()
team.hire(
[
SimpleCoder(),
SimpleTester(),
SimpleReviewer(),
]
)
team.invest(investment=investment)
team.run_project(idea)
await team.run(n_round=n_round)
if __name__ == "__main__":
fire.Fire(main)
这节课来学习Metagpt的一个多动作多智能体的实战案例-狼人杀游戏。
游戏规则
狼人杀是一款多人参与的社交推理游戏，游戏中的角色分为狼人、村民和特殊角色三大类。基本规则如下：
- 角色分配：游戏开始前，每位玩家随机分配一个角色，包括狼人、普通村民和具有特殊能力的神职村民（如预言家、女巫、猎人等）。
- 游戏流程：游戏分为夜晚和白天两个阶段。夜晚，狼人睁眼并杀害一名玩家；白天，所有玩家讨论并投票处决一名玩家。这个过程会不断重复，直到满足某个胜利条件。
- 胜利条件：游戏的胜利条件分为狼人阵营胜利和村民阵营胜利。 狼人胜利：狼人数量等于村民数量时，狼人阵营获胜 村民胜利：所有狼人被找出并处决，村民阵营获胜
Metagpt多智能体代码核心关注三部分：
- 角色（Role）-智能体的角色
- 动作（Action）-角色对应的动作
- 交互环境（Environment）-串联各角色的消息实现智能体间的信息交互
定义角色
1.角色包括：村民、狼人、守卫、先知、巫师、主持人
2.角色框架 - BasePlayer，该类封装了角色的基本行为和属性，所有的角色都继承自这个类，从这个类中派生。其基本属性和初始化如下：
- 首先角色都需要监听 InstructSpeak 动作产生的消息：self._watch([InstructSpeak])
- 角色的行为设置：self.set_actions(capable_actions)，包括设置进来的 special_actions 和 Speak Action。
定义动作
主持人 Moderator 的主要职责是：开始游戏、主持流程、解析角色发言和宣布游戏结果。
村民继承自 BasePlayer，其拥有 Speak 行为。
狼人除了能 Speak （继承自 BasePlayer）外，拥有特殊技能 Hunt。狼人在白天时，要伪装成好人说话，所以，还有个额外的Action：Impersonate。狼人就两个动作：一个是夜晚干人，二是白天伪装成好人发言。
守卫的特殊技能：Protect，保护人。
先知的特殊技能：Verify，验证其它角色的身份。
巫师有两个特殊技能：Save 和 Poison，救人和毒人。
**夜晚共同的Action - NighttimeWhispers，**这个 Action 的设定是在夜晚的时候进行悄悄地思考和发言。大部分的Action都继承自一个 NighttimeWhispers。
定义环境
环境就是用来在各角色之间进行消息传递的。另外还有 round_cnt 来控制最大交互轮数。WerewolfExtEnv 也有更新游戏和各角色状态的作用。可以大体看下环境的封装：
class WerewolfGame(Team):
"""Use the "software company paradigm" to hold a werewolf game"""
env: Optional[WerewolfEnv] = None
def __init__(self, context: Context = None, **data: Any):
super(Team, self).__init__(**data)
ctx = context or Context()
if not self.env:
self.env = WerewolfEnv(context=ctx)
else:
self.env.context = ctx # The `env` object is allocated by deserialization
class WerewolfEnv(WerewolfExtEnv, Environment):
round_cnt: int = Field(default=0)
class WerewolfExtEnv(ExtEnv):
model_config = ConfigDict(arbitrary_types_allowed=True)
players_state: dict[str, tuple[str, RoleState]] = Field(
default_factory=dict, description="the player's role type and state by player_name"
)
round_idx: int = Field(default=0) # the current round
step_idx: int = Field(default=0) # the current step of current round
eval_step_idx: list[int] = Field(default=[])
per_round_steps: int = Field(default=len(STEP_INSTRUCTIONS))
# game global states
game_setup: str = Field(default="", description="game setup including role and its num")
special_role_players: list[str] = Field(default=[])
winner: Optional[str] = Field(default=None)
win_reason: Optional[str] = Field(default=None)
witch_poison_left: int = Field(default=1, description="should be 1 or 0")
witch_antidote_left: int = Field(default=1, description="should be 1 or 0")
# game current round states, a round is from closing your eyes to the next time you close your eyes
round_hunts: dict[str, str] = Field(default_factory=dict, description="nighttime wolf hunt result")
round_votes: dict[str, str] = Field(
default_factory=dict, description="daytime all players vote result, key=voter, value=voted one"
)
player_hunted: Optional[str] = Field(default=None)
player_protected: Optional[str] = Field(default=None)
is_hunted_player_saved: bool = Field(default=False)
player_poisoned: Optional[str] = Field(default=None)
player_current_dead: list[str] = Field(default=[])
代码运行
运行过程大致为：
- 运行代码，游戏开始，角色分配
- 主持人走流程，黑夜守卫说话
- 狼人杀人
- 重复类似上述流程，直至游戏结束。
动手操作：
创建一个werewolf.py的文件运行代码详情如下：
##运行代码详情
#导入角色和游戏相关依赖
import asyncio
import fire
from metagpt.ext.werewolf.roles import Guard, Moderator, Seer, Villager, Werewolf, Witch//守卫 主持人 先知 村民 狼人 巫师
from metagpt.ext.werewolf.roles.human_player import prepare_human_player
from metagpt.ext.werewolf.werewolf_game import WerewolfGame
from metagpt.logs import logger
#由于MetaGPT是异步框架，使用asyncio启动游戏
async def start_game(
investment: float = 3.0,
n_round: int = 5,#建议n_round值设置小一点
shuffle: bool = True,
add_human: bool = False,
use_reflection: bool = True,
use_experience: bool = False,
use_memory_selection: bool = False,
new_experience_version: str = "",
):
game = WerewolfGame()
#初始化游戏设置
game_setup, players = game.env.init_game_setup(
role_uniq_objs=[Villager, Werewolf, Guard, Seer, Witch],#设置游戏玩家职业
num_werewolf=2,
num_villager=2,
shuffle=shuffle,#是否打乱职业顺序，默认打乱
add_human=add_human,#设置真人也参与游戏
use_reflection=use_reflection,#是否让智能体对对局信息反思，默认开启
use_experience=use_experience,#是否让智能体根据过去行为优化自身动作，默认关闭
use_memory_selection=use_memory_selection,
new_experience_version=new_experience_version,
prepare_human_player=prepare_human_player,
)
logger.info(f"{game_setup}")
players = [Moderator()] + players#主持人加入游戏
game.hire(players)
game.invest(investment)
game.run_project(game_setup)#主持人广播游戏情况
await game.run(n_round=n_round)
def main(
investment: float = 20.0,
n_round: int = 12,//运行前建议将此处n_round修改小一点，否则对钱包不友好！！！
shuffle: bool = True,
add_human: bool = False,
use_reflection: bool = True,
use_experience: bool = False,
use_memory_selection: bool = False,
new_experience_version: str = "",
):
asyncio.run(
start_game(
investment,
n_round,
shuffle,
add_human,
use_reflection,
use_experience,
use_memory_selection,
new_experience_version,
)
)
if __name__ == "__main__":
fire.Fire(main)
在命令行输入python werewolf.py ,终端会打印出对局相关消息。
项目资料： 1.metagpt斯坦福虚拟小镇
2.原版斯坦福小镇
斯坦福小镇中构建了一个虚拟的RPG世界，AI在其中可以自由探索、相互合作、发展友情、举办活动、构建家庭。本节将用Metagpt提供的模块展现斯坦福小镇的互动环境。
快速开始
前期准备：
为了方便GA（ generative_agents ）的前端对接数据（避免改动它那块的代码），可在启动_run_st_game.py_加上_temp_storage_path_指向_generative_agents_对应的_temp_storage_路径。比如
python3 run\_st\_game.py --temp\_storage\_path path/to/ga/temp\_storage xxx
或将const.py下的
STORAGE_PATH = EXAMPLE_PATH.joinpath("storage")
TEMP_STORAGE_PATH = EXAMPLE_PATH.joinpath("temp_storage")
更新为
STORAGE_PATH = Path("{path/to/ga/storage}")
TEMP_STORAGE_PATH = Path("{path/to/ga/temp_storage}")
这样可用实现不改变GA代码情况下，实现仿真数据的对接。不然得修改GA的代码来适配MG的输出路径。
如果你不想从0开始启动，拷贝generative_agents/environment/frontend_server/storage/下的其他仿真目录到examples/stanford_town/storage，并选择一个目录名作为fork_sim_code。
后端服务启动：
执行入口为：
python3 run\_st\_game.py "Host a open lunch party at 13:00 pm" "base\_the\_ville\_isabella\_maria\_klaus" "test\_sim" 10
或者
python3 run\_st\_game.py "Host a open lunch party at 13:00 pm" "base\_the\_ville\_isabella\_maria\_klaus" "test\_sim" 10 --temp\_storage\_path path/to/ga/temp\_storage
_idea_为用户给第一个Agent的用户心声，并通过这个心声进行传播，看最后多智能体是否达到举办、参加活动的目标。
cd examples/stanford_town
python run_st_game.py "Host a open lunch party at 13:00 pm" "base_the_ville_isabella_maria_klaus" "test_sim" --temp_storage_path "temp_storage"
前端服务启动：
进入generative_agents所在的项目目录
进入environment/frontend_server，使用python3 manage.py runserver启动前端服务。 访问http://localhost:8000/simulator_home 进入当前的仿真界面。
快速开始：
在命令行中输入下面指令可以快速开始模拟：
cd examples/stanford_town
python run_st_game.py "Host a open lunch party at 13:00 pm" "base_the_ville_isabella_maria_klaus" "test_sim" --temp_storage_path "temp_storage"
有几个重要参数如下所示：
- idea:将传给小镇第一位居民，模拟由此开始
- fork_sim_code：可以沿用过去的模拟结果，相当于一套居民状况模板，存放在examples/stanford_town/storage。也可以用原版斯坦福小镇的其他模板
- sim_code:当前模拟结果保存的文件夹命名，模拟中会不断更新该文件夹
- temp_storage_path:存储模拟的step
小镇环境讲解
小镇用Metagpt的环境模块来实现交互逻辑，让每个角色可以与环境交互，获取观察并更新状态。具体代码
动手尝试：Metagpt斯坦福虚拟小镇模拟使用
- 导入环境相关类并初始化环境
这里选取斯坦福小镇自带的Maze
from metagpt.environment.stanford_town.stanford_town_ext_env import StanfordTownExtEnv
from metagpt.environment.stanford_town.env_space import (
EnvAction,
EnvActionType,
EnvObsParams,
EnvObsType,
)
from metagpt.ext.stanford_town.utils.const.const import MAZE_ASSET_PATH
env=StanfordTownExtEnv(maze_asset_path="/path/to/MAZE_ASSET_PATH)
- 观察环境
这里我们选取小镇地图坐标的（72，14），即设定的伊莎贝拉初始位置作为案例，可以通过传给环境观察类型来获取需要的信息
obs, _ = env.reset() # 得到完整观察值
path_tiles=env.observe(EnvObsParams(obs_type=EnvObsType.TILE_PATH, coord=(72, 14)))#可以查看当前坐标地址，如可以发现伊莎贝拉初始在自己公寓的主卧床上
get_titles=env.observe(EnvObsParams(obs_type=EnvObsType.GET_TITLE, coord=(72, 14)))#可以查看当前坐标的详细观察值
nearby_tiles = env.observe(
EnvObsParams(
obs_type=EnvObsType.TILE_NBR, coord=(72, 14), vision_radius=10
)
) # 得到局部观察值，当前位置(200, 300)视野内的其他网格信息
- 执行动作
动作说明
action = EnvAction(action_type=EnvActionType.RM_TITLE_SUB_EVENT, coord=(72, 14), subject="Isabella Rodriguez") # 初始化一组动作值，删除指定位置主语为subject的事件,事件event=["the Ville:Isabella Rodriguez's apartment:main room:bed","Isabella Rodriguez","is","sleep"]
obs, _, _, _, info = env.step(action) # 执行动作并得到新的完整观察
单个动作的单智能体
使用现有的智能体完成单个动作
import asyncio
from metagpt.roles.product_manager import ProductManager
# 定义提示信息
prompt = """
# Role: The software development team
## Background :
I am a software development team.
Now we need to develop a brushing program with html, js, vue3, and element-plus.
Brushing questions can give people a deeper grasp of the knowledge points involved in the questions.
## Profile:
- author: Li Wei
- version: 0.1
- Language: Chinese
- description: I'm a software development team.
## Goals:
- Development requirements document for developing a problem brushing program using HTML, JS, VUE3, and element-plus.
## Constrains:
1. The final delivery program is an HTML single file, and no other files.
2. The question type includes at least 2 true/false questions, 2 multiple-choice questions, and 2 fill-in-the-blank questions.
3. The content of the question is related to the basic theory of artificial intelligence agent.
4. At least 10 sample questions will be given in the brushing process.
5. Write the question in the form of a list in the script section of the HTML file.
6. Vue3 and element-plus are introduced in the header part of HTML in the form of CDN.
## Skills:
1. Strong JS language development ability
2. Familiar with the use of VUE3 and element-plus
3. Have a good understanding of the basic theory of artificial intelligence
4. Have a typographic aesthetic, and use serial numbers, indentations, dividers, and line breaks to beautify the typography of information
Please combine the above requirements to improve the development requirements document of the brushing program.
"""
async def main():
# 初始化角色
role = ProductManager()
# 运行角色并获取结果
result = await role.run(prompt)
# 打印结果
print(result)
if __name__ == "__main__":
# 运行异步函数
asyncio.run(main())
这个方法最简单，只需要按照实际要求，修改 prompt 部分，就可以完成单动作单智能体。
定制智能体
如果一个智能体能够执行某些动作（无论是由LLM驱动还是其他方式），它就具有一定的用途。简单来说，我们定义智能体应该具备哪些行为，为智能体配备这些能力，我们就拥有了一个简单可用的智能体！
假设我们想用自然语言编写代码，并想让一个智能体为我们做这件事。让我们称这个智能体为 SimpleCoder，我们需要两个步骤来让它工作：
- 定义一个编写代码的动作
- 为智能体配备这个动作
定义动作
在 MetaGPT 中，类 Action 是动作的逻辑抽象。用户可以通过简单地调用 self._aask 函数令 LLM 赋予这个动作能力，即这个函数将在底层调用 LLM api。
from metagpt.actions import Action
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction} and provide two runnnable test cases.
Return ```python your_code_here ```with NO other texts,
your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
rsp = await self._aask(prompt)
code_text = SimpleWriteCode.parse_code(rsp)
return code_text
@staticmethod
def parse_code(rsp):
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1) if match else rsp
return code_text
定义角色
在 MetaGPT 中，Role 类是智能体的逻辑抽象。一个 Role 能执行特定的 Action，拥有记忆、思考并采用各种策略行动。基本上，它充当一个将所有这些组件联系在一起的凝聚实体。目前，让我们只关注一个执行动作的智能体，并看看如何定义一个最简单的 Role。
在这个示例中，我们创建了一个 SimpleCoder，它能够根据人类的自然语言描述编写代码。步骤如下：
- 我们为其指定一个名称和配置文件。
- 我们使用 self._init_action 函数为其配备期望的动作 SimpleWriteCode。
- 我们覆盖 _act 函数，其中包含智能体具体行动逻辑。我们写入，我们的智能体将从最新的记忆中获取人类指令，运行配备的动作，MetaGPT将其作为待办事项 (self.rc.todo) 在幕后处理，最后返回一个完整的消息。
import re
import os
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
class SimpleCoder(Role):
name: str = "Alice"
profile: str = "SimpleCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteCode])
async def _act(self) -> Message:
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
todo = self.rc.todo # todo will be SimpleWriteCode()
msg = self.get_memories(k=1)[0] # find the most recent messages
code_text = await todo.run(msg.content)
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
运行这个角色
现在我们可以让我们的智能体开始工作，只需初始化它并使用一个起始消息运行它。
async def main():
msg = "write a function that calculates the sum of a list"
role = SimpleCoder()
logger.info(msg)
result = await role.run(msg)
logger.info(result)
return result
rtn = await main()
完整代码
import asyncio
import re
import logging
from metagpt.actions import Action
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 定义动作
class SimpleWriteCode(Action):
PROMPT_TEMPLATE: str = """
Write a python function that can {instruction} and provide two runnable test cases.
Return ```python your_code_here ``` with NO other texts.
Your code:
"""
name: str = "SimpleWriteCode"
async def run(self, instruction: str):
# 格式化提示
prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)
# 调用 LLM 生成代码
rsp = await self._aask(prompt)
# 解析代码
code_text = SimpleWriteCode.parse_code(rsp)
return code_text
@staticmethod
def parse_code(rsp: str) -> str:
# 使用正则表达式提取代码块
pattern = r"```python(.*)```"
match = re.search(pattern, rsp, re.DOTALL)
code_text = match.group(1).strip() if match else rsp
return code_text
# 定义角色
class SimpleCoder(Role):
name: str = "Alice"
profile: str = "SimpleCoder"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.set_actions([SimpleWriteCode]) # 设置动作
async def _act(self) -> Message:
# 记录日志
logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
# 获取待执行的动作
todo = self.rc.todo # todo 是 SimpleWriteCode 的实例
# 从记忆中获取最新的人类指令
msg = self.get_memories(k=1)[0]
# 执行动作
code_text = await todo.run(msg.content)
# 返回消息
msg = Message(content=code_text, role=self.profile, cause_by=type(todo))
return msg
# 主程序
async def main():
# 定义任务
msg = "write a function that calculates the sum of a list"
# 初始化角色
role = SimpleCoder()
# 记录任务
logger.info(f"Task: {msg}")
# 运行角色并获取结果
result = await role.run(msg)
# 记录结果
logger.info(f"Generated Code:\n{result.content}")
return result
# 运行异步函数
if __name__ == "__main__":
asyncio.run(main())
执行结果
