如果你最近在折腾 Agent Harness大概率撞上过同一个坑Harness 本身是真好用但模型端点几乎是被焊死的。我最初只是想把手头一个常跑批量任务的 Harness从默认的贵价模型换成国内高性价比的开源模型结果发现光改 base_url 和 API Key 完全不够——它不仅要校验 Key还会去请求一个固定路径的模型列表消息协议也和国内模型服务商默认给的接口对不上。折腾两天之后我干脆自己写了个本地网关也就是 cn-llm-router它在本机监听一个端口对 Harness 发来的所有模型请求做三件事——把模型名和鉴权信息改写成目标服务商认识的形态做一层请求与响应协议翻译再原样转发流式请求则按协议规则做逐块的格式转换。这篇文章会把网关的完整设计、核心实现、接入方式和踩坑记录都摊开讲给同样被模型绑定问题折磨的人一个可以直接抄作业的方案。这个项目适合这几类人正在用 Harness 做 Agent 开发但觉得默认模型太贵的人想接国产模型、又不想改 Harness 源码的人团队里希望把模型接入做成可配置、可切换、可统计的基础设施的人。下面从最开始的需求聊起。1. 为什么需要 cn-llm-router一次模型切换引发的连环报错1.1 从改个 base_url 就能用到三连报错我一开始对换模型的难度严重低估。当时想着国内开源模型普遍提供 Chat Completions 风格接口Harness 也是走标准模型接口把地址指过去、填上 Key 不就行了结果实际操作时Harness 启动阶段先给我来了一个 401 Unauthorized紧接着报 ModelNotFound最后请求路径都不对。我把日志打开仔细看才发现 Harness 客户端在底层走的是 Messages 风格的接口路径是 /v1/messages而我配置的是 /v1/chat/completions它做鉴权也不是简单发一个 Bearer Token而是用一个本地登录态去换临时凭证。换句话说我面对的不是改一行配置的问题而是两个接口规范 两套鉴权逻辑的兼容问题。这个现象其实很普遍。国外主流的 Harness 工具在设计时默认就是绑定自家模型服务的所以代码里充满了针对自家接口的特殊处理系统提示词放在独立字段、工具调用用 input_schema 描述、流式返回用分块事件而不是纯文本增量。国内模型平台倒是一直在往兼容方向靠但兼容的多是 Chat Completions 风格。两边各说各话中间如果不加一个翻译层你怎么换模型都会翻车。1.2 国外 Harness 的绑定到底藏在哪我把实际拆解后发现绑定主要由四层构成缺一层都跑不通协议层Harness 与模型服务之间的请求响应格式包括多轮消息结构、系统提示词放哪、工具调用如何描述、函数参数如何传递。两个风格在这几个点上差异非常明显。鉴权层Harness 默认走自己的账号体系或者要求特定格式的 Key。即使你填一个服务商给的 Key它也可能在本地做格式校验格式不对直接拒绝启动。而国内服务商普遍只认Authorization: Bearer key。模型元数据层Harness 初始化时会主动请求一次模型列表用于校验你配置的模型名是否存在同时读取模型的上下文窗口长度。这一步被很多开发者忽略了但其实是最容易报模型不存在的根因。策略层Harness 会按模型名的语义推断能力。在同一套 Harness 里不同模型会被分配到不同任务比如复杂代码生成用旗舰模型简单问答用轻量模型。如果模型名映射做不好性能预期会不准。想绕开这四层绑定最干净的做法不是去改 Harness 源码而是在它和模型服务商之间放一个看得懂两边规矩的网关。1.3 为什么是本地路由网关而不是改源码或上商业网关我认真比较过三条路各自都有明显短板。改 Harness 源码是最直接的想法但代价也最大。Harness 版本更新频繁每次升级都要处理冲突插件生态也可能因为改动而失效。我更希望能把模型接入和 Harness 主程序解耦让升级不碰路由逻辑。上商业 API 网关则是另一个极端。功能确实全但数据要经过第三方服务很多团队对代码和对话数据出本地这件事有硬要求此外商业网关按量计费对个人开发者和小团队来说是一笔额外开销配置也偏重。本地网关就舒服得多监听 127.0.0.1 的某个端口请求完全不出本机数据安全可控协议转换、模型映射、故障转移、用量统计这些逻辑集中在一个进程里Harness 那边只需要把模型服务地址指到本机后续想换服务商改配置就行不用碰代码。它的本质可以理解成一个多国语言转接头——一边听懂 Harness 的方言一边听懂国内模型服务商的普通话中间做实时翻译。2. 整体设计拆解cn-llm-router 需要具备哪些能力2.1 一张模型路由表加一层协议翻译器cn-llm-router 的定位不是一个重型平台而是一个模型寻址表 协议翻译器。核心思想是配置驱动所有模型映射、服务商信息、路由策略都在一个 YAML 文件里维护程序本身只做机械的查表和转换。这样做的好处是日常操作里 90% 的变更都不需要改代码只需要改配置并重启服务。进程内部主要有五个模块入口层对外暴露两个接口端点一个兼容 Chat Completions 风格一个兼容 Messages 风格分别对应不同 Harness 的默认行为。解析层从请求体里取出模型名、消息内容、工具定义、采样参数统一成内部对象。路由层根据模型别名查映射表找到目标服务商和真实模型 ID再按配置决定是否需要走备用服务商。转换层把内部对象转成目标服务商的请求格式响应回来后再转成 Harness 认识的格式。治理层记录 token 用量、错误率、延迟做并发限流。2.2 双向协议转换是核心难点协议转换不是简单改几个字段名而是要把整个消息模型对齐。Messages 风格和 Chat Completions 风格至少有四个关键差异需要处理。第一系统提示词的位置前者单独放在顶层 system 字段后者需要塞进 messages 数组的第一条且 role 标记为 system。第二工具调用的描述方式前者用一个input_schema描述参数结构后者叫parameters而且在函数类型里还嵌套了一层对象。第三多轮对话中工具结果回传前者用 tool_result 块后者用 role 为 tool 的消息并且需要携带 tool_call_id 跟之前的函数调用关联。第四流式输出前者的每一块内容都有明确的类型标记比如文本增量、工具参数增量后者的增量则统一包在 choices 数组的 delta 字段里。每一个差异背后都对应实际场景。比如工具调用Harness 让模型生成一个调用某函数的指令模型返回的内容里包含函数名和参数转换层必须把参数从 JSON 对象转成 JSON 字符串再包进 function 字段否则对端模型服务会直接解析失败。很多初版实现就是栽在这种小细节上的。2.3 为什么用 FastAPI httpx 就够不引入重型组件有过分布式系统经验的人可能会想模型路由这种活是不是得上消息队列、Redis、Kubernetes我的判断是这个场景绝大多数情况下不需要。Harness 本来就是一个单机或小规模工具QPS 不高核心瓶颈在模型服务的网络延迟上而不是网关的计算能力。用 Python 的 FastAPI 做入口用 httpx 做异步转发单进程就能扛住日常使用结构简单部署就是python router.py一行命令。异步非阻塞是必须的因为模型接口的响应时间从几秒到几十秒不等长连接很多。FastAPI 的 async 机制配合 httpx.AsyncClient 能很好地处理这类耗时请求不会阻塞其他请求。引入消息队列反而会增加部署心智负担对个人开发者来说得不偿失。3. 核心实现一个能跑的 cn-llm-router 长什么样3.1 目录结构与主服务骨架我项目的目录结构很简单四个文件足以描述核心逻辑cn-llm-router/ ├── config.example.yaml # 配置示例服务商、模型映射、路由策略 ├── router.py # FastAPI 主服务与请求入口 ├── convert.py # 协议转换逻辑 ├── providers.py # 转发、故障转移、限流 └── requirements.txt # 依赖清单入口层只需要两个 POST 接口。核心逻辑是读请求体解析模型名查路由表做转换转发再转换响应。# router.py 关键骨架 from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import httpx app FastAPI() app.post(/v1/chat/completions) async def chat_completions(request: Request): body await request.json() return await forward_request(body, stylechat) app.post(/v1/messages) async def messages(request: Request): body await request.json() return await forward_request(body, stylemessages)forward_request 内部先确定目标服务商再调用 providers 模块发送请求。对于流式请求返回 StreamingResponse 并设置正确的 Content-Type。3.2 鉴权改写把 Harness 的 Token吃掉这是最早遇到也是最容易解决的问题。Harness 那边通常要求填一个 Key 才能启动但这个 Key 不一定能直接用在国产模型服务商上。最简单可靠的办法给 Harness 填一个格式合法的占位 Key比如sk-local-gateway网关收到请求后忽略入站请求的 Authorization替换成目标服务商的真实 Key。def build_outbound_headers(inbound_headers: dict, provider: dict) - dict: return { Authorization: fBearer {os.environ[provider[api_key_env]]}, Content-Type: application/json, }真实 Key 不写进 YAML而是通过环境变量注入避免配置文件泄露。这个方案的好处是Harness 的鉴权纯属摆设真正管 Key 的是网关。3.3 协议转换中最容易错的三个细节第一个是 system 字段。Messages 风格允许 system 是一个字符串或内容块数组Chat Completions 风格要求 role 为 system 的消息。转换时要保证 system 排在消息序列最前面。def anthropic_to_openai(body: dict) - dict: messages [] if body.get(system): messages.append({role: system, content: body[system]}) # 继续处理其他消息第二个是工具定义。Messages 风格的 tool 结构里字段名是input_schemaChat Completions 风格是parameters。转换时把input_schema原样塞进parameters再包一层 function 对象。第三个是 max_tokens。Messages 风格强制要求这个字段不填直接报错Chat Completions 风格则是可选。网关在转换时要主动补一个默认值否则 Harness 不传国内模型服务商就有可能拒绝请求。3.4 模型探测与名称映射/v1/models 的坑很多 Harness 在启动阶段会先调用模型列表接口校验配置的模型名是否合法。如果不接管这个接口Harness 会拿默认模型名去探测而国内服务商根本没有这个模型直接报 ModelNotFound。解决办法是在网关里同样实现一个 /v1/models 端点返回一份伪造的模型列表。列表里包含 Harness 认识的模型名对应字段如 id、context_window 等要按真实映射后的模型能力填写。这一步很关键因为 Harness 会依据 context_window 决定该给模型塞多少历史上下文填太小影响效果填太大又会触发服务商超限。模型名映射在配置里要支持一对多一个 Harness 侧模型名可以对应多个目标服务商模型按权重或优先级分发。routes: - match: heavy-model targets: - provider: provider_a model: reasoner-model weight: 80 - provider: provider_b model: max-model weight: 20这种设计让给不同任务分配不同模型变得非常灵活。3.5 流式 SSE 的逐块转换与透传Agent 场景几乎全程流式网关对流式的处理质量直接决定体验。Chat Completions 风格的流式响应是 text/event-stream每一行是data: {json}最后一行是data: [DONE]。Messages 风格则是多种事件交替message_start、content_block_start、content_block_delta、content_block_stop、message_delta、message_stop。双向转换的要点是事件对齐。从 Messages 风格转到 Chat Completions 风格时要把 content_block_delta 里的文本增量映射到 delta.content把工具参数增量累积成 delta.tool_calls 里的 arguments 增量结束时用 message_delta 里的 stop_reason 映射 finish_reason。反向转换同理。async def stream_translate_anthropic_to_openai(resp): async for line in resp.aiter_lines(): if not line.startswith(data:): continue payload json.loads(line[5:].strip()) if payload[type] content_block_delta: delta payload[delta] if delta[type] text_delta: yield fdata: {json.dumps({choices: [{delta: {content: delta[text]}}]})}\n\n elif delta[type] input_json_delta: # 累积后作为 function.arguments 增量送出 yield build_tool_call_delta(delta[partial_json])流式转换时还要注意超时和心跳。模型服务有时思考时间很长超过默认超时时间连接就被掐断。网关要把上游超时时间调大并给 Harness 侧持续发送注释行或空事件保活避免前端提前断开。3.6 故障转移与限流让路由真正聪明起来多模型服务商接入后故障转移就成了刚需。有的服务商高峰期会限流返回 429有的会偶尔 5xx。网关要做的是依次尝试服务商列表把 429、500、502、503 视为可转移故障换下一个服务商其他错误则直接返回给调用方。async def call_with_fallback(payload, provider_list): for provider in provider_list: try: resp await send(provider, payload) if resp.status_code in (429, 500, 502, 503): continue return resp except httpx.TransportError: continue raise GatewayError(all providers failed)限流方面用 asyncio.Semaphore 给每个服务商配一个并发上限防止 Harness 的并行任务把某个服务商瞬间打爆。3.7 用量统计不统计就没法优化成本换模型的核心动力是省钱但省没省、省多少需要数据说话。网关在每个请求完成后把 prompt_tokens、completion_tokens、模型名、服务商写进一行 JSONL 日志。我后来加了一个简单的内存聚合函数可以按Harness 侧模型别名分组统计当日 token 消耗在 /metrics 端点返回。这样哪个模型贵、哪个模型跑得多一眼就能看出来之后调整路由权重就有依据。4. 接入 Harness配置文件与 5 分钟冒烟测试4.1 完整的配置文件示例配置文件是网关和真实场景之间的桥梁我贴一份我实际在用的结构略去敏感信息listen: host: 127.0.0.1 port: 8321 providers: provider_a: type: openai_compatible base_url: https://api.your-model-service-a.example.com/v1 api_key_env: MODEL_SERVICE_A_KEY timeout: 600 max_concurrency: 16 provider_b: type: openai_compatible base_url: https://api.your-model-service-b.example.com/v1 api_key_env: MODEL_SERVICE_B_KEY timeout: 600 max_concurrency: 8 routes: - match: harness-heavy context_window: 64000 targets: - provider: provider_a model: reasoner-model weight: 80 - provider: provider_b model: light-chat-model weight: 20 - match: harness-light context_window: 32000 targets: - provider: provider_a model: light-chat-model weight: 100base_url 指向服务商的兼容端点模型名是服务商真实模型 ID。这样配置的好处是Harness 侧看到的永远是 harness-heavy、harness-light 这类逻辑名底层随便换。4.2 把 Harness 指到本地网关的三种方式不同 Harness 的接入方式略有差异但大体可以归为三类总有一款适合你环境变量法很多 Harness 支持通过环境变量覆盖模型服务地址和 Key。最常见的模式是设置类似 MODEL_BASE_URL 和 MODEL_API_KEY 的变量。CLI 参数法Harness 在启动时接受模型参数指定一个模型名。配置文件法Harness 有配置文件在配置里声明自定义模型服务指向 http://127.0.0.1:8321/v1。无论哪种核心都是让 Harness 认为本机网关就是模型服务商。这样就绕开了默认官方的账号体系。4.3 冒烟测试curl 验证路由和工具调用配置完别急着进 Harness先用 curl 把网关打一遍。我最常用的三个测试命令如下。先测模型列表探测curl http://127.0.0.1:8321/v1/models再测普通对话curl http://127.0.0.1:8321/v1/chat/completions \ -H Content-Type: application/json \ -d {model:harness-light,messages:[{role:user,content:hi}]}最后测带工具调用的流式请求。这步最容易暴露协议转换问题所以我会用一个包含函数定义和强制工具选择的请求来验证curl http://127.0.0.1:8321/v1/messages \ -H Content-Type: application/json \ -d { model:harness-heavy, max_tokens: 512, stream: true, tools: [{ name: get_weather, description: 获取天气, input_schema: { type: object, properties: {city: {type: string}} } }], tool_choice: {type: tool, name: get_weather}, messages: [{role: user, content: 北京天气怎么样}] }返回里如果能看到完整的工具调用块和参数 JSON说明协议转换链路是通的再进 Harness 就基本稳了。5. 生产实践避坑清单5.1 高频故障速查表我整理了一张表格都是实际跑出来的经验按出现频率排序现象根因解决方案启动即报 401Harness 要求特定格式 Key网关忽略入站鉴权头替换为服务商真实 Key报 ModelNotFound/v1/models 没被接管或模型名不匹配网关实现模型列表接口返回配置里所有别名工具调用返回参数为空input_schema 未正确转成 parameters检查 convert 模块中的工具定义转换逻辑流式对话中途断流上游超时时间过短调大 httpx timeout网关侧发送保活注释行模型不接上下文context_window 配置过小在模型路由表里按真实能力填 context_window某个服务商频繁 429并发过高触发限流配置 max_concurrency启动故障转移响应内容乱码或截断SSE 解析没处理多行 JSON确保按行读取并只处理 data: 前缀的行5.2 离线与局域网部署要点很多团队会在完全离线的环境里跑 Harness。这个场景下国产模型服务可能换成本地部署的模型运行时。好消息是网关完全不受影响——只需要把 provider 的 base_url 指向本机或内网某台服务器上的推理服务地址。如果推理服务提供的是 Chat Completions 兼容接口网关连代码都不用改只改配置。局域网场景要注意的是其他机器要访问网关监听地址就不能写成 127.0.0.1要改成 0.0.0.0并确认防火墙放行对应端口。另外内网部署环境通常没有外网 DNS配置文件里不要写域名直接写内网 IP 最稳。5.3 长期运行的管理经验网关跑了几个月后我总结出三条维护经验。第一配置版本化。路由表就是模型接入的历史记录谁在什么时候把流量从哪个服务商切到了另一个在 Git 历史里都查得到。出问题的时候回滚非常方便。第二Key 轮换要设计成零停机。真实 Key 通过环境变量注入重启进程即可生效。如果 Key 写在 YAML 里轮换时还得担心文件泄露和权限问题不如环境变量干净。第三token 日志要及时归档。日志文件按天切割再写个简单脚本按周汇总。我后来就是靠这份日志发现某个轻量模型实际成本比旗舰模型还高因为它的错误重试次数太多光是重试 token 就把省下的钱吃掉了。写在最后的个人体会就我自己的使用体验而言cn-llm-router 最值钱的部分不是协议转换代码本身而是把模型选择权从工具手里夺回来这件事。过去换模型要改 Harness、求插件、等社区适配现在只是改一行 YAML重启网关任务流自动走新模型。我建议如果你也打算做类似的东西第一版千万别追求功能齐全先跑通最简单的对话和流式再逐步加工具调用、故障转移、统计。等跑顺了之后可以继续扩展的方向我比较看好两个一是把评测逻辑挂到网关里让模型切换前先自动跑一组样例集对比效果二是接入更细粒度的任务级路由比如系统提示词里出现代码关键字就走重模型否则走轻模型。这个思路一旦跑通模型成本还能再降一个台阶。 
长河君 · 2026/10/26 12:58:40 · 阅读 1.2k 打造本地模型网关：轻松破解Agent Harness模型绑定
如果你最近在折腾 Agent Harness大概率撞上过同一个坑Harness 本身是真好用但模型端点几乎是被焊死的。我最初只是想把手头一个常跑批量任务的 Harness从默认的贵价模型换成国内高性价比的开源模型结果发现光改 base_url 和 API Key 完全不够——它不仅要校验 Key还会去请求一个固定路径的模型列表消息协议也和国内模型服务商默认给的接口对不上。折腾两天之后我干脆自己写了个本地网关也就是 cn-llm-router它在本机监听一个端口对 Harness 发来的所有模型请求做三件事——把模型名和鉴权信息改写成目标服务商认识的形态做一层请求与响应协议翻译再原样转发流式请求则按协议规则做逐块的格式转换。这篇文章会把网关的完整设计、核心实现、接入方式和踩坑记录都摊开讲给同样被模型绑定问题折磨的人一个可以直接抄作业的方案。这个项目适合这几类人正在用 Harness 做 Agent 开发但觉得默认模型太贵的人想接国产模型、又不想改 Harness 源码的人团队里希望把模型接入做成可配置、可切换、可统计的基础设施的人。下面从最开始的需求聊起。1. 为什么需要 cn-llm-router一次模型切换引发的连环报错1.1 从改个 base_url 就能用到三连报错我一开始对换模型的难度严重低估。当时想着国内开源模型普遍提供 Chat Completions 风格接口Harness 也是走标准模型接口把地址指过去、填上 Key 不就行了结果实际操作时Harness 启动阶段先给我来了一个 401 Unauthorized紧接着报 ModelNotFound最后请求路径都不对。我把日志打开仔细看才发现 Harness 客户端在底层走的是 Messages 风格的接口路径是 /v1/messages而我配置的是 /v1/chat/completions它做鉴权也不是简单发一个 Bearer Token而是用一个本地登录态去换临时凭证。换句话说我面对的不是改一行配置的问题而是两个接口规范 两套鉴权逻辑的兼容问题。这个现象其实很普遍。国外主流的 Harness 工具在设计时默认就是绑定自家模型服务的所以代码里充满了针对自家接口的特殊处理系统提示词放在独立字段、工具调用用 input_schema 描述、流式返回用分块事件而不是纯文本增量。国内模型平台倒是一直在往兼容方向靠但兼容的多是 Chat Completions 风格。两边各说各话中间如果不加一个翻译层你怎么换模型都会翻车。1.2 国外 Harness 的绑定到底藏在哪我把实际拆解后发现绑定主要由四层构成缺一层都跑不通协议层Harness 与模型服务之间的请求响应格式包括多轮消息结构、系统提示词放哪、工具调用如何描述、函数参数如何传递。两个风格在这几个点上差异非常明显。鉴权层Harness 默认走自己的账号体系或者要求特定格式的 Key。即使你填一个服务商给的 Key它也可能在本地做格式校验格式不对直接拒绝启动。而国内服务商普遍只认Authorization: Bearer key。模型元数据层Harness 初始化时会主动请求一次模型列表用于校验你配置的模型名是否存在同时读取模型的上下文窗口长度。这一步被很多开发者忽略了但其实是最容易报模型不存在的根因。策略层Harness 会按模型名的语义推断能力。在同一套 Harness 里不同模型会被分配到不同任务比如复杂代码生成用旗舰模型简单问答用轻量模型。如果模型名映射做不好性能预期会不准。想绕开这四层绑定最干净的做法不是去改 Harness 源码而是在它和模型服务商之间放一个看得懂两边规矩的网关。1.3 为什么是本地路由网关而不是改源码或上商业网关我认真比较过三条路各自都有明显短板。改 Harness 源码是最直接的想法但代价也最大。Harness 版本更新频繁每次升级都要处理冲突插件生态也可能因为改动而失效。我更希望能把模型接入和 Harness 主程序解耦让升级不碰路由逻辑。上商业 API 网关则是另一个极端。功能确实全但数据要经过第三方服务很多团队对代码和对话数据出本地这件事有硬要求此外商业网关按量计费对个人开发者和小团队来说是一笔额外开销配置也偏重。本地网关就舒服得多监听 127.0.0.1 的某个端口请求完全不出本机数据安全可控协议转换、模型映射、故障转移、用量统计这些逻辑集中在一个进程里Harness 那边只需要把模型服务地址指到本机后续想换服务商改配置就行不用碰代码。它的本质可以理解成一个多国语言转接头——一边听懂 Harness 的方言一边听懂国内模型服务商的普通话中间做实时翻译。2. 整体设计拆解cn-llm-router 需要具备哪些能力2.1 一张模型路由表加一层协议翻译器cn-llm-router 的定位不是一个重型平台而是一个模型寻址表 协议翻译器。核心思想是配置驱动所有模型映射、服务商信息、路由策略都在一个 YAML 文件里维护程序本身只做机械的查表和转换。这样做的好处是日常操作里 90% 的变更都不需要改代码只需要改配置并重启服务。进程内部主要有五个模块入口层对外暴露两个接口端点一个兼容 Chat Completions 风格一个兼容 Messages 风格分别对应不同 Harness 的默认行为。解析层从请求体里取出模型名、消息内容、工具定义、采样参数统一成内部对象。路由层根据模型别名查映射表找到目标服务商和真实模型 ID再按配置决定是否需要走备用服务商。转换层把内部对象转成目标服务商的请求格式响应回来后再转成 Harness 认识的格式。治理层记录 token 用量、错误率、延迟做并发限流。2.2 双向协议转换是核心难点协议转换不是简单改几个字段名而是要把整个消息模型对齐。Messages 风格和 Chat Completions 风格至少有四个关键差异需要处理。第一系统提示词的位置前者单独放在顶层 system 字段后者需要塞进 messages 数组的第一条且 role 标记为 system。第二工具调用的描述方式前者用一个input_schema描述参数结构后者叫parameters而且在函数类型里还嵌套了一层对象。第三多轮对话中工具结果回传前者用 tool_result 块后者用 role 为 tool 的消息并且需要携带 tool_call_id 跟之前的函数调用关联。第四流式输出前者的每一块内容都有明确的类型标记比如文本增量、工具参数增量后者的增量则统一包在 choices 数组的 delta 字段里。每一个差异背后都对应实际场景。比如工具调用Harness 让模型生成一个调用某函数的指令模型返回的内容里包含函数名和参数转换层必须把参数从 JSON 对象转成 JSON 字符串再包进 function 字段否则对端模型服务会直接解析失败。很多初版实现就是栽在这种小细节上的。2.3 为什么用 FastAPI httpx 就够不引入重型组件有过分布式系统经验的人可能会想模型路由这种活是不是得上消息队列、Redis、Kubernetes我的判断是这个场景绝大多数情况下不需要。Harness 本来就是一个单机或小规模工具QPS 不高核心瓶颈在模型服务的网络延迟上而不是网关的计算能力。用 Python 的 FastAPI 做入口用 httpx 做异步转发单进程就能扛住日常使用结构简单部署就是python router.py一行命令。异步非阻塞是必须的因为模型接口的响应时间从几秒到几十秒不等长连接很多。FastAPI 的 async 机制配合 httpx.AsyncClient 能很好地处理这类耗时请求不会阻塞其他请求。引入消息队列反而会增加部署心智负担对个人开发者来说得不偿失。3. 核心实现一个能跑的 cn-llm-router 长什么样3.1 目录结构与主服务骨架我项目的目录结构很简单四个文件足以描述核心逻辑cn-llm-router/ ├── config.example.yaml # 配置示例服务商、模型映射、路由策略 ├── router.py # FastAPI 主服务与请求入口 ├── convert.py # 协议转换逻辑 ├── providers.py # 转发、故障转移、限流 └── requirements.txt # 依赖清单入口层只需要两个 POST 接口。核心逻辑是读请求体解析模型名查路由表做转换转发再转换响应。# router.py 关键骨架 from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import httpx app FastAPI() app.post(/v1/chat/completions) async def chat_completions(request: Request): body await request.json() return await forward_request(body, stylechat) app.post(/v1/messages) async def messages(request: Request): body await request.json() return await forward_request(body, stylemessages)forward_request 内部先确定目标服务商再调用 providers 模块发送请求。对于流式请求返回 StreamingResponse 并设置正确的 Content-Type。3.2 鉴权改写把 Harness 的 Token吃掉这是最早遇到也是最容易解决的问题。Harness 那边通常要求填一个 Key 才能启动但这个 Key 不一定能直接用在国产模型服务商上。最简单可靠的办法给 Harness 填一个格式合法的占位 Key比如sk-local-gateway网关收到请求后忽略入站请求的 Authorization替换成目标服务商的真实 Key。def build_outbound_headers(inbound_headers: dict, provider: dict) - dict: return { Authorization: fBearer {os.environ[provider[api_key_env]]}, Content-Type: application/json, }真实 Key 不写进 YAML而是通过环境变量注入避免配置文件泄露。这个方案的好处是Harness 的鉴权纯属摆设真正管 Key 的是网关。3.3 协议转换中最容易错的三个细节第一个是 system 字段。Messages 风格允许 system 是一个字符串或内容块数组Chat Completions 风格要求 role 为 system 的消息。转换时要保证 system 排在消息序列最前面。def anthropic_to_openai(body: dict) - dict: messages [] if body.get(system): messages.append({role: system, content: body[system]}) # 继续处理其他消息第二个是工具定义。Messages 风格的 tool 结构里字段名是input_schemaChat Completions 风格是parameters。转换时把input_schema原样塞进parameters再包一层 function 对象。第三个是 max_tokens。Messages 风格强制要求这个字段不填直接报错Chat Completions 风格则是可选。网关在转换时要主动补一个默认值否则 Harness 不传国内模型服务商就有可能拒绝请求。3.4 模型探测与名称映射/v1/models 的坑很多 Harness 在启动阶段会先调用模型列表接口校验配置的模型名是否合法。如果不接管这个接口Harness 会拿默认模型名去探测而国内服务商根本没有这个模型直接报 ModelNotFound。解决办法是在网关里同样实现一个 /v1/models 端点返回一份伪造的模型列表。列表里包含 Harness 认识的模型名对应字段如 id、context_window 等要按真实映射后的模型能力填写。这一步很关键因为 Harness 会依据 context_window 决定该给模型塞多少历史上下文填太小影响效果填太大又会触发服务商超限。模型名映射在配置里要支持一对多一个 Harness 侧模型名可以对应多个目标服务商模型按权重或优先级分发。routes: - match: heavy-model targets: - provider: provider_a model: reasoner-model weight: 80 - provider: provider_b model: max-model weight: 20这种设计让给不同任务分配不同模型变得非常灵活。3.5 流式 SSE 的逐块转换与透传Agent 场景几乎全程流式网关对流式的处理质量直接决定体验。Chat Completions 风格的流式响应是 text/event-stream每一行是data: {json}最后一行是data: [DONE]。Messages 风格则是多种事件交替message_start、content_block_start、content_block_delta、content_block_stop、message_delta、message_stop。双向转换的要点是事件对齐。从 Messages 风格转到 Chat Completions 风格时要把 content_block_delta 里的文本增量映射到 delta.content把工具参数增量累积成 delta.tool_calls 里的 arguments 增量结束时用 message_delta 里的 stop_reason 映射 finish_reason。反向转换同理。async def stream_translate_anthropic_to_openai(resp): async for line in resp.aiter_lines(): if not line.startswith(data:): continue payload json.loads(line[5:].strip()) if payload[type] content_block_delta: delta payload[delta] if delta[type] text_delta: yield fdata: {json.dumps({choices: [{delta: {content: delta[text]}}]})}\n\n elif delta[type] input_json_delta: # 累积后作为 function.arguments 增量送出 yield build_tool_call_delta(delta[partial_json])流式转换时还要注意超时和心跳。模型服务有时思考时间很长超过默认超时时间连接就被掐断。网关要把上游超时时间调大并给 Harness 侧持续发送注释行或空事件保活避免前端提前断开。3.6 故障转移与限流让路由真正聪明起来多模型服务商接入后故障转移就成了刚需。有的服务商高峰期会限流返回 429有的会偶尔 5xx。网关要做的是依次尝试服务商列表把 429、500、502、503 视为可转移故障换下一个服务商其他错误则直接返回给调用方。async def call_with_fallback(payload, provider_list): for provider in provider_list: try: resp await send(provider, payload) if resp.status_code in (429, 500, 502, 503): continue return resp except httpx.TransportError: continue raise GatewayError(all providers failed)限流方面用 asyncio.Semaphore 给每个服务商配一个并发上限防止 Harness 的并行任务把某个服务商瞬间打爆。3.7 用量统计不统计就没法优化成本换模型的核心动力是省钱但省没省、省多少需要数据说话。网关在每个请求完成后把 prompt_tokens、completion_tokens、模型名、服务商写进一行 JSONL 日志。我后来加了一个简单的内存聚合函数可以按Harness 侧模型别名分组统计当日 token 消耗在 /metrics 端点返回。这样哪个模型贵、哪个模型跑得多一眼就能看出来之后调整路由权重就有依据。4. 接入 Harness配置文件与 5 分钟冒烟测试4.1 完整的配置文件示例配置文件是网关和真实场景之间的桥梁我贴一份我实际在用的结构略去敏感信息listen: host: 127.0.0.1 port: 8321 providers: provider_a: type: openai_compatible base_url: https://api.your-model-service-a.example.com/v1 api_key_env: MODEL_SERVICE_A_KEY timeout: 600 max_concurrency: 16 provider_b: type: openai_compatible base_url: https://api.your-model-service-b.example.com/v1 api_key_env: MODEL_SERVICE_B_KEY timeout: 600 max_concurrency: 8 routes: - match: harness-heavy context_window: 64000 targets: - provider: provider_a model: reasoner-model weight: 80 - provider: provider_b model: light-chat-model weight: 20 - match: harness-light context_window: 32000 targets: - provider: provider_a model: light-chat-model weight: 100base_url 指向服务商的兼容端点模型名是服务商真实模型 ID。这样配置的好处是Harness 侧看到的永远是 harness-heavy、harness-light 这类逻辑名底层随便换。4.2 把 Harness 指到本地网关的三种方式不同 Harness 的接入方式略有差异但大体可以归为三类总有一款适合你环境变量法很多 Harness 支持通过环境变量覆盖模型服务地址和 Key。最常见的模式是设置类似 MODEL_BASE_URL 和 MODEL_API_KEY 的变量。CLI 参数法Harness 在启动时接受模型参数指定一个模型名。配置文件法Harness 有配置文件在配置里声明自定义模型服务指向 http://127.0.0.1:8321/v1。无论哪种核心都是让 Harness 认为本机网关就是模型服务商。这样就绕开了默认官方的账号体系。4.3 冒烟测试curl 验证路由和工具调用配置完别急着进 Harness先用 curl 把网关打一遍。我最常用的三个测试命令如下。先测模型列表探测curl http://127.0.0.1:8321/v1/models再测普通对话curl http://127.0.0.1:8321/v1/chat/completions \ -H Content-Type: application/json \ -d {model:harness-light,messages:[{role:user,content:hi}]}最后测带工具调用的流式请求。这步最容易暴露协议转换问题所以我会用一个包含函数定义和强制工具选择的请求来验证curl http://127.0.0.1:8321/v1/messages \ -H Content-Type: application/json \ -d { model:harness-heavy, max_tokens: 512, stream: true, tools: [{ name: get_weather, description: 获取天气, input_schema: { type: object, properties: {city: {type: string}} } }], tool_choice: {type: tool, name: get_weather}, messages: [{role: user, content: 北京天气怎么样}] }返回里如果能看到完整的工具调用块和参数 JSON说明协议转换链路是通的再进 Harness 就基本稳了。5. 生产实践避坑清单5.1 高频故障速查表我整理了一张表格都是实际跑出来的经验按出现频率排序现象根因解决方案启动即报 401Harness 要求特定格式 Key网关忽略入站鉴权头替换为服务商真实 Key报 ModelNotFound/v1/models 没被接管或模型名不匹配网关实现模型列表接口返回配置里所有别名工具调用返回参数为空input_schema 未正确转成 parameters检查 convert 模块中的工具定义转换逻辑流式对话中途断流上游超时时间过短调大 httpx timeout网关侧发送保活注释行模型不接上下文context_window 配置过小在模型路由表里按真实能力填 context_window某个服务商频繁 429并发过高触发限流配置 max_concurrency启动故障转移响应内容乱码或截断SSE 解析没处理多行 JSON确保按行读取并只处理 data: 前缀的行5.2 离线与局域网部署要点很多团队会在完全离线的环境里跑 Harness。这个场景下国产模型服务可能换成本地部署的模型运行时。好消息是网关完全不受影响——只需要把 provider 的 base_url 指向本机或内网某台服务器上的推理服务地址。如果推理服务提供的是 Chat Completions 兼容接口网关连代码都不用改只改配置。局域网场景要注意的是其他机器要访问网关监听地址就不能写成 127.0.0.1要改成 0.0.0.0并确认防火墙放行对应端口。另外内网部署环境通常没有外网 DNS配置文件里不要写域名直接写内网 IP 最稳。5.3 长期运行的管理经验网关跑了几个月后我总结出三条维护经验。第一配置版本化。路由表就是模型接入的历史记录谁在什么时候把流量从哪个服务商切到了另一个在 Git 历史里都查得到。出问题的时候回滚非常方便。第二Key 轮换要设计成零停机。真实 Key 通过环境变量注入重启进程即可生效。如果 Key 写在 YAML 里轮换时还得担心文件泄露和权限问题不如环境变量干净。第三token 日志要及时归档。日志文件按天切割再写个简单脚本按周汇总。我后来就是靠这份日志发现某个轻量模型实际成本比旗舰模型还高因为它的错误重试次数太多光是重试 token 就把省下的钱吃掉了。写在最后的个人体会就我自己的使用体验而言cn-llm-router 最值钱的部分不是协议转换代码本身而是把模型选择权从工具手里夺回来这件事。过去换模型要改 Harness、求插件、等社区适配现在只是改一行 YAML重启网关任务流自动走新模型。我建议如果你也打算做类似的东西第一版千万别追求功能齐全先跑通最简单的对话和流式再逐步加工具调用、故障转移、统计。等跑顺了之后可以继续扩展的方向我比较看好两个一是把评测逻辑挂到网关里让模型切换前先自动跑一组样例集对比效果二是接入更细粒度的任务级路由比如系统提示词里出现代码关键字就走重模型否则走轻模型。这个思路一旦跑通模型成本还能再降一个台阶。