接到需求时我愣了一下:一套接口,又要走本地模型,又要走云端API,还要能随时切换。第一反应是这需求是不是有点折腾,但仔细一想,这个场景现在越来越常见——敏感数据想在本地处理,复杂推理想交给云端大模型,成本还要控在预算内。真正落地时最大的坎不是模型效果,而是接口层怎么设计。这篇文章我就把用一套OpenAI兼容协议同时调度Ollama本地模型和多家云端API的完整过程写出来,包括配置、路由、降级、代码实现和踩坑记录,按顺序照做就能跑通。
本文适合正在做AI应用落地、被本地和云端模型切换折磨的开发者。不管你用的是FastAPI还是Node.js,核心思路都通用:用一套标准协议屏蔽底层差异,用模型别名把“业务要什么”和“背后用哪个模型”完全解耦,再用一层轻量网关把调度、降级、流式透传都收住。
1. 本地与云端混用的真实痛点:为什么业务代码会被模型拖死
1.1 一次让我彻底崩溃的业务改造
上个月我在做一个内部知识库问答系统,业务逻辑本身不复杂:用户提问后,系统先做意图识别和关键词抽取,再决定走哪个检索分支,最后把检索结果拼进prompt让大模型生成答案。
一开始所有请求都走云端API,代码写得很痛快,一个SDK搞定。结果上线两周后问题来了:意图识别和标题抽取这种高频轻量任务,按token计费的成本非常高。而且有些内部文档不能出内网,必须用本地模型处理。于是产品经理提了个需求:同一套业务代码,有些模型调用走本地,有些走云端,还要能随时切。
我原以为只是把云端API的base_url换成Ollama的地址就行,真正动手才发现完全不是这么回事。业务代码里已经写死了云端SDK的请求格式、鉴权方式、异常处理逻辑,换成本地模型后接口路径不一样、认证方式不一样、返回结构也有细微差异。更要命的是模型名不一样,云端叫deepseek-chat,本地叫deepseek-r1:7b,业务侧如果写死模型名,以后换模型又得改代码。
1.2 本地模型与云端API的差异,远不止“延迟”和“价格”
要把本地和云端统一调度起来,首先得搞清楚它们到底差在哪。我整理了一张对比表,这是我做技术选型和路由设计的基础:
| 维度 | 本地模型 | 云端API |
|---|---|---|
| 延迟 | 取决于GPU,通常百毫秒到秒级,无网络波动 | 受网络和厂商排队影响,通常秒级 |
| 成本 | 硬件一次性投入,边际成本极低,基本接近0 | 按token计费,高频调用成本增长很快 |
| 数据安全 | 数据不出内网,可处理敏感数据 | 数据要发给第三方,有合规和隐私风险 |
| 模型能力 | 参数量小(7B/13B居多),复杂推理弱 | 参数量大(100B+),复杂推理和长上下文能力强 |
| 可用性 | 依赖本机资源,OOM、断电、显存不足都可能挂 | 依赖厂商SLA,可能限流、故障,但整体可用性高 |
| 上下文长度 | 通常4k/8k/16k,长文档受限 | 动辄32k/128k/200k,长文本处理能力强 |
这表说明一件事:本地和云端不是替代关系,是互补关系。轻量高频任务适合放本地,省钱又保隐私;复杂推理、长文档总结适合放云端,效果好但需要控制调用量。既然要互补,调度层就必须能够根据路由规则自动决定请求去向,而不是靠开发者在代码里写死。
1.3 不用统一接口,后面会踩哪些坑
如果只是临时用一下,直接在业务代码里写两套调用逻辑也不是不行,但长期下来会很痛苦:
- 业务代码里混着两套SDK、两套异常处理、两套超时配置,维护成本翻倍;
- 换个模型要改代码重新发布,测试回归流程走一遍,半天就过去了;
- 无法自动降级。本地模型OOM了,云端也不一定能接得住;云端限流了,本地也没办法兜底;
- 监控和成本统计是两本账,想搞清楚“哪个环节花了多少钱”非常麻烦。
所以我的结论很明确:必须在业务代码和具体模型之间插一层调度代理,让上游只认识一套接口、一个模型名。下面先讲协议选型,这是整个方案的地基。
2. 统一调度的破局点:让所有模型先讲同一种“语言”
2.1 为什么我选中OpenAI兼容协议,而不是各家私有协议
选接口协议是个战略决策,选错了后面适配工作量非常大。我的选择是OpenAI兼容协议,原因有四个:
- 事实标准:无论是Ollama、vLLM、LM Studio这些本地推理工具,还是DeepSeek、阿里云百炼、智谱这些国内云厂商,几乎都原生提供OpenAI兼容端点。也就是说,大家都愿意向这个协议靠拢。
- 生态最全:LangChain、LlamaIndex、Dify、FastGPT等主流框架默认就能对接OpenAI兼容接口,基座替换成本极低。
- SDK成熟:OpenAI官方Python/Node SDK支持度最好,改个base_url和api_key就能切换目标,不需要额外引包。
- 排错资料多:遇到问题搜索时,OpenAI兼容协议的报错信息在网上有大量讨论,排查效率高。
你可以把OpenAI兼容协议理解为AI接口界的USB-C。过去各家充电口都不一样,出门得带一堆线,现在接口统一了,只要设备支持USB-C,一根线就能通用。这里也一样,只要模型服务提供商愿意暴露一个OpenAI兼容端点,上层就能用统一方式调用。
2.2 OpenAI兼容协议的核心构成:端点、请求体、鉴权
要自己写网关,核心协议的四个要素必须吃透:
- 端点路径:通常是
/v1/chat/completions,聊天补全主入口;还有/v1/models用于列出可用模型。 - 请求体:主要字段包括
model(模型名)、messages(对话消息数组)、temperature(采样温度)、max_tokens(生成上限)、stream(是否流式返回)。 - 返回体:非流式时是完整JSON,里面有
choices[0].message.content(生成内容)和usage(token用量);流式时是SSE格式的增量事件流。 - 鉴权方式:HTTP Header里加
Authorization: Bearer <api_key>。本地Ollama不校验key,但格式上必须带上这个字段,方便统一处理。
这里有个细节很关键:请求体里的model字段,是整个调度路由的入口。网关截获请求后,第一件事就是读这个字段,然后根据别名映射决定把请求转发到哪里。所以这个字段的值不能随便用,应该在团队内统一约定。
一个标准Chat Completion请求长这样:
{ "model": "balanced", "messages": [ {"role": "system", "content": "你是一个知识库问答助手"}, {"role": "user", "content": "请总结这份合同的关键条款"} ], "temperature": 0.3, "max_tokens": 1024, "stream": false }业务代码只认这个格式,至于model是本地还是云端、实际背后的模型叫什么,业务完全不关心。这就是隔离的威力。
3. 本地侧接入:让Ollama和vLLM暴露一个标准端点
3.1 Ollama接入:最适合单机起步的方案
本地模型部署,我最推荐先用Ollama,因为对新手最友好,命令少、坑也少。以deepseek-r1:7b为例,完整接入步骤是这样的:
第一步,安装Ollama(macOS和Linux都支持,Windows也有安装包),然后设置两个环境变量:
# 允许局域网内其他机器访问Ollama export OLLAMA_HOST=0.0.0.0:11434 # 指定模型存储目录,建议放数据盘 export OLLAMA_MODELS=/data/ollama/models第二步,拉取模型并启动服务:
ollama pull deepseek-r1:7b ollama serveollama serve启动后,Ollama会同时监听/api和/v1两套端点。其中/v1就是OpenAI兼容端点,支持/v1/models和/v1/chat/completions。
第三步,用curl验证OpenAI兼容端点是否正常:
curl http://127.0.0.1:11434/v1/models curl http://127.0.0.1:11434/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "deepseek-r1:7b", "messages": [{"role": "user", "content": "你好,简单介绍一下自己"}], "stream": false }'能正常返回模型回复,就说明这个端点已经可以作为网关的后端provider了。
这里有个我踩过的坑提醒一下:Ollama默认的context length(上下文长度)通常比较保守,实际任务里如果prompt很长,容易出现“生成内容截断”甚至直接报错。解决办法是创建一个自定义Modelfile,显式设置num_ctx。比如:
cat <<EOF > Modelfile FROM deepseek-r1:7b PARAMETER num_ctx 16384 EOF ollama create deepseek-r1-16k -f ./Modelfile后续请求里直接用deepseek-r1-16k这个模型名就行。这一步不做,长文档类任务基本跑不通。
3.2 vLLM接入:更适合GPU服务器的选择
Ollama的最大问题是并发吞吐一般,如果团队有多人同时用,或者QPS要求较高,我更推荐上vLLM。vLLM的OpenAI兼容服务启动命令也很简单:
python -m vllm.entrypoints.openai.api_server \ --model /data/models/deepseek-r1-7b \ --served-model-name local-llm \ --port 8000 \ --tensor-parallel-size 1 \ --max-model-len 16384这里有两个参数值得多花一句话解释:
--served-model-name:给模型起一个对外暴露的名字。这非常有用,因为在网关的配置里,provider内部的模型名和对外暴露的模型名可以完全分离,以后换真实模型文件时,对外名字可以保持不变。--max-model-len:控制上下文长度,按实际显存和任务需求调整,不要超过显卡能承受的极限。
vLLM的优势在学术界和工业界已经有大量验证:PagedAttention显存管理、continuous batching请求调度,吞吐量比朴素的推理脚本高一个数量级。如果你的模型文件是AWQ/GPTQ量化格式,vLLM也支持加载。
3.3 本地端点验证清单:上线前必须确认的几件事
本地服务挂起来不等于能用,我建议在上配置之前先按这个清单过一遍:
- 非流式请求返回正常,响应时间是预期量级;
- 流式请求正常,SSE事件持续输出没有中断;
- 连续并发2-3个请求,确认不会OOM、不会互相阻塞;
- 模型是否支持function calling/tools调用?Ollama和vLLM对tools的支持程度跟模型本身强相关,7B小模型经常表现不稳定,这个直接影响路由设计,后面细说;
- 上下文长度是否符合业务需求,超出后是截断还是报错,要做到心里有数。
这些验证做完,本地侧就绪,接下来处理云端API侧。
4. 云端API适配层:用模型别名把多家厂商收进一个名字空间
4.1 各家云端API的OpenAI兼容情况
好消息是,现在国内外的云厂商基本都提供了OpenAI兼容端点,这大大减少了适配工作量。我整理了一份主流厂商的接入参数:
| 厂商 | OpenAI兼容端点 | 典型模型名 | 备注 |
|---|---|---|---|
| OpenAI | https://api.openai.com/v1 | gpt-4o、gpt-4o-mini | 原生接口 |
| DeepSeek | https://api.deepseek.com/v1 | deepseek-chat、deepseek-reasoner | 完整兼容 |
| 阿里云百炼 | https://dashscope.aliyuncs.com/compatible-mode/v1 | qwen-plus、qwen-max | 需要开启兼容模式 |
| 智谱 | https://open.bigmodel.cn/api/paas/v4 | glm-4-plus、glm-4-flash | v4版本基本兼容 |
你会发现,只要厂商提供OpenAI兼容端点,适配层根本不需要做协议转换,真正要做的只有三件事:统一鉴权Header格式、统一请求体格式、统一模型命名空间。前两个基本上白拿,重点在模型命名空间上。
4.2 模型别名的设计:业务名字和真实模型彻底解耦
我强烈建议不要直接在业务代码里写具体的模型名,比如deepseek-chat、gpt-4o这类。原因很简单:模型名是厂商的产品名,是会变的;业务语义才是稳定的。今天你用qwen-max,明天想换deepseek-reasoner,如果业务代码里写死了模型名,又得改代码。
我的做法是引入一层“模型别名”,按业务用途和SLA分级:
fast:高频轻量任务,默认路由到本地7B模型,省钱;balanced:通用推理任务,默认路由到云端大模型,效果好;ultra:复杂推理、长文档任务,路由到最强云端模型,能不用就不用。
业务代码里永远写model: "fast"或model: "balanced",至于fast背后是deepseek-r1:7b还是以后换成的qwen3-8b,那是网关配置文件里的一个字段而已。
用配置驱动而不是代码驱动,这是整个方案里最值得坚持的设计原则。配置文件长这样:
# 模型路由配置 providers: local-ollama: type: local base_url: http://127.0.0.1:11434/v1 api_key: ollama model: deepseek-r1-16k max_context: 16384 health_check_path: /v1/models cloud-deepseek: type: cloud base_url: https://api.deepseek.com/v1 api_key: ${DEEPSEEK_API_KEY} model: deepseek-chat max_context: 65536 cloud-openai-gpt4o: type: cloud base_url: https://api.openai.com/v1 api_key: ${OPENAI_API_KEY} model: gpt-4o max_context: 128000 routes: fast: primary: local-ollama fallback: cloud-deepseek balanced: primary: cloud-deepseek fallback: cloud-openai-gpt4o ultra: primary: cloud-openai-gpt4o注意到几个设计细节:
- API Key从环境变量读取(
${DEEPSEEK_API_KEY}),不要硬编码进配置文件,避免误提交到Git仓库; - 每个provider维护自己的
max_context,网关在做长文本路由时要有能力根据这个值判断当前请求是否超限; routes里的primary和fallback是实现自动降级的关键,这个下面细讲。
4.3 兼容层要不要写代码?什么时候需要自定义适配器
大部分情况,配置就够用了。但有一个场景需要写代码适配:厂商的OpenAI兼容端点做得不完整,或者参数语义有差异。比如某些厂商不支持frequency_penalty,或者对max_tokens有特殊限制。如果网关直接把请求体透传过去,可能会收到4xx错误。
解决方法是:在网关里做一次“参数清洗”,根据provider配置动态调整请求体字段。举例来说,如果provider配置里声明supports_temperature: false,网关转发前就把temperature删掉;如果max_tokens上限是4096,而请求里传了8192,就得在网关层截断或告警。
这些规则可以做成独立适配器模块,每个厂商一个类,但绝大多数情况下不推荐一开始就写。先把最小闭环跑通,遇到真实报错再补适配器,才是务实的做法。
5. 调度网关的完整实现:从转发到降级一次说清
5.1 网关选型:自研轻量服务还是用开源项目
本地和云端的provider都就绪后,核心问题来了:这一层调度网关怎么做?
市面上已经有一些优秀的开源方案,比如one-api、new-api、LiteLLM Gateway。它们开箱即用,自带配额管理、key管理、多渠道聚合,适合团队内部快速搭一个模型代理平台。我自己也用过,确实省了不少时间。
但如果你和我一样,需要精确控制路由策略、要给网关嵌入内部监控体系、要做非常细的上下文截断或成本控制,自研一个轻量网关其实代码量并不大。下面我用FastAPI写一个最小可用版本,只做四件事:接收OpenAI兼容请求、解析模型别名、转发到真实provider、失败时自动降级。
5.2 网关主逻辑:配置加载与路由分发
先定义配置结构。我还是用YAML,方便非开发同学直接改,不用碰代码。初始化时把配置文件读进来,构建一个routes字典,key是模型别名,value是对应的路由规则。
import os import yaml def load_config(path="gateway.yaml"): with open(path, "r", encoding="utf-8") as f: config = yaml.safe_load(f) providers = config.get("providers", {}) # 将环境变量形式的 api_key 替换为真实值 for name, p in providers.items(): if isinstance(p.get("api_key"), str) and p["api_key"].startswith("${"): env_name = p["api_key"].strip("${} ") p["api_key"] = os.getenv(env_name, "") return config然后写路由函数,这一步是整个网关最核心的地方:根据model字段查路由表,找到primary和fallback,然后决定请求往哪儿发。如果模型别名不存在,直接返回4xx,避免把错误请求透传到后端。
def resolve_route(model_alias: str): config = load_config() route = config["routes"].get(model_alias) if not route: return None providers = config["providers"] primary = providers.get(route["primary"]) fallback = providers.get(route.get("fallback", "")) if route.get("fallback") else None return primary, fallback5.3 请求转发:非流式与流式透传都怎么实现
网关接收到请求后,要把OpenAI兼容请求原样转发到目标provider,只是把请求里的model字段替换成provider的真实模型名。这个替换非常关键,否则provider会报“model not found”。
非流式转发实现如下:
import httpx from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse, StreamingResponse app = FastAPI() async def forward_non_stream(provider, body): target_url = provider["base_url"].rstrip("/") + "/chat/completions" body = dict(body) body["model"] = provider["model"] # 替换为 provider 真实模型名 headers = { "Authorization": f"Bearer {provider['api_key']}", "Content-Type": "application/json", } async with httpx.AsyncClient(timeout=300) as client: resp = await client.post(target_url, json=body, headers=headers) return JSONResponse(status_code=resp.status_code, content=resp.json())流式转发稍有不同,要启用SSE流式透传,把provider返回的增量内容一段一段喂给上游调用方。用一个异步生成器处理:
async def forward_stream(provider, body): target_url = provider["base_url"].rstrip("/") + "/chat/completions" body = dict(body) body["model"] = provider["model"] body["stream"] = True headers = { "Authorization": f"Bearer {provider['api_key']}", "Content-Type": "application/json", } async with httpx.AsyncClient(timeout=600) as client: async with client.stream("POST", target_url, json=body, headers=headers) as resp: if resp.status_code >= 400: error_text = await resp.aread() yield json.dumps({"error": {"message": error_text.decode("utf-8")}}) return async for line in resp.aiter_lines(): if line.strip(): yield line + "\n"注意到流式这里有个细节:即使后端出错了,在流式场景下也还是要把错误信息以SSE格式返回给上游,因为直接抛异常会导致客户端挂起。这是我在实际联调时踩过的坑。
主入口函数把路由解析和非流式/流式转发串起来,同时负责异常捕获和降级逻辑:
@app.post("/v1/chat/completions") async def chat_completions(request: Request): body = await request.json() model_alias = body.get("model", "") resolved = resolve_route(model_alias) if not resolved: raise HTTPException(status_code=404, detail=f"unknown model alias: {model_alias}") primary, fallback = resolved try: if body.get("stream"): return StreamingResponse(forward_stream(primary, body), media_type="text/event-stream") return await forward_non_stream(primary, body) except Exception as e: if fallback: print(f"primary provider failed, fallback to {fallback['base_url']}: {e}") if body.get("stream"): return StreamingResponse(forward_stream(fallback, body), media_type="text/event-stream") return await forward_non_stream(fallback, body) raise HTTPException(status_code=502, detail=f"provider error: {str(e)}")这段逻辑短小但是完整。你的业务代码只需指向这个网关的地址(比如http://gateway-host:8000/v1),传什么模型名都行,网关会照顾后面的一切。
5.4 降级与重试策略:不要让一个失败拖垮全局
降级这里有个很重要的分寸问题:不是所有失败都适合自动切到云端。
我的原则是:
- 超时/连接失败(provider完全不可用):可以自动降级到fallback;
- 4xx错误(请求本身有问题):不要降级,直接把错误返回给调用方,否则会把错误请求发到另一个provider,浪费一次调用;
- 本地OOM/显存不足:可以降级到云端,但要在日志里打上标记,方便后续给本地服务加容量或重启;
- 模型内容安全风险或输出异常:不要自动重试,需要人工介入。
另外必须给网关加一个全局并发限制。本地模型的并发能力远弱于云端,特别是Ollama在GPU上跑一个7B模型时,同时进来三四个请求就可能排队或OOM。我在网关里用一个asyncio.Semaphore控制每个provider的最大并发数,超出时直接返回429或排队等待:
import asyncio # 每个 provider 一个信号量,按配置里 concurrency 字段初始化 semaphores = {} def get_semaphore(provider_name): if provider_name not in semaphores: semaphores[provider_name] = asyncio.Semaphore(2) # 从配置读取 return semaphores[provider_name]这一步不做,本地模型服务很容易在高并发下被打挂,而且Ollama进程本身不一定能自动恢复,还得手动重启,非常麻烦。
6. 从跑通到跑稳:实测数据、典型坑和几条建议
6.1 我实测的一组参考数据
我在自己的环境中做了一组简单对比测试。测试任务是用模型做一段合同文本的摘要,prompt约1500个中文字符。机器配置是一张RTX 4090 24GB,本地模型是deepseek-r1:7b(int4量化版本),云端模型是deepseek-chat。
| 指标 | 本地7B(RTX 4090) | 云端deepseek-chat |
|---|---|---|
| 单次请求延迟 | 约2-4秒 | 约1.5-3秒 |
| 单次token成本 | 约0元(电费和硬件折旧不算) | 约0.003-0.005元 |
| 输出质量 | 一般,偶有事实性错误 | 较好,逻辑更连贯 |
| 处理敏感数据 | 可以 | 原则上不可以 |
| 长文本(>8k) | 吃力,需要截断 | 轻松 |
实测结论和我的设计预期一致:轻量任务用本地模型,不仅省钱,延迟也不吃亏;但涉及复杂推理和长文档,本地7B还是扛不住。所以路由规则的配置很关键,不能一刀切。
6.2 几个一定要提前知道的坑
这个方案跑通不难,跑稳需要跨过几道坎,我逐个说一下。
第一,本地模型的function calling支持程度差异很大。Ollama和vLLM都支持tools参数,但模型本身如果不擅长工具调用,经常会出现“该调的没调,不该调的瞎调”。7B小模型尤其明显。我的建议是:需要function calling的任务,尽量不要路由到本地7B,至少等模型能力提升或换用更大参数量的本地模型。
第二,上下文长度不一致会导致部分请求失败。本地模型max_context只有16k,云端模型可以到128k,如果用户一次性上传超长文档并走了本地模型,就会报错。我建议网关在路由时加一个判断:如果请求中的预估token数超过了provider的max_context,自动改路由到云端更高配模型,而不是让请求失败。
第三,本地模型冷启动延迟非常高。Ollama第一次请求一个模型时,需要把权重从磁盘加载到显存,这个过程可能要等几十秒。网关的超时时间如果设成10秒,会直接判定请求失败。解决方法是:启动时主动预热一次,或者把超时设置给足,并且把这条路径纳入降级策略。
第四,流式响应在本地模型上更容易中断。Ollama的流式输出如果推理中途出错,连接会直接断开,但SSE协议本身没有结束时标记,客户端会一直转圈。处理办法是在网关层监控流是否持续产生数据,如果一段时间没有数据就主动掐断连接并返回错误。
第五,成本控制不能只靠路由,还得靠“不重试”的纪律。上面说过,不是所有失败都适合降级。如果网关把所有异常都自动重试到云端,那成本会快速失控。我在生产环境的策略是:本地失败降级到云端,云端失败直接返回错误,让调用方决定是否重试。
6.3 网关上线后,我还做了两件增值的事
网关跑通之后,我顺手做了两件事,收益很大。
第一是加了详细的请求日志,记录每次调用的模型别名、实际provider、延迟、耗时、token消耗、是否发生降级。有了这些数据,我就能清楚地看到“fast路由每天处理了多少请求,多少次走了fallback云端”。这既是对成本的审计,也是调整模型的依据。
第二是加了一个健康检查接口,持续探测所有provider的可用性。本地Ollama如果挂了,健康检查能第一时间发现并在监控里告警;同时网关还可以通过配置把路由权重临时切换到云端,保证业务连续性。
这两件事做出来后,这套统一调度体系才算完整:不只是“能把请求发出去”,而是“知道请求走到哪儿了、为什么失败、成本花在哪了”。
如果只让我给一条最核心的建议,那就是先把模型别名机制定好,不要急着写代码。别名是整条链路的共同语言:产品经理说“轻量任务”,开发写model: fast,网关配置定义fast的走向。这个解耦一旦做好,后续不管加新模型、切厂商,还是做成本控制、故障降级,都是在配置层面就能解决的事情,不用再动业务代码。这也是本地模型和云端API混用这件事,真正值得投入的地方。