Python 异步核心
1、协程
1、可以暂停，恢复执行的函数
2、不占用额外的线程，超级清凉
3、AI 项目用来：同时调用多个大模型、不卡接口
2、事件循环
1、协程的调度器
2、负责:谁暂停、谁恢复、谁执行
3、，Python 自动管理
3、async/awiat
async:声明这是有个异步函数
await:暂停等待异步任务完成
1.定义异步函数
async def 函数名():
return 结果
await 函数名()
3.普通函数不能直接用awaut
4,异步函数不能直接调用，必须扔进事件循环
基础1demo
import asyncio async def llm_request(prompt): print(f"开始请求大模型:{prompt}") await asyncio.sleep(2) return f"AI回复：{prompt} → 这是测试回答" async def main(): result = await llm_request("你好") print(result) #启动事件循环 asyncio.run(main())基础demo2
import asyncio async def llm_1(): await asyncio.sleep(2) return "大模型1返回结果 " async def llm_2(): await asyncio.sleep(2) return ("大模型2返回结果") async def main(): 并发执行,总耗时 = 2秒,不是4秒 res1,res2= await asyncio.gather(llm_1, llm_2) #协程并发 print(res1) print(res2) asyncio.run(main())示例3demo
import asyncio # 异步协程核心库（用来跑异步函数） import aiohttp # 异步HTTP请求库（用来调用AI接口，相当于异步版requests） async def fetch_llm_response(prompt): async with aiohttp.ClientSession as session: #async with = 异步上下文管理器 ClientSession = 异步请求会话（相当于打开浏览器） async with session.post(url,json={"prompt":prompt}) as resp: return await resp.json() async def main(): result = await fetch_llm_response("介绍rag") asyncio.run(main())异步的核心知识
1.异步任务Task是什么
async def 只是定义协程，不会自己执行
asyncio.create_task 把协程包装成后台任务，立刻进入事件循环调度
适合：后台并行跑多个任务、不阻塞主流程
2.asyncio.gatjcer 并发精髓
1、传入多个协程/任务，同时并发执行
2、所有任务全部完成才返回结果列表
3、总耗时 = 最慢的那个任务耗时，
3.异步超时 asyncio.wait_for
大模型接口经常超时，企业开发必须加：
- 设定最大等待时间
- 超时自动抛出异常，防止接口卡死
4.异步异常捕获
普通的 try except 可以用 单必须抱在await外面
用于：大模型请求失败、网络异常、自动降级返回兜底文案
Demo：Task 手动创建任务
import asyncio async def llm_task(name,delay): print(f"任务{name}开始执行") await asyncio.sleep(delay) print(f"任务{name}执行完成") return f"{name}结果" async def main(): #创建后台任务，立刻开始跑 t1 = asyncio.create_task(llm_task("模型A", 2)) t2 = asyncio.create_task(llm_task("模型B", 3)) #等待所有任务 res1 = await t1 res2 = await t2 print("最终结果：", res1, res2) asyncio.run(main()) 两个任务同时启动，总耗时约 3 秒，不是 5 秒demo2：批量并发
import asyncio async def call_llm(prompt,cost): await asyncio.sleep(cost) return f"【{prompt}】回答完成，耗时{cost}s" async def main(): # 批量并发三个大模型请求 tasks = [ call_llm("生成测试用例", 2), call_llm("文档测试摘要", 3), call_llm("代码解释", 1.5), ] # 并发等待全部完成 results = await asyncio.gather(*tasks) for res in results: print(res) asyncio.run(main())demo3异步超时控制（大模型接口必备）
import asyncio async def slow_llm(): await asyncio.sleep(5) return "大模型正常返回结果" async def main(): try: res = await asyncio.wait_for(slow_llm(), timeout=3) print(res) except asyncio.TimeoutError: print("请求超时：大模型响应太慢，触发兜底逻辑") asyncio.run(main())、实战代码 4：异步全局异常捕获 + 降级兜底
import asyncio async def call_api(): await asyncio.sleep(1) raise Exception("网络连接失败") async def main(): try: res = await call_api() print(res) except Exception as e: print(e) #业务降级，返回默认兜底回答 print("返回兜底文案：当前服务繁忙，请稍后再试") asyncio.run(main())i