Claude Code 接入 DeepSeek API 的架构设计与实现

1次阅读

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

直接调用 DeepSeek API 时开发者常面临三个核心问题：

速率限制的暴力破解 ：DeepSeek 对免费 tier 通常设置 5-10QPS 的限制，突发流量会导致大量 429 错误。传统线性重试会加剧服务压力
错误处理的复杂性 ：API 可能返回 5xx 服务端错误、网络抖动、签名过期等多种异常，需要分类处理
资源利用率低下 ：单次请求 -RTT 模式在高延迟环境下（跨地域访问）吞吐量急剧下降

通过压测工具比较两种协议在 AWS c5.large 实例上的表现（单位：ms）：

指标	REST/HTTPS	gRPC
平均延迟	120	65
P99 延迟	410	180
连接开销	高	低
二进制支持	部分	完整

注：gRPC 需要额外处理 HTTP/2 连接复用和负载均衡

import hashlib
import hmac
import time

class AuthGenerator:
    """
    生成带签名的请求头
    :param api_key: 从安全存储获取的密钥
    :param secret: 通过 KMS 解密的 secret
    """
    def __init__(self, api_key: str, secret: str):
        self.api_key = api_key
        self.secret = secret.encode('utf-8')

    def generate_headers(self, body: str) -> dict:
        timestamp = str(int(time.time()))
        sign = hmac.new(
            self.secret, 
            f"{timestamp}{body}".encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

        return {
            "X-API-KEY": self.api_key,
            "X-TIMESTAMP": timestamp,
            "X-SIGNATURE": sign
        }

from dataclasses import dataclass
from typing import List
import asyncio

@dataclass
class BatchRequest:
    max_batch_size: int = 50  # 根据 API 文档调整
    max_wait_ms: int = 200   # 批次等待时间

    def __post_init__(self):
        self.queue = asyncio.Queue()
        self.semaphore = asyncio.Semaphore(10)  # 并发控制

    async def add_request(self, item):
        """非阻塞式添加请求到批次"""
        await self.queue.put(item)

    async def process_batch(self):
        """触发批次处理的核心方法"""
        batch = []
        start_time = asyncio.get_event_loop().time()

        while True:
            elapsed = (asyncio.get_event_loop().time() - start_time) * 1000

            # 达到最大等待时间或批次上限时立即处理
            if batch and (elapsed >= self.max_wait_ms or len(batch) >= self.max_batch_size):
                async with self.semaphore:
                    await self._send_to_api(batch)
                batch.clear()
                start_time = asyncio.get_event_loop().time()

            try:
                # 非阻塞获取队列元素
                item = self.queue.get_nowait()
                batch.append(item)
            except asyncio.QueueEmpty:
                if batch:
                    continue
                await asyncio.sleep(0.01)  # 避免空转

import random

class RetryPolicy:
    """
    基于 TCP 拥塞控制算法的重试策略
    :param max_retries: 最大重试次数
    :param base_delay: 基础延迟 (s)
    """
    def __init__(self, max_retries=5, base_delay=1):
        self.max_retries = max_retries
        self.base_delay = base_delay

    async def execute_with_retry(self, func, *args):
        retry_count = 0
        last_error = None

        while retry_count <= self.max_retries:
            try:
                return await func(*args)
            except APIError as e:
                last_error = e
                if not self._should_retry(e):
                    break

                delay = self._calculate_delay(retry_count)
                await asyncio.sleep(delay)
                retry_count += 1

        raise RetryExhaustedError(last_error)

    def _calculate_delay(self, retry_count):
        # 加入随机抖动避免惊群效应
        return min(self.base_delay * (2 ** retry_count) + random.uniform(0, 1),
            60  # 最大不超过 60s
        )

    def _should_retry(self, error):
        return error.status_code in (429, 500, 502, 503, 504)

在 4C8G 的云服务器上测试（单位：请求 / 秒）：

场景	单线程	多线程 (8)	协程 (1000)
简单请求	12	85	320
批量请求 (50)	18	130	950
带重试请求	8	60	210

注：测试使用模拟 API，真实环境建议用生产流量影子测试

使用 AWS Secrets Manager 或 HashiCorp Vault 动态获取密钥
内存中的密钥有效时间控制在 5-10 分钟
禁止将密钥写入日志 / 配置文件

采用 Hystrix 风格的熔断策略：

# 基于滑动窗口的熔断检测
class CircuitBreaker:
    def __init__(self, 
                 failure_threshold=0.5, 
                 recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = "CLOSED"
        self.metrics = SlidingWindowMetrics(60)  # 60 秒窗口

    async def execute(self, func):
        if self.state == "OPEN":
            raise CircuitOpenError()

        try:
            result = await func()
            self.metrics.record_success()
            return result
        except Exception as e:
            self.metrics.record_failure()
            self._check_thresholds()
            raise

    def _check_thresholds(self):
        if self.metrics.failure_rate() > self.failure_threshold:
            self.state = "OPEN"
            asyncio.get_event_loop().call_later(
                self.recovery_timeout, 
                lambda: setattr(self, 'state', 'HALF_OPEN')
            )