Claude代码无缝切换本地模型的工程实践与避坑指南

1次阅读

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

在实际项目中同时使用 Claude API 和本地 LLM 模型时，开发者常遇到三个致命问题：

接口规范不一致：Claude 的响应是标准的 RESTful JSON 结构，而本地模型可能返回 ProtoBuf 甚至自定义二进制格式
延迟特性差异：Claude API 有稳定的 SLA 保证（通常 P99<2s），但本地模型的响应时间会受硬件资源影响剧烈波动
错误处理割裂：API 服务有完善的 429/503 错误码，而本地模型可能直接抛出 CUDA 内存错误

@startuml
class Client {+sendRequest()
}

interface ModelAdapter {+preprocess()
  +postprocess()}

class ClaudeAdapter {+convertToAPIFormat()
}

class LocalModelAdapter {+handleStreaming()
}

Client --> ModelAdapter
ModelAdapter <|-- ClaudeAdapter
ModelAdapter <|-- LocalModelAdapter
@enduml

关键设计原则：

请求统一转换为 PromptRequest 对象，包含：
文本内容
温度参数
最大 token 数
响应统一包装为ModelResponse，标准化字段包括：
生成文本
消耗 token 数
计算耗时

def route_request(request):
    # 实时计算各后端权重
    claude_weight = calculate_weight(
        current_qps=claude_metrics.qps,
        error_rate=claude_metrics.error_rate
    )

    local_weight = calculate_weight(
        current_qps=local_metrics.qps,
        error_rate=local_metrics.error_rate
    )

    # 基于权重的随机路由
    if random.random() < claude_weight / (claude_weight + local_weight):
        return claude_adapter.execute(request)
    else:
        return local_adapter.execute(request)

class ModelProxy:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries

    def __call__(self, func):
        @wraps(func)
        def wrapped(*args, **kwargs):
            attempt = 0
            base_delay = 0.5

            while attempt < self.max_retries:
                try:
                    response = func(*args, **kwargs)
                    return self._standardize_response(response)
                except (RateLimitError, ModelTimeoutError) as e:
                    attempt += 1
                    sleep_time = base_delay * (2 ** attempt)
                    time.sleep(min(sleep_time, 5))  # 上限 5 秒
                except ModelCriticalError as e:
                    raise ServiceUnavailableError from e

            raise MaxRetryError(f"Failed after {self.max_retries} attempts")

        return wrapped

    def _standardize_response(self, raw_response) -> ModelResponse:
        """将不同来源的响应转换为标准格式"""
        return ModelResponse(text=raw_response.get('generated_text', ''),
            tokens=raw_response.get('usage', {}).get('total_tokens', 0),
            latency=raw_response.get('latency_ms', 0)
        )

指标	Claude API	本地模型（A100）	本地模型（T4）
平均延迟	1.2s	0.8s	2.5s
P90 延迟	1.5s	1.1s	3.8s
P99 延迟	1.9s	1.6s	6.2s

# JIT 编译优化示例（PyTorch）model = torch.jit.script(
    model,
    example_inputs={'input_ids': torch.randint(0, 100, (1, 32)),
        'attention_mask': torch.ones((1, 32))
    }
)

现象：切换模型后对话上下文丢失

解决方案：

def transfer_context(claude_history, local_format):
    return [
        {'role': 'user' if msg['role'] == 'human' else 'assistant',
            'content': msg['text']
        }
        for msg in claude_history
    ]

现象：Claude 使用 SSE 而本地模型是长连接

适配代码：

async def convert_stream(original_stream):
    async for chunk in original_stream:
        if isinstance(chunk, bytes):
            yield f"data: {chunk.decode()}\n\n"
        else:
            yield f"data: {json.dumps(chunk)}\n\n"

现象：连续切换后显存不释放

根治方案：

import gc

def cleanup():
    torch.cuda.empty_cache()
    gc.collect()

    # 重要：清空 CUDA 事件池
    for p in torch.cuda._C._cuda_getAllStreams():
        torch.cuda._C._cuda_streamSynchronize(p)