httpx 与韧性设计：AI 应用的 HTTP 客户端实践

本文适合谁：有 Java OkHttp/RestTemplate/Resilience4j 使用经验的工程师，想构建生产级 LLM API 客户端的开发者。读完本篇，你能配置合理的超时、实现正确的重试策略，处理 LLM API 的限流场景。

调用 LLM API 是一种特殊的 HTTP 请求：延迟高（几秒甚至十几秒）、响应大（流式输出持续传输）、失败模式多样（429 限流、503 过载、网络抖动）。为这类场景设计 HTTP 客户端，远比处理普通 REST API 复杂。

本文以 httpx 为核心，系统介绍 AI 应用的 HTTP 客户端实践，涵盖超时配置、重试策略、流式响应和生产级封装。

1.1 为什么不用 requests

超时控制、重试策略、熔断器、降级处理四层防护架构

requests 是 Python HTTP 客户端的事实标准，但在 AI 应用场景下有几个根本限制：

python

# requests 的核心限制演示
import requests

# 限制一：不支持 async（阻塞事件循环）
# async def call_api():
#     response = requests.get(url)  # 这会阻塞整个事件循环！

# 限制二：连接池配置不灵活
session = requests.Session()
# 默认连接池大小为 10，无法按 host 独立配置
# 无法设置 connect_timeout 和 read_timeout 分离

# 限制三：不支持 HTTP/2（LLM API 大多支持 HTTP/2 多路复用）
# requests 只支持 HTTP/1.1

特性	requests	httpx
async 支持	否	是（`AsyncClient`）
HTTP/2（HTTP 协议的第二版，支持多路复用，同一连接可并行发多个请求，比 HTTP/1.1 更高效）	否	是（需安装 `httpx[http2]`）
超时粒度	connect + read	connect / read / write / pool 分离
连接池	基础	精细可配置（limits 参数）
流式响应	有限	原生 `stream()` 支持
类型标注	部分	完整

1.2 httpx 核心：AsyncClient 与超时配置

1.2.1 连接池与基础配置

python

import httpx
from typing import AsyncIterator

# 推荐做法：复用 AsyncClient，而非每次请求新建
# 连接池在 Client 对象上，每次 new Client 会销毁连接池

# 不好的做法（每次请求新建 Client，无法复用连接）
async def bad_practice(url: str) -> dict:
    async with httpx.AsyncClient() as client:  # 每次都新建、销毁
        response = await client.get(url)
        return response.json()

# 好的做法：在应用生命周期内复用 Client
class LLMHttpClient:
    def __init__(self):
        self._client: httpx.AsyncClient | None = None

    async def __aenter__(self):
        self._client = httpx.AsyncClient(
            # 超时配置：四个维度独立配置
            timeout=httpx.Timeout(
                connect=5.0,    # TCP 连接建立的超时（秒）
                read=120.0,     # 等待服务器响应数据的超时（流式响应需要足够长）
                write=10.0,     # 发送请求体的超时
                pool=5.0,       # 从连接池获取连接的超时
            ),
            # 连接池配置
            limits=httpx.Limits(
                max_connections=50,          # 总连接数上限
                max_keepalive_connections=20, # 保活连接数（复用）
                keepalive_expiry=30.0,        # 保活连接最大空闲时间（秒）
            ),
            # HTTP/2 支持（需要 pip install httpx[http2]）
            http2=True,
            # 基础 URL，后续请求只需要写路径
            base_url="https://api.openai.com",
        )
        return self

    async def __aexit__(self, *args):
        if self._client:
            await self._client.aclose()

1.2.2 connect_timeout vs read_timeout vs pool_timeout

这三个超时的含义和配置逻辑完全不同，混淆会导致要么超时太快（正常请求被杀死），要么超时太慢（故障请求长时间卡住）：

python

# 各超时的实际含义和推荐值（针对 LLM API 场景）
TIMEOUT_CONFIG = httpx.Timeout(
    connect=5.0,
    # 含义：与服务器建立 TCP 连接的等待时间
    # 建议值：3～10 秒。超过 10 秒几乎可以确定是网络问题

    read=180.0,
    # 含义：成功连接后，等待服务器返回任何数据的时间
    # 对 LLM API：非流式 = 等待完整响应（可能 30s+）；流式 = 等待第一个 chunk
    # 建议值：非流式 60～180s；流式可以更短（30s），因为流式响应应该很快开始

    write=10.0,
    # 含义：发送请求体到服务器的时间（请求 body 很大时才会超时）
    # 建议值：10～30s

    pool=5.0,
    # 含义：从连接池获取可用连接的等待时间
    # 连接池满时会等待其他请求完成归还连接
    # 建议值：3～10s，太长说明并发量超过连接池配置
)

1.3 重试策略：tenacity 实现指数退避

tenacity：Python 的重试库，通过装饰器方式给函数添加自动重试逻辑，支持指数退避、最大重试次数等策略。

LLM API 的失败有两类：可重试失败（网络抖动、429 限流（Too Many Requests，请求太频繁被服务端拒绝）、503 临时过载）和不可重试失败（400 参数错误、401 认证失败、内容政策拒绝）。

重试策略的关键是精准区分这两类错误：

python

# pip install tenacity
import asyncio
import httpx
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    retry_if_exception,
    before_sleep_log,
    after_log,
)
import logging

logger = logging.getLogger(__name__)

# 定义哪些错误可以重试
def is_retryable_error(exception: Exception) -> bool:
    """判断异常是否值得重试"""
    if isinstance(exception, httpx.TimeoutException):
        return True  # 超时：可重试
    if isinstance(exception, httpx.ConnectError):
        return True  # 连接失败：可重试
    if isinstance(exception, httpx.HTTPStatusError):
        # 429 Too Many Requests：限流，可重试
        # 502/503/504：服务器临时不可用，可重试
        return exception.response.status_code in {429, 502, 503, 504}
    return False


@retry(
    # 最多重试 5 次（加上首次共 6 次尝试）
    stop=stop_after_attempt(5),

    # 指数退避：等待时间 = min(2^attempt * multiplier, max)
    # 指数退避：每次失败后等待时间翻倍（第1次: 1s，第2次: 2s，第3次: 4s，第4次: 8s，最大 60s）
    # 加入随机抖动（Jitter）可避免多个客户端同时重试造成的"雷群效应"
    wait=wait_exponential(multiplier=1, min=1, max=60),

    # 只有满足条件的异常才重试
    retry=retry_if_exception(is_retryable_error),

    # 重试前记录日志
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
async def call_llm_with_retry(
    client: httpx.AsyncClient,
    payload: dict,
) -> dict:
    """
    带重试的 LLM API 调用。
    tenacity 的 @retry 装饰器会自动处理重试逻辑。
    """
    response = await client.post(
        "/v1/chat/completions",
        json=payload,
        headers={"Authorization": f"Bearer {API_KEY}"},
    )

    # 主动抛出 HTTP 错误（触发 retry 机制）
    response.raise_for_status()

    return response.json()


# 针对 429 限流的特殊处理：尊重 Retry-After 响应头
async def call_with_rate_limit_respect(
    client: httpx.AsyncClient,
    payload: dict,
    max_retries: int = 5,
) -> dict:
    """
    处理 429 限流：读取 Retry-After 响应头，等待指定时间后重试。
    比固定指数退避更精准，避免不必要的等待。
    """
    for attempt in range(max_retries):
        try:
            response = await client.post("/v1/chat/completions", json=payload)
            response.raise_for_status()
            return response.json()

        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                # 读取服务器建议的等待时间
                retry_after = e.response.headers.get("retry-after", "5")
                wait_seconds = float(retry_after)

                logger.warning(
                    f"Rate limited (429), waiting {wait_seconds}s before retry "
                    f"(attempt {attempt + 1}/{max_retries})"
                )
                await asyncio.sleep(wait_seconds)
            elif e.response.status_code in {502, 503}:
                # 服务不可用：指数退避
                wait = min(2 ** attempt, 60)
                await asyncio.sleep(wait)
            else:
                raise  # 不可重试的错误直接抛出

    raise RuntimeError(f"Failed after {max_retries} attempts")

1.4 流式响应：client.stream() 逐块处理 SSE

LLM API 的流式响应使用 Server-Sent Events（SSE，服务端主动推送格式，服务器边生成边发送数据，客户端持续接收）格式。每个数据块是一行 data: {...} 或 data: [DONE]。

python

import json
from collections.abc import AsyncIterator

async def stream_chat_completion(
    client: httpx.AsyncClient,
    messages: list[dict],
    model: str = "gpt-4o",
) -> AsyncIterator[str]:
    """
    流式调用 LLM API，逐 token 生成输出。
    使用 async generator，调用方可以用 async for 消费。
    """
    payload = {
        "model": model,
        "messages": messages,
        "stream": True,
    }

    async with client.stream(
        "POST",
        "/v1/chat/completions",
        json=payload,
        headers={"Authorization": f"Bearer {API_KEY}"},
    ) as response:
        response.raise_for_status()

        async for line in response.aiter_lines():
            if not line.startswith("data: "):
                continue

            data_str = line[6:]  # 去掉 "data: " 前缀

            if data_str == "[DONE]":
                break

            try:
                data = json.loads(data_str)
            except json.JSONDecodeError:
                continue

            # 提取增量内容
            delta = data.get("choices", [{}])[0].get("delta", {})
            content = delta.get("content")
            if content:
                yield content


# 调用方：实时打印流式输出
async def print_streaming_response():
    async with httpx.AsyncClient(
        base_url="https://api.openai.com",
        timeout=httpx.Timeout(connect=5.0, read=180.0, write=10.0, pool=5.0),
    ) as client:
        messages = [{"role": "user", "content": "Explain quantum computing in 3 sentences."}]

        async for token in stream_chat_completion(client, messages):
            print(token, end="", flush=True)

        print()  # 换行

1.5 中间件模式：统一请求日志、鉴权注入、响应校验

httpx 支持通过自定义 Auth 类和事件钩子实现类似中间件的功能：

python

import time
import httpx
from typing import Generator

# 方式一：自定义 Auth 类（鉴权注入）
class BearerTokenAuth(httpx.Auth):
    """统一注入 Bearer Token，支持 token 刷新"""

    def __init__(self, token: str):
        self._token = token

    def auth_flow(self, request: httpx.Request) -> Generator:
        request.headers["Authorization"] = f"Bearer {self._token}"
        request.headers["X-Request-ID"] = f"req_{int(time.time() * 1000)}"
        response = yield request

        # 如果 401，可以在这里刷新 token 并重试
        if response.status_code == 401:
            self._token = self._refresh_token()
            request.headers["Authorization"] = f"Bearer {self._token}"
            yield request

    def _refresh_token(self) -> str:
        # 实际场景：调用 token 刷新接口
        return "new_token"


# 方式二：Event Hooks（请求/响应日志）
class RequestLogger:
    """记录每个请求的详细日志，用于调试和性能分析"""

    def __init__(self):
        self._request_times: dict[str, float] = {}

    def log_request(self, request: httpx.Request) -> None:
        request_id = request.headers.get("X-Request-ID", "unknown")
        self._request_times[request_id] = time.time()
        logger.debug(
            "HTTP Request",
            extra={
                "method": request.method,
                "url": str(request.url),
                "request_id": request_id,
            },
        )

    def log_response(self, response: httpx.Response) -> None:
        request_id = response.request.headers.get("X-Request-ID", "unknown")
        start_time = self._request_times.pop(request_id, time.time())
        elapsed = time.time() - start_time

        logger.info(
            "HTTP Response",
            extra={
                "status_code": response.status_code,
                "elapsed_ms": round(elapsed * 1000),
                "request_id": request_id,
                "content_length": len(response.content) if not response.is_stream_consumed else "streamed",
            },
        )


# 将日志和鉴权组合使用
def create_llm_client(api_key: str) -> httpx.AsyncClient:
    request_logger = RequestLogger()

    return httpx.AsyncClient(
        base_url="https://api.openai.com",
        auth=BearerTokenAuth(api_key),
        timeout=httpx.Timeout(connect=5.0, read=180.0, write=10.0, pool=5.0),
        limits=httpx.Limits(
            max_connections=30,
            max_keepalive_connections=10,
            keepalive_expiry=30.0,
        ),
        http2=True,
        event_hooks={
            "request": [request_logger.log_request],
            "response": [request_logger.log_response],
        },
    )

1.6 连接池调优：避免把 LLM API 打爆

python

# 连接池配置的经验公式
# max_connections ≈ 并发请求峰值 * 1.2（留 20% 余量）
# max_keepalive_connections ≈ 平均并发请求数
# keepalive_expiry 建议 20～60s（太长浪费服务器资源，太短频繁重建连接）

# 场景一：低并发（开发/测试环境）
dev_limits = httpx.Limits(
    max_connections=10,
    max_keepalive_connections=5,
    keepalive_expiry=20.0,
)

# 场景二：中等并发（生产 API 服务，每秒 10-20 请求）
prod_limits = httpx.Limits(
    max_connections=30,
    max_keepalive_connections=15,
    keepalive_expiry=30.0,
)

# 场景三：高并发（批处理系统，每秒 50+ 请求）
batch_limits = httpx.Limits(
    max_connections=100,
    max_keepalive_connections=50,
    keepalive_expiry=60.0,
)

1.7 完整示例：生产级 LLM API 客户端封装

python

import asyncio
import json
import logging
import time
from collections.abc import AsyncIterator
from dataclasses import dataclass

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception

logger = logging.getLogger(__name__)

@dataclass
class ChatMessage:
    role: str
    content: str

@dataclass
class ChatResponse:
    content: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: float


class ProductionLLMClient:
    """
    生产级 LLM HTTP 客户端。
    封装了超时配置、重试策略、流式响应、日志记录。
    """

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.openai.com",
        default_model: str = "gpt-4o",
        max_retries: int = 3,
        max_connections: int = 30,
    ):
        self._api_key = api_key
        self._default_model = default_model
        self._max_retries = max_retries

        self._client = httpx.AsyncClient(
            base_url=base_url,
            auth=BearerTokenAuth(api_key),
            timeout=httpx.Timeout(
                connect=5.0,
                read=180.0,
                write=10.0,
                pool=5.0,
            ),
            limits=httpx.Limits(
                max_connections=max_connections,
                max_keepalive_connections=max_connections // 2,
                keepalive_expiry=30.0,
            ),
            http2=True,
            headers={
                "Content-Type": "application/json",
                "User-Agent": "MyApp/1.0 (httpx)",
            },
            event_hooks={
                "request": [self._log_request],
                "response": [self._log_response],
            },
        )
        self._request_times: dict[str, float] = {}

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()

    async def close(self):
        await self._client.aclose()

    def _log_request(self, request: httpx.Request) -> None:
        req_id = request.headers.get("X-Request-ID", "")
        self._request_times[req_id] = time.time()
        logger.debug(f"→ {request.method} {request.url.path}")

    def _log_response(self, response: httpx.Response) -> None:
        req_id = response.request.headers.get("X-Request-ID", "")
        start = self._request_times.pop(req_id, time.time())
        elapsed_ms = (time.time() - start) * 1000
        logger.debug(f"← {response.status_code} ({elapsed_ms:.0f}ms)")

    def _is_retryable(self, exc: Exception) -> bool:
        if isinstance(exc, (httpx.TimeoutException, httpx.ConnectError)):
            return True
        if isinstance(exc, httpx.HTTPStatusError):
            return exc.response.status_code in {429, 500, 502, 503, 504}
        return False

    async def chat(
        self,
        messages: list[ChatMessage],
        model: str | None = None,
        temperature: float = 0.0,
        max_tokens: int = 2048,
    ) -> ChatResponse:
        """非流式聊天接口，带自动重试"""
        payload = {
            "model": model or self._default_model,
            "messages": [{"role": m.role, "content": m.content} for m in messages],
            "temperature": temperature,
            "max_tokens": max_tokens,
        }

        start = time.time()

        for attempt in range(self._max_retries + 1):
            try:
                response = await self._client.post(
                    "/v1/chat/completions",
                    json=payload,
                    headers={"X-Request-ID": f"req_{int(time.time()*1000)}_{attempt}"},
                )
                response.raise_for_status()
                data = response.json()

                latency_ms = (time.time() - start) * 1000
                choice = data["choices"][0]
                usage = data.get("usage", {})

                return ChatResponse(
                    content=choice["message"]["content"],
                    model=data["model"],
                    prompt_tokens=usage.get("prompt_tokens", 0),
                    completion_tokens=usage.get("completion_tokens", 0),
                    latency_ms=latency_ms,
                )

            except Exception as exc:
                if not self._is_retryable(exc) or attempt == self._max_retries:
                    logger.error(f"LLM call failed after {attempt + 1} attempts: {exc}")
                    raise

                # 429 尊重 Retry-After
                wait = 5.0
                if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code == 429:
                    wait = float(exc.response.headers.get("retry-after", wait))
                else:
                    wait = min(2 ** attempt * 1.0, 30.0)

                logger.warning(f"Retry {attempt + 1}/{self._max_retries} in {wait:.1f}s: {exc}")
                await asyncio.sleep(wait)

        raise RuntimeError("Unreachable")

    async def stream_chat(
        self,
        messages: list[ChatMessage],
        model: str | None = None,
    ) -> AsyncIterator[str]:
        """流式聊天接口，逐 token yield"""
        payload = {
            "model": model or self._default_model,
            "messages": [{"role": m.role, "content": m.content} for m in messages],
            "stream": True,
        }

        async with self._client.stream(
            "POST",
            "/v1/chat/completions",
            json=payload,
        ) as response:
            response.raise_for_status()

            async for line in response.aiter_lines():
                if not line.startswith("data: "):
                    continue
                data_str = line[6:]
                if data_str == "[DONE]":
                    break
                try:
                    data = json.loads(data_str)
                    content = data["choices"][0]["delta"].get("content", "")
                    if content:
                        yield content
                except (json.JSONDecodeError, KeyError):
                    continue


# 使用示例
async def main():
    import os

    async with ProductionLLMClient(api_key=os.environ["OPENAI_API_KEY"]) as client:
        # 非流式调用
        response = await client.chat(
            messages=[
                ChatMessage(role="system", content="You are a helpful assistant."),
                ChatMessage(role="user", content="What is the capital of France?"),
            ]
        )
        print(f"Response: {response.content}")
        print(f"Tokens: {response.prompt_tokens}+{response.completion_tokens}")
        print(f"Latency: {response.latency_ms:.0f}ms")

        # 流式调用
        print("\nStreaming:")
        async for token in client.stream_chat(
            messages=[ChatMessage(role="user", content="Count to 5 slowly.")]
        ):
            print(token, end="", flush=True)
        print()


if __name__ == "__main__":
    asyncio.run(main())

1.8 小结

超时要分层配置：connect_timeout（5s）、read_timeout（LLM 非流式 120s+）、pool_timeout（5s），三个数值对应三个不同的故障原因。429/5xx 可重试，400/401 不可重试，错误分类决定了重试策略的有效性。指数退避加 Jitter 能避免雷群效应，tenacity 的 wait_exponential 内置了这个逻辑。429 限流时要读取 Retry-After 响应头，等服务器建议的时间，不要盲目固定等待。连接池也不是越大越好——超过 API 并发限制反而会触发更多 429，根据业务流量合理设置 max_connections。

关键配置项	推荐值（LLM API 场景）	说明
connect_timeout	5s	超过说明网络有问题
read_timeout	120～180s	覆盖 LLM 长时间推理
pool_timeout	5s	超过说明并发量超出配置
max_connections	峰值并发 × 1.2	留余量避免 pool_timeout
max_retries	3～5	更多没有意义，只会增加延迟
退避等待上限	30～60s	超过 60s 用户体验已经很差