大模型 API 工程化开发：OpenAI 与 Claude API 最佳实践
本文从工程化视角深入讲解如何高效、稳定、低成本地集成 OpenAI 和 Claude 等大模型 API，涵盖异步调用、流式处理、重试降级、成本监控等核心环节，提供可直接落地的生产级代码。
为什么需要工程化封装？
很多开发者在初次调用大模型 API 时，往往直接使用官方 SDK 的简单示例：
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "你好"}]
)
这种写法在 demo 阶段没问题，但一到生产环境就会暴露各种问题：
- 超时未处理：网络抖动导致请求卡住
- 重试缺失：偶发 500/503 错误直接抛异常
- 流式未支持：用户等待太久体验差
- 成本失控：没有 Token 用量监控
- 多厂商切换困难：OpenAI 和 Claude API 差异大
本文将带你构建一套生产级的大模型 API 调用框架。
核心设计目标
- 统一接口：一套代码适配 OpenAI、Claude、Azure 等多厂商
- 高可用：自动重试、降级、熔断
- 高性能：异步 + 流式 + 连接池
- 可观测：延迟、Token、成本实时监控
- 易扩展：新模型接入只需配置，不改代码
工程化代码实战
1. 统一抽象层
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncGenerator, Optional, List, Dict, Any
import asyncio
import time
import functools
@dataclass
class LLMResponse:
"""统一响应格式"""
content: str
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
latency_ms: float
finish_reason: Optional[str] = None
@dataclass
class LLMConfig:
"""模型配置"""
provider: str # openai / anthropic / azure
model: str
api_key: str
base_url: Optional[str] = None
max_retries: int = 3
timeout: float = 60.0
max_tokens: int = 4096
temperature: float = 0.7
class BaseLLMClient(ABC):
"""大模型客户端抽象基类"""
def __init__(self, config: LLMConfig):
self.config = config
self._call_count = 0
self._total_tokens = 0
self._total_latency = 0.0
@abstractmethod
async def chat(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse:
"""非流式对话"""
pass
@abstractmethod
async def chat_stream(self, messages: List[Dict[str, str]], **kwargs) -> AsyncGenerator[str, None]:
"""流式对话"""
pass
def get_stats(self) -> Dict[str, Any]:
"""获取调用统计"""
return {
"calls": self._call_count,
"total_tokens": self._total_tokens,
"avg_latency_ms": self._total_latency / max(self._call_count, 1),
"est_cost_usd": self._estimate_cost()
}
def _estimate_cost(self) -> float:
"""估算成本（简化版）"""
pricing = {
"gpt-4o": {"input": 2.5, "output": 10.0}, # $/1M tokens
"gpt-4o-mini": {"input": 0.15, "output": 0.6},
"claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
"claude-3-haiku": {"input": 0.25, "output": 1.25},
}
p = pricing.get(self.config.model, {"input": 1.0, "output": 1.0})
# 按输出占比估算
return (self._total_tokens * 0.3 * p["input"] + self._total_tokens * 0.7 * p["output"]) / 1_000_000
2. OpenAI 实现
from openai import AsyncOpenAI
import openai
class OpenAIClient(BaseLLMClient):
"""OpenAI 生产级客户端"""
def __init__(self, config: LLMConfig):
super().__init__(config)
self.client = AsyncOpenAI(
api_key=config.api_key,
base_url=config.base_url,
timeout=config.timeout,
max_retries=0 # 我们自己处理重试
)
async def chat(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse:
start = time.perf_counter()
@self._with_retry
async def _call():
return await self.client.chat.completions.create(
model=self.config.model,
messages=messages,
max_tokens=kwargs.get("max_tokens", self.config.max_tokens),
temperature=kwargs.get("temperature", self.config.temperature),
)
resp = await _call()
latency = (time.perf_counter() - start) * 1000
self._call_count += 1
self._total_tokens += resp.usage.total_tokens if resp.usage else 0
self._total_latency += latency
return LLMResponse(
content=resp.choices[0].message.content or "",
model=resp.model,
prompt_tokens=resp.usage.prompt_tokens if resp.usage else 0,
completion_tokens=resp.usage.completion_tokens if resp.usage else 0,
total_tokens=resp.usage.total_tokens if resp.usage else 0,
latency_ms=latency,
finish_reason=resp.choices[0].finish_reason
)
async def chat_stream(self, messages: List[Dict[str, str]], **kwargs) -> AsyncGenerator[str, None]:
stream = await self.client.chat.completions.create(
model=self.config.model,
messages=messages,
max_tokens=kwargs.get("max_tokens", self.config.max_tokens),
temperature=kwargs.get("temperature", self.config.temperature),
stream=True
)
async for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
def _with_retry(self, func):
"""重试装饰器"""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(self.config.max_retries + 1):
try:
return await func(*args, **kwargs)
except (openai.RateLimitError, openai.APITimeoutError) as e:
last_exception = e
wait = min(2 ** attempt, 30) # 指数退避，最大30秒
print(f"⚠️ 请求失败，{wait}秒后重试... ({attempt+1}/{self.config.max_retries})")
await asyncio.sleep(wait)
except openai.APIError as e:
if e.status_code in [500, 502, 503, 504]:
last_exception = e
await asyncio.sleep(1)
continue
raise
raise last_exception
return wrapper
3. Claude 实现
from anthropic import AsyncAnthropic
import anthropic
class ClaudeClient(BaseLLMClient):
"""Anthropic Claude 生产级客户端"""
def __init__(self, config: LLMConfig):
super().__init__(config)
self.client = AsyncAnthropic(
api_key=config.api_key,
base_url=config.base_url,
timeout=config.timeout,
max_retries=0
)
async def chat(self, messages: List[Dict[str, str]], **kwargs) -> LLMResponse:
start = time.perf_counter()
# Claude 使用 system 参数，需要转换消息格式
system_msg = ""
user_messages = []
for msg in messages:
if msg["role"] == "system":
system_msg = msg["content"]
else:
user_messages.append(msg)
@self._with_retry
async def _call():
return await self.client.messages.create(
model=self.config.model,
max_tokens=kwargs.get("max_tokens", self.config.max_tokens),
temperature=kwargs.get("temperature", self.config.temperature),
system=system_msg or None,
messages=user_messages
)
resp = await _call()
latency = (time.perf_counter() - start) * 1000
self._call_count += 1
self._total_tokens += resp.usage.input_tokens + resp.usage.output_tokens
self._total_latency += latency
return LLMResponse(
content=resp.content[0].text if resp.content else "",
model=resp.model,
prompt_tokens=resp.usage.input_tokens,
completion_tokens=resp.usage.output_tokens,
total_tokens=resp.usage.input_tokens + resp.usage.output_tokens,
latency_ms=latency,
finish_reason=resp.stop_reason
)
async def chat_stream(self, messages: List[Dict[str, str]], **kwargs) -> AsyncGenerator[str, None]:
system_msg = ""
user_messages = []
for msg in messages:
if msg["role"] == "system":
system_msg = msg["content"]
else:
user_messages.append(msg)
stream = await self.client.messages.create(
model=self.config.model,
max_tokens=kwargs.get("max_tokens", self.config.max_tokens),
temperature=kwargs.get("temperature", self.config.temperature),
system=system_msg or None,
messages=user_messages,
stream=True
)
async for event in stream:
if event.type == "content_block_delta":
yield event.delta.text
def _with_retry(self, func):
"""重试装饰器"""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(self.config.max_retries + 1):
try:
return await func(*args, **kwargs)
except anthropic.RateLimitError as e:
last_exception = e
wait = min(2 ** attempt, 30)
print(f"⚠️ Claude 限流，{wait}秒后重试...")
await asyncio.sleep(wait)
except anthropic.APIError as e:
if hasattr(e, 'status_code') and e.status_code in [500, 502, 503, 529]:
last_exception = e
await asyncio.sleep(1)
continue
raise
raise last_exception
return wrapper
4. 工厂与多厂商路由
class LLMFactory:
"""大模型客户端工厂"""
_clients: Dict[str, BaseLLMClient] = {}
@classmethod
def register(cls, provider: str, client_class: type):
cls._providers[provider] = client_class
_providers = {
"openai": OpenAIClient,
"anthropic": ClaudeClient,
}
@classmethod
def create(cls, config: LLMConfig) -> BaseLLMClient:
client_class = cls._providers.get(config.provider)
if not client_class:
raise ValueError(f"不支持的供应商: {config.provider}")
return client_class(config)
@classmethod
def get_or_create(cls, config: LLMConfig) -> BaseLLMClient:
key = f"{config.provider}:{config.model}"
if key not in cls._clients:
cls._clients[key] = cls.create(config)
return cls._clients[key]
class LLMRouter:
"""智能路由：按成本、延迟、可用性选择模型"""
def __init__(self, configs: List[LLMConfig]):
self.clients = {cfg.model: LLMFactory.create(cfg) for cfg in configs}
self.fallback_order = ["gpt-4o-mini", "claude-3-haiku", "gpt-4o", "claude-3-5-sonnet"]
async def chat_with_fallback(self, messages: List[Dict[str, str]], preferred: str = None, **kwargs) -> LLMResponse:
"""带自动降级的对话"""
models = [preferred] if preferred else []
models += [m for m in self.fallback_order if m not in models]
last_error = None
for model in models:
if model not in self.clients:
continue
try:
print(f"🔄 尝试使用 {model}...")
return await self.clients[model].chat(messages, **kwargs)
except Exception as e:
print(f"❌ {model} 失败: {e}")
last_error = e
raise last_error or RuntimeError("所有模型均不可用")
5. 完整使用示例
import asyncio
async def main():
# 配置多个模型
configs = [
LLMConfig(provider="openai", model="gpt-4o", api_key="sk-xxx"),
LLMConfig(provider="openai", model="gpt-4o-mini", api_key="sk-xxx"),
LLMConfig(provider="anthropic", model="claude-3-5-sonnet", api_key="sk-ant-xxx"),
]
router = LLMRouter(configs)
messages = [
{"role": "system", "content": "你是一位技术专家"},
{"role": "user", "content": "解释异步编程中的事件循环机制"}
]
# 流式输出
print("🚀 流式输出：")
client = LLMFactory.create(configs[0])
async for chunk in client.chat_stream(messages):
print(chunk, end="", flush=True)
print("\n")
# 带降级的非流式调用
print("🛡️ 带降级保护的非流式调用：")
resp = await router.chat_with_fallback(messages, preferred="gpt-4o")
print(f"模型: {resp.model}")
print(f"延迟: {resp.latency_ms:.0f}ms")
print(f"Token: {resp.total_tokens}")
print(f"内容: {resp.content[:200]}...")
# 打印统计
print("\n📊 调用统计：")
for name, client in router.clients.items():
stats = client.get_stats()
print(f" {name}: {stats}")
if __name__ == "__main__":
asyncio.run(main())
进阶：成本与性能监控
import json
from datetime import datetime
class CostMonitor:
"""成本监控器"""
def __init__(self, budget_usd: float = 100.0):
self.budget = budget_usd
self.daily_usage: Dict[str, float] = {}
def record(self, model: str, tokens: int, latency_ms: float):
today = datetime.now().strftime("%Y-%m-%d")
cost = self._calc_cost(model, tokens)
self.daily_usage[today] = self.daily_usage.get(today, 0) + cost
if self.daily_usage[today] > self.budget:
print(f"🚨 警告：今日成本 ${self.daily_usage[today]:.2f} 已超预算 ${self.budget}")
def _calc_cost(self, model: str, tokens: int) -> float:
pricing = {
"gpt-4o": 5.0 / 1_000_000, # 平均单价
"gpt-4o-mini": 0.3 / 1_000_000,
"claude-3-5-sonnet": 7.5 / 1_000_000,
}
return tokens * pricing.get(model, 1.0 / 1_000_000)
def report(self):
print("\n📈 成本报告:")
for day, cost in sorted(self.daily_usage.items())[-7:]:
bar = "█" * int(cost / self.budget * 20)
print(f" {day}: ${cost:.2f} {bar}")
常见问题 FAQ
Q: 如何处理长文本超出 Token 限制？
A: 使用 tiktoken 预先计算 Token 数，超长时采用分段摘要、RAG 检索或切换到支持长上下文的模型（如 Claude 3 支持 200K）。
Q: 流式输出时如何统计 Token？ A: OpenAI 流式响应不包含 usage，需要额外调用非流式接口估算，或使用 tiktoken 本地计算。
Q: 如何限制并发请求数？
A: 使用 asyncio.Semaphore 控制并发：async with asyncio.Semaphore(10)。
Q: 不同厂商的 message 格式差异如何解决？ A: 统一抽象层内部做格式转换（如本文 ClaudeClient 将 system role 转为 system 参数）。
Q: 生产环境应该用同步还是异步？ A: 推荐异步。asyncio 可以高效处理 I/O 密集型的大模型 API 调用，避免线程开销。
总结
本文从统一抽象、重试降级、流式处理、成本监控四个维度，构建了一套生产级的大模型 API 工程化框架。核心要点：
- 抽象基类隔离厂商差异
- 重试装饰器保障高可用
- 智能路由实现自动降级
- 统计监控把控成本
按此框架封装后，你的应用可以在 OpenAI、Claude、Azure 之间无缝切换，同时具备生产环境所需的稳定性和可观测性。