Claude API 与 OpenRouter 集成实战：从技术选型到生产环境避坑指南

1次阅读

共计 2236 个字符，预计需要花费 6 分钟才能阅读完成。

在直接使用 Claude API 时，开发者常遇到三个核心挑战：

版本碎片化 ：Claude 3 系列包含 Sonnet/Opus/Haiku 等多个子版本，各版本接口参数存在差异
计费复杂化 ：不同模型的 tokens 计费标准不同（如 Claude 3 Opus 价格是 Haiku 的 5 倍）
切换成本高 ：业务需要同时使用多个模型时，需维护多套代码逻辑

传统架构与 OpenRouter 方案对比：

flowchart LR
    A[客户端] -->| 直接调用 | B[Claude API]
    A -->| 统一接入 | C[OpenRouter]
    C --> D[Claude 3 Sonnet]
    C --> E[Claude 3 Opus]
    C --> F[其他模型]

import aiohttp
from typing import Literal

ModelType = Literal['claude-3-sonnet', 'claude-3-opus']

async def chat_completion(
    model: ModelType,
    messages: list[dict],
    temperature: float = 0.7
) -> dict:
    """
    :param model: 模型标识符
    :param messages: 对话消息列表
    :return: 标准化响应 {
        'content': str,
        'input_tokens': int,
        'output_tokens': int
    }
    """headers = {'Authorization': f'Bearer {OPENROUTER_KEY}','HTTP-Referer':'YOUR_APP_URL',  # 必填'X-Title':'Your App Name'  # 可选
    }

    payload = {
        "model": model,
        "messages": messages,
        "temperature": temperature
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(
            'https://openrouter.ai/api/v1/chat/completions',
            json=payload,
            headers=headers
        ) as resp:
            data = await resp.json()

            # 令牌计算（含 20% 的 OpenRouter 额外开销）input_tokens = data['usage']['prompt_tokens'] * 1.2
            output_tokens = data['usage']['completion_tokens'] * 1.2

            return {'content': data['choices'][0]['message']['content'],
                'input_tokens': int(input_tokens),
                'output_tokens': int(output_tokens)
            }

异步请求处理 ：
使用 aiohttp 实现非阻塞调用
建议配置 5-10 的连接池大小

动态模型路由 ：

def model_router(query: str) -> ModelType:
    """根据查询复杂度自动选择模型"""
    if len(query) > 1000:
        return 'claude-3-opus'
    return 'claude-3-sonnet'

流式响应处理 ：

async for chunk in resp.content.iter_chunks():
    print(chunk.decode('utf-8'), end='')

连接池配置 ：

conn = aiohttp.TCPConnector(
    limit=20,  # 最大连接数
    force_close=True,  # 避免连接泄漏
    enable_cleanup_closed=True  # 自动清理
)

批处理模式 ：

async with session.post(..., json={"messages": [msg1, msg2, msg3],  # 多条消息合并
    "stream": False  # 必须关闭流式
}) as resp:

密钥轮换 ：
每月自动轮换 API Key
使用环境变量注入密钥

请求签名 ：

import hmac

def gen_signature(payload: str) -> str:
    return hmac.new(SECRET_KEY.encode(),
        payload.encode(),
        'sha256'
    ).hexdigest()

现象：升级模型版本后输出风格突变

解决方案 ：
1. 在请求中明确指定版本号

model = 'claude-3-sonnet-20240229'  # 完整版本号

2. 保留旧版本测试环境

现象：返回 429 状态码

解决方案 ：
1. 实现指数退避重试

async def retry_request(max_retries=3):
    for i in range(max_retries):
        try:
            return await make_request()
        except aiohttp.ClientError:
            await asyncio.sleep(2 ** i)