跳至内容

Python中的异步编程#

如果你是Python异步编程的新手，这个页面就是为你准备的。

在LlamaIndex中，许多操作和函数支持异步执行。这使您能够同时运行多个操作而不会阻塞主线程，在许多情况下有助于提高整体吞吐量和性能。

以下是您应该理解的一些关键概念：

1. asyncio基础#

  • 事件循环: 事件循环负责处理异步操作的调度和执行。它会持续检查并执行任务(协程)。所有的异步操作都通过这个循环运行，且每个线程只能有一个事件循环。

  • asyncio.run(): 该函数是运行异步程序的入口点。它会创建并管理事件循环，并在完成后进行清理。请注意，该函数设计为每个线程仅调用一次。像FastAPI这样的框架会为您运行事件循环，而其他框架则需要您自行运行。

  • 异步 + Python 笔记本: Python笔记本是一个特殊情况，事件循环已经在运行。这意味着您不需要自己调用asyncio.run()，可以直接调用和等待异步函数。

2. 异步函数与 await#

  • 定义异步函数: 使用async def语法来定义一个异步函数(协程)。调用异步函数时不会立即执行，而是返回一个需要被调度运行的协程对象。

  • 使用await: 在异步函数内部，await用于暂停该函数的执行，直到等待的任务完成。当你编写await some_fn()时，函数会将控制权交还给事件循环，以便可以调度和运行其他任务。一次只有一个异步函数执行，它们通过await进行协作式让步。

3. 并发机制解析#

  • 协作式并发: 虽然您可以调度多个异步任务，但同一时间只有一个任务在运行。这与真正的并行不同，后者是多个任务同时执行。当一个任务遇到await时，它会暂停执行以便另一个任务可以运行。这使得异步程序非常适合I/O密集型任务，例如调用LLM和其他服务的API，这类任务经常需要等待。

  • 非真正的并行性: Asyncio实现了并发但不会并行运行任务。对于需要并行执行的CPU密集型工作，请考虑使用线程或多进程。LlamaIndex在大多数情况下通常避免使用多进程，并将其留给用户自行实现，因为要以安全高效的方式实现可能会比较复杂。

4. 处理阻塞（同步）代码#

  • asyncio.to_thread(): 有时您需要在不冻结异步程序的情况下运行同步（阻塞）代码。asyncio.to_thread()将阻塞代码卸载到单独的线程中，允许事件循环继续处理其他任务。请谨慎使用，因为它会增加一些开销，并可能使调试更具挑战性。

  • 替代方案：执行器: 您可能还会遇到使用loop.run_in_executor()来处理阻塞函数的情况。

5. 实际示例#

以下是一个演示如何使用asyncio编写和运行异步函数的示例：

python
import asyncio


async def fetch_data(delay):
    print(f"Started fetching data with {delay}s delay")

    # Simulates I/O-bound work, such as network operation
    await asyncio.sleep(delay)

    print("Finished fetching data")
    return f"Data after {delay}s"


async def main():
    print("Starting main")

    # Schedule two tasks concurrently
    task1 = asyncio.create_task(fetch_data(2))
    task2 = asyncio.create_task(fetch_data(3))

    # Wait until both tasks complete
    result1, result2 = await asyncio.gather(task1, task2)

    print(result1)
    print(result2)
    print("Main complete")


if name == "main":
    asyncio.run(main())
优云智算