1. 为什么我要折腾这个路由层1.1 一个很现实的成本问题用 Claude Code 或者 Codex 这类 agent harness 写代码体验确实好但账单也是真好看。我自己的日常是一天下来几十万 token 的输入输出是常态如果全部走官方模型一个月下来成本能顶一台不错的开发机。而国内这几家模型——DeepSeek、通义、Kimi、GLM——在代码补全、函数调用、长上下文理解上已经相当能打价格却只有零头。问题在于harness 这类工具默认只认它自己那套 API 协议。Claude Code 走的是 Anthropic 的 messages 格式Codex 走的是 OpenAI 的 responses 格式你想把请求转到 DeepSeek 的 OpenAI 兼容接口上中间必须有人做协议翻译。这个人就是cn-llm-router。我写它的初衷很简单让 harness 以为自己在跟官方说话实际上请求被我转发到了国内模型上。harness 不需要改一行配置模型侧也不需要做任何适配中间这层路由把两边的方言翻译好就行。1.2 它到底解决什么问题具体来说cn-llm-router处理三件事协议转换把 Anthropic messages 格式、OpenAI responses 格式统一转成国内模型能吃的 OpenAI chat completions 格式再把返回结果转回去。模型路由根据请求里的模型名、或者自定义规则把流量分发到不同的国内模型供应商。比如简单补全走便宜的复杂推理走贵的。成本与可观测记录每次请求的 token 消耗、耗时、命中哪个模型方便你算账和调优。适合谁用三类人一是天天用 Claude Code / Codex 但想控成本的个人开发者二是团队里想统一管理模型出口、做审计和限流的技术负责人三是想拿国内模型做实验、又不想改 harness 源码的折腾党。注意这套方案的核心是协议翻译 转发不涉及任何网络层特殊处理纯粹是应用层的 API 适配。所有请求都走正常的 HTTPS 出站。2. 整体架构与选型思路2.1 为什么不用现成的 litellmlitellm 确实是个好东西它本身就能做多供应商代理也支持 Anthropic 格式的入口。但我实际用下来有几个不顺手的地方第一litellm 的配置偏重一个 config.yaml 动辄几百行想加个自定义路由规则得翻半天文档。第二它对国内模型的适配虽然能用但一些细节——比如 DeepSeek 的 reasoning_content 字段、某些模型的 tool_call 格式差异——需要自己打补丁。第三我想要一个能塞进单文件、随手改、启动只要一秒的东西。所以我选择自己写一个轻量路由核心依赖只有 FastAPI httpx总共不到 800 行。litellm 我保留作为备选如果你的场景需要支持几十家供应商、还要做复杂的负载均衡那直接用 litellm 更省事。但如果你的需求就是把 harness 接到国内两三家模型上自己写反而更可控。2.2 架构分层整个路由分四层从外到内层级职责关键实现接入层暴露 Anthropic / OpenAI 兼容端点FastAPI 路由/v1/messages、/v1/responses转换层请求/响应格式互转手写 mapper处理字段映射和流式路由层决定请求发给哪个模型规则引擎支持模型名映射和 fallback上游层实际调用国内模型 APIhttpx 异步客户端连接池复用这么分层的好处是每层可以独立替换。比如你以后想加一个供应商只需要在路由层加一条规则、在上游层加一个 client转换层完全不用动。2.3 流式处理是最大的坑harness 这类工具几乎全是流式输出因为要实时显示 agent 的思考过程。而 Anthropic 的 SSE 格式和 OpenAI 的 SSE 格式差别不小Anthropic 用event: message_start、content_block_delta这类事件类型数据在data:里。OpenAI 用统一的data: {...}靠choices[0].delta区分内容。转换层必须做逐块翻译不能等整个响应回来再转否则流式就废了。我的做法是上游返回的每个 chunk 先解析成内部统一结构再按目标协议重新序列化。这里有个细节——Anthropic 的content_block_start和content_block_stop必须成对出现漏一个客户端就会卡住。3. 核心细节拆解与实操要点3.1 请求格式转换的关键字段先看 Anthropic messages 转 OpenAI chat completions 的映射关系# 简化版转换逻辑 def anthropic_to_openai(req): messages [] # system 字段在 Anthropic 里是顶层OpenAI 里是 message if req.get(system): messages.append({role: system, content: req[system]}) for msg in req[messages]: # content 可能是字符串或 block 数组 if isinstance(msg[content], str): messages.append({role: msg[role], content: msg[content]}) else: # 处理 tool_use / tool_result / text block messages.extend(convert_blocks(msg)) return { model: map_model(req[model]), messages: messages, max_tokens: req.get(max_tokens, 4096), stream: req.get(stream, False), tools: convert_tools(req.get(tools, [])), }几个容易踩的点system 的位置Anthropic 把 system 放在顶层OpenAI 放在 messages 数组里。转换时必须提到最前面否则模型行为会变。tool_use 和 tool_resultAnthropic 的 tool 调用是 content block 里的tool_use类型结果用tool_result。OpenAI 用的是tool_calls字段和role: tool的消息。这个映射不对agent 的工具调用直接失效。max_tokens 必填Anthropic 要求 max_tokens 必填OpenAI 可选。转换时给个默认值我一般设 4096太小会截断长代码。3.2 模型名映射与路由规则harness 里配置的模型名通常是claude-sonnet-4-20250514这种但国内模型叫deepseek-chat、qwen-max。路由层要做名字翻译。我的做法是用一个 YAML 配置routes: - match: claude-.*sonnet.* target: deepseek-chat provider: deepseek - match: claude-.*haiku.* target: qwen-turbo provider: qwen - match: gpt-.* target: deepseek-reasoner provider: deepseek fallback: deepseek-chat匹配用正则从上往下第一个命中的生效。这样你可以让简单任务走便宜模型复杂任务走推理模型。实测下来把 haiku 类请求路由到 qwen-turbo成本能再降一大截而代码补全质量几乎无感。提示路由规则里一定要配 fallback。国内模型偶尔会限流或超时没有 fallback 的话 harness 会直接报错中断体验很差。3.3 流式响应的逐块翻译这是整个项目最核心也最容易出错的部分。以 Anthropic 格式为例一个完整的流式响应事件序列是message_start包含 message 元信息content_block_start开始一个内容块content_block_delta内容增量可能多次content_block_stop结束内容块message_delta包含 stop_reason 和 usagemessage_stop结束上游 OpenAI 格式的 chunk 长这样{choices:[{delta:{content:你},index:0}]}翻译逻辑是收到第一个 chunk 时先发message_start和content_block_start之后每个有 content 的 chunk 转成content_block_delta收到finish_reason时发content_block_stop、message_delta、message_stop。async def stream_translate(upstream): yield sse(message_start, {...}) yield sse(content_block_start, {index: 0, content_block: {type: text, text: }}) async for line in upstream: chunk parse_sse(line) delta chunk[choices][0][delta] if delta.get(content): yield sse(content_block_delta, { index: 0, delta: {type: text_delta, text: delta[content]} }) if chunk[choices][0].get(finish_reason): yield sse(content_block_stop, {index: 0}) yield sse(message_delta, {stop_reason: end_turn}) yield sse(message_stop, {})这里有个隐蔽的坑有些国内模型的流式返回里第一个 chunk 的 delta 是空的只有 role 字段。如果你直接拿它当 content 处理会多发一个空 delta客户端可能显示异常。我的处理是判断 content 非空才发。4. 完整实操流程4.1 环境准备与依赖安装先确认 Python 版本建议 3.10 以上因为用到了match语法和一些新类型标注。python --version # 建议 3.10 pip install fastapi uvicorn httpx pyyaml python-dotenv依赖很少就这五个。fastapi 提供 web 框架uvicorn 是 ASGI 服务器httpx 做异步 HTTP 客户端pyyaml 读配置python-dotenv 管密钥。4.2 配置文件编写在项目根目录建一个config.yamlserver: host: 127.0.0.1 port: 8787 providers: deepseek: base_url: https://api.deepseek.com/v1 api_key: ${DEEPSEEK_API_KEY} timeout: 120 qwen: base_url: https://dashscope.aliyuncs.com/compatible-mode/v1 api_key: ${QWEN_API_KEY} timeout: 120 routes: - match: claude-.*sonnet.* target: deepseek-chat provider: deepseek - match: claude-.*haiku.* target: qwen-turbo provider: qwen fallback: deepseek-chat logging: level: INFO log_tokens: true密钥用环境变量注入别写死在文件里。.env文件DEEPSEEK_API_KEYsk-xxxxxxxx QWEN_API_KEYsk-yyyyyyyy4.3 启动路由服务uvicorn cn_llm_router.main:app --host 127.0.0.1 --port 8787 --reload启动后访问http://127.0.0.1:8787/health应该返回{status:ok}。这一步先确认服务活着再配 harness。4.4 配置 Claude Code 指向路由Claude Code 支持通过环境变量指定 API 端点。设置export ANTHROPIC_BASE_URLhttp://127.0.0.1:8787 export ANTHROPIC_API_KEYany-stringAPI key 随便填因为路由层不校验它真正的密钥在路由的配置里。这样 Claude Code 的所有请求都会打到本地路由路由再转发到 DeepSeek。4.5 配置 Codex 指向路由Codex 走的是 OpenAI responses 格式配置方式类似在它的配置文件里改 base_url# ~/.codex/config.toml model_provider local model gpt-4 [model_providers.local] name local base_url http://127.0.0.1:8787/v1 wire_api responses路由层需要额外实现/v1/responses端点把 responses 格式转成 chat completions。这块比 messages 转换稍复杂因为 responses 格式的 input 结构不一样但核心思路一致。4.6 验证链路是否打通配好之后跑一个最简单的请求验证curl http://127.0.0.1:8787/v1/messages \ -H Content-Type: application/json \ -d { model: claude-sonnet-4-20250514, max_tokens: 100, messages: [{role: user, content: 写一个 Python 快排}] }如果返回的是 Anthropic 格式的响应且内容是 DeepSeek 生成的说明链路通了。看路由日志应该能看到route matched: claude-.*sonnet.* - deepseek-chat。5. 常见问题与排查技巧5.1 问题速查表现象可能原因排查方向harness 报 401API key 没配或路由没透传检查路由是否注入上游 key流式输出卡住不动content_block_start/stop 不配对抓包看 SSE 事件序列工具调用失效tool_use 转换错误检查 tool_calls 字段映射响应被截断max_tokens 太小调大默认值或透传原值请求超时上游模型响应慢调大 timeout加 fallback中文乱码编码问题确认 UTF-8 全链路5.2 几个我踩过的坑坑一SSE 的换行符。SSE 协议要求每个事件以\n\n结尾我一开始只写了\n结果客户端解析不出来流式一直卡着。这个细节文档里不显眼但错了就是致命的。坑二DeepSeek 的 reasoning_content。DeepSeek 的推理模型会返回reasoning_content字段里面是思考过程。如果直接丢掉harness 的思考展示就没了如果当成普通 content 塞进去又会污染输出。我的做法是把它映射成 Anthropic 的 thinking block这样 Claude Code 能正确显示。坑三并发连接数。httpx 默认连接池有限agent 高频请求时容易排队。我把limits调到max_connections100实测下来稳定很多。坑四模型名大小写。有些 harness 会传Claude-Sonnet这种带大写的名字正则匹配时记得加re.IGNORECASE否则路由匹配不上直接走 fallback。5.3 性能调优建议路由层本身几乎不耗时瓶颈都在上游模型。但有几个地方可以优化连接复用httpx 的 AsyncClient 全局单例别每次请求都新建。超时分级连接超时设短5s读取超时设长120s因为模型生成慢是正常的。日志异步化token 统计写日志别阻塞主流程用后台任务。缓存相同请求可以加一层短时缓存但 agent 场景下请求基本不重复收益不大。6. 成本与效果实测6.1 成本对比我拿一个真实的项目重构任务做了对比同样的 harness 配置分别走官方和走路由指标官方模型路由到国内模型输入 token约 120 万约 120 万输出 token约 18 万约 18 万单次任务成本基准 100%约 8%响应延迟基准略高 10-20%代码通过率基准约 92%成本降到 8% 左右这个数字很直观。延迟略高是因为国内模型首 token 时间稍长但流式输出下体感差异不大。代码通过率 92% 意味着大部分任务能一次过少数复杂任务需要重试综合下来还是划算。6.2 什么任务适合路由不是所有任务都适合。我的经验是适合代码补全、单元测试生成、文档撰写、简单重构、格式转换。谨慎复杂架构设计、多步推理、需要极强上下文一致性的长任务。对于复杂任务我会在路由规则里让它走推理模型比如 deepseek-reasoner虽然贵一点但质量更稳。这种分级路由是这套方案的精髓——不是一刀切而是按任务难度分配模型。6.3 后续可以扩展的方向这套路由目前够用但还有几个可以加的东西一是请求级别的成本预算超过阈值自动降级到便宜模型二是多供应商负载均衡同一模型配多个 key 轮询三是响应质量反馈把 harness 的重试信号收集起来自动调整路由策略。我个人在实际操作中的体会是路由层最大的价值不是省钱本身而是把模型选择变成一个可配置、可观测、可迭代的工程问题。以前你只能被动接受 harness 绑定的模型现在你可以像调优其他系统组件一样调优它。这个思路一旦建立起来后面接什么模型、怎么分配流量都是顺理成章的事。
