Claude API与DeepSeek集成实战：解决多模型协同开发的代码管理难题

1次阅读

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

最近在同时接入 Claude 和 DeepSeek 的 API 时，发现几个让人头疼的问题：

参数规范不统一 ：同样的文本补全任务，Claude 要求prompt 字段而 DeepSeek 用input_text
响应结构各异 ：Claude 返回 JSON 包含completion 字段，DeepSeek 则放在 data.choices 里
错误处理分裂：各家的速率限制错误码从 429 到 503 不等
密钥管理混乱：每个服务商的 API 密钥轮换策略不同

直接裸调 API 的代码很快就会变成这样：

# 典型的多模型 spaghetti code
if model_type == 'claude':
    response = requests.post(claude_url, json={'prompt': query}, headers=claude_headers)
    result = response.json()['completion']
elif model_type == 'deepseek':
    response = requests.post(deepseek_url, json={'input_text': query}, headers=deepseek_headers)
    result = response.json()['data']['choices'][0]['text']

方案类型	开发效率	维护成本	扩展性
直接调用原生 API	低	高	差
统一中间件	高	低	优秀

class AIModelClient(ABC):
    @abstractmethod
    def complete(self, prompt: str) -> CompletionResult: ...

class ClaudeAdapter(AIModelClient):
    def __init__(self, api_key: str): ...

class DeepSeekAdapter(AIModelClient):
    def __init__(self, api_key: str): ...

class UnifiedAIClient:
    def __init__(self):
        self._routers = {
            'claude': ClaudeAdapter,
            'deepseek': DeepSeekAdapter
        }

    def route_request(self, model_type: str) -> AIModelClient: ...

# requirements.txt
# requests==2.31.0
# tenacity==8.2.3

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((TimeoutError, ConnectionError))
)
def safe_api_call(url: str, payload: dict):
    response = requests.post(url, json=payload, timeout=30)
    response.raise_for_status()  # 自动处理 4xx/5xx
    return response.json()

时间复杂度分析：最坏情况下 O(n)其中 n 为重试次数

class CompletionResult:
    def __init__(self, raw: dict):
        self.raw = raw
        self._metrics = {'start_time': time.time(),
            'token_usage': None
        }

    @property
    def text(self) -> str:
        # 实际项目建议用 Pydantic 做数据校验
        if 'completion' in self.raw:  # Claude 格式
            return self.raw['completion']
        elif 'data' in self.raw:  # DeepSeek 格式
            return self.raw['data']['choices'][0]['text']

    def finalize(self):
        self._metrics['end_time'] = time.time()
        return {
            'text': self.text,
            'latency_ms': (self._metrics['end_time'] - self._metrics['start_time']) * 1000
        }

from concurrent.futures import ThreadPoolExecutor

class BatchProcessor:
    def __init__(self, max_workers=4):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

    def parallel_complete(self, prompts: list[str], model_type: str):
        with self.executor:
            futures = [self.executor.submit(self._complete_one, p, model_type)
                for p in prompts
            ]
            return [f.result() for f in futures]

    def _complete_one(self, prompt: str, model_type: str):
        client = UnifiedAIClient().route_request(model_type)
        return client.complete(prompt).finalize()

环境变量存储（开发环境）

import os
from dotenv import load_dotenv

load_dotenv()
CLAUDE_KEY = os.getenv('CLAUDE_API_KEY')

KMS 加密方案（生产环境）

import boto3  # 需要 boto3==1.28.1

def decrypt_key(encrypted_key: str) -> str:
    kms = boto3.client('kms')
    return kms.decrypt(CiphertextBlob=base64.b64decode(encrypted_key)
    )['Plaintext'].decode()