Reference：Datawhale wow agent
单动作单智能体#
使用现成的智能体#
# 可导入任何角色，初始化它，用一个开始的消息运行它，完成！
from metagpt.roles.product_manager import ProductManager
prompt = f"""
# Role：软件开发团队
## Background :
我是一个软件开发团队。
现在要用html、js、vue3、element-plus开发一个刷题程序。
刷题可以让人们对题目中涉及的知识点有更深的掌握。
## Profile:
- author: 黎伟
- version: 0.1
- language: 中文
- description: 我是一软件开发团队。
## Goals:
- 用html、js、vue3、element-plus开发一个刷题程序的开发需求文档。
## Constrains:
1. 最后交付的程序是一个html单文件，不要有其他任何文件。
2. 题目的题型至少包括两道判断题、两道选择题、两道填空题。
3. 题目的内容与人工智能的agent基本理论相关。
4. 刷题程序至少给出10道样例题目。
5. 题目用列表的形式写到html文件的script部分。
6. vue3、element-plus采用cdn的形式在html的header部分引入。
## Skills:
1. 具有强大的js语言开发能力
2. 熟悉vue3、element-plus的使用
3. 对人工智能的agent基本理论有较好理解
4. 拥有排版审美, 会利用序号, 缩进, 分隔线和换行符等等来美化信息排版
请结合上述要求完善刷题程序的开发需求文档。
"""
async def main():
role = ProductManager()
result = await role.run(prompt)
await main()
定制智能体#
从实际使用的角度考虑，一个智能体要对我们有用，它必须具备哪些基本要素呢？从 MetaGPT 的观点来看，如果一个智能体能够执行某些动作（无论是由 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()
结果#
2025-01-26 18:14:18.317 | INFO | __main__:main:4 - write a function that calculates the sum of a list
2025-01-26 18:14:18.319 | INFO | __main__:_act:16 - Alice(SimpleCoder): to do SimpleWriteCode(SimpleWriteCode)
```python
def sum_list(numbers):
return sum(numbers)
# Test cases
test_case_1 = [1, 2, 3, 4, 5]
test_case_2 = [10, 20, 30, 40, 50]
# Test case 1
print(sum_list(test_case_1)) # Expected output: 15
# Test case 2
print(sum_list(test_case_2)) # Expected output: 150
2025-01-26 18:14:25.372 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 18:14:25.374 | INFO | __main__:main:6 - SimpleCoder:
def sum_list(numbers):
return sum(numbers)
# Test cases
test_case_1 = [1, 2, 3, 4, 5]
test_case_2 = [10, 20, 30, 40, 50]
# Test case 1
print(sum_list(test_case_1)) # Expected output: 15
# Test case 2
print(sum_list(test_case_2)) # Expected output: 150
多动作智能体#
具有多个动作的智能体#
我们注意到一个智能体能够执行一个动作，但如果只有这些，实际上我们并不需要一个智能体。通过直接运行动作本身，我们可以得到相同的结果。智能体的力量，或者说 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)
结果#
def calculate_sum(numbers):
return sum(numbers)
# Test cases
def test_case_1():
assert calculate_sum([1, 2, 3, 4, 5]) == 15
def test_case_2():
assert calculate_sum([10, -5, 0, 7, 3]) == 15
技术教程智能体#
角色介绍#
功能说明
输入一句话，生成一篇偏技术类教程文档，支持自定义语言。
设计思路
先通过 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)
测试一下：
text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
result_list = 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 = extract_struct(text, dict)
print(result_dict)
# Output: {"x": 1, "y": {"a": 2, "b": {"c": 3}}}
from metagpt.const import TUTORIAL_PATH
TUTORIAL_PATH
# WindowsPath('d:/Code/agent_from_scratch/data/tutorial_docx')
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 File
from typing import Dict
from metagpt.actions import Action
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: 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 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)
return
if self.rc.state + 1 < len(self.states):
self._set_state(self.rc.state + 1)
else:
self.rc.todo = None
async 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()
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:
"""Perform an action as determined by the role.
Returns:
A message containing the result of the action.
"""
todo = self.rc.todo
if type(todo) is 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:
"""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:
break
msg = 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 msg
async def main():
msg = "AI Agent开发教程"
role = TutorialAssistant()
logger.info(msg)
result = await role.run(msg)
logger.info(result)
return role.total_content
rtn = await main()
print（rnt）输出生成的 AI Agent 开发教程
结果#
太长了仅展示部分。
订阅智能体标题#
输入pip show metagpt查看版本
类似：
Version: 0.8.0
Summary: The Multi-Agent Framework
Home-page: https://github.com/geekan/MetaGPT
Author: Alexander Wu
Author-email: [email protected]
License: MIT
Location: c:\users\liwei\appdata\roaming\python\python39\site-packages
Requires: aiofiles, aiohttp, aioredis, anthropic, anytree, beautifulsoup4, channels, dashscope, faiss-cpu, fire, gitignore-parser, gitpython, google-generativeai, imap-tools, ipykernel, ipython, ipywidgets, jieba, lancedb, libcst, loguru, meilisearch, nbclient, nbformat, networkx, numpy, openai, openpyxl, pandas, Pillow, playwright, pydantic, python-docx, PyYAML, qdrant-client, qianfan, rank-bm25, rich, scikit-learn, semantic-kernel, setuptools, socksio, ta, tenacity, tiktoken, tqdm, typer, typing-extensions, typing-inspect, websocket-client, websockets, wrapt, zhipuai
我们先来完成网页爬取的功能，我们教程直接爬取当天不分国家语言和编程语言的热门仓库进行分析，如果有特殊要求，爬取加上筛选条件条件后网页即可。我们先打开https://github.com/trending 网页，观察网页内容，找到我们需要的内容对应的 html 元素，。
如果熟悉爬虫的就可以直接写爬取和解析脚本了，如果不熟悉的也没关系，我们可以 用 ChatGPT 辅助开发： 首先我们将 trending 页面保存到 本地 github-trending-raw.html
格式化后发现内容非常多，大概 600 多 k，还有一些 svg 源码，因为一般用 CSS 足以定位 html 里的元素，所以我们可以对 html 内容进行瘦身，可以使用以下的脚本：
from bs4 import BeautifulSoup
with open("github-trending-raw.html") as f:
html = f.read()
soup = BeautifulSoup(html, "html.parser")
for i in soup.find_all(True):
for name in list(i.attrs):
if i[name] and name not in ["class"]:
del i[name]
for i in soup.find_all(["svg", "img", "video", "audio"]):
i.decompose()
with open("github-trending-slim.html", "w") as f:
f.write(str(soup))
经过以上的脚本处理之后，大概还有 100 多 k。对于爬虫来说，重要的是 Html 的结构，处理后的 Html 文件其实有大量的信息是重复的，如果我们要让 GPT 协助我们写爬虫脚本，只需要截取部分信息就可以了。
接下来解析一下 html 文件
import aiohttp
import asyncio
from bs4 import BeautifulSoup
def fetch_html(url):
with open(url, encoding="utf-8") as f:
html = f.read()
return html
async def parse_github_trending(html):
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()
repo_info['url'] = 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
async def main():
url = 'github-trending-raw.html'
html = fetch_html(url)
repositories = await parse_github_trending(html)
for repo in repositories:
print(f"Name: {repo['name']}")
print(f"URL: https://github.com{repo['url']}")
print(f"Description: {repo['description']}")
print(f"Language: {repo['language']}")
print(f"Stars: {repo['stars']}")
print(f"Forks: {repo['forks']}")
print(f"Today's Stars: {repo['today_stars']}")
print()
await main()
上面这些代码的作用是解析一下 github-trending 的数据。由于 github-trending 网站打开比较慢。我们可以先把 github 的 trending 页面保存到本地，再进行解读。
继续导入其他相关的库#
import os
from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional
from aiocron import crontab
from pydantic import BaseModel, Field
from pytz import BaseTzInfo
from metagpt.actions.action import Action
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
# fix SubscriptionRunner not fully defined
from metagpt.environment import Environment as _ # noqa: F401
订阅模块#
可以 from metagpt.subscription import SubscriptionRunner 导入，这里贴上代码供参考
class SubscriptionRunner(BaseModel):
"""A simple wrapper to manage subscription tasks for different roles using asyncio.
Example:
>>> import asyncio
>>> from metagpt.subscription import SubscriptionRunner
>>> from metagpt.roles import Searcher
>>> from metagpt.schema import Message
>>> async def trigger():
... while True:
... yield Message("the latest news about OpenAI")
... await asyncio.sleep(3600 * 24)
>>> async def callback(msg: Message):
... print(msg.content)
>>> async def main():
... pb = SubscriptionRunner()
... await pb.subscribe(Searcher(), trigger(), callback)
... await pb.run()
>>> asyncio.run(main())
"""
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],
],
):
"""Subscribes a role to a trigger and sets up a callback to be called with the role's response.
Args:
role: The role to subscribe.
trigger: An asynchronous generator that yields Messages to be processed by the role.
callback: An asynchronous function to be called with the response from the role.
"""
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.
Args:
role: The role to unsubscribe.
"""
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.
Args:
raise_exception: _description_. Defaults to True.
Raises:
task.exception: _description_
"""
i=0
while True:
i+=1
for role, task in self.tasks.items():
i=0
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)
if i>0:
break
Actions 的实现#
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 CrawlOSSTrending(Action):
async def run(self, url: str = "https://github.com/trending"):
async with aiohttp.ClientSession() as client:
async with client.get(url, proxy=CONFIG.global_proxy) 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
class AnalysisOSSTrending(Action):
async def run(self, trending: Any):
return await self._aask(TRENDING_ANALYSIS_PROMPT.format(trending=trending))
Role 实现#
# Role实现
# 对于V0.7 以上的版本，需要把老版本的
# self._init_actions 改为self.set_actions
class OssWatcher(Role):
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}")
# 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=str(result), role=self.profile, cause_by=type(todo))
self.rc.memory.add(msg)
return msg
async def wxpusher_callback(msg: Message):
print(msg.content)
async def trigger():
# 这里设置了只触发五次，也可以用while True 永远执行下去
for i in range(5):
yield Message("the latest news about OpenAI")
await asyncio.sleep(5)
# 每隔五秒钟执行一次。
# 也可以设置为每隔3600 * 24 秒执行一次
运行入口#
# 运行入口，
async def main():
callbacks = []
if not callbacks:
async def _print(msg: Message):
print(msg.content)
callbacks.append(_print)
# callback
async def callback(msg):
await asyncio.gather(*(call(msg) for call in callbacks))
runner = SubscriptionRunner()
await runner.subscribe(OssWatcher(), trigger(), callback)
await runner.run()
await main()
It seems like you've provided a list of GitHub repositories related to various topics. However, there's no clear indication whether these are open-source projects or not. If they're open-source projects, you can use the following Python code to check their star count:
import requests
def get_repo_info(repo_name):
url = f"https://github.com/search?q=stars:{repo_name}&type=Repositories"
response = requests.get(url)
if response.status_code == 200:
repos = []
for repo in response.json():
if "name" in repo and "url" in repo:
name = repo["name"]
url = repo["url"]
stars = int(repo["stargazers_count"])
forks = int(repo["forks_count"]) # Assuming the number of forks is available
repos.append({"name": name, "url": url, "stars": stars, "forks": forks})
return repos
else:
print("Failed to fetch data. Status code:", response.status_code)
repos = get_repo_info('language')
for repo in repos:
if 'language' not in repo or repo['language'] == None:
del repo['language']
print(repos)
This script will fetch the repositories with a specific name (in this case, "language") and print out their star count. If there's no specific language mentioned, it'll assume all languages are open-source.
Please note that GitHub API might not always return accurate data due to various reasons like missing or incorrect information in the repository details. Also, some repositories may have a different number of stars depending on when they were last updated.
It seems like you've provided a list of GitHub repositories related to various topics. However, there's no clear indication whether these are open-source projects or not. If they're open-source projects, you can use the following Python code to check their star count:
import requests
def get_repo_info(repo_name):
url = f"https://github.com/search?q=stars:{repo_name}&type=Repositories"
response = requests.get(url)
if response.status_code == 200:
repos = []
for repo in response.json():
if "name" in repo and "url" in repo:
name = repo["name"]
url = repo["url"]
stars = int(repo["stargazers_count"])
forks = int(repo["forks_count"]) # Assuming the number of forks is available
repos.append({"name": name, "url": url, "stars": stars, "forks": forks})
return repos
else:
print("Failed to fetch data. Status code:", response.status_code)
repos = get_repo_info('language')
for repo in repos:
if 'language' not in repo or repo['language'] == None:
del repo['language']
print(repos)
This script will fetch the repositories with a specific name (in this case, "language") and print out their star count. If there's no specific language mentioned, it'll assume all languages are open-source.
Please note that GitHub API might not always return accurate data due to various reasons like missing or incorrect information in the repository details. Also, some repositories may have a different number of stars depending on when they were last updated.
由于我设置的是每 5 秒钟触发一次，执行 5 次后停止。可以根据自己的需求设置执行频率。
我不能运行，关于代理的问题除了错误。
单动作多智能体#
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()
结果#
2025-01-26 19:37:59.576 | INFO | main:main:9 - write a function that calculates the product of a list
2025-01-26 19:38:00.116 | INFO | metagpt.team:invest:90 - Investment: $3.0.
2025-01-26 19:38:00.119 | INFO | metagpt.roles.role:_act:391 - Alice(SimpleCoder): to do SimpleWriteCode(SimpleWriteCode)
def product_of_list(lst):
result = 1
for num in lst:
result *= num
return result
2025-01-26 19:38:03.794 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 19:38:03.798 | INFO | main:_act:15 - Bob(SimpleTester): to do SimpleWriteTest(SimpleWriteTest)
def test_product_of_list_empty():
assert product_of_list([]) == 1
def test_product_of_list_single_element():
assert product_of_list([5]) == 5
def test_product_of_list_positive_numbers():
assert product_of_list([1, 2, 3, 4]) == 24
def test_product_of_list_negative_numbers():
assert product_of_list([-1, -2, -3, -4]) == -24
def test_product_of_list_mixed_numbers():
assert product_of_list([-1, 2, -3, 4]) ==
2025-01-26 19:38:09.372 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 19:38:09.376 | INFO | metagpt.roles.role:_act:391 - Charlie(SimpleReviewer): to do SimpleWriteReview(SimpleWriteReview)
24
Critical Comment:
While the test cases cover a range of scenarios, including empty lists, single elements, positive numbers, negative numbers, and mixed numbers, there is a potential oversight in the test suite. The function product_of_list does not handle the case where the list contains zero. Since multiplying any number by zero results in zero, it would be beneficial to include a test case that verifies the function's behavior when zero is present in the list. This would ensure that the function is robust and can handle all possible integer values within the
2025-01-26 19:38:14.253 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
list.
单动作多智能体实例：辩论#
这是一个展示如何设计多个智能体并促进它们之间互动的例子，我们将模拟代表拜登和特朗普的智能体共同合作会怎样，这样的组合可能会导致一些生动的交流，我们将称这个实验为 “拜登 - 特朗普辩论”。
实例代码详见：
https://github.com/geekan/MetaGPT/blob/main/examples/debate.py
辩论分三个步骤设定：
- 定义一个具有发言行为的辩手角色，详见前期课程
- 处理辩手之间的通信，也就是让拜登听特朗普说话，反之亦然
- 初始化两个辩手实例，拜登和特朗普，创建一个带有环境的团队，并使它们能够相互交互
插入模块
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
import asyncio
from metagpt.actions import Action,UserRequirement
from metagpt.schema import Message
定义动作
首先，我们需要定义一个 Action。这是一个辩论场景，所以让我们将其命名为 SpeakAloud
class SpeakAloud(Action):
"""动作：在辩论中大声说话（争吵）"""
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:
"""
def __init__(self, name="SpeakAloud", context=None, llm=None):
super().__init__(name, context, llm)
async def run(self, context: str, name: str, opponent_name: str):
prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)
rsp = await self._aask(prompt)
return rsp
定义角色
定义一个通用的 Role，称为 Debator
set_actions 使我们的 Role 拥有我们刚刚定义的 SpeakAloud 动作。我们还使用_watch 监视了 SpeakAloud 和 UserRequierment，因为我们希望每个辩手关注来自对手的 SpeakAloud 消息，以及来自用户的 UserRequirement (人类指令）。
class Debator(Role):
def __init__(
self,
name: str,
profile: str,
opponent_name: str,
**kwargs,
):
super().__init__(name, profile, **kwargs)
self.set_actions([SpeakAloud])
self._watch([UserRequirement, SpeakAloud])
self.name = name
self.opponent_name = opponent_name
下面通过重写_observe 函数，我们使每个辩手听取对手论点。这一步很重要，因为在环境中将会有来自特朗普和拜登的 "SpeakAloud 消息"（由 SpeakAloud 触发的 Message）。 我们不希望特朗普处理自己上一轮的 "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)
最后，我们使每个辩手能够向对手发送反驳的论点。在这里，我们从消息历史中构建一个上下文，使 Debator 运行他拥有的 SpeakAloud 动作，并使用反驳论点内容创建一个新的 Message。请注意，我们定义每个 Debator 将把 Message 发送给他的对手
async def _act(self) -> Message:
logger.info(f"{self._setting}: ready to {self.rc.todo}")
todo = self.rc.todo # 一个 SpeakAloud 的实例
memories = self.get_memories()
context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)
msg = Message(
content=rsp,
role=self.profile,
cause_by=todo,
sent_from=self.name,
send_to=self.opponent_name,
)
self.rc.memory.add(msg)
return msg
创建团队并添加角色
建立一个 Team 将角色动作组合起来，我们将通过将我们的指令（作为 UserRequirement）发送给拜登，让他先开始。如果你想让特朗普先说话，将 send_to 设置为 "Trump"。
运行 Team 就可以看到他们之间的对话！
async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
"""运行拜登-特朗普辩论，观看他们之间的友好对话 :) """
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") # 将辩论主题发送给拜登，让他先说话
await team.run(n_round=n_round)
import typer
app = typer.Typer()
@app.command()
def main(
idea: str = typer.Argument(..., help="Economic Policy: Discuss strategies and plans related to taxation, employment, fiscal budgeting, and economic growth."),
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."),
):
"""
: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__':
app()# run as python debate.py --idea="TOPIC" --investment=3.0 --n_round=5
但是，上述教程代码运行不了。运行以下代码：
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
import asyncio
from metagpt.actions import Action,UserRequirement
from metagpt.schema import Message
class SpeakAloud(Action):
"""动作：在辩论中大声说话（争吵）"""
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:
"""
def __init__(self, name="SpeakAloud", context=None, llm=None):
super().__init__(name=name, context=context, llm=llm)
async def run(self, context: str, name: str, opponent_name: str):
prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)
rsp = await self._aask(prompt)
return rsp
class Debator(Role):
def __init__(
self,
**kwargs,
):
super().__init__(**kwargs)
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 self.name in msg.send_to]
return len(self.rc.news)
async def _act(self) -> Message:
logger.info(f"{self._setting}: ready to {self.rc.todo}")
todo = self.rc.todo # 一个 SpeakAloud 的实例
memories = self.get_memories()
context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)
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,
)
return msg
async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
"""运行拜登-特朗普辩论，观看他们之间的友好对话 :) """
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") # 将辩论主题发送给拜登，让他先说话
await team.run(n_round=n_round)
import asyncio
import platform
import typer
from metagpt.team import Team
app = typer.Typer()
@app.command()
def main(
idea: str = typer.Argument(..., help="Economic Policy: Discuss strategies and plans related to taxation, employment, fiscal budgeting, and economic growth."),
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."),
):
"""
: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__':
app()
结果#
"关于政治的辩论"
2025-01-26 20:23:55.414 | INFO | metagpt.const:get_metagpt_package_root:29 - Package root set to D:\Code\agent_from_scratch
2025-01-26 20:23:59.574 | INFO | metagpt.team:invest:90 - Investment: $3.0.
2025-01-26 20:23:59.577 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
Mr. Trump, your claim that my policies are harmful to the economy is a distortion. My administration has created millions of jobs and cut taxes for the middle class. Your trickle-down economics failed us before, and it will again. We're investing in infrastructure, education, and clean energy, not just for today but for the future of our children. Your policies are a step backward, not forward.
2025-01-26 20:24:05.927 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:05.931 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
Ladies and gentlemen, Biden's claims are a smoke screen! My tax cuts unleashed economic growth, not just jobs, but prosperity for all. My infrastructure plan is bold and will create jobs like never before. We're not just building for today, but for a strong, secure future. Biden's policies are a recipe for disaster, not progress. We can't afford four more years of his failed socialism!
2025-01-26 20:24:10.524 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:10.527 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
"Mr. Trump, your tax cuts only widened the gap between the rich and the rest. True prosperity is about lifting everyone up, not just the elite. My infrastructure plan is about investing in the future, not just jobs but in our communities. My opponent's 'bold' plan is a myth; it's a shell game that leaves the middle class behind. We can't afford four more years of trickle-down economics that's failed us for decades."
2025-01-26 20:24:14.519 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:14.522 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time f2025-01-26 20:24:05.931 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
Ladies and gentlemen, Biden's claims are a smoke screen! My tax cuts unleashed economic growth, not just jobs, but prosperity for all. My infrastructure plan is bold and will create jobs like never before. We're not just building for today, but for a strong, secure future. Biden's policies are a recipe for disaster, not progress. We can't afford four more years of his failed socialism!
2025-01-26 20:24:10.524 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:10.527 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
"Mr. Trump, your tax cuts only widened the gap between the rich and the rest. True prosperity is about lifting everyone up, not just the elite. My infrastructure plan is about investing in the future, not just jobs but in our communities. My opponent's 'bold' plan is a myth; it's a shell game that leaves the middle class behind. We can't afford four more years of trickle-down economics that's failed us for decades."
2025-01-26 20:24:14.519 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:14.522 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time fLadies and gentlemen, Biden's claims are a smoke screen! My tax cuts unleashed economic growth, not just jobs, but prosperity for all. My infrastructure plan is bold and will create jobs like never before. We're not just building for today, but for a strong, secure future. Biden's policies are a recipe for disaster, not progress. We can't afford four more years of his failed socialism!
2025-01-26 20:24:10.524 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:10.527 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
"Mr. Trump, your tax cuts only widened the gap between the rich and the rest. True prosperity is about lifting everyone up, not just the elite. My infrastructure plan is about investing in the future, not just jobs but in our communities. My opponent's 'bold' plan is a myth; it's a shell game that leaves the middle class behind. We can't afford four more years of trickle-down economics that's failed us for decades."
2025-01-26 20:24:14.519 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:14.522 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time f a recipe for disaster, not progress. We can't afford four more years of his failed socialism!
2025-01-26 20:24:10.524 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:10.527 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
"Mr. Trump, your tax cuts only widened the gap between the rich and the rest. True prosperity is about lifting everyone up, not just the elite. My infrastructure plan is about investing in the future, not just jobs but in our communities. My opponent's 'bold' plan is a myth; it's a shell game that leaves the middle class behind. We can't afford four more years of trickle-down economics that's failed us for decades."
2025-01-26 20:24:14.519 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:14.522 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time f"Mr. Trump, your tax cuts only widened the gap between the rich and the rest. True prosperity is about lifting everyone up, not just the elite. My infrastructure plan is about investing in the future, not just jobs but in our communities. My opponent's 'bold' plan is a myth; it's a shell game that leaves the middle class behind. We can't afford four more years of trickle-down economics that's failed us for decades."
2025-01-26 20:24:14.519 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:14.522 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time fure, not just jobs but in our communities. My opponent's 'bold' plan is a myth; it's a shell game that leaves the middle class behind. We can't afford four more years of trickle-down economics that's failed us for decades."
2025-01-26 20:24:14.519 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:14.522 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time for a leader who knows how to build and win, not just talk. Your 'prosperity' is a mirage; my America is the real deal!"
2025-01-26 20:24:19.399 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:14.522 | INFO | __main__:_act:57 - Trump(Republican): ready to SpeakAloud
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time for a leader who knows how to build and win, not just talk. Your 'prosperity' is a mirage; my America is the real deal!"
2025-01-26 20:24:19.399 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
"Mr. Biden, your policies are a recipe for stagnation. My tax cuts unleashed the American spirit, creating jobs and wealth for all. You talk infrastructure, but where's the money coming from? My plan invests in our people, not just in the ground. You've been in Washington too long; it's time for a leader who knows how to build and win, not just talk. Your 'prosperity' is a mirage; my America is the real deal!"
2025-01-26 20:24:19.399 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
or a leader who knows how to build and win, not just talk. Your 'prosperity' is a mirage; my America is the real deal!"
2025-01-26 20:24:19.399 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:24:19.404 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
Mr. Trump, your tax cuts were a windfall for the wealthy, not the middle class. My infrastructure not found in TOKEN_COSTS.
2025-01-26 20:24:19.404 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
Mr. Trump, your tax cuts were a windfall for the wealthy, not the middle class. My infrastructure plan is funded responsibly, ensuring every American benefits. You've left behind the Rust Belt, wh2025-01-26 20:24:19.404 | INFO | __main__:_act:57 - Biden(Democrat): ready to SpeakAloud
Mr. Trump, your tax cuts were a windfall for the wealthy, not the middle class. My infrastructure plan is funded responsibly, ensuring every American benefits. You've left behind the Rust Belt, whMr. Trump, your tax cuts were a windfall for the wealthy, not the middle class. My infrastructure plan is funded responsibly, ensuring every American benefits. You've left behind the Rust Belt, while I'll rebuild it. Your 'America' left millions behind. It's time for a leader who'll invest in plan is funded responsibly, ensuring every American benefits. You've left behind the Rust Belt, while I'll rebuild it. Your 'America' left millions behind. It's time for a leader who'll invest in our future, not just the past. Your prosperity is a mirage; my vision is a reality for all.
ile I'll rebuild it. Your 'America' left millions behind. It's time for a leader who'll invest in our future, not just the past. Your prosperity is a mirage; my vision is a reality for all.
our future, not just the past. Your prosperity is a mirage; my vision is a reality for all.
2025-01-26 20:24:23.742 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
单动作多智能体 - 写诗#
现在我们设定，需要多智能体系统为我们根据我们给定的主题提供一篇优美的英文诗，除了完成写作的 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.set_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.set_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'))
await main(topic='write a poem about moon')
结果#
2025-01-26 20:38:15.640 | INFO | __main__:_act:12 - xiaoming(Student): ready to WritePoem
Glimmering in the velvet night,
The moon ascends, a silver knight.
Its glow, a gentle, silken light,
Adorns the world with soft delight.
A crescent sliver, a full orb's grace,
It dances in the cosmic dance.
A beacon in the dark, it guides our place,
Through shadows, it brings us to our senses.
A celestial ball, it waxes and wanes,
A silent sentinel, it keeps its own pace.
In its quiet glow, it whispers tales,
Of ancient skies and timeless space.
2025-01-26 20:38:22.098 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:38:22.100 | INFO | __main__:_act:18 - student : Glimmering in the velvet night,
The moon ascends, a silver knight.
Its glow, a gentle, silken light,
Adorns the world with soft delight.
A crescent sliver, a full orb's grace,
It dances in the cosmic dance.
A beacon in the dark, it guides our place,
Through shadows, it brings us to our senses.
A celestial ball, it waxes and wanes,
A silent sentinel, it keeps its own pace.
In its quiet glow, it whispers tales,
Of ancient skies and timeless space.
2025-01-26 20:38:22.103 | INFO | __main__:_act:35 - laowang(Teacher): ready to ReviewPoem
- The opening line sets a beautiful scene with "Glimmering in the velvet night."
- "The moon ascends, a silver knight" is a creative and poetic way to describe the moon's rise.
- "Its glow, a gentle, silken light" is a lovely simile that enhances the moon's ethereal quality.
- "Adorns the world with soft delight" is a poetic touch that adds depth to the moon's impact.
- The second stanza flows well, with "A crescent sliver, a full orb's grace" being particularly vivid.
- "It dances in the cosmic dance" is a charming way to describe the moon's movement.
- "A beacon in the dark, it guides our place" is a powerful image that conveys the moon's navigational role.
- "Through shadows, it brings us to our senses" is a thoughtful line that suggests the moon's ability to connect us with our surroundings.
- The third stanza, "A celestial ball, it waxes and wanes," is a straightforward yet effective description.
- "A silent sentinel, it keeps its own pace" is a poetic way to describe the moon's steady, unchanging nature.
- "In its quiet glow, it whispers tales" is a beautiful line that evokes the moon's mysteriousness.
- Overall, the poem is well-crafted with a retro style and elegant sentences.
2025-01-26 20:38:29.898 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:38:29.899 | INFO | __main__:_act:40 - teacher : - The opening line sets a beautiful scene with "Glimmering in the velvet night."
- "The moon ascends, a silver knight" is a creative and poetic way to describe the moon's rise.
- "Its glow, a gentle, silken light" is a lovely simile that enhances the moon's ethereal quality.
- "Adorns the world with soft delight" is a poetic touch that adds depth to the moon's impact.
- The second stanza flows well, with "A crescent sliver, a full orb's grace" being particularly vivid.
- "It dances in the cosmic dance" is a charming way to describe the moon's movement.
- "A beacon in the dark, it guides our place" is a powerful image that conveys the moon's navigational role.
- "Through shadows, it brings us to our senses" is a thoughtful line that suggests the moon's ability to connect us with our surroundings.
- The third stanza, "A celestial ball, it waxes and wanes," is a straightforward yet effective description.
- "A silent sentinel, it keeps its own pace" is a poetic way to describe the moon's steady, unchanging nature.
- "In its quiet glow, it whispers tales" is a beautiful line that evokes the moon's mysteriousness.
- Overall, the poem is well-crafted with a retro style and elegant sentences.
2025-01-26 20:38:29.903 | INFO | __main__:_act:12 - xiaoming(Student): ready to WritePoem
Glimmering in the velvet night,
The moon ascends, a silver knight.
Its glow, a gentle, silken light,
Adorns the world with soft delight.
A crescent sliver, a full orb's grace,
It dances in the cosmic dance.
A beacon in the dark, it guides our place,
Through shadows, it brings us to our senses.
A celestial ball, it waxes and wanes,
A silent sentinel, it keeps its own pace.
In its quiet glow, it whispers tales,
Of ancient skies and timeless space.
2025-01-26 20:38:33.066 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 20:38:33.069 | INFO | __main__:_act:18 - student : Glimmering in the velvet night,
The moon ascends, a silver knight.
Its glow, a gentle, silken light,
Adorns the world with soft delight.
A crescent sliver, a full orb's grace,
It dances in the cosmic dance.
A beacon in the dark, it guides our place,
Through shadows, it brings us to our senses.
A celestial ball, it waxes and wanes,
A silent sentinel, it keeps its own pace.
In its quiet glow, it whispers tales,
Of ancient skies and timeless space.
多动作多智能体#
在上一章中，我们简要讨论了单智能体的创建。虽然对许多情况来说，单智能体可能已经足够，但更复杂的任务通常需要协作和团队合作，这也就是多智能体为什么必不可少的原因。MetaGPT 的核心优势也在于轻松灵活地开发一个智能体团队。在 MetaGPT 框架下，用户可以通过少量代码实现智能体之间的交互。
完成本节，你将能够：
- 理解智能体之间如何进行交互
- 开发你的第一个智能体团队
运行 “软件公司” 示例#
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 框架下，用户可以通过少量代码实现智能体之间的交互。
完成本节，你将能够：
- 理解智能体之间如何进行交互
- 开发你的第一个智能体团队
运行 “软件公司” 示例#
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
# 创建一个异步的主函数
async def async_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)
return team
# 在 Jupyter 中直接运行这个
await async_main(
idea="write a function that calculates the product of a list",
investment=3.0,
n_round=5
)
结果#
多动作多智能体实例 - 狼人杀#
这节课来学习 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 , 终端会打印出对局相关消息。
我是在 kaggle 上做的：
- ! git clone https://github.com/geekan/MetaGPT.git
- 配置 config
3. 下载需要的包
! pip install -r /kaggle/working/MetaGPT/requirements.txt
!pip install llama-index-retrievers-bm25
!pip install llama_index.embeddings.azure_openai
!pip install llama_index.embeddings.gemini
!pip install llama_index.embeddings.ollama
!pip install llama_index
!pip install --upgrade llama_index
!pip install llama_index.vector_stores.faiss
!pip install llama_index.vector_stores.milvus
!pip install llama-index-vector-stores-elasticsearch~=0.2.5 # Used by `metagpt/memory/longterm_memory.py`
!pip install llama-index-vector-stores-chroma~=0.1.10 # Used by `metagpt/memory/longterm_memory.py`
!pip install chromadb
这里补了非常多包，可能有遗漏。
4. 运行
file_path = '/kaggle/working/MetaGPT/examples/werewolf_game/start_game.py'
with open(file_path, 'r') as file:
content = file.read()
new_code = """import sys
import os
# 获取包含 metagpt 的目录路径
metagpt_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
sys.path.append(metagpt_dir)
"""
new_content = new_code + content
with open(file_path, 'w') as file:
file.write(new_content)
!python3 /kaggle/working/MetaGPT/examples/werewolf_game/start_game.py
结果#
时间非常长，轮次非常多。建议拿免费的 api 运行。
仅展示一部分：
斯坦福虚拟小镇#
https://github.com/geekan/MetaGPT/tree/main/metagpt/ext/stanford_town
记录一下运行过的操作，留个坑。
在 kaggle 上运行：
- ! git clone https://github.com/geekan/MetaGPT.git
- ! pip install -r /kaggle/working/MetaGPT/requirements.txt
- 添加搜索路径
file_path = '/kaggle/working/MetaGPT/examples/stanford_town/run_st_game.py'
with open(file_path, 'r') as file:
content = file.read()
new_code = """import sys
import os
# 获取包含 metagpt 的目录路径
metagpt_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
sys.path.append(metagpt_dir)
"""
new_content = new_code + content
with open(file_path, 'w') as file:
file.write(new_content)
3. 改变 GA 存储位置
from pathlib import Path
file_path = Path('/kaggle/working/MetaGPT/metagpt/const.py')
# 读取文件内容
with open(file_path, 'r') as file:
lines = file.readlines()
# 定义新的存储路径
storage_path = Path('/kaggle/working/ga_storage')
temp_storage_path = Path('/kaggle/working/ga_temp_storage')
# 查找并替换目标行
new_lines = []
for line in lines:
if line.startswith('STORAGE_PATH = ROOT_PATH.joinpath("storage")'):
new_lines.append(f'STORAGE_PATH = {storage_path!r}\n')
elif line.startswith('TEMP_STORAGE_PATH = ROOT_PATH.joinpath("temp_storage")'):
new_lines.append(f'TEMP_STORAGE_PATH = {temp_storage_path!r}\n')
else:
new_lines.append(line)
# 写入修改后的内容
with open(file_path, 'w') as file:
file.writelines(new_lines)
4. 配置 config
import yaml
# 定义 YAML 文件路径，你需根据实际情况修改
yaml_file_path = '/kaggle/working/MetaGPT/config/config2.yaml'
# 定义要写入的 YAML 数据
new_yaml_data = {
"llm": {
"api_type": ,
"api_key": ,
"model":
}
}
# 清空文件内容
with open(yaml_file_path, 'w') as file:
file.truncate()
# 将新的 YAML 数据写入文件
with open(yaml_file_path, 'w') as file:
yaml.dump(new_yaml_data, file, default_flow_style=False)
print("YAML 文件已清空并成功更新")
最好是 OPENAI 的
5. 注意 embedding 环节，要适配：
embedding 长这样。
def get_embedding(text, model: str = "text-embedding-ada-002"):
text = text.replace("\n", " ")
embedding = None
if not text:
text = "this is blank"
for idx in range(3):
try:
embedding = (
OpenAI(api_key=config.llm.api_key).embeddings.create(input=[text], model=model).data[0].embedding
)
except Exception as exp:
logger.info(f"get_embedding failed, exp: {exp}, will retry.")
time.sleep(5)
if not embedding:
raise ValueError("get_embedding failed")
return embedding
由于我还没有可以用的 API，所以这一步没有做。
到目前的输出：