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行代码，就是你从“玩具项目”迈向“生产系统”的第一步。