用Python封装LLM API：从零构建高可用AI服务

为什么需要封装LLM API？

在构建生产级AI应用时，直接硬编码调用OpenAI或Anthropic等厂商的HTTP接口会导致代码耦合度高、异常处理缺失且难以扩展。封装LLM API的核心目标是创建一层抽象服务层，将网络通信、错误恢复、资源管理等逻辑与业务逻辑解耦，从而提升系统的可维护性与高可用性。

核心设计目标

• 异步非阻塞：利用Python的asyncio处理并发请求，避免线程阻塞。
• 自动重试：针对429（限流）和5xx错误实施指数退避策略。
• 本地缓存：减少重复请求成本，降低延迟。

技术选型与项目结构

推荐组合使用 FastAPI（Web框架）、httpx（异步HTTP客户端）和 aiomemory 或 redis（缓存）。项目结构应遵循分层架构：

1. llm_client.py：底层HTTP通信封装。
2. service.py：业务逻辑、重试与缓存管理。
3. main.py：FastAPI路由入口。

实现高可用LLM客户端

1. 异步HTTP基础封装

首先使用httpx创建异步客户端实例，配置基础超时时间。避免每次请求都新建连接，使用连接池复用TCP连接以优化性能。

import httpx
class LLMHttpClient:
    def __init__(self):
        self.client = httpx.AsyncClient(
            base_url="https://api.openai.com/v1",
            timeout=30.0,
            headers={"Authorization": "Bearer sk-..."}
        )

2. 集成指数退避重试机制

网络不稳定或触发厂商限流是常见故障。使用 tenacity 库装饰器实现自动重试，设置最大重试次数和等待间隔递增策略。

from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=4, max=10))
async def request_completion(self, payload: dict) -> dict:
    response = await self.client.post("/chat/completions", json=payload)
    if response.status_code in (429, 500, 503):
        raise LLMServiceError(f"Server error: {response.status_code}")
    return response.json()

3. 添加语义缓存层

对于相同的用户提问，若结果未变，可直接返回缓存。使用Redis存储键值对，键为输入参数的哈希值，值为JSON序列化的响应。设置TTL（生存时间）为1小时，平衡一致性与性能。

构建FastAPI服务接口

将上述客户端封装进FastAPI路由，添加全局异常处理器，确保当LLM服务不可用时，API能返回标准的HTTP 503错误码及友好提示，而不是抛出原始堆栈信息。

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    try:
        result = await llm_service.chat(request)
        return {"status": "success", "data": result}
    except LLMServiceError:
        raise HTTPException(status_code=503, detail="LLM Service Unavailable")

最佳实践与监控

• 日志记录：使用JSON格式化日志，记录每次请求的耗时、Token消耗及重试次数，便于后续分析。
• 密钥管理：严禁在代码中硬编码API Key，使用环境变量或Vault服务读取。
• 压测验证：使用Locust进行并发压测，观察P99延迟和错误率，调整连接池大小和超时参数。

通过这种分层封装模式，你的AI服务将从“脆弱的手工脚本”进化为“鲁棒的生产级组件”，有效应对高并发场景下的各种边缘情况。