MetaGPT中的教程助手：TutorialAssistant
1. 提示词
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.
"""
)
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}".
"""
)
2. 两个动作： 写文档大纲 (WriteDirectory) 和文档内容 (WriteContent)
from typing import Dict
from metagpt.actions import Action
from metagpt.utils.common import OutputParser
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"]}]}.
"""
prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
resp = await self._aask(prompt=prompt)
return OutputParser.extract_struct(resp, dict)
class WriteContent(Action):
"""Action class for writing tutorial content.
Args:
name: The name of the action.
directory: The content to write.
language: The language to output, default is "Chinese".
"""
name: str = "WriteContent"
directory: dict = dict()
language: str = "Chinese"
async def run(self, topic: str, *args, **kwargs) -> str:
"""Execute the action to write document content according to the directory and topic.
Args:
topic: The tutorial topic.
Returns:
The written tutorial content.
"""
prompt = CONTENT_PROMPT.format(topic=topic, language=self.language, directory=self.directory)
return await self._aask(prompt=prompt)
3. 教程助手定义
具体逻辑：第一步写文档目录，第二步，根据文档目录写每一章的内容，第三步，将内容写入markdown文件中
from datetime import datetime
from typing import Dict
from metagpt.const import get_metagpt_root
from metagpt.logs import logger
from metagpt.roles.role import Role
from metagpt.schema import Message
from metagpt.utils.file import File
TUTORIAL_PATH = get_metagpt_root()/"data"/"tutorial_docx"
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="by_order")
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(self.actions)
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.max_react_loop = len(self.actions)
return Message()
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:
msg = await super().react()
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"))
msg.content = str(root_path / f"{self.main_title}.md")
return msg
2024-12-12 15:39:46.005 | INFO | metagpt.const:get_metagpt_package_root:21 - Package root set to d:\llm\metagpt
4. 示例
async def main():
topic = "Write a tutorial about python asyncio"
role = TutorialAssistant(language="Chinese")
await role.run(topic)
await main()
输出：
{
"title": "Python Asyncio 异步编程教程",
"directory": [
{
"一、异步编程基础": [
"1.1 异步编程概念",
"1.2 Python中的异步历史",
"1.3 异步编程的优势"
]
},
{
"二、Asyncio入门": [
"2.1 安装与配置",
"2.2 Event Loop事件循环",
"2.3 coroutine协程",
"2.4 Future对象",
"2.5 使用async和await"
]
},
{
"三、Asyncio高级应用": [
"3.1 并发执行与同步",
"3.2 异步网络请求",
"3.3 异步文件操作",
"3.4 异步数据库操作",
"3.5 异步队列"
]
},
{
"四、常见问题与解决方案": [
"4.1 协程间通信",
"4.2 异步异常处理",
"4.3 性能优化",
"4.4 兼容性问题"
]
},
{
"五、实战案例": [
"5.1 异步爬虫",
"5.2 异步Web服务",
"5.3 异步数据处理"
]
},
{
"六、附录": [
"6.1 常用库与资源",
"6.2 最佳实践",
"6.3 进一步学习"
]
}
]
}
2024-12-12 15:39:57.769 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4 not found in TOKEN_COSTS.
2024-12-12 15:39:57.770 | INFO | __main__:_act:72 - {'title': 'Python Asyncio 异步编程教程', 'directory': [{'一、异步编程基础': ['1.1 异步编程概念', '1.2 Python中的异步历史', '1.3 异步编程的优势']}, {'二、Asyncio入门': ['2.1 安装与配置', '2.2 Event Loop事件循环', '2.3 coroutine协程', '2.4 Future对象', '2.5 使用async和await']}, {'三、Asyncio高级应用': ['3.1 并发执行与同步', '3.2 异步网络请求', '3.3 异步文件操作', '3.4 异步数据库操作', '3.5 异步队列']}, {'四、常见问题与解决方案': ['4.1 协程间通信', '4.2 异步异常处理', '4.3 性能优化', '4.4 兼容性问题']}, {'五、实战案例': ['5.1 异步爬虫', '5.2 异步Web服务', '5.3 异步数据处理']}, {'六、附录': ['6.1 常用库与资源', '6.2 最佳实践', '6.3 进一步学习']}]}
# 一、异步编程基础
## 1.1 异步编程概念
异步编程是一种编程范式，允许程序在等待某些操作完成（如I/O操作）时继续执行其他任务。与传统的同步编程不同，同步编程在等待操作完成时会阻塞当前任务的执行。异步编程可以提高程序的效率和响应性。
## 1.2 Python中的异步历史
Python的异步编程经历了几个阶段的发展：
- Python 2.x时代：主要通过`threading`和`multiprocessing`模块实现并发。
- Python 3.0至3.3：引入了`yield`关键字和`generators`，可以通过生成器实现简单的协同程序。
- Python 3.4：引入了`asyncio`库，提供了编写单线程并发代码的框架。
- Python 3.5：引入了`async`和`await`关键字，使得异步编程更加直观和易于编写。
## 1.3 异步编程的优势
异步编程具有以下优势：
1. 提高I/O密集型应用的性能：在等待I/O操作完成时，CPU可以切换到其他任务，从而提高资源利用率。
2. 更好的资源利用：异步编程允许使用单线程管理多个并发任务，减少了线程或进程的开销。
3. 提升用户体验：在Web应用和用户界面中，异步编程可以避免因长时间操作导致的界面冻结。
以下是一个简单的使用`asyncio`的例子：
```python
import asyncio
async def hello_world():
print("Hello")
await asyncio.sleep(1) # 模拟I/O操作
print("World")
# 运行异步函数
asyncio.run(hello_world())
```
请注意，上述代码应在Python 3.7及以上版本运行，因为`asyncio.run()`函数是在Python 3.7中引入的。在之前的版本中，需要使用`asyncio.get_event_loop().run_until_complete(hello_world())`来运行异步函数
2024-12-12 15:40:15.460 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4 not found in TOKEN_COSTS.
2024-12-12 15:40:15.462 | INFO | __main__:_act:75 - # 一、异步编程基础
## 1.1 异步编程概念
异步编程是一种编程范式，允许程序在等待某些操作完成（如I/O操作）时继续执行其他任务。与传统的同步编程不同，同步编程在等待操作完成时会阻塞当前任务的执行。异步编程可以提高程序的效率和响应性。
## 1.2 Python中的异步历史
Python的异步编程经历了几个阶段的发展：
- Python 2.x时代：主要通过`threading`和`multiprocessing`模块实现并发。
- Python 3.0至3.3：引入了`yield`关键字和`generators`，可以通过生成器实现简单的协同程序。
- Python 3.4：引入了`asyncio`库，提供了编写单线程并发代码的框架。
- Python 3.5：引入了`async`和`await`关键字，使得异步编程更加直观和易于编写。
## 1.3 异步编程的优势
异步编程具有以下优势：
1. 提高I/O密集型应用的性能：在等待I/O操作完成时，CPU可以切换到其他任务，从而提高资源利用率。
2. 更好的资源利用：异步编程允许使用单线程管理多个并发任务，减少了线程或进程的开销。
3. 提升用户体验：在Web应用和用户界面中，异步编程可以避免因长时间操作导致的界面冻结。
以下是一个简单的使用`asyncio`的例子：
```python
import asyncio
async def hello_world():
print("Hello")
await asyncio.sleep(1) # 模拟I/O操作
print("World")
# 运行异步函数
asyncio.run(hello_world())
```
请注意，上述代码应在Python 3.7及以上版本运行，因为`asyncio.run()`函数是在Python 3.7中引入的。在之前的版本中，需要使用`asyncio.get_event_loop().run_until_complete(hello_world())`来运行异步函数。
。
# 二、Asyncio入门
## 2.1 安装与配置
Asyncio 是 Python 的一个库，用于编写单线程并发代码，通过异步/等待语法。要在 Python 中使用 asyncio，你需要 Python 3.5 或更高版本。
### 安装
Asyncio 随 Python 3.4+ 一起安装，无需额外安装。
### 配置
无需特别的配置，直接在 Python 脚本中导入即可使用。
```python
import asyncio
```
## 2.2 Event Loop事件循环
事件循环是 asyncio 程序的核心，它用来调度和执行任务。
```python
# 示例：创建事件循环
loop = asyncio.get_event_loop()
try:
loop.run_forever()
finally:
loop.close()
```
## 2.3 coroutine协程
协程是使用 `async def` 定义的函数，它是异步编程的核心。
```python
# 示例：定义协程
async def main():
print('Hello')
await asyncio.sleep(1)
print('World')
# 运行协程
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 2.4 Future对象
Future 对象表示尚未完成的计算，它用来与异步执行的结果进行交互。
```python
# 示例：创建 Future 对象
async def main():
# 获取当前事件循环
loop = asyncio.get_event_loop()
# 创建一个 Future 对象
future = loop.create_future()
# 模拟异步操作，并在一段时间后设置结果
async def set_future():
await asyncio.sleep(1)
future.set_result('Future is done!')
# 启动任务
loop.create_task(set_future())
# 等待 Future 完成
result = await future
print(result)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 2.5 使用async和await
`async` 和 `await` 关键字用于编写异步代码。
```python
# 示例：使用 async 和 await
async def say_after(delay, what):
await asyncio.sleep(delay)
print(what)
async def main():
print(f"started at {time.strftime('%X')}")
await say_after(1, 'hello')
await say_after(2, 'world')
print(f"finished at {time.strftime('%X')}")
asyncio.run(main())
```
请注意，上面的示例中使用了 `asyncio.run(main())`，这是 Python 3.7+ 推荐的运行协程的方式，代替了手动获取事件循环和关闭事件循环的做法。`asyncio.run()` 应该只被调用一次，通常在主程序入口处。
2024-12-12 15:40:36.272 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4 not found in TOKEN_COSTS.
2024-12-12 15:40:36.274 | INFO | __main__:_act:75 - # 二、Asyncio入门
## 2.1 安装与配置
Asyncio 是 Python 的一个库，用于编写单线程并发代码，通过异步/等待语法。要在 Python 中使用 asyncio，你需要 Python 3.5 或更高版本。
### 安装
Asyncio 随 Python 3.4+ 一起安装，无需额外安装。
### 配置
无需特别的配置，直接在 Python 脚本中导入即可使用。
```python
import asyncio
```
## 2.2 Event Loop事件循环
事件循环是 asyncio 程序的核心，它用来调度和执行任务。
```python
# 示例：创建事件循环
loop = asyncio.get_event_loop()
try:
loop.run_forever()
finally:
loop.close()
```
## 2.3 coroutine协程
协程是使用 `async def` 定义的函数，它是异步编程的核心。
```python
# 示例：定义协程
async def main():
print('Hello')
await asyncio.sleep(1)
print('World')
# 运行协程
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 2.4 Future对象
Future 对象表示尚未完成的计算，它用来与异步执行的结果进行交互。
```python
# 示例：创建 Future 对象
async def main():
# 获取当前事件循环
loop = asyncio.get_event_loop()
# 创建一个 Future 对象
future = loop.create_future()
# 模拟异步操作，并在一段时间后设置结果
async def set_future():
await asyncio.sleep(1)
future.set_result('Future is done!')
# 启动任务
loop.create_task(set_future())
# 等待 Future 完成
result = await future
print(result)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 2.5 使用async和await
`async` 和 `await` 关键字用于编写异步代码。
```python
# 示例：使用 async 和 await
async def say_after(delay, what):
await asyncio.sleep(delay)
print(what)
async def main():
print(f"started at {time.strftime('%X')}")
await say_after(1, 'hello')
await say_after(2, 'world')
print(f"finished at {time.strftime('%X')}")
asyncio.run(main())
```
请注意，上面的示例中使用了 `asyncio.run(main())`，这是 Python 3.7+ 推荐的运行协程的方式，代替了手动获取事件循环和关闭事件循环的做法。`asyncio.run()` 应该只被调用一次，通常在主程序入口处。
# 三、Asyncio高级应用
## 3.1 并发执行与同步
`asyncio` 是 Python 的一个异步I/O框架，用于编写单线程并发代码，通过异步/等待语法。
### 并发执行
`asyncio` 允许我们并发地运行多个协程。以下是一个简单的并发执行示例：
```python
import asyncio
async def say_after(delay, what):
await asyncio.sleep(delay)
print(what)
async def main():
print(f"started at {time.strftime('%X')}")
# 创建两个协程对象，不立即运行
coroutine1 = say_after(1, 'hello')
coroutine2 = say_after(2, 'world')
# 等待两个协程并发执行
await asyncio.gather(coroutine1, coroutine2)
print(f"finished at {time.strftime('%X')}")
# 运行主协程
asyncio.run(main())
```
### 同步
在并发编程中，同步是确保多个任务按预期执行的机制。`asyncio` 提供了 `asyncio.Lock()` 和 `asyncio.Event()` 等同步原语。
以下是一个使用 `asyncio.Lock()` 的示例：
```python
import asyncio
async def worker(lock, worker_id):
print(f"Worker {worker_id} waiting for the lock")
async with lock:
print(f"Worker {worker_id} has the lock")
await asyncio.sleep(1)
print(f"Worker {worker_id} released the lock")
async def main():
# 创建一个锁
lock = asyncio.Lock()
# 创建并运行三个工作协程
tasks = [asyncio.create_task(worker(lock, i)) for i in range(3)]
await asyncio.gather(*tasks)
# 运行主协程
asyncio.run(main())
```
## 3.2 异步网络请求
`aiohttp` 是一个支持 `asyncio` 的异步HTTP网络库。
```python
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
html = await fetch(session, 'http://python.org')
print(html)
# 运行主协程
asyncio.run(main())
```
## 3.3 异步文件操作
`asyncio` 允许你以异步方式执行文件操作。
```python
import asyncio
import aiofiles
async def write_to_file(filename, content):
async with aiofiles.open(filename, 'w') as f:
await f.write(content)
async def main():
await write_to_file('example.txt', 'Hello, world!')
# 运行主协程
asyncio.run(main())
```
## 3.4 异步数据库操作
对于数据库操作，可以使用像 `aiomysql` 或 `aiopg` 这样的异步驱动。
以下是使用 `aiomysql` 的简单示例：
```python
import asyncio
import aiomysql
async def query_db(loop, dbname, user, password):
async with aiomysql.create_pool(host='127.0.0.1', port=3306,
user=user, password=password,
db=dbname, loop=loop) as pool:
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT 1")
result = await cur.fetchone()
print(result)
loop = asyncio.get_event_loop()
loop.run_until_complete(query_db(loop, 'test', 'user', 'password'))
```
## 3.5 异步队列
`asyncio` 提供了 `asyncio.Queue`，这是一个异步的先进先出（FIFO）队列。
```python
import asyncio
async def worker(q):
while True:
item = await q.get()
print(f'Worker got item: {item}')
# 模拟处理项目
await asyncio.sleep(1)
q.task_done()
async def main():
q = asyncio.Queue(maxsize=3)
# 启动三个工作协程
tasks = []
for i in range(3):
task = asyncio.create_task(worker(q))
tasks.append(task)
# 生产项目
for item in range(10):
await q.put(item)
# 等待队列清空
await q.join()
# 取消工作协程
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
# 运行主协程
asyncio.run(main())
```
请注意，以上示例代码遵循了代码规范，并包含了必要的注释，以方便理解。
2024-12-12 15:41:09.514 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4 not found in TOKEN_COSTS.
2024-12-12 15:41:09.515 | INFO | __main__:_act:75 - # 三、Asyncio高级应用
## 3.1 并发执行与同步
`asyncio` 是 Python 的一个异步I/O框架，用于编写单线程并发代码，通过异步/等待语法。
### 并发执行
`asyncio` 允许我们并发地运行多个协程。以下是一个简单的并发执行示例：
```python
import asyncio
async def say_after(delay, what):
await asyncio.sleep(delay)
print(what)
async def main():
print(f"started at {time.strftime('%X')}")
# 创建两个协程对象，不立即运行
coroutine1 = say_after(1, 'hello')
coroutine2 = say_after(2, 'world')
# 等待两个协程并发执行
await asyncio.gather(coroutine1, coroutine2)
print(f"finished at {time.strftime('%X')}")
# 运行主协程
asyncio.run(main())
```
### 同步
在并发编程中，同步是确保多个任务按预期执行的机制。`asyncio` 提供了 `asyncio.Lock()` 和 `asyncio.Event()` 等同步原语。
以下是一个使用 `asyncio.Lock()` 的示例：
```python
import asyncio
async def worker(lock, worker_id):
print(f"Worker {worker_id} waiting for the lock")
async with lock:
print(f"Worker {worker_id} has the lock")
await asyncio.sleep(1)
print(f"Worker {worker_id} released the lock")
async def main():
# 创建一个锁
lock = asyncio.Lock()
# 创建并运行三个工作协程
tasks = [asyncio.create_task(worker(lock, i)) for i in range(3)]
await asyncio.gather(*tasks)
# 运行主协程
asyncio.run(main())
```
## 3.2 异步网络请求
`aiohttp` 是一个支持 `asyncio` 的异步HTTP网络库。
```python
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
html = await fetch(session, 'http://python.org')
print(html)
# 运行主协程
asyncio.run(main())
```
## 3.3 异步文件操作
`asyncio` 允许你以异步方式执行文件操作。
```python
import asyncio
import aiofiles
async def write_to_file(filename, content):
async with aiofiles.open(filename, 'w') as f:
await f.write(content)
async def main():
await write_to_file('example.txt', 'Hello, world!')
# 运行主协程
asyncio.run(main())
```
## 3.4 异步数据库操作
对于数据库操作，可以使用像 `aiomysql` 或 `aiopg` 这样的异步驱动。
以下是使用 `aiomysql` 的简单示例：
```python
import asyncio
import aiomysql
async def query_db(loop, dbname, user, password):
async with aiomysql.create_pool(host='127.0.0.1', port=3306,
user=user, password=password,
db=dbname, loop=loop) as pool:
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT 1")
result = await cur.fetchone()
print(result)
loop = asyncio.get_event_loop()
loop.run_until_complete(query_db(loop, 'test', 'user', 'password'))
```
## 3.5 异步队列
`asyncio` 提供了 `asyncio.Queue`，这是一个异步的先进先出（FIFO）队列。
```python
import asyncio
async def worker(q):
while True:
item = await q.get()
print(f'Worker got item: {item}')
# 模拟处理项目
await asyncio.sleep(1)
q.task_done()
async def main():
q = asyncio.Queue(maxsize=3)
# 启动三个工作协程
tasks = []
for i in range(3):
task = asyncio.create_task(worker(q))
tasks.append(task)
# 生产项目
for item in range(10):
await q.put(item)
# 等待队列清空
await q.join()
# 取消工作协程
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
# 运行主协程
asyncio.run(main())
```
请注意，以上示例代码遵循了代码规范，并包含了必要的注释，以方便理解。
# 四、常见问题与解决方案
## 4.1 协程间通信
在Python的`asyncio`中，协程之间的通信通常使用`asyncio.Queue`或者`asyncio.Event`来实现。
### 示例：使用`asyncio.Queue`
```python
import asyncio
async def producer(queue):
for i in range(5):
print(f"Producing item {i}")
await queue.put(i)
await queue.put(None) # 用来标记生产结束
async def consumer(queue):
while True:
item = await queue.get()
if item is None:
break
print(f"Consuming item {item}")
await asyncio.sleep(1) # 模拟消费时间
queue.task_done()
queue = asyncio.Queue()
producer_coro = producer(queue)
consumer_coro = consumer(queue)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(producer_coro, consumer_coro))
loop.close()
```
## 4.2 异步异常处理
在异步编程中，异常处理非常重要。可以使用`try-except`块来捕获和处理异常。
### 示例：异步异常处理
```python
import asyncio
async def async_function_with_error():
raise ValueError("An error occurred")
async def main():
try:
await async_function_with_error()
except ValueError as e:
print(f"Caught an exception: {e}")
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 4.3 性能优化
性能优化通常涉及减少协程之间的等待时间，以及减少不必要的上下文切换。
- 使用`asyncio.gather`来并发运行多个协程。
- 使用`await`关键字仅在你确实需要等待某个操作完成时。
### 示例：并发运行多个协程
```python
import asyncio
async def fast_function():
await asyncio.sleep(1)
async def main():
await asyncio.gather(
fast_function(),
fast_function(),
fast_function()
)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 4.4 兼容性问题
由于`asyncio`是Python 3.4+的一部分，如果你需要支持更早的Python版本，可能会遇到兼容性问题。
- 使用`asyncio`之前，检查Python版本。
- 使用`pip`安装`asyncio`的第三方包，如`aiolib`，来支持旧版本的Python。
确保你的代码在支持的版本范围内运行良好，并且遵循最佳实践来减少兼容性问题。
---
请注意，以上内容遵循了Markdown的语法规范，并提供了相应的代码示例。代码示例遵循了标准的Python代码规范，并包含了必要的
2024-12-12 15:41:28.223 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4 not found in TOKEN_COSTS.
注释。
2024-12-12 15:41:28.226 | INFO | __main__:_act:75 - # 四、常见问题与解决方案
## 4.1 协程间通信
在Python的`asyncio`中，协程之间的通信通常使用`asyncio.Queue`或者`asyncio.Event`来实现。
### 示例：使用`asyncio.Queue`
```python
import asyncio
async def producer(queue):
for i in range(5):
print(f"Producing item {i}")
await queue.put(i)
await queue.put(None) # 用来标记生产结束
async def consumer(queue):
while True:
item = await queue.get()
if item is None:
break
print(f"Consuming item {item}")
await asyncio.sleep(1) # 模拟消费时间
queue.task_done()
queue = asyncio.Queue()
producer_coro = producer(queue)
consumer_coro = consumer(queue)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(producer_coro, consumer_coro))
loop.close()
```
## 4.2 异步异常处理
在异步编程中，异常处理非常重要。可以使用`try-except`块来捕获和处理异常。
### 示例：异步异常处理
```python
import asyncio
async def async_function_with_error():
raise ValueError("An error occurred")
async def main():
try:
await async_function_with_error()
except ValueError as e:
print(f"Caught an exception: {e}")
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 4.3 性能优化
性能优化通常涉及减少协程之间的等待时间，以及减少不必要的上下文切换。
- 使用`asyncio.gather`来并发运行多个协程。
- 使用`await`关键字仅在你确实需要等待某个操作完成时。
### 示例：并发运行多个协程
```python
import asyncio
async def fast_function():
await asyncio.sleep(1)
async def main():
await asyncio.gather(
fast_function(),
fast_function(),
fast_function()
)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()
```
## 4.4 兼容性问题
由于`asyncio`是Python 3.4+的一部分，如果你需要支持更早的Python版本，可能会遇到兼容性问题。
- 使用`asyncio`之前，检查Python版本。
- 使用`pip`安装`asyncio`的第三方包，如`aiolib`，来支持旧版本的Python。
确保你的代码在支持的版本范围内运行良好，并且遵循最佳实践来减少兼容性问题。
---
请注意，以上内容遵循了Markdown的语法规范，并提供了相应的代码示例。代码示例遵循了标准的Python代码规范，并包含了必要的注释。
# 五、实战案例
## 5.1 异步爬虫
异步IO在爬虫领域有广泛的应用，因为它可以显著提高爬取效率，特别是在处理网络请求时。以下是一个使用`asyncio`和`aiohttp`的异步爬虫示例。
```python
import asyncio
import aiohttp
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
url = 'http://example.com'
html = await fetch(session, url)
print(html)
if __name__ == '__main__':
asyncio.run(main())
```
在这个示例中，`fetch` 函数是一个异步的HTTP请求处理函数，它使用了`aiohttp`库来异步获取网页内容。`main` 函数初始化了一个`aiohttp.ClientSession`，然后调用`fetch`函数，并打印出获取的HTML内容。
## 5.2 异步Web服务
`asyncio`也可以用于创建异步的Web服务。下面是一个使用`aiohttp`创建的简单异步Web服务示例。
```python
from aiohttp import web
async def handle(request):
name = request.match_info.get('name', "Anonymous")
text = f"Hello, {name}!"
return web.Response(text=text)
app = web.Application()
app.add_routes([
web.get('/', handle),
web.get('/{name}', handle)
])
web.run_app(app)
```
在这个例子中，我们创建了一个简单的Web应用程序，它响应根URL和带名字的路径。`handle` 函数是一个异步处理函数，它根据URL路径中的名字返回响应。
## 5.3 异步数据处理
在数据处理任务中，如果任务可以并行处理，`asyncio`也能提供帮助。以下是一个使用`asyncio`进行异步文件读写的例子。
```python
import asyncio
import aiofiles
async def write_data(file_path, data):
async with aiofiles.open(file_path, 'w') as f:
await f.write(data)
async def read_data(file_path):
async with aiofiles.open(file_path, 'r') as f:
data = await f.read()
print(data)
async def main():
file_path = 'example.txt'
data_to_write = '异步文件写入示例'
# 写入数据
await write_data(file_path, data_to_write)
# 读取数据
await read_data(file_path)
if __name__ == '__main__':
asyncio.run(main())
```
在这个例子中，`write_data` 函数异步地将数据写入文件，而`read_data` 函数从文件异步读取数据。通过这种方式，可以在处理I/O密集型任务时提高效率。
2024-12-12 15:41:46.964 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4 not found in TOKEN_COSTS.
2024-12-12 15:41:46.965 | INFO | __main__:_act:75 - # 五、实战案例
## 5.1 异步爬虫
异步IO在爬虫领域有广泛的应用，因为它可以显著提高爬取效率，特别是在处理网络请求时。以下是一个使用`asyncio`和`aiohttp`的异步爬虫示例。
```python
import asyncio
import aiohttp
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
url = 'http://example.com'
html = await fetch(session, url)
print(html)
if __name__ == '__main__':
asyncio.run(main())
```
在这个示例中，`fetch` 函数是一个异步的HTTP请求处理函数，它使用了`aiohttp`库来异步获取网页内容。`main` 函数初始化了一个`aiohttp.ClientSession`，然后调用`fetch`函数，并打印出获取的HTML内容。
## 5.2 异步Web服务
`asyncio`也可以用于创建异步的Web服务。下面是一个使用`aiohttp`创建的简单异步Web服务示例。
```python
from aiohttp import web
async def handle(request):
name = request.match_info.get('name', "Anonymous")
text = f"Hello, {name}!"
return web.Response(text=text)
app = web.Application()
app.add_routes([
web.get('/', handle),
web.get('/{name}', handle)
])
web.run_app(app)
```
在这个例子中，我们创建了一个简单的Web应用程序，它响应根URL和带名字的路径。`handle` 函数是一个异步处理函数，它根据URL路径中的名字返回响应。
## 5.3 异步数据处理
在数据处理任务中，如果任务可以并行处理，`asyncio`也能提供帮助。以下是一个使用`asyncio`进行异步文件读写的例子。
```python
import asyncio
import aiofiles
async def write_data(file_path, data):
async with aiofiles.open(file_path, 'w') as f:
await f.write(data)
async def read_data(file_path):
async with aiofiles.open(file_path, 'r') as f:
data = await f.read()
print(data)
async def main():
file_path = 'example.txt'
data_to_write = '异步文件写入示例'
# 写入数据
await write_data(file_path, data_to_write)
# 读取数据
await read_data(file_path)
if __name__ == '__main__':
asyncio.run(main())
```
在这个例子中，`write_data` 函数异步地将数据写入文件，而`read_data` 函数从文件异步读取数据。通过这种方式，可以在处理I/O密集型任务时提高效率。
# 六、附录
## 6.1 常用库与资源
Python 的 `asyncio` 是一个用于编写并发代码的库，使用async/await语法。以下是一些常用的库和资源，可以帮助你更好地理解和应用 `asyncio`。
### 常用库
- `aiohttp`：用于异步HTTP网络请求。
- `aiomysql` / `aiopg`：用于异步操作MySQL和PostgreSQL数据库。
- `aioredis`：用于异步操作Redis。
- `aiofiles`：用于异步文件操作。
### 资源
- [官方文档](https://docs.python.org/3/library/asyncio.html)：提供了最权威的 `asyncio` 使用指南。
- [Asyncio GitHub 仓库](https://github.com/python/asyncio)：可以找到最新的信息和一些示例。
## 6.2 最佳实践
在使用 `asyncio` 进行编程时，以下是一些最佳实践：
1. **避免阻塞操作**：确保你的异步函数不会进行阻塞操作，如长时间的计算或等待输入/输出。
2. **合理使用并发**：不要盲目地创建大量协程，合理地管理和调度它们可以避免资源浪费和提高效率。
3. **错误处理**：使用 `try-except` 块来处理可能出现的异常，确保程序的健壮性。
### 代码示例
```python
import asyncio
async def fetch_data(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
urls = ['http://example.com', 'http://example.org']
tasks = [fetch_data(url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
if __name__ == '__main__':
asyncio.run(main())
```
## 6.3 进一步学习
想要进一步学习 `asyncio`，以下是一些建议：
- **理解事件循环**：深入了解 `asyncio` 的事件循环是如何工作的，这是 `asyncio` 的核心概念。
- **并发模型**：研究 `asyncio` 的并发模型，包括协程、任务和未来。
- **性能调优**：学习如何对基于 `asyncio` 的应用进行性能分析和调优。
通过这些资源的帮助，你可以更深入地掌握 `asyncio`，并在实践中更有效地
2024-12-12 15:42:04.027 | WARNING | metagpt.utils.cost_manager:update_cost:49 - Model GLM-4 not found in TOKEN_COSTS.
使用它。
2024-12-12 15:42:04.030 | INFO | __main__:_act:75 - # 六、附录
## 6.1 常用库与资源
Python 的 `asyncio` 是一个用于编写并发代码的库，使用async/await语法。以下是一些常用的库和资源，可以帮助你更好地理解和应用 `asyncio`。
### 常用库
- `aiohttp`：用于异步HTTP网络请求。
- `aiomysql` / `aiopg`：用于异步操作MySQL和PostgreSQL数据库。
- `aioredis`：用于异步操作Redis。
- `aiofiles`：用于异步文件操作。
### 资源
- [官方文档](https://docs.python.org/3/library/asyncio.html)：提供了最权威的 `asyncio` 使用指南。
- [Asyncio GitHub 仓库](https://github.com/python/asyncio)：可以找到最新的信息和一些示例。
## 6.2 最佳实践
在使用 `asyncio` 进行编程时，以下是一些最佳实践：
1. **避免阻塞操作**：确保你的异步函数不会进行阻塞操作，如长时间的计算或等待输入/输出。
2. **合理使用并发**：不要盲目地创建大量协程，合理地管理和调度它们可以避免资源浪费和提高效率。
3. **错误处理**：使用 `try-except` 块来处理可能出现的异常，确保程序的健壮性。
### 代码示例
```python
import asyncio
async def fetch_data(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
urls = ['http://example.com', 'http://example.org']
tasks = [fetch_data(url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
if __name__ == '__main__':
asyncio.run(main())
```
## 6.3 进一步学习
想要进一步学习 `asyncio`，以下是一些建议：
- **理解事件循环**：深入了解 `asyncio` 的事件循环是如何工作的，这是 `asyncio` 的核心概念。
- **并发模型**：研究 `asyncio` 的并发模型，包括协程、任务和未来。
- **性能调优**：学习如何对基于 `asyncio` 的应用进行性能分析和调优。
通过这些资源的帮助，你可以更深入地掌握 `asyncio`，并在实践中更有效地使用它。
参考链接：https://github.com/geekan/MetaGPT
如果有任何问题，欢迎在评论区提问。
原文地址：https://blog.csdn.net/qq_41472205/article/details/144428414
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！