Python 异步核心

1、协程1、可以暂停恢复执行的函数2、不占用额外的线程超级清凉3、AI 项目用来同时调用多个大模型、不卡接口2、事件循环1、协程的调度器2、负责:谁暂停、谁恢复、谁执行3、Python 自动管理3、async/awiatasync:声明这是有个异步函数await:暂停等待异步任务完成1.定义异步函数async def 函数名():return 结果await 函数名()3.普通函数不能直接用awaut4,异步函数不能直接调用必须扔进事件循环基础1demoimport asyncio async def llm_request(prompt): print(f开始请求大模型:{prompt}) await asyncio.sleep(2) return fAI回复{prompt} → 这是测试回答 async def main(): result await llm_request(你好) print(result) #启动事件循环 asyncio.run(main())基础demo2import 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())示例3demoimport 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外面用于大模型请求失败、网络异常、自动降级返回兜底文案DemoTask 手动创建任务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(), timeout3) 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