恭喜你前四天你已经打通了 AI 应用的“逻辑链路”。今天第 5 天我们要进入工程性能的分水岭从“线性思维”转向“并发思维”。“理解高并发与非阻塞”是区分“写脚本的”和“做系统的”关键标志。第一站解密asyncio—— 为什么要学它1. 什么是asyncioasyncio是 Python 用于编写并发代码的库。同步 (Synchronous)像排队买奶茶前面的人没拿到你就得在那干等着什么也干不了。异步 (Asynchronous)像在餐厅扫码点餐点完后你可以继续和朋友聊天、刷手机等菜好了服务员会叫你。2. 为什么要学它场景痛点大模型的响应通常很慢有时需要几秒甚至十几秒。如果你在同步模式下连续处理 10 个文档程序会一个接一个等待总耗时 10 × 等待时间。异步优势使用异步你可以同时发出 10 个请求。程序在等待 AI 返回结果的“空窗期”会去处理其他任务。总耗时 ≈ 最慢的那个请求的时间。职业进阶现代 AI 框架如 FastAPI, LangChain底层全是异步。看不懂async/await你将很难阅读和重构复杂的 AI 项目源码。第二站核心概念——三个关键词event loop(事件循环)它是大管家负责调度任务。谁准备好了菜好了它就让谁执行。async def定义一个“协程”函数。这种函数调用时不会立即执行而是返回一个协程对象。await挂起标志。告诉管家“我现在要等 AI 回复了你先去忙别的吧等回复到了再回来叫我。” 今日最终里程碑编写异步 AI 请求脚本由于requests库是同步的不支持异步。我们需要安装aiohttp异步 HTTP 库或直接使用 OpenAI 的异步 SDK。为了让你以后能无缝对接各大模型我们直接学习官方异步 SDK的用法。1. 安装异步环境在终端执行pip install openai2. 编写异步请求代码新建文件day5_async_bot.pyimportasyncioimporttimefromopenaiimportAsyncOpenAI# 导入异步客户端# 1. 初始化异步客户端 (以硅基流动为例)clientAsyncOpenAI(api_key你的API_KEY,base_urlhttps://api.siliconflow.cn/v1)asyncdefask_ai(task_id,question): 这是一个异步协程函数 print(f 任务{task_id}: 发出请求 -{question[:15]}...)start_timetime.time()# 2. 使用 await 挂起请求等待期间 CPU 会去处理其他任务responseawaitclient.chat.completions.create(modelQwen/Qwen2.5-7B-Instruct,messages[{role:user,content:question}])end_timetime.time()contentresponse.choices[0].message.contentprint(f✅ 任务{task_id}: 耗时{end_time-start_time:.2f}s | 返回:{content[:20]}...)asyncdefmain():print(--- 异步并发测试开始 ---)start_totaltime.time()# 3. 同时启动 3 个不同的任务# 注意这里只是创建了任务并没有阻塞运行tasks[ask_ai(1,请解释什么是高分子材料的蠕变现象),ask_ai(2,如何提高 CPI 薄膜的耐热性),ask_ai(3,总结聚酰亚胺在航空航天领域的三个应用。)]# 4. 并发执行所有任务awaitasyncio.gather(*tasks)end_totaltime.time()print(f\n✨ 所有任务完成总耗时:{end_total-start_total:.2f}s)print(提示你会发现总耗时远小于 3 个任务耗时的总和。)if__name____main__:# 5. 启动事件循环asyncio.run(main())验证环节体会“非阻塞”观察日志你会发现任务 1、2、3 的“发出请求”是几乎同时打印出来的而不是等 1 完了才打印 2。对比耗时同步模式耗时 ≈ 3s 3s 3s 9s异步模式耗时 ≈max(3s, 3.2s, 2.8s) ≈ 3.2s使用场景感悟想象一下第 6 周你要做“合同批量审核”如果不学异步老板让你审 100 份合同你可能要让老板等 10 分钟用了异步可能只需 30 秒。 学习技巧不要掉进细节深渊异步编程有很多高级特性如Lock,Queue,Semaphore今天请全部忽略。 你只需要记住这个公式async def定义函数 await调用耗时操作 asyncio.gather批量并发。