1. 项目概述为什么我们需要给LLM调用“上保险”直接调用大模型API比如OpenAI的ChatCompletion或者国内各种模型的接口是很多开发者上手LLM应用的第一步。代码简单到几行就能跑通这给了我们一种“大模型调用很简单”的错觉。但只要你把这样的代码放到生产环境或者跑一个需要连续处理上百条请求的脚本各种幺蛾子就会接踵而至网络突然波动一下API返回一个超时错误模型服务端因为负载过高给你抛回一个“rate limit”的429状态码甚至有时候模型本身会抽风返回一些无法解析的乱码或者结构错误的JSON。这时候如果你的代码只是简单的一个requests.post()外面套个try...except那整个流程就卡死了。用户看到的是“服务不可用”而你半夜会被报警电话叫醒手忙脚乱地去重启服务或者重跑脚本。这其实就是“裸调”大模型——没有任何弹性处理机制的调用方式脆弱得就像在裸奔。所以这个项目的核心价值就是给裸奔的LLM调用穿上“盔甲”。我们用大约60行Python代码实现三个最核心的弹性机制重试、超时和降级。重试负责在遇到临时性失败如网络抖动、服务端限流时自动再试几次提高单次请求的最终成功率。超时负责设置一个等待上限防止某个慢请求拖死整个线程或进程。降级则是最后的保障当所有重试都失败后提供一个保底结果比如返回一个预设的友好错误提示或者切换到一个更简单、更稳定的备用模型确保主流程不会彻底中断。这不仅仅是让代码更健壮更是工程思维的体现。在真实的生产系统中对待外部服务尤其是像LLM API这种可能昂贵且不稳定的服务的调用必须假设它可能失败并为此做好准备。接下来我会把这套机制的每一个零件拆开讲清楚原理并给你一份可以直接复制粘贴使用的代码。2. 核心设计思路构建一个健壮的LLM客户端在开始写代码之前我们先要理清楚一个健壮的LLM调用客户端应该长什么样。我们不能简单地在业务逻辑里到处写for i in range(3): try...except...那样代码会又臭又长难以维护。我们的目标是设计一个可复用的“装饰器”或“包装器”它能够以最小侵入的方式增强任何一个LLM调用函数。2.1 功能定义与边界我们的robust_llm_call包装器需要实现以下核心功能可配置的重试机制允许用户设置最大重试次数、重试的延迟策略例如固定延迟、指数退避。它应该只对特定的、可恢复的异常进行重试比如网络超时、连接错误、服务端返回5xx错误或429请求过多错误。对于客户端错误如4xxAPI密钥错误重试是没有意义的应该立即失败。双重超时控制这里有两个层面的超时。一是连接/读取超时即单次HTTP请求等待响应的最长时间这通常由requests库或httpx库的timeout参数控制。二是整体超时即从第一次尝试开始到所有重试结束或成功的总时间上限。防止一个不断重试的请求无限期占用资源。优雅的降级策略当重试耗尽仍失败时不能直接抛出一个异常让上游崩溃。我们需要一个降级回调函数。这个函数可以很简单比如返回一个“服务暂时不可用”的字符串也可以很复杂比如切换到一个本地运行的轻量级模型如ChatGLM-6B INT4或者从缓存中提取一个近似答案。2.2 技术选型为什么是Tenacity和Httpx要实现重试逻辑最原始的做法是自己写循环和time.sleep。但这需要处理很多细节比如异常类型的判断、退避算法的实现等。更好的选择是使用专门的重试库。Python里最流行的就是tenacity。它通过装饰器的方式工作配置灵活功能强大可以轻松实现我们想要的“带指数退避的、针对特定异常的重试”。对于HTTP客户端标准库的requests固然好用但它在异步支持上有所欠缺。考虑到现代LLM应用可能会在异步框架如FastAPI中使用我们选择httpx。它提供了与requests几乎相同的同步接口并且原生支持异步为未来的扩展留有余地。它的超时配置也更清晰。所以我们的技术栈就定为tenacityhttpx。当然如果你坚持用requests整体思路完全不变只是代码略有调整。2.3 架构设计装饰器模式的应用我们将采用装饰器模式来构建核心功能。装饰器就像一个包装盒你把你原来的LLM调用函数放进去它返回一个新的、具备了重试、超时、降级能力的函数。这样做的好处是解耦弹性逻辑和业务逻辑构造prompt、解析响应完全分离。可复用这个装饰器可以用来包装任何类似的远程服务调用函数。可测试可以单独测试装饰器的逻辑也可以测试包装后的函数。基本的使用形态会是这样retry_with_timeout_and_fallback(...) def call_openai_api(prompt: str) - str: # 原始的、裸调的API调用逻辑 response httpx.post(...) return response.json()[choices][0][message][content] # 调用时这个函数已经自带了“盔甲” result call_openai_api(你好世界)3. 核心代码逐行解析与实现现在我们进入实战环节把这60行左右的代码一行行写出来并解释每一部分的意图和细节。3.1 环境准备与依赖安装首先你需要安装必要的库。打开你的终端执行pip install tenacity httpx如果你使用Poetry或Pipenv等依赖管理工具请将它们加入到你的项目依赖文件中。注意在生产环境中务必使用requirements.txt或pyproject.toml精确锁定版本避免因库版本更新导致意外行为。例如可以指定tenacity8.2.0,9.0.0。3.2 构建核心装饰器函数我们将创建一个名为retry_with_timeout_and_fallback的函数它本身是一个返回装饰器的函数即“装饰器工厂”以便我们能够传入配置参数。import httpx from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log, ) import logging from typing import Callable, Any, Optional import asyncio from functools import wraps # 设置一个日志记录器方便观察重试行为 logger logging.getLogger(__name__) def retry_with_timeout_and_fallback( max_retries: int 3, request_timeout: float 30.0, total_timeout: float 120.0, fallback_func: Optional[Callable[[Exception, dict], Any]] None, ): 一个为LLM API调用添加重试、超时和降级功能的装饰器工厂。 参数: max_retries: 最大重试次数不包括第一次尝试。 request_timeout: 单次HTTP请求的超时时间秒。 total_timeout: 整个操作含所有重试的最大总耗时秒。 fallback_func: 降级函数。接受两个参数最后捕获的异常以及调用时的关键字参数字典。 必须返回一个值作为降级结果。 # 定义我们需要重试的异常类型 retryable_exceptions ( httpx.RequestError, # 包含所有网络相关错误连接超时、读取超时等 httpx.HTTPStatusError, # 对于4xx/5xx状态码我们需要进一步筛选 ) def _is_retryable_http_error(exc: httpx.HTTPStatusError) - bool: 判断一个HTTP状态码错误是否应该重试。 # 429 请求过多是典型的可重试错误配合退避 # 5xx 服务器内部错误通常是暂时的 # 408 请求超时有时也可重试 return exc.response.status_code in {408, 429, 500, 502, 503, 504} # 创建 tenacity 的重试装饰器 tenacity_retry retry( # 停止条件达到最大重试次数或总时间超时 stop(stop_after_attempt(max_retries 1) | _stop_after_total_timeout(total_timeout)), # 等待策略指数退避最小1秒最大30秒。避免“惊群”效应。 waitwait_exponential(multiplier1, min1, max30), # 重试条件异常是指定的可重试类型并且如果是HTTP错误状态码符合要求 retry( retry_if_exception_type(retryable_exceptions) retry_if_exception(_is_retryable_exception) ), # 在每次重试睡眠前记录日志 before_sleepbefore_sleep_log(logger, logging.WARNING), reraiseTrue, # 当所有重试耗尽后重新抛出最后的异常 ) def decorator(func: Callable) - Callable: wraps(func) def wrapper(*args, **kwargs): # 记录开始时间用于计算总耗时 start_time asyncio.get_event_loop().time() if asyncio.iscoroutinefunction(func) else time.time() last_exception None # 定义一个内部函数它将被 tenacity 装饰 tenacity_retry def _retryable_call(): nonlocal last_exception try: # 在这里我们为每次尝试注入超时设置。 # 假设被装饰的函数接受一个 timeout 参数或者我们通过其他方式传递。 # 更通用的做法是如果函数使用 httpx我们可以在其上下文中设置超时。 # 为了简化我们假设被装饰的函数内部已经处理了 request_timeout。 # 实际上我们需要更精细的控制这引出了下一个章节的“上下文管理器”模式。 return func(*args, **kwargs) except Exception as e: last_exception e raise # 重新抛出异常让 tenacity 捕获并决定是否重试 try: return _retryable_call() except Exception as final_exc: # 如果所有重试都失败了执行降级逻辑 if fallback_func is not None: logger.error(f所有重试均失败执行降级函数。最后异常: {final_exc}) # 将调用时的参数通常是prompt等传递给降级函数 call_kwargs kwargs if kwargs else dict(zip(func.__code__.co_varnames, args)) return fallback_func(final_exc, call_kwargs) else: # 没有设置降级则直接抛出异常 logger.critical(fLLM调用失败且未设置降级异常上抛: {final_exc}) raise return wrapper return decorator上面的代码是一个框架但里面有几个关键点需要解释和补全_stop_after_total_timeout函数tenacity没有内置的“总超时”停止条件我们需要自己实现一个。这涉及到在每次重试前检查从开始到现在是否超过了total_timeout。_is_retryable_exception函数我们需要一个统一的函数来判断一个异常是否应该重试。它需要处理httpx.HTTPStatusError检查其状态码。超时参数的传递上面的简化代码假设被装饰的函数自己处理超时。但在实际中我们希望装饰器能统一控制单次请求的超时。这要求我们对被装饰的函数有更多了解或者采用更通用的模式。3.3 实现缺失的辅助函数与完整逻辑让我们补全这些缺失的部分并采用一个更实用的设计我们将装饰器设计为主要与一个执行实际HTTP调用的内部函数配合使用。这个内部函数接收一个配置好的httpx.Client或httpx.AsyncClient对象。import time from tenacity import RetryCallState, stop_base from typing import Type class stop_after_total_timeout(stop_base): Tenacity停止条件总耗时超过指定时间。 def __init__(self, total_timeout: float): self.total_timeout total_timeout self.start_time None def __call__(self, retry_state: RetryCallState) - bool: if self.start_time is None: self.start_time time.time() return (time.time() - self.start_time) self.total_timeout def _is_retryable_exception(exception: Exception) - bool: 判断异常是否属于可重试类型。 if isinstance(exception, httpx.RequestError): # 所有网络请求错误都重试 return True if isinstance(exception, httpx.HTTPStatusError): # 只对特定的HTTP状态码进行重试 retryable_codes {408, 429, 500, 502, 503, 504} return exception.response.status_code in retryable_codes # 其他异常如业务逻辑错误、JSON解析错误不重试 return False现在我们可以编写一个更完整、更模块化的版本。我们将核心的HTTP调用逻辑分离出来让装饰器专注于重试和降级策略。def robust_llm_call( api_endpoint: str, api_key: str, max_retries: int 3, request_timeout: float 30.0, total_timeout: float 120.0, fallback_response: Any 抱歉AI服务暂时不可用请稍后再试。, ): 创建一个配置了重试、超时和降级的LLM调用函数。 这是一个更面向过程的、易于理解的实现。 参数: api_endpoint: LLM API的端点URL。 api_key: API密钥。 ... (其他参数同上) ... fallback_response: 降级时直接返回的固定值。可以是字符串、字典等。 # 配置HTTP客户端设置默认超时和认证头 client httpx.Client( timeoutrequest_timeout, headers{ Authorization: fBearer {api_key}, Content-Type: application/json, } ) # 定义实际执行请求的函数 def _make_request(payload: dict) - httpx.Response: 执行单次HTTP请求。 return client.post(api_endpoint, jsonpayload) # 使用 tenacity 装饰这个内部函数 retry( stop(stop_after_attempt(max_retries 1) | stop_after_total_timeout(total_timeout)), waitwait_exponential(multiplier1, min1, max30), retryretry_if_exception(_is_retryable_exception), before_sleepbefore_sleep_log(logger, logging.WARNING), reraiseTrue, ) def _retryable_make_request(payload: dict) - httpx.Response: return _make_request(payload) # 这是最终暴露给用户的函数 def call_llm(prompt: str, **extra_params) - Any: 调用LLM的主函数。 参数: prompt: 输入的提示词。 **extra_params: 其他传递给API的参数字典如model, temperature等。 返回: LLM的响应文本或降级结果。 payload { model: extra_params.get(model, gpt-3.5-turbo), messages: [{role: user, content: prompt}], **extra_params # 允许覆盖或添加其他参数 } last_exception None try: response _retryable_make_request(payload) response.raise_for_status() # 如果状态码不是2xx抛出HTTPStatusError result response.json() # 这里需要根据实际API的响应结构进行解析以下是一个OpenAI格式的示例 return result[choices][0][message][content].strip() except Exception as e: last_exception e logger.error(fLLM调用最终失败: {e}) # 执行降级 logger.warning(触发降级返回预设响应。) return fallback_response return call_llm # 返回这个配置好的函数这个版本更清晰。我们通过robust_llm_call这个工厂函数传入API配置和弹性策略参数它返回一个可以直接使用的call_llm函数。这个函数内部已经集成了所有健壮性逻辑。3.4 使用示例与效果演示让我们看看如何在实际中使用它# 配置并创建你的强化版LLM调用器 my_llm_caller robust_llm_call( api_endpointhttps://api.openai.com/v1/chat/completions, api_keyyour-api-key-here, max_retries2, request_timeout15.0, total_timeout45.0, fallback_response[服务降级] 当前无法获取AI回复请稍后重试或联系客服。 ) # 像调用普通函数一样使用它 try: answer my_llm_caller(请用Python写一个快速排序函数, modelgpt-4, temperature0.7) print(fAI回复{answer}) except Exception as e: # 只有在降级函数也未设置或自身出错时才会走到这里 print(f调用完全失败: {e}) # 模拟一个会失败的场景比如使用一个无效的端点 unreliable_caller robust_llm_call( api_endpointhttps://httpstat.us/503, # 一个总是返回503状态码的服务 api_keydummy, max_retries2, fallback_response服务繁忙已启用降级模式。 ) result unreliable_caller(你好) print(result) # 输出: “服务繁忙已启用降级模式。”运行这段代码当模拟的API持续返回503错误时你会看到日志中打印出重试的警告信息并在两次重试后返回我们预设的降级响应而不会导致程序崩溃。4. 高级话题与生产级优化上面的代码已经是一个可用的版本但要用于生产环境还需要考虑更多细节。4.1 异步支持现代Python网络应用几乎都是异步的。我们的代码需要支持async/await。幸运的是httpx和tenacity都支持异步。import asyncio import httpx from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential, retry_if_exception async def robust_async_llm_call(api_endpoint: str, api_key: str, **kwargs): client httpx.AsyncClient( timeouthttpx.Timeout(kwargs.get(request_timeout, 30.0)), headers{Authorization: fBearer {api_key}}, ) async def _make_request(payload): return await client.post(api_endpoint, jsonpayload) async for attempt in AsyncRetrying( stopstop_after_attempt(kwargs.get(max_retries, 3) 1), waitwait_exponential(multiplier1, min1, max30), retryretry_if_exception(_is_retryable_exception), reraiseTrue, ): with attempt: response await _make_request(payload) response.raise_for_status() return response.json() # 如果重试循环结束非break说明所有重试都失败了 return kwargs.get(fallback_response)4.2 更精细的降级策略固定字符串降级是最简单的。更高级的降级策略可能包括缓存降级返回上一次对同一问题成功的、未过期的回答。模型降级当主模型如GPT-4不可用时自动尝试调用次一级的模型如GPT-3.5-Turbo或本地模型。规则降级对于一些简单、模式固定的查询如“你好”、“谢谢”直接返回预定义的规则答案完全不走网络请求。这需要你将fallback_response参数从一个固定值替换为一个可调用的函数这个函数能接收到原始的请求参数prompt等从而做出更智能的决策。4.3 监控与度量在生产中你还需要知道这套机制运行得怎么样。你需要记录重试率有多少比例的请求触发了重试失败率在重试后最终失败触发降级的比例是多少延迟分布引入重试和退避后请求的P50、P95、P99延迟增加了多少你可以在装饰器内部添加打点逻辑将数据发送到像Prometheus、StatsD这样的监控系统或者至少记录到结构化日志中。4.4 与现有框架集成如果你在使用LangChain、LlamaIndex等LLM应用框架它们通常有自己的重试和超时配置。例如LangChain的LLM类可以直接设置max_retries和request_timeout参数。我们的方法更底层、更通用适用于任何自定义的API调用或者在这些框架的配置不够灵活时进行补充。5. 常见问题与排查技巧实录在实际使用中你可能会遇到下面这些问题。这里记录了我踩过的坑和解决方法。5.1 重试风暴与退避策略问题一开始我用了固定延迟重试比如每次失败后等2秒。当某个服务端节点出现问题时所有客户端都在同一时间重试导致服务端在恢复的瞬间又被海量重试请求打垮形成“重试风暴”。解决这就是为什么一定要用指数退避wait_exponential。它让每次重试的等待时间指数级增加1s, 2s, 4s, 8s...并加上随机抖动jitter使得客户端的重试时间点分散开极大地缓解了对服务端的冲击。5.2 哪些异常应该重试判断准则一个黄金法则是只重试那些有希望成功的临时性故障。必须重试网络连接错误httpx.ConnectError、读超时httpx.ReadTimeout、HTTP 429请求过多、HTTP 5xx服务器内部错误。这些错误通常是暂时的。不应重试HTTP 4xx客户端错误如401未授权、403禁止访问、404未找到、400错误请求。这些错误通常意味着你的请求本身有问题重试多少次都没用反而会增加负载和成本。务必在_is_retryable_exception函数中仔细过滤。5.3 超时设置多少合适单次请求超时request_timeout这需要根据你调用的模型和上下文长度来评估。一个经验值是简单对话GPT-3.51k tokens10-15秒。复杂任务或长上下文GPT-48k tokens30-60秒甚至更长。总超时total_timeout这应该是(max_retries 1) * request_timeout的1.5到2倍。为指数退避的等待时间留出余量。例如重试3次单次超时15秒那么总超时可以设为(31)*15*1.5 ≈ 90秒。5.4 降级函数本身出错了怎么办问题降级逻辑可能依赖另一个服务如查询缓存数据库如果这个服务也挂了降级函数会抛出异常导致整个流程依然失败。解决降级函数内部必须有非常强的容错能力最好能做到“自包含”。最简单的实现是在降级函数外面再套一层try...except确保它无论如何都能返回一个兜底值。def super_safe_fallback(exc, kwargs): try: # 尝试一些智能降级逻辑... # return smart_result pass except Exception: # 如果智能降级也失败返回最硬的兜底 return 系统繁忙请稍后再试。5.5 如何测试重试和降级逻辑你不能总等着真实API出错来测试。你需要模拟故障。使用Mock在单元测试中使用unittest.mock来模拟httpx.Client.post方法让它第一次调用抛出httpx.ReadTimeout第二次调用成功。验证你的函数是否重试了并且最终返回了正确结果。使用模拟服务启动一个简单的HTTP服务器比如用fastapi在特定端点上编程控制返回不同的状态码如200, 429, 500来测试你的重试逻辑。网络模拟工具使用像toxiproxy这样的工具在本地模拟网络延迟、中断和丢包进行集成测试。5.6 关于API成本与重试的权衡重要提醒重试会增加你的API调用成本特别是对于按token收费的模型。如果一次请求因为网络问题在传输中途失败服务器端可能已经处理了部分计算。无脑重试可能导致你为同一个任务付费多次。建议对于非等幂非幂等的操作尤其是那些会改变服务器状态的虽然LLM completion通常是幂等的重试要格外小心。可以考虑在重试前先判断错误类型。如果是明显的客户端错误4xx或请求内容错误不应重试。设置一个合理的max_retries通常2-3次已经足够。不要设置得过高。6. 完整代码整合与最终建议最后我将一个同步版本的、相对完整的工具函数整合如下。你可以将它保存为一个独立的模块如llm_utils.py然后在项目中导入使用。# llm_utils.py import httpx import time import logging from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception, before_sleep_log, stop_base, ) from typing import Any, Callable, Optional from functools import wraps logger logging.getLogger(__name__) class StopAfterTotalTimeout(stop_base): def __init__(self, total_timeout: float): self.total_timeout total_timeout self.start_time None def __call__(self, retry_state): if self.start_time is None: self.start_time time.time() return (time.time() - self.start_time) self.total_timeout def _is_retryable_exception(exception: Exception) - bool: 判断异常是否可重试。 if isinstance(exception, httpx.RequestError): return True if isinstance(exception, httpx.HTTPStatusError): retryable_codes {408, 429, 500, 502, 503, 504} return exception.response.status_code in retryable_codes return False def create_robust_llm_caller( api_endpoint: str, api_key: str, default_model: str gpt-3.5-turbo, max_retries: int 2, request_timeout: float 20.0, total_timeout: float 60.0, fallback_handler: Optional[Callable[[Exception, dict], Any]] None, ): 创建并返回一个健壮的LLM同步调用函数。 # 创建具有默认超时的客户端 client httpx.Client( timeoutrequest_timeout, headers{ Authorization: fBearer {api_key}, Content-Type: application/json, }, ) # 定义实际请求函数并应用重试逻辑 retry( stop(stop_after_attempt(max_retries 1) | StopAfterTotalTimeout(total_timeout)), waitwait_exponential(multiplier1, min1, max20), retryretry_if_exception(_is_retryable_exception), before_sleepbefore_sleep_log(logger, logging.INFO), reraiseTrue, ) def _make_request_with_retry(payload: dict) - httpx.Response: logger.debug(fSending request to {api_endpoint}, payload keys: {list(payload.keys())}) response client.post(api_endpoint, jsonpayload) # 注意raise_for_status() 会在状态码非2xx时抛出HTTPStatusError # 这个异常会被我们的 _is_retryable_exception 函数判断。 response.raise_for_status() return response def call_llm(prompt: str, **extra_params) - Any: 核心调用函数。 payload { model: extra_params.pop(model, default_model), messages: [{role: user, content: prompt}], **extra_params, } last_exception None try: response _make_request_with_retry(payload) result response.json() # 解析响应这里需要适配你的具体API # 以OpenAI格式为例 if choices in result and len(result[choices]) 0: return result[choices][0][message][content].strip() else: # 如果响应格式不符合预期视为不可重试的业务错误 raise ValueError(fUnexpected API response format: {result}) except Exception as e: last_exception e logger.error(fLLM call failed after retries: {e}, exc_infoTrue) # 执行降级 if fallback_handler is not None: try: return fallback_handler(last_exception, {prompt: prompt, **extra_params}) except Exception as fallback_error: logger.critical(fFallback handler also failed: {fallback_error}) # 降级函数自身失败返回一个最基础的硬编码值 return [System] Service temporarily unavailable. else: # 没有设置降级抛出异常 if last_exception: raise last_exception else: raise RuntimeError(LLM call failed for unknown reason.) return call_llm # 示例一个简单的固定消息降级函数 def simple_fallback(exc: Exception, context: dict) - str: prompt context.get(prompt, ) # 你可以根据异常类型或prompt内容决定不同的降级消息 if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code 429: return 请求过于频繁请稍后重试。 return f无法处理您的请求{prompt[:30]}...。请检查网络或稍后再试。 # 使用示例 if __name__ __main__: # 配置日志 logging.basicConfig(levellogging.INFO) # 创建调用器请替换为真实的API信息 caller create_robust_llm_caller( api_endpointhttps://api.openai.com/v1/chat/completions, api_keysk-..., max_retries2, fallback_handlersimple_fallback, ) # 进行调用 try: response caller(你好请介绍一下你自己。, temperature0.5) print(f成功: {response}) except Exception as e: # 只有在无降级且所有重试失败时才会到达这里 print(f调用彻底失败: {e})最终建议这套“重试超时降级”的机制是你构建可靠AI应用的基础设施。它不能保证100%成功但能将偶发故障对用户的影响降到最低。在实际项目中你还可以将它和断路器模式、限流、监控报警结合起来形成一个更具弹性的系统。记住面对不稳定的外部服务永远要做最坏的打算并为此做好准备。这60行代码就是你从“玩具项目”迈向“生产系统”的第一步。 Python实现LLM API调用重试、超时与降级机制
1. 项目概述为什么我们需要给LLM调用“上保险”直接调用大模型API比如OpenAI的ChatCompletion或者国内各种模型的接口是很多开发者上手LLM应用的第一步。代码简单到几行就能跑通这给了我们一种“大模型调用很简单”的错觉。但只要你把这样的代码放到生产环境或者跑一个需要连续处理上百条请求的脚本各种幺蛾子就会接踵而至网络突然波动一下API返回一个超时错误模型服务端因为负载过高给你抛回一个“rate limit”的429状态码甚至有时候模型本身会抽风返回一些无法解析的乱码或者结构错误的JSON。这时候如果你的代码只是简单的一个requests.post()外面套个try...except那整个流程就卡死了。用户看到的是“服务不可用”而你半夜会被报警电话叫醒手忙脚乱地去重启服务或者重跑脚本。这其实就是“裸调”大模型——没有任何弹性处理机制的调用方式脆弱得就像在裸奔。所以这个项目的核心价值就是给裸奔的LLM调用穿上“盔甲”。我们用大约60行Python代码实现三个最核心的弹性机制重试、超时和降级。重试负责在遇到临时性失败如网络抖动、服务端限流时自动再试几次提高单次请求的最终成功率。超时负责设置一个等待上限防止某个慢请求拖死整个线程或进程。降级则是最后的保障当所有重试都失败后提供一个保底结果比如返回一个预设的友好错误提示或者切换到一个更简单、更稳定的备用模型确保主流程不会彻底中断。这不仅仅是让代码更健壮更是工程思维的体现。在真实的生产系统中对待外部服务尤其是像LLM API这种可能昂贵且不稳定的服务的调用必须假设它可能失败并为此做好准备。接下来我会把这套机制的每一个零件拆开讲清楚原理并给你一份可以直接复制粘贴使用的代码。2. 核心设计思路构建一个健壮的LLM客户端在开始写代码之前我们先要理清楚一个健壮的LLM调用客户端应该长什么样。我们不能简单地在业务逻辑里到处写for i in range(3): try...except...那样代码会又臭又长难以维护。我们的目标是设计一个可复用的“装饰器”或“包装器”它能够以最小侵入的方式增强任何一个LLM调用函数。2.1 功能定义与边界我们的robust_llm_call包装器需要实现以下核心功能可配置的重试机制允许用户设置最大重试次数、重试的延迟策略例如固定延迟、指数退避。它应该只对特定的、可恢复的异常进行重试比如网络超时、连接错误、服务端返回5xx错误或429请求过多错误。对于客户端错误如4xxAPI密钥错误重试是没有意义的应该立即失败。双重超时控制这里有两个层面的超时。一是连接/读取超时即单次HTTP请求等待响应的最长时间这通常由requests库或httpx库的timeout参数控制。二是整体超时即从第一次尝试开始到所有重试结束或成功的总时间上限。防止一个不断重试的请求无限期占用资源。优雅的降级策略当重试耗尽仍失败时不能直接抛出一个异常让上游崩溃。我们需要一个降级回调函数。这个函数可以很简单比如返回一个“服务暂时不可用”的字符串也可以很复杂比如切换到一个本地运行的轻量级模型如ChatGLM-6B INT4或者从缓存中提取一个近似答案。2.2 技术选型为什么是Tenacity和Httpx要实现重试逻辑最原始的做法是自己写循环和time.sleep。但这需要处理很多细节比如异常类型的判断、退避算法的实现等。更好的选择是使用专门的重试库。Python里最流行的就是tenacity。它通过装饰器的方式工作配置灵活功能强大可以轻松实现我们想要的“带指数退避的、针对特定异常的重试”。对于HTTP客户端标准库的requests固然好用但它在异步支持上有所欠缺。考虑到现代LLM应用可能会在异步框架如FastAPI中使用我们选择httpx。它提供了与requests几乎相同的同步接口并且原生支持异步为未来的扩展留有余地。它的超时配置也更清晰。所以我们的技术栈就定为tenacityhttpx。当然如果你坚持用requests整体思路完全不变只是代码略有调整。2.3 架构设计装饰器模式的应用我们将采用装饰器模式来构建核心功能。装饰器就像一个包装盒你把你原来的LLM调用函数放进去它返回一个新的、具备了重试、超时、降级能力的函数。这样做的好处是解耦弹性逻辑和业务逻辑构造prompt、解析响应完全分离。可复用这个装饰器可以用来包装任何类似的远程服务调用函数。可测试可以单独测试装饰器的逻辑也可以测试包装后的函数。基本的使用形态会是这样retry_with_timeout_and_fallback(...) def call_openai_api(prompt: str) - str: # 原始的、裸调的API调用逻辑 response httpx.post(...) return response.json()[choices][0][message][content] # 调用时这个函数已经自带了“盔甲” result call_openai_api(你好世界)3. 核心代码逐行解析与实现现在我们进入实战环节把这60行左右的代码一行行写出来并解释每一部分的意图和细节。3.1 环境准备与依赖安装首先你需要安装必要的库。打开你的终端执行pip install tenacity httpx如果你使用Poetry或Pipenv等依赖管理工具请将它们加入到你的项目依赖文件中。注意在生产环境中务必使用requirements.txt或pyproject.toml精确锁定版本避免因库版本更新导致意外行为。例如可以指定tenacity8.2.0,9.0.0。3.2 构建核心装饰器函数我们将创建一个名为retry_with_timeout_and_fallback的函数它本身是一个返回装饰器的函数即“装饰器工厂”以便我们能够传入配置参数。import httpx from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log, ) import logging from typing import Callable, Any, Optional import asyncio from functools import wraps # 设置一个日志记录器方便观察重试行为 logger logging.getLogger(__name__) def retry_with_timeout_and_fallback( max_retries: int 3, request_timeout: float 30.0, total_timeout: float 120.0, fallback_func: Optional[Callable[[Exception, dict], Any]] None, ): 一个为LLM API调用添加重试、超时和降级功能的装饰器工厂。 参数: max_retries: 最大重试次数不包括第一次尝试。 request_timeout: 单次HTTP请求的超时时间秒。 total_timeout: 整个操作含所有重试的最大总耗时秒。 fallback_func: 降级函数。接受两个参数最后捕获的异常以及调用时的关键字参数字典。 必须返回一个值作为降级结果。 # 定义我们需要重试的异常类型 retryable_exceptions ( httpx.RequestError, # 包含所有网络相关错误连接超时、读取超时等 httpx.HTTPStatusError, # 对于4xx/5xx状态码我们需要进一步筛选 ) def _is_retryable_http_error(exc: httpx.HTTPStatusError) - bool: 判断一个HTTP状态码错误是否应该重试。 # 429 请求过多是典型的可重试错误配合退避 # 5xx 服务器内部错误通常是暂时的 # 408 请求超时有时也可重试 return exc.response.status_code in {408, 429, 500, 502, 503, 504} # 创建 tenacity 的重试装饰器 tenacity_retry retry( # 停止条件达到最大重试次数或总时间超时 stop(stop_after_attempt(max_retries 1) | _stop_after_total_timeout(total_timeout)), # 等待策略指数退避最小1秒最大30秒。避免“惊群”效应。 waitwait_exponential(multiplier1, min1, max30), # 重试条件异常是指定的可重试类型并且如果是HTTP错误状态码符合要求 retry( retry_if_exception_type(retryable_exceptions) retry_if_exception(_is_retryable_exception) ), # 在每次重试睡眠前记录日志 before_sleepbefore_sleep_log(logger, logging.WARNING), reraiseTrue, # 当所有重试耗尽后重新抛出最后的异常 ) def decorator(func: Callable) - Callable: wraps(func) def wrapper(*args, **kwargs): # 记录开始时间用于计算总耗时 start_time asyncio.get_event_loop().time() if asyncio.iscoroutinefunction(func) else time.time() last_exception None # 定义一个内部函数它将被 tenacity 装饰 tenacity_retry def _retryable_call(): nonlocal last_exception try: # 在这里我们为每次尝试注入超时设置。 # 假设被装饰的函数接受一个 timeout 参数或者我们通过其他方式传递。 # 更通用的做法是如果函数使用 httpx我们可以在其上下文中设置超时。 # 为了简化我们假设被装饰的函数内部已经处理了 request_timeout。 # 实际上我们需要更精细的控制这引出了下一个章节的“上下文管理器”模式。 return func(*args, **kwargs) except Exception as e: last_exception e raise # 重新抛出异常让 tenacity 捕获并决定是否重试 try: return _retryable_call() except Exception as final_exc: # 如果所有重试都失败了执行降级逻辑 if fallback_func is not None: logger.error(f所有重试均失败执行降级函数。最后异常: {final_exc}) # 将调用时的参数通常是prompt等传递给降级函数 call_kwargs kwargs if kwargs else dict(zip(func.__code__.co_varnames, args)) return fallback_func(final_exc, call_kwargs) else: # 没有设置降级则直接抛出异常 logger.critical(fLLM调用失败且未设置降级异常上抛: {final_exc}) raise return wrapper return decorator上面的代码是一个框架但里面有几个关键点需要解释和补全_stop_after_total_timeout函数tenacity没有内置的“总超时”停止条件我们需要自己实现一个。这涉及到在每次重试前检查从开始到现在是否超过了total_timeout。_is_retryable_exception函数我们需要一个统一的函数来判断一个异常是否应该重试。它需要处理httpx.HTTPStatusError检查其状态码。超时参数的传递上面的简化代码假设被装饰的函数自己处理超时。但在实际中我们希望装饰器能统一控制单次请求的超时。这要求我们对被装饰的函数有更多了解或者采用更通用的模式。3.3 实现缺失的辅助函数与完整逻辑让我们补全这些缺失的部分并采用一个更实用的设计我们将装饰器设计为主要与一个执行实际HTTP调用的内部函数配合使用。这个内部函数接收一个配置好的httpx.Client或httpx.AsyncClient对象。import time from tenacity import RetryCallState, stop_base from typing import Type class stop_after_total_timeout(stop_base): Tenacity停止条件总耗时超过指定时间。 def __init__(self, total_timeout: float): self.total_timeout total_timeout self.start_time None def __call__(self, retry_state: RetryCallState) - bool: if self.start_time is None: self.start_time time.time() return (time.time() - self.start_time) self.total_timeout def _is_retryable_exception(exception: Exception) - bool: 判断异常是否属于可重试类型。 if isinstance(exception, httpx.RequestError): # 所有网络请求错误都重试 return True if isinstance(exception, httpx.HTTPStatusError): # 只对特定的HTTP状态码进行重试 retryable_codes {408, 429, 500, 502, 503, 504} return exception.response.status_code in retryable_codes # 其他异常如业务逻辑错误、JSON解析错误不重试 return False现在我们可以编写一个更完整、更模块化的版本。我们将核心的HTTP调用逻辑分离出来让装饰器专注于重试和降级策略。def robust_llm_call( api_endpoint: str, api_key: str, max_retries: int 3, request_timeout: float 30.0, total_timeout: float 120.0, fallback_response: Any 抱歉AI服务暂时不可用请稍后再试。, ): 创建一个配置了重试、超时和降级的LLM调用函数。 这是一个更面向过程的、易于理解的实现。 参数: api_endpoint: LLM API的端点URL。 api_key: API密钥。 ... (其他参数同上) ... fallback_response: 降级时直接返回的固定值。可以是字符串、字典等。 # 配置HTTP客户端设置默认超时和认证头 client httpx.Client( timeoutrequest_timeout, headers{ Authorization: fBearer {api_key}, Content-Type: application/json, } ) # 定义实际执行请求的函数 def _make_request(payload: dict) - httpx.Response: 执行单次HTTP请求。 return client.post(api_endpoint, jsonpayload) # 使用 tenacity 装饰这个内部函数 retry( stop(stop_after_attempt(max_retries 1) | stop_after_total_timeout(total_timeout)), waitwait_exponential(multiplier1, min1, max30), retryretry_if_exception(_is_retryable_exception), before_sleepbefore_sleep_log(logger, logging.WARNING), reraiseTrue, ) def _retryable_make_request(payload: dict) - httpx.Response: return _make_request(payload) # 这是最终暴露给用户的函数 def call_llm(prompt: str, **extra_params) - Any: 调用LLM的主函数。 参数: prompt: 输入的提示词。 **extra_params: 其他传递给API的参数字典如model, temperature等。 返回: LLM的响应文本或降级结果。 payload { model: extra_params.get(model, gpt-3.5-turbo), messages: [{role: user, content: prompt}], **extra_params # 允许覆盖或添加其他参数 } last_exception None try: response _retryable_make_request(payload) response.raise_for_status() # 如果状态码不是2xx抛出HTTPStatusError result response.json() # 这里需要根据实际API的响应结构进行解析以下是一个OpenAI格式的示例 return result[choices][0][message][content].strip() except Exception as e: last_exception e logger.error(fLLM调用最终失败: {e}) # 执行降级 logger.warning(触发降级返回预设响应。) return fallback_response return call_llm # 返回这个配置好的函数这个版本更清晰。我们通过robust_llm_call这个工厂函数传入API配置和弹性策略参数它返回一个可以直接使用的call_llm函数。这个函数内部已经集成了所有健壮性逻辑。3.4 使用示例与效果演示让我们看看如何在实际中使用它# 配置并创建你的强化版LLM调用器 my_llm_caller robust_llm_call( api_endpointhttps://api.openai.com/v1/chat/completions, api_keyyour-api-key-here, max_retries2, request_timeout15.0, total_timeout45.0, fallback_response[服务降级] 当前无法获取AI回复请稍后重试或联系客服。 ) # 像调用普通函数一样使用它 try: answer my_llm_caller(请用Python写一个快速排序函数, modelgpt-4, temperature0.7) print(fAI回复{answer}) except Exception as e: # 只有在降级函数也未设置或自身出错时才会走到这里 print(f调用完全失败: {e}) # 模拟一个会失败的场景比如使用一个无效的端点 unreliable_caller robust_llm_call( api_endpointhttps://httpstat.us/503, # 一个总是返回503状态码的服务 api_keydummy, max_retries2, fallback_response服务繁忙已启用降级模式。 ) result unreliable_caller(你好) print(result) # 输出: “服务繁忙已启用降级模式。”运行这段代码当模拟的API持续返回503错误时你会看到日志中打印出重试的警告信息并在两次重试后返回我们预设的降级响应而不会导致程序崩溃。4. 高级话题与生产级优化上面的代码已经是一个可用的版本但要用于生产环境还需要考虑更多细节。4.1 异步支持现代Python网络应用几乎都是异步的。我们的代码需要支持async/await。幸运的是httpx和tenacity都支持异步。import asyncio import httpx from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential, retry_if_exception async def robust_async_llm_call(api_endpoint: str, api_key: str, **kwargs): client httpx.AsyncClient( timeouthttpx.Timeout(kwargs.get(request_timeout, 30.0)), headers{Authorization: fBearer {api_key}}, ) async def _make_request(payload): return await client.post(api_endpoint, jsonpayload) async for attempt in AsyncRetrying( stopstop_after_attempt(kwargs.get(max_retries, 3) 1), waitwait_exponential(multiplier1, min1, max30), retryretry_if_exception(_is_retryable_exception), reraiseTrue, ): with attempt: response await _make_request(payload) response.raise_for_status() return response.json() # 如果重试循环结束非break说明所有重试都失败了 return kwargs.get(fallback_response)4.2 更精细的降级策略固定字符串降级是最简单的。更高级的降级策略可能包括缓存降级返回上一次对同一问题成功的、未过期的回答。模型降级当主模型如GPT-4不可用时自动尝试调用次一级的模型如GPT-3.5-Turbo或本地模型。规则降级对于一些简单、模式固定的查询如“你好”、“谢谢”直接返回预定义的规则答案完全不走网络请求。这需要你将fallback_response参数从一个固定值替换为一个可调用的函数这个函数能接收到原始的请求参数prompt等从而做出更智能的决策。4.3 监控与度量在生产中你还需要知道这套机制运行得怎么样。你需要记录重试率有多少比例的请求触发了重试失败率在重试后最终失败触发降级的比例是多少延迟分布引入重试和退避后请求的P50、P95、P99延迟增加了多少你可以在装饰器内部添加打点逻辑将数据发送到像Prometheus、StatsD这样的监控系统或者至少记录到结构化日志中。4.4 与现有框架集成如果你在使用LangChain、LlamaIndex等LLM应用框架它们通常有自己的重试和超时配置。例如LangChain的LLM类可以直接设置max_retries和request_timeout参数。我们的方法更底层、更通用适用于任何自定义的API调用或者在这些框架的配置不够灵活时进行补充。5. 常见问题与排查技巧实录在实际使用中你可能会遇到下面这些问题。这里记录了我踩过的坑和解决方法。5.1 重试风暴与退避策略问题一开始我用了固定延迟重试比如每次失败后等2秒。当某个服务端节点出现问题时所有客户端都在同一时间重试导致服务端在恢复的瞬间又被海量重试请求打垮形成“重试风暴”。解决这就是为什么一定要用指数退避wait_exponential。它让每次重试的等待时间指数级增加1s, 2s, 4s, 8s...并加上随机抖动jitter使得客户端的重试时间点分散开极大地缓解了对服务端的冲击。5.2 哪些异常应该重试判断准则一个黄金法则是只重试那些有希望成功的临时性故障。必须重试网络连接错误httpx.ConnectError、读超时httpx.ReadTimeout、HTTP 429请求过多、HTTP 5xx服务器内部错误。这些错误通常是暂时的。不应重试HTTP 4xx客户端错误如401未授权、403禁止访问、404未找到、400错误请求。这些错误通常意味着你的请求本身有问题重试多少次都没用反而会增加负载和成本。务必在_is_retryable_exception函数中仔细过滤。5.3 超时设置多少合适单次请求超时request_timeout这需要根据你调用的模型和上下文长度来评估。一个经验值是简单对话GPT-3.51k tokens10-15秒。复杂任务或长上下文GPT-48k tokens30-60秒甚至更长。总超时total_timeout这应该是(max_retries 1) * request_timeout的1.5到2倍。为指数退避的等待时间留出余量。例如重试3次单次超时15秒那么总超时可以设为(31)*15*1.5 ≈ 90秒。5.4 降级函数本身出错了怎么办问题降级逻辑可能依赖另一个服务如查询缓存数据库如果这个服务也挂了降级函数会抛出异常导致整个流程依然失败。解决降级函数内部必须有非常强的容错能力最好能做到“自包含”。最简单的实现是在降级函数外面再套一层try...except确保它无论如何都能返回一个兜底值。def super_safe_fallback(exc, kwargs): try: # 尝试一些智能降级逻辑... # return smart_result pass except Exception: # 如果智能降级也失败返回最硬的兜底 return 系统繁忙请稍后再试。5.5 如何测试重试和降级逻辑你不能总等着真实API出错来测试。你需要模拟故障。使用Mock在单元测试中使用unittest.mock来模拟httpx.Client.post方法让它第一次调用抛出httpx.ReadTimeout第二次调用成功。验证你的函数是否重试了并且最终返回了正确结果。使用模拟服务启动一个简单的HTTP服务器比如用fastapi在特定端点上编程控制返回不同的状态码如200, 429, 500来测试你的重试逻辑。网络模拟工具使用像toxiproxy这样的工具在本地模拟网络延迟、中断和丢包进行集成测试。5.6 关于API成本与重试的权衡重要提醒重试会增加你的API调用成本特别是对于按token收费的模型。如果一次请求因为网络问题在传输中途失败服务器端可能已经处理了部分计算。无脑重试可能导致你为同一个任务付费多次。建议对于非等幂非幂等的操作尤其是那些会改变服务器状态的虽然LLM completion通常是幂等的重试要格外小心。可以考虑在重试前先判断错误类型。如果是明显的客户端错误4xx或请求内容错误不应重试。设置一个合理的max_retries通常2-3次已经足够。不要设置得过高。6. 完整代码整合与最终建议最后我将一个同步版本的、相对完整的工具函数整合如下。你可以将它保存为一个独立的模块如llm_utils.py然后在项目中导入使用。# llm_utils.py import httpx import time import logging from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception, before_sleep_log, stop_base, ) from typing import Any, Callable, Optional from functools import wraps logger logging.getLogger(__name__) class StopAfterTotalTimeout(stop_base): def __init__(self, total_timeout: float): self.total_timeout total_timeout self.start_time None def __call__(self, retry_state): if self.start_time is None: self.start_time time.time() return (time.time() - self.start_time) self.total_timeout def _is_retryable_exception(exception: Exception) - bool: 判断异常是否可重试。 if isinstance(exception, httpx.RequestError): return True if isinstance(exception, httpx.HTTPStatusError): retryable_codes {408, 429, 500, 502, 503, 504} return exception.response.status_code in retryable_codes return False def create_robust_llm_caller( api_endpoint: str, api_key: str, default_model: str gpt-3.5-turbo, max_retries: int 2, request_timeout: float 20.0, total_timeout: float 60.0, fallback_handler: Optional[Callable[[Exception, dict], Any]] None, ): 创建并返回一个健壮的LLM同步调用函数。 # 创建具有默认超时的客户端 client httpx.Client( timeoutrequest_timeout, headers{ Authorization: fBearer {api_key}, Content-Type: application/json, }, ) # 定义实际请求函数并应用重试逻辑 retry( stop(stop_after_attempt(max_retries 1) | StopAfterTotalTimeout(total_timeout)), waitwait_exponential(multiplier1, min1, max20), retryretry_if_exception(_is_retryable_exception), before_sleepbefore_sleep_log(logger, logging.INFO), reraiseTrue, ) def _make_request_with_retry(payload: dict) - httpx.Response: logger.debug(fSending request to {api_endpoint}, payload keys: {list(payload.keys())}) response client.post(api_endpoint, jsonpayload) # 注意raise_for_status() 会在状态码非2xx时抛出HTTPStatusError # 这个异常会被我们的 _is_retryable_exception 函数判断。 response.raise_for_status() return response def call_llm(prompt: str, **extra_params) - Any: 核心调用函数。 payload { model: extra_params.pop(model, default_model), messages: [{role: user, content: prompt}], **extra_params, } last_exception None try: response _make_request_with_retry(payload) result response.json() # 解析响应这里需要适配你的具体API # 以OpenAI格式为例 if choices in result and len(result[choices]) 0: return result[choices][0][message][content].strip() else: # 如果响应格式不符合预期视为不可重试的业务错误 raise ValueError(fUnexpected API response format: {result}) except Exception as e: last_exception e logger.error(fLLM call failed after retries: {e}, exc_infoTrue) # 执行降级 if fallback_handler is not None: try: return fallback_handler(last_exception, {prompt: prompt, **extra_params}) except Exception as fallback_error: logger.critical(fFallback handler also failed: {fallback_error}) # 降级函数自身失败返回一个最基础的硬编码值 return [System] Service temporarily unavailable. else: # 没有设置降级抛出异常 if last_exception: raise last_exception else: raise RuntimeError(LLM call failed for unknown reason.) return call_llm # 示例一个简单的固定消息降级函数 def simple_fallback(exc: Exception, context: dict) - str: prompt context.get(prompt, ) # 你可以根据异常类型或prompt内容决定不同的降级消息 if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code 429: return 请求过于频繁请稍后重试。 return f无法处理您的请求{prompt[:30]}...。请检查网络或稍后再试。 # 使用示例 if __name__ __main__: # 配置日志 logging.basicConfig(levellogging.INFO) # 创建调用器请替换为真实的API信息 caller create_robust_llm_caller( api_endpointhttps://api.openai.com/v1/chat/completions, api_keysk-..., max_retries2, fallback_handlersimple_fallback, ) # 进行调用 try: response caller(你好请介绍一下你自己。, temperature0.5) print(f成功: {response}) except Exception as e: # 只有在无降级且所有重试失败时才会到达这里 print(f调用彻底失败: {e})最终建议这套“重试超时降级”的机制是你构建可靠AI应用的基础设施。它不能保证100%成功但能将偶发故障对用户的影响降到最低。在实际项目中你还可以将它和断路器模式、限流、监控报警结合起来形成一个更具弹性的系统。记住面对不稳定的外部服务永远要做最坏的打算并为此做好准备。这60行代码就是你从“玩具项目”迈向“生产系统”的第一步。