Claude Code接入本地大模型：从原理到落地的技术实践

1次阅读

没有评论

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

随着大模型技术的快速发展，越来越多的企业开始将大模型能力集成到自己的业务系统中。相比于直接调用云端 API，本地部署大模型具有以下优势：

数据安全性更高，敏感数据无需外传
减少网络延迟，提升响应速度
可针对特定业务场景进行定制优化
长期使用成本更低

在将 Claude Code 接入本地大模型时，我们主要有两种接入方式可选：

优点：

实现简单，易于调试
兼容性好，几乎所有语言都能支持
适合小型项目或原型开发

缺点：

序列化 / 反序列化开销较大
长连接支持不佳
性能相对较低

优点：

基于 HTTP/2，性能更高
支持双向流式传输
协议缓冲区 (Protobuf) 序列化效率高

缺点：

实现复杂度较高
调试不如 REST 直观
需要额外生成客户端代码

对于大多数生产环境，特别是高并发场景，建议优先考虑 gRPC 方式。

import requests
import json

class ClaudeLocalModel:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.headers = {
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {api_key}'
        }

    def generate_text(self, prompt: str, max_tokens=100):
        """
        调用本地模型生成文本
        :param prompt: 输入的提示文本
        :param max_tokens: 最大生成 token 数
        :return: 生成的文本结果
        """payload = {'prompt': prompt,'max_tokens': max_tokens}

        try:
            response = requests.post(f'{self.base_url}/generate',
                headers=self.headers,
                data=json.dumps(payload),
                timeout=30
            )
            response.raise_for_status()
            return response.json()['text']
        except requests.exceptions.RequestException as e:
            print(f'API 请求失败: {e}')
            return None

首先需要定义 proto 文件：

syntax = "proto3";

service ClaudeService {rpc GenerateText (TextRequest) returns (TextResponse);
}

message TextRequest {
    string prompt = 1;
    int32 max_tokens = 2;
}

message TextResponse {string text = 1;}

然后使用 grpcio-tools 生成客户端代码，最终实现类如下：

import grpc
import claude_pb2
import claude_pb2_grpc

class ClaudeGRPCClient:
    def __init__(self, host: str, port: int):
        self.channel = grpc.insecure_channel(f'{host}:{port}')
        self.stub = claude_pb2_grpc.ClaudeServiceStub(self.channel)

    def generate_text(self, prompt: str, max_tokens=100):
        """
        gRPC 方式调用本地模型
        :param prompt: 输入提示
        :param max_tokens: 最大 token 数
        :return: 生成的文本
        """
        request = claude_pb2.TextRequest(
            prompt=prompt,
            max_tokens=max_tokens
        )

        try:
            response = self.stub.GenerateText(request, timeout=30)
            return response.text
        except grpc.RpcError as e:
            print(f'gRPC 调用失败: {e.code()}: {e.details()}')
            return None

对于多个相似的请求，可以合并为一个批处理请求，减少网络开销：

def batch_generate_text(self, prompts: list, max_tokens=100):
    """
    批量生成文本
    :param prompts: 提示文本列表
    :param max_tokens: 最大 token 数
    :return: 生成文本列表
    """
    batch_request = claude_pb2.BatchTextRequest(
        prompts=prompts,
        max_tokens=max_tokens
    )

    try:
        response = self.stub.BatchGenerateText(batch_request, timeout=60)
        return response.texts
    except grpc.RpcError as e:
        print(f'批量请求失败: {e.code()}: {e.details()}')
        return None

使用异步 IO 提升并发性能：

import asyncio

class AsyncClaudeClient:
    def __init__(self, host: str, port: int):
        self.channel = grpc.aio.insecure_channel(f'{host}:{port}')
        self.stub = claude_pb2_grpc.ClaudeServiceStub(self.channel)

    async def generate_text(self, prompt: str, max_tokens=100):
        request = claude_pb2.TextRequest(
            prompt=prompt,
            max_tokens=max_tokens
        )

        try:
            response = await self.stub.GenerateText(request, timeout=30)
            return response.text
        except grpc.RpcError as e:
            print(f'异步调用失败: {e.code()}: {e.details()}')
            return None

实现简单的 LRU 缓存：

from functools import lru_cache

class CachedClaudeClient(ClaudeGRPCClient):
    @lru_cache(maxsize=1000)
    def generate_text(self, prompt: str, max_tokens=100):
        return super().generate_text(prompt, max_tokens)