异步并发asyncio包的介绍与用法示例

asyncio 是 Python 标准库里的异步 I/O 包，不需要额外安装。它主要用来写"并发等待型"程序，比如网络请求、数据库访问、文件/接口调用、LangChain 的 ainvoke 等。

它适合的场景是：程序大部分时间在"等外部结果"，而不是一直占用 CPU 计算。

核心概念

• async def：定义异步函数，也叫协程 coroutine
• await：等待一个异步操作完成，等待期间事件循环可以去执行别的任务
• asyncio.run()：启动事件循环，运行入口协程
• asyncio.create_task()：创建并发任务
• asyncio.gather()：并发执行多个协程，并收集结果
• asyncio.sleep()：异步等待，不会阻塞整个程序

最小例子

import asyncio


async def say_hello():
    print("开始")
    await asyncio.sleep(1)
    print("你好 asyncio")


asyncio.run(say_hello())

重点是这一行：

await asyncio.sleep(1)

它不是让整个 Python 程序卡住 1 秒，而是告诉事件循环："我这里要等 1 秒，你可以先去执行别的异步任务。"

更实用的 demo：并发执行多个任务

import asyncio
import time


async def download(name: str, seconds: int):
    print(f"{name} 开始下载，预计 {seconds} 秒")
    await asyncio.sleep(seconds)
    print(f"{name} 下载完成")
    return f"{name} 的结果"


async def main():
    start = time.perf_counter()

    results = await asyncio.gather(
        download("任务 A", 3),
        download("任务 B", 2),
        download("任务 C", 1),
    )

    end = time.perf_counter()

    print("所有结果：", results)
    print(f"总耗时：{end - start:.2f} 秒")


asyncio.run(main())

大概输出：

任务 A 开始下载，预计 3 秒
任务 B 开始下载，预计 2 秒
任务 C 开始下载，预计 1 秒
任务 C 下载完成
任务 B 下载完成
任务 A 下载完成
所有结果： ['任务 A 的结果', '任务 B 的结果', '任务 C 的结果']
总耗时：3.00 秒

如果不用 asyncio，三个任务顺序执行会花：

3 + 2 + 1 = 6 秒

用了 asyncio.gather() 后，它们并发等待，总耗时约等于最慢的那个任务，也就是 3 秒。

create_task() 的写法

import asyncio


async def worker(name: str, seconds: int):
    print(f"{name} start")
    await asyncio.sleep(seconds)
    print(f"{name} done")
    return name


async def main():
    task1 = asyncio.create_task(worker("A", 2))
    task2 = asyncio.create_task(worker("B", 1))

    print("任务已经创建，可以继续做别的事")

    result1 = await task1
    result2 = await task2

    print(result1, result2)


asyncio.run(main())

create_task() 的意思是：先把协程丢给事件循环调度执行，后面再用 await 拿结果。

结合 LangChain ainvoke 的用法

如果在 LangChain 里看到类似：

result = await model.ainvoke("你好")

通常就要放在异步函数里：

import asyncio


async def main():
    result = await model.ainvoke("介绍一下 Python asyncio")
    print(result)


asyncio.run(main())

如果要同时发起多个 LLM 调用，可以这样：

import asyncio


async def main():
    prompts = [
        "一句话介绍 Python",
        "一句话介绍 asyncio",
        "一句话介绍 LangChain",
    ]

    results = await asyncio.gather(
        *[model.ainvoke(prompt) for prompt in prompts]
    )

    for result in results:
        print(result)


asyncio.run(main())

总结

asyncio 用来让多个"需要等待"的任务并发执行，写法核心就是 async def + await + asyncio.run()，需要并发时再配合 gather() 或 create_task()。