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包装器需要实现以下核心功能：

1. 可配置的重试机制：允许用户设置最大重试次数、重试的延迟策略（例如，固定延迟、指数退避）。它应该只对特定的、可恢复的异常进行重试，比如网络超时、连接错误、服务端返回5xx错误或429（请求过多）错误。对于客户端错误（如4xx，API密钥错误），重试是没有意义的，应该立即失败。
2. 双重超时控制：这里有两个层面的超时。一是连接/读取超时，即单次HTTP请求等待响应的最长时间，这通常由requests库或httpx库的timeout参数控制。二是整体超时，即从第一次尝试开始，到所有重试结束（或成功）的总时间上限。防止一个不断重试的请求无限期占用资源。
3. 优雅的降级策略：当重试耗尽仍失败时，不能直接抛出一个异常让上游崩溃。我们需要一个降级回调函数。这个函数可以很简单，比如返回一个“服务暂时不可用”的字符串；也可以很复杂，比如切换到一个本地运行的轻量级模型（如ChatGLM-6B INT4），或者从缓存中提取一个近似答案。

2.2 技术选型：为什么是Tenacity和Httpx？

要实现重试逻辑，最原始的做法是自己写循环和time.sleep。但这需要处理很多细节，比如异常类型的判断、退避算法的实现等。更好的选择是使用专门的重试库。Python里最流行的就是tenacity。它通过装饰器的方式工作，配置灵活，功能强大，可以轻松实现我们想要的“带指数退避的、针对特定异常的重试”。

对于HTTP客户端，标准库的requests固然好用，但它在异步支持上有所欠缺。考虑到现代LLM应用可能会在异步框架（如FastAPI）中使用，我们选择httpx。它提供了与requests几乎相同的同步接口，并且原生支持异步，为未来的扩展留有余地。它的超时配置也更清晰。

所以，我们的技术栈就定为：tenacity+httpx。当然，如果你坚持用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精确锁定版本，避免因库版本更新导致意外行为。例如，可以指定tenacity>=8.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秒。避免“惊群”效应。 wait=wait_exponential(multiplier=1, min=1, max=30), # 重试条件：异常是指定的可重试类型，并且如果是HTTP错误，状态码符合要求 retry=( retry_if_exception_type(retryable_exceptions) & retry_if_exception(_is_retryable_exception) ), # 在每次重试睡眠前记录日志 before_sleep=before_sleep_log(logger, logging.WARNING), reraise=True, # 当所有重试耗尽后，重新抛出最后的异常 ) 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(f"LLM调用失败且未设置降级，异常上抛: {final_exc}") raise return wrapper return decorator

上面的代码是一个框架，但里面有几个关键点需要解释和补全：

1. _stop_after_total_timeout函数：tenacity没有内置的“总超时”停止条件，我们需要自己实现一个。这涉及到在每次重试前检查从开始到现在是否超过了total_timeout。
2. _is_retryable_exception函数：我们需要一个统一的函数来判断一个异常是否应该重试。它需要处理httpx.HTTPStatusError，检查其状态码。
3. 超时参数的传递：上面的简化代码假设被装饰的函数自己处理超时。但在实际中，我们希望装饰器能统一控制单次请求的超时。这要求我们对被装饰的函数有更多了解，或者采用更通用的模式。

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( timeout=request_timeout, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } ) # 定义实际执行请求的函数 def _make_request(payload: dict) -> httpx.Response: """执行单次HTTP请求。""" return client.post(api_endpoint, json=payload) # 使用 tenacity 装饰这个内部函数 @retry( stop=(stop_after_attempt(max_retries + 1) | stop_after_total_timeout(total_timeout)), wait=wait_exponential(multiplier=1, min=1, max=30), retry=retry_if_exception(_is_retryable_exception), before_sleep=before_sleep_log(logger, logging.WARNING), reraise=True, ) 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(f"LLM调用最终失败: {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_endpoint="https://api.openai.com/v1/chat/completions", api_key="your-api-key-here", max_retries=2, request_timeout=15.0, total_timeout=45.0, fallback_response="[服务降级] 当前无法获取AI回复，请稍后重试或联系客服。" ) # 像调用普通函数一样使用它 try: answer = my_llm_caller("请用Python写一个快速排序函数", model="gpt-4", temperature=0.7) print(f"AI回复：{answer}") except Exception as e: # 只有在降级函数也未设置或自身出错时，才会走到这里 print(f"调用完全失败: {e}") # 模拟一个会失败的场景（比如使用一个无效的端点） unreliable_caller = robust_llm_call( api_endpoint="https://httpstat.us/503", # 一个总是返回503状态码的服务 api_key="dummy", max_retries=2, 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( timeout=httpx.Timeout(kwargs.get('request_timeout', 30.0)), headers={"Authorization": f"Bearer {api_key}"}, ) async def _make_request(payload): return await client.post(api_endpoint, json=payload) async for attempt in AsyncRetrying( stop=stop_after_attempt(kwargs.get('max_retries', 3) + 1), wait=wait_exponential(multiplier=1, min=1, max=30), retry=retry_if_exception(_is_retryable_exception), reraise=True, ): 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.5，<1k tokens）：10-15秒。
• 复杂任务或长上下文（GPT-4，>8k tokens）：30-60秒甚至更长。总超时（total_timeout）：这应该是(max_retries + 1) * request_timeout的1.5到2倍。为指数退避的等待时间留出余量。例如，重试3次，单次超时15秒，那么总超时可以设为(3+1)*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出错来测试。你需要模拟故障。

1. 使用Mock：在单元测试中，使用unittest.mock来模拟httpx.Client.post方法，让它第一次调用抛出httpx.ReadTimeout，第二次调用成功。验证你的函数是否重试了，并且最终返回了正确结果。
2. 使用模拟服务：启动一个简单的HTTP服务器（比如用fastapi），在特定端点上编程控制返回不同的状态码（如200, 429, 500）来测试你的重试逻辑。
3. 网络模拟工具：使用像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( timeout=request_timeout, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, ) # 定义实际请求函数，并应用重试逻辑 @retry( stop=(stop_after_attempt(max_retries + 1) | StopAfterTotalTimeout(total_timeout)), wait=wait_exponential(multiplier=1, min=1, max=20), retry=retry_if_exception(_is_retryable_exception), before_sleep=before_sleep_log(logger, logging.INFO), reraise=True, ) def _make_request_with_retry(payload: dict) -> httpx.Response: logger.debug(f"Sending request to {api_endpoint}, payload keys: {list(payload.keys())}") response = client.post(api_endpoint, json=payload) # 注意：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(f"Unexpected API response format: {result}") except Exception as e: last_exception = e logger.error(f"LLM call failed after retries: {e}", exc_info=True) # 执行降级 if fallback_handler is not None: try: return fallback_handler(last_exception, {"prompt": prompt, **extra_params}) except Exception as fallback_error: logger.critical(f"Fallback 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(level=logging.INFO) # 创建调用器（请替换为真实的API信息） caller = create_robust_llm_caller( api_endpoint="https://api.openai.com/v1/chat/completions", api_key="sk-...", max_retries=2, fallback_handler=simple_fallback, ) # 进行调用 try: response = caller("你好，请介绍一下你自己。", temperature=0.5) print(f"成功: {response}") except Exception as e: # 只有在无降级且所有重试失败时才会到达这里 print(f"调用彻底失败: {e}")

最终建议：这套“重试+超时+降级”的机制，是你构建可靠AI应用的基础设施。它不能保证100%成功，但能将偶发故障对用户的影响降到最低。在实际项目中，你还可以将它和断路器模式、限流、监控报警结合起来，形成一个更具弹性的系统。记住，面对不稳定的外部服务，永远要做最坏的打算，并为此做好准备。这60行代码，就是你从“玩具项目”迈向“生产系统”的第一步。