Claude Code Router与Ollama集成实战：构建高效AI服务路由方案

1次阅读

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

在构建 AI 服务路由系统时，我们常遇到以下核心挑战：

模型冷启动延迟：大型 AI 模型加载需要数分钟，导致突发流量时响应延迟飙升
GPU 资源竞争：多模型共享 GPU 时显存溢出和算力争抢问题频发
请求排队堆积：传统轮询策略导致长尾请求阻塞整个队列

特性	Claude Code Router	Nginx	Envoy
模型感知	支持	无	无
动态权重	内置	需 Lua 扩展	需 WASM 插件
热加载	<200ms	>1s	>500ms
协议支持	HTTP/gRPC	HTTP	HTTP/gRPC

轻量化容器：单个容器镜像 <100MB（对比传统 Docker 镜像 >2GB）
快速启动：冷启动时间从分钟级降至秒级
资源隔离：通过 cgroups v2 实现精细化的 GPU 显存划分

graph TD
    A[客户端] --> B[Claude Router]
    B --> C{模型在线?}
    C -->| 是 | D[Ollama 容器]
    C -->| 否 | E[启动新容器]
    D --> F[返回推理结果]
    E --> D
    style B fill:#f9f,stroke:#333
    style D fill:#bbf,stroke:#333

def calculate_weight(container):
    # 综合考量容器负载和模型特性
    load_score = 0.7 * container.cpu_util + 0.3 * container.mem_util
    model_factor = 1.0 / (1 + math.exp(-container.model.priority))
    return model_factor / (load_score + 0.1)  # 避免除以零

class AIServiceRouter:
    def __init__(self, ollama_host='localhost:11434'):
        self.containers = []  # 活跃容器列表
        self.health_check_interval = 30

    async def route_request(self, model_name, input_data):
        # 健康检查过滤
        healthy_containers = [c for c in self.containers 
                             if c.is_healthy and c.model == model_name]

        if not healthy_containers:
            container = await self.launch_container(model_name)
            healthy_containers.append(container)

        # 动态权重选择
        selected = max(healthy_containers, key=calculate_weight)
        try:
            return await selected.inference(input_data)
        except OllamaError as e:
            logger.error(f"推理失败: {e}")
            self.containers.remove(selected)
            raise

JWT 验证：

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload["sub"]
    except JWTError:
        raise HTTPException(status_code=403, detail="凭证无效")

速率限制：

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/api/v1/predict")
@limiter.limit("100/minute")
async def predict_endpoint(request: Request, data: ModelInput):
    return await router.route_request(data.model_name, data.input)

并发数	传统方案(p99)	本方案(p99)	资源节省
50	3200ms	1200ms	38%
100	超时	1800ms	52%
200	服务崩溃	2400ms	61%

# 不同 batch size 下的吞吐量测试
for bs in [1, 4, 8, 16]:
    start = time.time()
    process_batch(requests, batch_size=bs)
    print(f"Batch {bs}: {len(requests)/(time.time()-start):.1f} req/s")

输出结果：
– Batch 1: 45 req/s
– Batch 4: 112 req/s
– Batch 8: 158 req/s
– Batch 16: 167 req/s（显存不足风险增加）