共计 2223 个字符，预计需要花费 6 分钟才能阅读完成。
Claude API 调用实战指南
背景与常见问题
初次集成 Claude API 时，开发者常遇到三类典型问题：
- 认证配置错误 ：密钥格式不正确、未及时刷新 JWT 令牌导致请求被拒绝
- 流式响应处理缺陷 ：未正确处理分块数据导致信息丢失，或未处理连接中断异常
- 突发流量管理不足 ：未预读速率限制头信息，遭遇 429 状态码时缺乏重试机制
技术方案对比
REST API
- 适用场景：单次请求 - 响应模式（如内容审核、文本摘要）
- 优势：实现简单，适合低频批量处理
- 劣势：长文本生成时需等待完整响应
WebSocket
- 适用场景：持续对话交互（如聊天机器人）
- 优势：支持双向实时通信
- 劣势：需要维护连接状态
核心实现详解
认证模块实现
import httpx
from datetime import datetime, timedelta
import jwt
class AuthManager:
def __init__(self, api_key):
self.api_key = api_key
self._token = None
self._expiry = None
def get_token(self):
if not self._token or datetime.utcnow() >= self._expiry:
self._refresh_token()
return self._token
def _refresh_token(self):
# JWT 有效期为 10 分钟，提前 2 分钟刷新
self._expiry = datetime.utcnow() + timedelta(minutes=8)
self._token = jwt.encode({
'iss': 'api-client',
'exp': self._expiry
}, self.api_key, algorithm='HS256')流式响应处理
async def stream_completion(prompt, max_tokens=500):
auth = AuthManager(os.getenv('CLAUDE_API_KEY'))
async with httpx.AsyncClient() as client:
try:
response = await client.post(
'https://api.claude.ai/v1/complete',
headers={'Authorization': f'Bearer {auth.get_token()}'},
json={'prompt': prompt, 'max_tokens': max_tokens, 'stream': True},
timeout=30.0
)
response.raise_for_status()
buffer = ""
async for chunk in response.aiter_text():
# 处理服务器可能发送的 keep-alive 空行
if chunk.strip():
buffer += chunk
yield buffer
buffer = ""
except httpx.ReadTimeout:
print("Warning: 流式响应超时，建议检查网络状况")
except httpx.HTTPStatusError as e:
print(f"API 请求失败: {e.response.status_code}")Token 用量计算
def count_tokens(text):
# 近似计算：英文按空格分词，中文按字符
import re
chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
english_words = len(re.findall(r'\w+', text))
return chinese_chars + english_words生产环境建议
重试策略实现
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(httpx.HTTPStatusError)
)
async def safe_api_call():
# 实现带退避的重试逻辑
pass敏感信息存储方案对比
| 方案 | 优点 | 缺点 |
|---|---|---|
| 环境变量 | 部署简单，容器友好 | 需重启服务才能更新 |
| 加密配置文件 | 支持动态更新 | 需处理密钥分发问题 |
| 密钥管理服务 | 最高安全性 | 增加架构复杂度 |
性能优化延伸
可设计本地缓存层降低调用成本：
- 对相同 prompt 的请求返回缓存结果
- 使用 LRU 策略管理缓存大小
- 为缓存设置 TTL 避免返回过时信息
- 记录缓存命中率监控优化效果
from diskcache import Cache
cache = Cache("./claude_cache")
def cached_completion(prompt):
key = hashlib.md5(prompt.encode()).hexdigest()
if key in cache:
return cache[key]
result = await stream_completion(prompt)
cache.set(key, result, expire=3600) # 缓存 1 小时
return result通过上述方案，开发者可构建具备生产可用性的 Claude API 集成方案，有效平衡功能完整性与系统稳定性。
正文完
