Skip to content

Python 异步编程指南#

如果你是 Python 异步编程的新手，本页内容将为你提供入门指导。

在 LlamaIndex 中，许多操作和函数都支持异步执行。这允许你同时运行多个操作而不会阻塞主线程，从而在多数情况下提升整体吞吐量和性能。

以下是需要理解的关键概念：

1. asyncio 基础#

  • 事件循环： 事件循环负责调度和执行异步操作。它会持续检查并执行任务（协程）。所有异步操作都通过该循环运行，且每个线程只能有一个事件循环。

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

  • 异步与 Python Notebook： Python Notebook 是特殊情况，其事件循环已在运行。这意味着你无需自行调用 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()： 有时需要在不冻结异步程序的情况下运行同步（阻塞）代码。该函数将阻塞代码转移到单独线程执行，使事件循环能继续处理其他任务。需谨慎使用，因为它会增加开销并可能加大调试难度。

  • 替代方案：执行器： 你也可以使用 loop.run_in_executor() 来处理阻塞函数。

5. 实践示例#

以下示例展示如何使用 asyncio 编写和运行异步函数：

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