import time
import random
from typing import Dict, Any, Optional
class ClaudeAPIHandler:
"""Claude API调用处理器，包含重试和错误处理"""
def __init__(self, client, max_retries=3):
self.client = client
self.max_retries = max_retries
def safe_completion(self, **kwargs) -> Optional[str]:
"""安全的API调用，包含重试机制"""
for attempt in range(self.max_retries):
try:
response = self.client.chat.completions.create(**kwargs)
return response.choices[0].message.content
except Exception as e:
error_type = type(e).__name__
if attempt == self.max_retries - 1:
print(f"API调用最终失败: {error_type}: {e}")
return None
# 根据错误类型调整重试策略
if "rate_limit" in str(e).lower():
delay = 2 ** attempt + random.uniform(0, 1)
print(f"遇到速率限制，等待 {delay:.2f} 秒后重试...")
time.sleep(delay)
elif "timeout" in str(e).lower():
delay = 1 + attempt * 0.5
print(f"请求超时，等待 {delay:.2f} 秒后重试...")
time.sleep(delay)
else:
print(f"遇到错误 {error_type}，立即重试...")
time.sleep(0.1)
return None
def batch_completion(self, prompts: list, **kwargs) -> list:
"""批量处理多个提示"""
results = []
for i, prompt in enumerate(prompts):
print(f"处理第 {i+1}/{len(prompts)} 个请求...")
kwargs_copy = kwargs.copy()
kwargs_copy['messages'] = [{"role": "user", "content": prompt}]
result = self.safe_completion(**kwargs_copy)
results.append(result)
# 避免过快请求
if i < len(prompts) - 1:
time.sleep(1)
return results
# 使用示例
handler = ClaudeAPIHandler(client)
# 单个请求
result = handler.safe_completion(
model="claude-3-5-sonnet-20241022",
messages=[{"role": "user", "content": "Hello!"}],
max_tokens=500
)
# 批量请求
prompts = [
"翻译：Hello world",
"总结：人工智能的发展历程",
"分析：电商发展趋势"
]
results = handler.batch_completion(
prompts,
model="claude-3-haiku-20240307",
max_tokens=300
)
