自建大模型统一网关：多模型路由、故障转移与成本治理的工程实战
当你的应用同时挂着 OpenAI、Anthropic、Gemini、DeepSeek、Qwen 五家 Key 的时候，真正折磨你的不是模型能力，而是「每次换模型都要改一坨 SDK、每家签名都不一样、账单月底爆炸、某个厂商半夜 503 全场雪崩」。LLM Gateway 就是把这一地鸡毛收拢成「一个地址、一把钥匙、一套协议」的中间层。本文从协议碎片化的病根讲起，拆解一个生产级网关必须解决的六类工程问题，并用手写 Go 网关 + Python 语义缓存把路由、熔断、重试、成本记账跑通。
一、背景：我们是怎么被「模型碎片化」拖垮的
2026 年的 AI 应用早就不满足于「调一个 GPT 就完事」。一个稍微像样的产品，后端往往是这样一幅图景：
- 闲聊、分类、抽取这类低价值请求，丢给便宜的小模型（DeepSeek-V3、Qwen、GLM）；
- 复杂推理、代码生成，才上 Claude / GPT 这类贵但强的模型；
- 国内合规流量走国产厂商，海外流量走国际厂商；
- 关键链路还要准备兜底：主力厂商限流或宕机时，能秒级切到备用厂商，不能让用户看到「502 Bad Gateway」。
听起来合理，落地就是一场灾难。根子在协议碎片化：
- OpenAI 用
/chat/completions，请求体是messages； - Anthropic 用
/messages，字段叫system而不是塞进 messages，且要求x-api-key头； - Gemini 是
contents+parts结构，连思维链都要单独字段； - 国内厂商更是各家一套签名算法，有的要
Authorization: Bearer，有的要把时间戳拼进 query 再 HMAC。
结果就是：业务代码里很快长出一堆 if provider == "anthropic" { ... } else if provider == "gemini" { ... }，密钥散落在十几个配置文件，计费逻辑和模型调用耦合在一起，谁也说不清「上个月那 8000 块钱到底烧在了哪些请求上」。
这正是 LLM Gateway 要解决的问题。它不是什么新概念——本质上就是传统 API Gateway 在 LLM 场景的特化：在应用和一堆异构模型之间，插一层统一协议 + 智能路由 + 弹性容错 + 成本治理的中间件。
1.1 这不是 PPT 概念，是 2026 年的真实热点
看几个有代表性的事实，证明这是当下真在发生的工程需求：
- OmniRoute（github.com/diegosouzapw/OmniRoute）：一个本地跑的 AI 网关，聚合 237 个 AI 服务商（其中 90+ 带免费额度），提供 13 种路由策略，并内置 RTK + Caveman 上下文压缩技术，号称能省 15%–95% 的 token。它用 npm / Docker / Desktop / ARM 多种形态分发，说明「开发者想把模型调度收口到本地」是刚需。
- LiteLLM：开源顶流 LLM 网关，统一 100+ 模型提供商的调用为 OpenAI 格式，支持预算控制、限流、负载均衡、成本追踪、虚拟 Key。生产落地案例显示，把多模型接入、智能路由、故障转移、成本精细化管控做扎实后，可把调用成本压低 70%，可用性从 99.5% 拉到 99.99%。
- New API：31.9k Stars 的 One API 增强版，原生支持 15+ 国产服务商，是大量国内团队的实际选择。
- 学术侧也在推进：FusionRoute（ICML 2026，CMU + Meta）提出了「专家路由 + 自我修正」的多 LLM 协作范式，说明路由策略本身正在从「规则硬编码」走向「模型驱动」。
行业数据也佐证了这点：KubeCon 2026 上披露，三分之二的生成式 AI 工作负载已经跑在 Kubernetes 上，K8s 正在从「基础设施」演变成「AI 操作系统」——而网关，就是这个操作系统里的流量调度内核。
所以，理解并亲手撸一个网关，不是玩具练习，是 2026 年后端/平台工程师的硬通货。
二、核心概念：一个生产级网关必须解决的六类问题
不要把网关想成一个「转发代理」就完了。真正生产级的 LLM 网关，要同时回答下面六个问题：
2.1 协议归一（Protocol Normalization）
对外只暴露一套 OpenAI 兼容的 REST 接口（业界事实标准），对内把各家协议翻译成对应厂商的格式。这意味着：业务侧永远只认 POST /v1/chat/completions + messages，换模型只需改一个字段，业务代码一行都不用动。
2.2 智能路由（Intelligent Routing）
不是所有问题都值得上 GPT-4 级别的模型。路由维度至少有四种：
- 成本优先：用历史单价把请求导向最便宜的及格模型；
- 延迟优先：基于每个厂商的 EWMA（指数加权移动平均）延迟，选当前最快的；
- 能力优先：代码生成走 Claude、长文摘要走便宜模型，按任务类型打标签路由；
- 负载均衡：轮询 / 加权轮询，避免把流量全压在一家。
2.3 故障转移与熔断（Failover & Circuit Breaker）
单个厂商 503 / 429 / 超时是常态，不是异常。网关要能做：
- 重试 + 退避：对幂等请求做有限次指数退避重试；
- 跨厂商 Fallback：主力挂了，自动切到候选厂商（注意流式场景要更谨慎）；
- 熔断：某厂商连续失败达到阈值，直接「开路」30 秒，不再浪费请求，进入半开探测。
2.4 成本治理（Cost Governance）
这是网关最大的隐性价值。要能：
- 在网关层做 token 计量，而不是事后看厂商账单；
- 按
虚拟 Key（Virtual Key）/ 团队 / 业务线做预算和分摊； - 超预算直接
402 Payment Required拒掉，防止「半夜账单刺客」。
2.5 可观测（Observability）
每一次调用都要留下：走的是哪家模型、耗时多少、花了多少 token、命中没命中缓存、失败原因是什么。没有这些，排障和成本优化就是盲人摸象。
2.6 安全（Security）
- 真实厂商 Key 只存在网关内存/密钥管理里，绝不出网；
- 上游应用拿的是网关发的虚拟 Key；
- 可选做 PII 脱敏、审计日志，满足合规。
这六条，就是网关的「功能边界」。下面我们用代码把它落出来。
三、架构分析：网关的分层设计
一个清晰的分层架构长这样（从上到下）：
┌─────────────────────────────────────────────┐
│ 接入层 (Edge) │
│ OpenAI 兼容 REST / SSE 流式 / 虚拟 Key 鉴权 │
├─────────────────────────────────────────────┤
│ 路由层 (Router) │
│ 成本 / 延迟 / 能力 / 轮询 策略 + 有序候选 │
├─────────────────────────────────────────────┤
│ 弹性层 (Resilience) │
│ 重试 · 退避 · 跨厂商 Fallback · 熔断 │
├─────────────────────────────────────────────┤
│ 适配器层 (Adapters) │
│ OpenAI / Anthropic / Gemini / 本地 vLLM │
├─────────────────────────────────────────────┤
│ 治理层 (Governance) │
│ 预算 · 配额 · 成本记账 · 分摊 │
├─────────────────────────────────────────────┤
│ 缓存层 (Cache) │
│ 精确缓存 · 语义缓存 · Prompt Cache │
├─────────────────────────────────────────────┤
│ 可观测层 (Observability) │
│ 日志 · Metrics · Trace · 评测回放 │
└─────────────────────────────────────────────┘
设计哲学：接入层薄、适配器薄、路由和治理厚。业务方只跟接入层打交道；厂商差异被锁死在适配器层；真正体现技术含量的，是路由策略怎么选、熔断阈值怎么定、成本怎么归因。
下面我用 Go 手写一个最小但真实可用的网关核心，覆盖协议归一、路由、熔断、Fallback、成本记账。生产环境你大概率会站在 LiteLLM / OmniRoute 的肩膀上，但理解这套内核，你才知道它们到底在替你做什么。
四、代码实战：从零撸一个 Go 网关内核
4.1 领域模型与 Provider 适配器
先定义统一请求/响应，以及把异构厂商归一化的 Provider 接口：
package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"time"
)
// 统一请求：业务方永远只发这一种结构
type Message struct {
Role string `json:"role"`
Content string `json:"content"`
}
type ChatRequest struct {
Model string `json:"model"`
Messages []Message `json:"messages"`
Stream bool `json:"stream"`
}
type ChatResponse struct {
ID string `json:"id"`
Model string `json:"model"` // 回填上真实提供方，便于归因
Choices []struct {
Message Message `json:"message"`
} `json:"choices"`
Usage struct {
PromptTokens int `json:"prompt_tokens"`
CompletionTokens int `json:"completion_tokens"`
TotalTokens int `json:"total_tokens"`
} `json:"usage"`
}
// Provider：把不同厂商协议归一为统一接口
type Provider interface {
Name() string
Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error)
CostPer1K(prompt, completion int) float64 // 每千 token 成本（美元）
}
OpenAIProvider 是一个具体的适配器，Anthropic / Gemini 只需在实现里把 ChatRequest 翻译成自家格式即可，对上完全透明：
type OpenAIProvider struct {
name string
baseURL string
apiKey string
costIn float64 // 每千 input token 单价
costOut float64 // 每千 output token 单价
client *http.Client
}
func (p *OpenAIProvider) Name() string { return p.name }
func (p *OpenAIProvider) CostPer1K(prompt, completion int) float64 {
return float64(prompt)/1000*p.costIn + float64(completion)/1000*p.costOut
}
func (p *OpenAIProvider) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) {
body, _ := json.Marshal(req)
httpReq, _ := http.NewRequestWithContext(ctx, http.MethodPost,
p.baseURL+"/chat/completions", bytes.NewReader(body))
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("Authorization", "Bearer "+p.apiKey)
start := time.Now()
resp, err := p.client.Do(httpReq)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("provider %s status %d: %s", p.name, resp.StatusCode, b)
}
var out ChatResponse
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
return nil, err
}
out.Model = p.name // 关键：标注真实来源，方便成本归因
_ = start
return &out, nil
}
4.2 熔断器：别让一个挂掉的厂商拖死全场
type Breaker struct {
mu sync.Mutex
failures int
threshold int
openUntil time.Time
}
func NewBreaker(threshold int) *Breaker {
return &Breaker{threshold: threshold}
}
func (b *Breaker) Allow() bool {
b.mu.Lock()
defer b.mu.Unlock()
return time.Now().After(b.openUntil) // 还在熔断期就拒绝
}
func (b *Breaker) Success() {
b.mu.Lock()
defer b.mu.Unlock()
b.failures = 0
b.openUntil = time.Time{}
}
func (b *Breaker) Fail() {
b.mu.Lock()
defer b.mu.Unlock()
b.failures++
if b.failures >= b.threshold {
b.openUntil = time.Now().Add(30 * time.Second) // 熔断 30s 后进入半开探测
}
}
4.3 路由器：把候选厂商排个序
type Strategy int
const (
StrategyCost Strategy = iota
StrategyLatency
StrategyRoundRobin
)
type Router struct {
strategy Strategy
rr uint64
latency map[string]float64 // 各厂商 EWMA 延迟（秒）
mu sync.RWMutex
}
func (r *Router) Observe(name string, d time.Duration) {
r.mu.Lock()
defer r.mu.Unlock()
v := d.Seconds()
if old, ok := r.latency[name]; ok {
r.latency[name] = old*0.7 + v*0.3 // EWMA 平滑
} else {
r.latency[name] = v
}
}
func (r *Router) Order(providers []Provider) []int {
n := len(providers)
idx := make([]int, n)
for i := range idx {
idx[i] = i
}
switch r.strategy {
case StrategyCost:
sort.SliceStable(idx, func(a, b int) bool {
// 用 1K prompt token 粗估单价排序
return providers[idx[a]].CostPer1K(1000, 0) <
providers[idx[b]].CostPer1K(1000, 0)
})
case StrategyLatency:
r.mu.RLock()
defer r.mu.RUnlock()
sort.SliceStable(idx, func(a, b int) bool {
return r.latency[providers[idx[a]].Name()] <
r.latency[providers[idx[b]].Name()]
})
case StrategyRoundRobin:
offset := int(atomic.AddUint64(&r.rr, 1)-1) % n
idx = append(idx[offset:], idx[:offset]...)
}
return idx
}
4.4 网关编排：路由 + 熔断 + Fallback + 成本记账
type Budget struct {
mu sync.Mutex
spent float64
limit float64
byModel map[string]float64
}
func (b *Budget) Record(name string, cost float64) {
b.mu.Lock()
defer b.mu.Unlock()
b.spent += cost
b.byModel[name] += cost
}
func (b *Budget) Allowed() bool {
b.mu.Lock()
defer b.mu.Unlock()
return b.spent < b.limit
}
type Gateway struct {
providers []Provider
breakers map[string]*Breaker
router *Router
budget *Budget
log *log.Logger
}
func (g *Gateway) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) {
order := g.router.Order(g.providers)
var lastErr error
for _, i := range order {
p := g.providers[i]
if br := g.breakers[p.Name()]; br != nil && !br.Allow() {
g.log.Printf("skip %s: breaker open", p.Name())
continue
}
start := time.Now()
resp, err := p.Chat(ctx, req)
if err != nil {
lastErr = err
if br := g.breakers[p.Name()]; br != nil {
br.Fail()
}
g.log.Printf("provider %s failed: %v; try next", p.Name(), err)
continue
}
if br := g.breakers[p.Name()]; br != nil {
br.Success()
}
// 成本记账 + 延迟观测
cost := p.CostPer1K(resp.Usage.PromptTokens, resp.Usage.CompletionTokens)
g.budget.Record(p.Name(), cost)
g.router.Observe(p.Name(), time.Since(start))
g.log.Printf("served by %s, tokens=%d, cost=$%.4f",
p.Name(), resp.Usage.TotalTokens, cost)
return resp, nil
}
return nil, fmt.Errorf("all providers failed, last err: %v", lastErr)
}
4.5 接入层：OpenAI 兼容 HTTP 入口
func (g *Gateway) Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/v1/chat/completions", func(w http.ResponseWriter, r *http.Request) {
if !g.budget.Allowed() {
http.Error(w, "budget exceeded", http.StatusPaymentRequired)
return
}
var req ChatRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
defer cancel()
resp, err := g.Chat(ctx, req)
if err != nil {
http.Error(w, err.Error(), http.StatusBadGateway)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
})
return mux
}
业务方从此只需：
curl http://gateway:8080/v1/chat/completions \
-H 'Authorization: Bearer <虚拟Key>' \
-d '{"model":"auto","messages":[{"role":"user","content":"你好"}]}'
model 填 auto，网关按策略选厂商；要指定厂商就填 claude-3-7 / deepseek-v3。业务代码彻底和厂商解耦。
工程提示：上面是「非流式」骨架。生产环境务必支持 SSE 流式转发（用
http.Flusher把上游 chunk 直接 pipe 给客户端），否则首 token 延迟会劝退用户。流式下 Fallback 要更谨慎——一旦开始吐字就不能换厂商，所以流式请求通常在「拿到第一个 chunk 之前」完成厂商选择。
五、性能优化：让网关又快又省的四把刀
网关自己要是成了瓶颈，前面全白搭。四个最值得投入的优化点：
5.1 流式转发 + 连接池
- 网关到厂商用长连接（
http.Transport配MaxIdleConns、IdleConnTimeout），避免每次握手； - 对外流式走 SSE，首 token 延迟从「等完整响应」降到「等第一个 chunk」；
- 给厂商 client 设合理的
Timeout，但业务侧用context单独控制总超时，两者不要打架。
5.2 语义缓存：重复问题零成本命中
大量真实流量是高度重复的（FAQ、内部知识问答、评测 prompt）。精确缓存（同样 hash 命中）覆盖率低，真正香的是语义缓存：把问题 embedding 化，余弦相似度超过阈值就直接返回历史答案。
import numpy as np
from openai import OpenAI
client = OpenAI()
_cache = [] # [(embedding: np.ndarray, answer: str)]
def _embed(text: str) -> np.ndarray:
r = client.embeddings.create(model="text-embedding-3-small", input=text)
return np.array(r.data[0].embedding, dtype=float)
def _cosine(a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))
def ask(prompt: str, threshold: float = 0.92):
q = _embed(prompt)
for emb, ans in _cache:
if _cosine(q, emb) >= threshold: # 语义近似即命中
return ans # 零推理成本
ans = call_llm(prompt) # 真去调模型
_cache.append((q, ans))
return ans
阈值 0.92 是经验起点，太低会答非所问，太高退化成精确缓存。生产用向量库（pgvector / Milvus）替代内存列表，否则缓存本身会成瓶颈。语义缓存通常能吃掉 30%–60% 的重复流量。
5.3 上下文压缩：把 token 账单砍下来
OmniRoute 敢说省 15%–95%，靠的就是上下文压缩。几类手段：
- Prompt Cache：Claude / Gemini / 部分 OpenAI 支持「前缀缓存」，稳定 system prompt 只计费一次；
- 滑动窗口 / 自动摘要：超长对话历史，旧轮次先做摘要再拼回；
- 结构化裁剪：RAG 场景只保留 top-k 片段，别把整本手册塞进 context。
5.4 地域就近 + 批处理
- 海外模型走离用户近的接入点，国内模型走国内节点，别让一次调用绕地球半圈；
- 可批量、非实时的任务（如批量抽取）做请求合并 / 异步队列，用便宜模型的闲时配额跑，进一步摊薄成本。
六、总结与展望：网关正在变成 AI 时代的「流量内核」
回到开头那个被五把 Key 折磨的场景。引入网关后，它的价值远不止「少写几行 SDK」：
- 业务解耦：模型是「能力池」里的资源，换厂商像换数据库连接池一样无感；
- 成本可控：网关层实时计量 + 预算拦截，月底不再有账单刺客；
- 韧性提升：熔断 + Fallback 把「单点 503」从「全场雪崩」降级为「轻微抖动」；
- 治理闭环：虚拟 Key、分摊、审计，让 AI 资源像云资源一样可被管理。
展望一下趋势，这个领域还在快速演化：
- 路由策略模型化：FusionRoute 这类工作已经证明，「该用哪个模型」本身可以用一个小模型来决策，并结合自我修正做多 LLM 协作——规则硬编码的路由器会逐步被「路由模型」取代。
- Agent 原生网关：随着 AI Agent 成为主流形态，网关要原生理解 tool-call、多轮规划、记忆存取，而不只是转发一次 chat。
- 混合部署常态化：敏感数据走本地 vLLM / Ollama，通用能力走云端，网关负责在这之间无缝调度，数据合规和成本兼得。
- 成为 AI 操作系统内核：当 2/3 的生成式 AI 负载跑在 K8s 上，网关就是这层「操作系统」里的调度器——它管的不只是流量，是算力、成本、合规与韧性的总和。
所以，无论你是后端、平台还是 AI 工程师，亲手理解并搭建一个 LLM 网关，都是 2026 年最值得投入的「底层能力」之一。站在 LiteLLM、OmniRoute 这些成熟项目肩上能让你快速上线，但只有自己撸过一遍路由、熔断、成本记账的内核，你才真正握住了 AI 基础设施的方向盘。
本文代码示例为最小可用骨架，生产环境请结合 LiteLLM / OmniRoute 等成熟方案，并补全流式 SSE、虚拟 Key 鉴权、向量化语义缓存与全链路 Trace。