Python 如何给 AI API 做请求限流：避免并发过高导致服务异常

当 AI 工具、自动化脚本或接口服务开始处理多个请求时单纯依赖接口本身往往不够。合理的限流可以控制并发、降低失败率也方便后续观察系统状态。为什么需要限流在本地测试时一个请求一个请求发送通常不会遇到明显问题。项目开始处理多个用户或批量任务后可能出现短时间请求数量过多接口返回限流错误本地线程或协程占用过高请求大量排队响应变慢失败后重复重试进一步放大压力限流的目标不是让请求变慢而是让系统保持在可控范围内。一、先区分并发和频率并发数表示同一时间正在处理多少个请求。例如同时允许 5 个任务访问 AI API。请求频率表示一段时间内发送多少次请求。例如每秒最多发送 2 次请求。两者解决的问题不完全相同。简单项目可以先控制并发批量任务还需要增加频率限制。二、用线程锁限制同步请求并发如果你的程序使用线程可以用Semaphore限制同时执行的任务数量fromthreadingimportSemaphorefromopenaiimportOpenAI limitSemaphore(3)clientOpenAI(api_keyyour-api-key,base_urlhttps://your-api-domain.com/v1,)defask_llm(prompt:str)-str:withlimit:responseclient.chat.completions.create(modelyour-model-name,messages[{role:user,content:prompt}],)returnresponse.choices[0].message.contentSemaphore(3)表示最多允许 3 个请求同时进入调用区域。三、异步项目中使用 asyncio.Semaphore如果项目使用异步代码可以用异步信号量importasynciofromopenaiimportAsyncOpenAI limitasyncio.Semaphore(3)clientAsyncOpenAI(api_keyyour-api-key,base_urlhttps://your-api-domain.com/v1,)asyncdefask_llm(prompt:str)-str:asyncwithlimit:responseawaitclient.chat.completions.create(modelyour-model-name,messages[{role:user,content:prompt}],)returnresponse.choices[0].message.content异步限流不会简单阻塞整个程序更适合聊天服务和批量任务。四、批量任务中配合 gather 使用批量执行时可以把并发控制放进统一函数asyncdefrun_one(prompt:str)-str:asyncwithlimit:responseawaitclient.chat.completions.create(modelyour-model-name,messages[{role:user,content:prompt}],)returnresponse.choices[0].message.contentasyncdefrun_batch(prompts:list[str])-list[str]:tasks[run_one(prompt)forpromptinprompts]returnawaitasyncio.gather(*tasks)即使一次创建很多任务真正进入 API 请求区域的数量也不会超过信号量限制。五、增加简单的请求间隔如果除了并发数还需要控制请求频率可以在请求前增加间隔importasyncio last_request_at0.0request_lockasyncio.Lock()asyncdefwait_interval(seconds:float0.5):globallast_request_atasyncwithrequest_lock:nowasyncio.get_running_loop().time()waitseconds-(now-last_request_at)ifwait0:awaitasyncio.sleep(wait)last_request_atasyncio.get_running_loop().time()调用时asyncwithlimit:awaitwait_interval(0.5)# 发起 API 请求这是一种简单的间隔控制适合小型脚本。大型服务通常需要使用队列或专门的限流组件。六、限流和重试不能互相放大一个常见问题是请求失败后立即重试结果产生更多请求。更合理的流程是限制并发遇到临时失败时等待使用有限次数重试重试仍失败时返回明确结果可以使用递增等待importasyncioasyncdefbackoff(attempt:int):awaitasyncio.sleep(min(2**attempt,10))等待时间不能无限增长也不能设置得过短。七、如何选择并发数量不要一开始就把并发设置得很大。建议从较小值开始观察看平均响应时间看失败率看限流错误数量看本机 CPU 和内存看任务排队时间如果失败率明显上升优先降低并发而不是继续增加重试次数。八、需要记录哪些数据为了判断限流是否有效建议记录请求开始时间请求结束时间当前模型并发数量响应耗时成功或失败重试次数错误类型有了这些数据才能知道瓶颈是在本地、网络还是上游接口。九、适合使用限流的场景批量文本处理AI 工具后端Agent 工作流自动化任务多用户聊天服务定时生成内容只要程序可能在短时间发出多个请求就值得考虑并发控制。十、结语AI API 限流的核心是让请求数量保持可控用信号量限制并发用间隔控制频率用退避减少重复压力用日志观察实际效果对于 Python 项目来说先从Semaphore或asyncio.Semaphore开始通常就能解决大部分基础场景。后续请求量变大再考虑队列、分布式限流和独立任务调度。免责声明本文内容仅用于技术交流与经验分享具体实现请结合项目实际情况调整。