本文来自《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. 实操任务
完成以下练习:
- 写两个异步函数,分别等待不同秒数后返回字符串。
- 用
asyncio.gather并发执行。 - 对比串行和并发的总耗时差异。
8. 自测题
- 异步适合解决什么问题,不适合解决什么问题?
- 为什么网络请求特别适合异步?
- 为什么不能在异步函数里再套一层
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 def 和 await
知道异步函数必须被异步地调用。
第二步:理解 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. 本章完成后你应该具备的能力
完成本章后,你应当做到:
- 能写出一个最小异步函数。
- 能解释
await的意义。 - 能用
asyncio.run启动脚本。 - 能看懂后续月份 1 里大部分异步 client 和路由代码。
22. 如果你卡在这里,先回看哪几章
- 基础函数还不熟:回看 01-Python最小入门.md
- 异常处理还不熟:回看 04-异常-文件-JSON.md
23. 从本章过渡到下一章的桥接说明
接下来进入 02-Python工程化基础 模块。
你现在已经具备了阅读后续月份 1 代码所需的最小语言基础。下一步不再是继续扩语言,而是把这些能力组织进一个规范项目里:使用 uv 建项目、配置依赖、组织目录、写日志、写模型、写测试。也就是从“会写代码”正式进入“会做项目”。