← Back to 文字

asyncio 基础

本文来自《AI 应用开发课程》月份 1 课程文档,已整理为网站文章版本。

学习目标

学完本节后,你应当能够:

  • 理解异步编程为什么会出现在 AI 应用中。
  • 写出最小 async def 函数。
  • 使用 await 调用异步操作。
  • asyncio.run 执行异步入口。
  • 对后续 httpx.AsyncClient 和 FastAPI 异步接口建立正确直觉。

前置知识

  • 已完成本模块前四节

1. 为什么你必须学异步

月份 1 后面至少会遇到三类等待:

  • 等待远程模型 API 返回
  • 等待网络请求
  • 等待流式分块返回

这些都属于 I/O 等待。异步能让程序在等待时更有效地安排执行。

2. 最小异步函数

import asyncio


async def say_hello() -> str:
    await asyncio.sleep(1)
    return "hello"


async def main() -> None:
    result = await say_hello()
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

关键点:

  • async def 定义异步函数
  • await 等待异步结果
  • asyncio.run 作为脚本入口执行事件循环

3. 同步与异步的直观区别

同步

  • 一步做完再做下一步
  • 等待网络时主流程停住

异步

  • 碰到 I/O 等待时把控制权交回事件循环
  • 更适合网络密集型任务

这不是“异步一定更快”,而是“异步更适合等待型任务”。

4. 并发示例

import asyncio


async def fetch_data(name: str, delay: int) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"


async def main() -> None:
    results = await asyncio.gather(
        fetch_data("task1", 1),
        fetch_data("task2", 2),
    )
    print(results)


if __name__ == "__main__":
    asyncio.run(main())

这就是后面并发请求多个接口或多个模型调用时的基础模式。

5. 你当前阶段只需要掌握的异步规则

  • 脚本入口使用 asyncio.run
  • 异步函数之间用 await
  • I/O 操作优先考虑异步版本库
  • 不要在异步函数里再调用 asyncio.run

6. 后续会怎么用到

在模型调用里

  • 使用 httpx.AsyncClient
  • 异步等待 API 返回

在 FastAPI 中

  • 使用 async def 定义 endpoint
  • 在 endpoint 中 await service.call_model()

7. 实操任务

完成以下练习:

  1. 写两个异步函数,分别等待不同秒数后返回字符串。
  2. asyncio.gather 并发执行。
  3. 对比串行和并发的总耗时差异。

8. 自测题

  1. 异步适合解决什么问题,不适合解决什么问题?
  2. 为什么网络请求特别适合异步?
  3. 为什么不能在异步函数里再套一层 asyncio.run

9. 作业与验收

作业:

  • 写一个 async_retry_demo.py,包含一个异步函数和一个简单重试逻辑。

验收标准:

  • 能成功运行
  • 你能指出哪一行是异步边界
  • 你能解释 await 为什么不能随便删

10. 常见错误

  • 忘记 await
  • 在同步函数里直接调用异步函数却不执行事件循环
  • 把 CPU 密集型任务误当成异步优化对象

11. 延伸阅读

  • Python 官方文档中的 asyncio
  • Real Python 的 async 教程

12. 本章与前文关系

前面几章解决的是:

  • 代码怎么写
  • 模块怎么拆
  • 数据怎么表达
  • 文件怎么读写
  • 错误怎么处理

现在开始进入月份 1 的第一块“真正与后续 AI 应用直接相连”的基础能力:异步。

后面你会频繁遇到这些操作:

  • 等待 DeepSeek API 返回
  • 等待 HTTP 请求完成
  • 等待流式输出逐块返回
  • 在 FastAPI 中编写异步 endpoint

这些都不是“纯计算”,而是“等待 I/O”。这正是 asyncio 最适合解决的问题。

13. 本章在研发助手项目中的位置

研发助手项目的几乎所有外部交互能力都和异步有关:

  • 模型调用 client
  • FastAPI 路由
  • 流式输出
  • 可能的重试逻辑

如果你这一章只停留在“知道 async def 语法”,后面会很难真正读懂 llm_client.py 和 FastAPI 代码。

14. 一个更稳的异步心智模型

很多初学者第一次接触异步时,容易陷入两个误区:

误区一:异步一定更快

不准确。异步更适合 I/O 等待场景,不一定适合 CPU 密集型计算。

误区二:异步就是多线程

也不准确。异步更像是:程序在等待外部结果时,把控制权交回事件循环,让其他任务继续推进。

对于月份 1,你当前只需要记住一句话:

模型调用、HTTP 请求、流式返回这类“等外部响应”的任务,非常适合异步。

15. 错误示例 vs 正确示例

错误示例:定义了异步函数,但没有运行事件循环

import asyncio


async def say_hello() -> str:
    await asyncio.sleep(1)
    return "hello"


result = say_hello()
print(result)

这不会打印真正结果,而是打印一个 coroutine 对象。

正确示例:通过 asyncio.run 执行入口

import asyncio


async def say_hello() -> str:
    await asyncio.sleep(1)
    return "hello"


async def main() -> None:
    result = await say_hello()
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

16. 完整文件级示例:async_retry_demo.py

"""演示异步函数、并发和简单重试。"""

from __future__ import annotations

import asyncio


async def unstable_task(task_name: str, fail_times: int, state: dict[str, int]) -> str:
    """模拟一个可能失败几次后才成功的异步任务。"""

    await asyncio.sleep(0.5)

    current_count = state.get(task_name, 0)
    if current_count < fail_times:
        state[task_name] = current_count + 1
        raise TimeoutError(f"{task_name} 暂时失败,第 {current_count + 1} 次")

    return f"{task_name} 成功完成"


async def retry_task(task_name: str, fail_times: int, retries: int = 3) -> str:
    """为异步任务增加最小重试逻辑。"""

    state: dict[str, int] = {}
    last_error: Exception | None = None

    for attempt in range(1, retries + 1):
        try:
            print(f"[{task_name}] 开始第 {attempt} 次尝试")
            result = await unstable_task(task_name, fail_times, state)
            print(f"[{task_name}] 执行成功")
            return result
        except TimeoutError as error:
            last_error = error
            print(f"[{task_name}] 发生超时: {error}")
            await asyncio.sleep(0.3)

    raise RuntimeError(f"{task_name} 重试后仍然失败") from last_error


async def main() -> None:
    """并发运行两个任务,观察它们的重试过程。"""

    results = await asyncio.gather(
        retry_task("task-1", fail_times=1),
        retry_task("task-2", fail_times=2),
    )
    print(results)


if __name__ == "__main__":
    asyncio.run(main())

17. 逐段解释这份完整示例

unstable_task

它模拟的是“外部服务偶发失败”的情况。后面模型 API 请求本质上就是这种外部 I/O 任务。

retry_task

它展示了月份 1 后面会经常用到的两个概念:

  • 并不是所有错误都要立即放弃
  • 但重试也必须有边界

asyncio.gather

它让多个异步任务并发执行。这会在后面你同时处理多个 I/O 请求时非常常见。

18. 从最小异步到工程异步的递进

第一步:理解 async defawait

知道异步函数必须被异步地调用。

第二步:理解 asyncio.run

知道脚本入口需要事件循环。

第三步:理解 gather

知道多个等待型任务可以并发推进。

第四步:把这种思维迁移到 HTTP 请求和 FastAPI

这正是后面 httpx.AsyncClient 和异步路由会用到的模式。

19. 为什么月份 1 只讲最小异步,而不讲一整套并发理论

因为你当前的目标不是成为 asyncio 专家,而是建立足够支撑后续 AI 应用开发的异步直觉。

对月份 1 来说,你真正必须掌握的是:

  • 何时使用 async
  • 何时使用 await
  • 何时使用 asyncio.run
  • 何时可以用 gather
  • 为什么不能随便重试所有错误

这已经足以支撑第 2 周到第 4 周的大部分代码阅读和编写。

20. 调试与排错:本章最常见问题

问题一:忘记 await

现象通常是:

  • 你得到 coroutine 对象
  • 或逻辑根本没有真正执行

问题二:在异步函数里再调用 asyncio.run

这通常会导致事件循环错误,尤其在 FastAPI 或某些交互环境中。

问题三:把不该重试的错误也重试

例如参数错误、配置错误,重试并不会让它 magically 成功。

问题四:误把异步当作“写法变化”,而不是“等待模型”

你要始终记住:异步的核心不是语法,而是如何处理等待。

21. 本章完成后你应该具备的能力

完成本章后,你应当做到:

  1. 能写出一个最小异步函数。
  2. 能解释 await 的意义。
  3. 能用 asyncio.run 启动脚本。
  4. 能看懂后续月份 1 里大部分异步 client 和路由代码。

22. 如果你卡在这里,先回看哪几章

23. 从本章过渡到下一章的桥接说明

接下来进入 02-Python工程化基础 模块。

你现在已经具备了阅读后续月份 1 代码所需的最小语言基础。下一步不再是继续扩语言,而是把这些能力组织进一个规范项目里:使用 uv 建项目、配置依赖、组织目录、写日志、写模型、写测试。也就是从“会写代码”正式进入“会做项目”。

Fin