《AI大模型应用开发实战从入门到精通共60篇》041、异步编程：用asyncio提升LLM应用的并发性能

041 异步编程用asyncio提升LLM应用的并发性能从一次线上事故说起凌晨两点告警电话把我从床上拽起来。监控显示我们的LLM对话服务响应时间从200ms飙到了8秒CPU负载却只有30%。查日志发现每次用户请求都在等上游的OpenAI接口返回——一个请求等2秒十个请求排队等20秒。同步阻塞典型的“一个人干活一群人围观”场景。这个问题的本质是LLM调用是I/O密集型操作网络延迟占大头。同步代码里线程在等待网络响应时啥也不干白白浪费CPU时间片。当时我盯着火焰图里那条长长的recvfrom调用链心想必须上异步了。asyncio到底在解决什么问题很多人把asyncio和“并发”划等号其实它解决的是等待时的资源复用。想象你在咖啡店排队点单同步模式是你站在柜台前等咖啡做好才让下一个人点异步模式是你点完单拿个号去旁边坐着服务员继续接待下一个人咖啡好了叫号。Python的GIL让多线程在CPU密集型任务上表现糟糕但对I/O密集型任务asyncio用单线程事件循环实现了比多线程更高效的并发。原因很简单线程切换有开销协程切换是用户态操作轻量得多。从同步到异步一个LLM调用的改造实录先看一个典型的同步LLM调用代码importrequestsimporttimedefcall_llm(prompt):# 这里踩过坑requests默认超时时间很长生产环境一定要设timeoutresprequests.post(https://api.openai.com/v1/chat/completions,json{model:gpt-3.5-turbo,messages:[{role:user,content:prompt}]},headers{Authorization:Bearer YOUR_KEY},timeout30)returnresp.json()# 串行调用10个请求要等20秒prompts[讲个笑话,写首诗,翻译成英文,...]results[]forpinprompts:results.append(call_llm(p))这段代码的问题每个call_llm都在阻塞等待网络响应10个请求依次执行总耗时 单个请求耗时 × 请求数。改成asyncio版本importasyncioimportaiohttp# 别用requests它不支持异步asyncdefcall_llm_async(session,prompt):# 注意aiohttp的session要复用别每次请求都创建新的asyncwithsession.post(https://api.openai.com/v1/chat/completions,json{model:gpt-3.5-turbo,messages:[{role:user,content:prompt}]},headers{Authorization:Bearer YOUR_KEY},timeoutaiohttp.ClientTimeout(total30))asresp:returnawaitresp.json()asyncdefmain():# 这里踩过坑connector参数控制连接池大小默认20并发高时要调大connectoraiohttp.TCPConnector(limit50)asyncwithaiohttp.ClientSession(connectorconnector)assession:prompts[讲个笑话,写首诗,翻译成英文,...]# 创建10个协程任务并发执行tasks[call_llm_async(session,p)forpinprompts]resultsawaitasyncio.gather(*tasks)returnresults# 运行事件循环starttime.time()resultsasyncio.run(main())print(f耗时{time.time()-start:.2f}秒)改造后10个请求几乎同时发出总耗时 ≈ 最慢的那个请求的耗时通常2-3秒。性能提升5-10倍。那些年我踩过的asyncio坑坑1在异步函数里调用同步阻塞代码asyncdefbad_example():# 别这样写time.sleep会阻塞整个事件循环time.sleep(5)returndoneasyncdefgood_example():# 正确做法用asyncio.sleep让出控制权awaitasyncio.sleep(5)returndone如果必须调用同步阻塞库比如某些数据库驱动用loop.run_in_executor把它扔到线程池里importconcurrent.futuresasyncdefcall_sync_db():loopasyncio.get_event_loop()# 这里踩过坑默认线程池大小是CPU核心数×5高并发场景要自定义withconcurrent.futures.ThreadPoolExecutor(max_workers20)aspool:resultawaitloop.run_in_executor(pool,sync_db_query)returnresult坑2忘记处理异常导致任务静默失败asyncio.gather默认会抛出第一个异常导致其他任务被取消。更稳妥的做法asyncdefsafe_gather():tasks[call_llm_async(session,p)forpinprompts]# return_exceptionsTrue让异常作为结果返回不会中断其他任务resultsawaitasyncio.gather(*tasks,return_exceptionsTrue)fori,rinenumerate(results):ifisinstance(r,Exception):print(f第{i}个请求失败{r})results[i]None# 用默认值替代returnresults坑3事件循环嵌套在Jupyter Notebook或某些框架里事件循环可能已经运行再调用asyncio.run()会报错。解决方案try:loopasyncio.get_running_loop()exceptRuntimeError:# 没有运行中的事件循环创建新的resultsasyncio.run(main())else:# 已有事件循环用create_taskresultsloop.run_until_complete(main())实战构建一个带限速的LLM并发调用器LLM API通常有速率限制比如每分钟60次并发太高会被限流甚至封号。我们需要一个带令牌桶的异步限速器importasyncioimporttimeclassRateLimiter:def__init__(self,max_calls,period60):self.max_callsmax_calls self.periodperiod self.tokensmax_calls self.last_refilltime.monotonic()self._lockasyncio.Lock()# 这里踩过坑必须用asyncio.Lock不是threading.Lockasyncdefacquire(self):asyncwithself._lock:nowtime.monotonic()elapsednow-self.last_refill# 按时间比例补充令牌self.tokensmin(self.max_calls,self.tokenselapsed*(self.max_calls/self.period))self.last_refillnowifself.tokens1:# 令牌不足计算需要等待的时间wait_time(1-self.tokens)*(self.period/self.max_calls)awaitasyncio.sleep(wait_time)self.tokens0else:self.tokens-1# 使用示例limiterRateLimiter(max_calls60,period60)# 每分钟60次asyncdefrate_limited_call(session,prompt):awaitlimiter.acquire()# 获取令牌可能等待returnawaitcall_llm_async(session,prompt)这个限速器用asyncio.Lock保证令牌操作的原子性用time.monotonic避免系统时间调整带来的问题。实际生产环境中我还会加上指数退避重试逻辑。性能调优从10倍到50倍基础异步改造能带来5-10倍提升但想榨干性能还需要几个技巧1. 连接复用aiohttp.ClientSession默认会复用TCP连接但要注意连接池大小。我一般根据API的并发限制来设置# 如果API允许100并发连接池设120留余量connectoraiohttp.TCPConnector(limit120,limit_per_host120)2. 请求合并对于流式输出SSE用aiohttp的content属性逐块读取避免等待完整响应asyncdefstream_llm(session,prompt):asyncwithsession.post(url,json{stream:True,...},headersheaders)asresp:asyncforchunkinresp.content:# 处理每个chunk实现打字机效果yieldchunk.decode()3. 使用uvloop加速uvloop是用Cython重写的事件循环性能比默认的asyncio事件循环快2倍importuvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())# 然后正常使用asyncio.run()注意uvloop在Windows上支持有限生产环境建议在Linux上部署。个人经验建议别盲目全盘异步化。如果你的服务主要是CPU计算比如图像处理、模型推理异步带来的收益有限反而增加代码复杂度。异步最适合I/O密集且并发高的场景。异步代码的调试是个噩梦。asyncio.gather里的异常堆栈经常指向事件循环内部而不是你的业务代码。我的做法是每个协程函数都加try/except用logging.exception记录完整上下文再重新抛出或返回错误码。注意内存泄漏。异步代码里容易忘记释放资源特别是aiohttp.ClientSession。始终用async with管理资源或者显式调用session.close()。监控要跟上。异步服务的性能指标和同步服务不同要关注事件循环的延迟event loop lag、协程队列长度、连接池使用率。我习惯在关键路径上埋点记录每个协程的等待时间和执行时间。从同步到异步的迁移策略不要一次性重写所有代码。先找出I/O瓶颈最严重的模块比如LLM调用、数据库查询用异步重写其他部分保持同步。用run_in_executor作为过渡桥梁。最后说一句asyncio不是银弹但对付LLM这种高延迟、高并发的I/O场景它确实是最趁手的工具。下次你的服务再因为同步阻塞而告警试试把requests换成aiohttp把time.sleep换成asyncio.sleep——你会回来感谢我的。