OmniRoute 深度拆解：一个端点接住 268 家模型提供商，AI 网关的路由决策链、多层回退与上下文压缩实战

一、为什么 2026 年每个 AI 团队都需要一层网关

如果你今年在做 AI 应用，大概率经历过这样的深夜：OpenAI 突然 529 过载，Claude 的 API 又开始 overloaded_error，你手忙脚乱地改环境变量切到 DeepSeek，结果发现请求体格式对不上、tool_calls 的字段名不一样、流式返回的 SSE 分帧也不同——半小时过去了，线上还在报错。

这不是个例，而是 2026 年 AI 工程的结构性痛点。模型提供商越来越多（OpenAI、Anthropic、Google、DeepSeek、Groq、Cohere、Mistral、xAI，再加上一堆自建 vLLM/Ollama），每家的 API 都号称"兼容 OpenAI 格式"，但真到边界情况全是坑。于是**AI 网关（AI Gateway / LLM Router）**成了新的基础设施拐点：把"接哪个模型、怎么接、接不上怎么办"这些脏活累活全部收敛到一层代理里。

OmniRoute 就是这波浪潮里冲上 GitHub Trending 的代表。它的定位一句话说清：一个 OpenAI 兼容端点，背后接住 268+ 提供商、500+ 模型，其中 90 多个有免费额度，在此之上叠加智能路由、多层回退和上下文压缩，让你的 AI 助手"永不掉线、成本可控"。

很多人第一眼看到"268 个提供商"会觉得是堆数字。但真正做过多模型接入的人会立刻明白它的价值：它把复杂性留给自己，把简单留给你。你的代码永远只对一个 base_url 说话，剩下的全是网关的事。

本文不做搬运式的功能罗列，而是以工程师视角把 AI 网关这套东西彻底拆开：从 OpenAI 兼容层的协议归一，到路由决策链的责任链设计，到多层回退的熔断逻辑，再到上下文压缩这个真正省钱的核心，最后给出自托管部署与可观测性的完整落地方案。所有关键点都配可运行的代码。读完你不仅懂 OmniRoute，还能自己撸一个精简版网关。

二、核心概念：AI 网关到底在解决什么

在动手看架构之前，先厘清 AI 网关承担的五个职责，这决定了后面所有设计。

2.1 协议归一（Protocol Normalization）

这是网关存在的第一性理由。虽然大家都说"兼容 OpenAI"，但实际差异无处不在：

• Anthropic 的 messages API 里 system 是独立字段，不放在 messages 数组
• Google Gemini 用 contents 和 parts，角色叫 model 不叫 assistant
• 工具调用：OpenAI 是 tool_calls，Anthropic 是 content 里的 tool_use block
• 流式：OpenAI 是 data: {json}\n\n 的 delta，Anthropic 是带 event 类型的 SSE

网关要做的，就是对外暴露一套标准（通常选 OpenAI Chat Completions），对内把请求翻译成各家原生格式，再把响应翻译回来。

2.2 智能路由（Intelligent Routing）

同一个请求，该发给谁？这是路由层的核心决策：

• 简单任务（改个错别字）用便宜的小模型
• 复杂推理用强模型
• 有的租户指定了模型
• 有的场景要求低延迟，就近选 Groq

2.3 多层回退（Multi-layer Fallback）

主模型挂了怎么办？网关维护一条回退链：gpt-4o → claude-3.7 → deepseek-v3，前面失败自动切后面，用户无感知。

2.4 上下文压缩（Context Compression）

这是最省钱的一环。长对话、大 RAG 上下文动辄几万 token，网关在转发前对上下文做裁剪/摘要/去重，token 支出能砍下一大截。

2.5 可观测与治理（Observability & Governance）

统一日志、成本核算、限流、密钥托管、审计——把这些散落各处的横切关注点收进网关。

理解了这五点，我们就能开始搭骨架了。

三、架构总览：请求在网关内部走了哪几步

先看一个请求的完整生命周期。假设客户端发来一个标准的 OpenAI 请求：

Client (OpenAI SDK)
      │  POST /v1/chat/completions
      ▼
┌─────────────────────────────────────────┐
│              AI Gateway                    │
│                                            │
│  1. Auth & RateLimit    鉴权 & 限流        │
│         ▼                                  │
│  2. Normalize In        入向归一化         │
│         ▼                                  │
│  3. Router.decide()     决策：用哪个模型   │
│         ▼                                  │
│  4. ContextCompressor   上下文压缩         │
│         ▼                                  │
│  5. Provider Adapter    翻译成原生格式     │
│         ▼                                  │
│  6. Execute + Fallback  执行 + 失败回退    │
│         ▼                                  │
│  7. Normalize Out       出向归一化         │
│         ▼                                  │
│  8. Observability       计费 & 日志        │
└─────────────────────────────────────────┘
      │  OpenAI-compatible response
      ▼
   Client

这套流水线的关键设计原则是每一层职责单一、可插拔。路由策略、压缩策略、回退策略都应该是独立可替换的模块，而不是塞在一个大 if-else 里。这正是 OmniRoute 能接住这么多提供商还不崩的工程秘密。

下面我们逐层用代码实现。技术栈选 Python + FastAPI（便于讲解，OmniRoute 本体用的是更偏性能的栈，但设计思想相通）。

四、协议归一层：把八家 API 拍平成一套

4.1 统一的内部请求模型

第一步，定义网关内部流转的中立数据结构，把所有入向请求先翻译成它：

from dataclasses import dataclass, field
from typing import Any, Literal, Optional

@dataclass
class Message:
    role: Literal["system", "user", "assistant", "tool"]
    content: str
    tool_calls: list[dict] = field(default_factory=list)
    tool_call_id: Optional[str] = None

@dataclass
class UnifiedRequest:
    """网关内部的中立请求表示，与任何厂商无关"""
    model: str                          # 逻辑模型名，如 "gpt-4o"
    messages: list[Message]
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    stream: bool = False
    tools: list[dict] = field(default_factory=list)
    # 网关自用的元数据
    tenant_id: Optional[str] = None
    scenario: Optional[str] = None      # 场景标签，用于路由
    metadata: dict = field(default_factory=dict)

4.2 Provider 适配器接口

每家提供商实现同一个抽象接口，这是可插拔的关键：

from abc import ABC, abstractmethod
from typing import AsyncIterator

@dataclass
class UnifiedResponse:
    content: str
    tool_calls: list[dict]
    prompt_tokens: int
    completion_tokens: int
    model: str
    finish_reason: str

class ProviderAdapter(ABC):
    """所有提供商必须实现的统一接口"""

    name: str
    supports_tools: bool = True
    supports_streaming: bool = True

    @abstractmethod
    def to_native(self, req: UnifiedRequest) -> dict:
        """把中立请求翻译成本厂商原生请求体"""

    @abstractmethod
    def from_native(self, raw: dict) -> UnifiedResponse:
        """把本厂商原生响应翻译回中立响应"""

    @abstractmethod
    async def invoke(self, req: UnifiedRequest) -> UnifiedResponse:
        """非流式调用"""

    @abstractmethod
    async def stream(self, req: UnifiedRequest) -> AsyncIterator[str]:
        """流式调用，yield 出 OpenAI 格式的 SSE 分片"""

4.3 一个真实适配器：Anthropic

来看协议归一最典型的坑——Anthropic 的 system 字段和 tool_use。这段代码展示了适配器到底在"翻译"什么：

import httpx

class AnthropicAdapter(ProviderAdapter):
    name = "anthropic"

    def __init__(self, api_key: str, model_map: dict[str, str]):
        self.api_key = api_key
        self.model_map = model_map   # 逻辑名 -> Anthropic 真实模型名
        self.client = httpx.AsyncClient(
            base_url="https://api.anthropic.com/v1",
            timeout=httpx.Timeout(60.0, connect=5.0),
        )

    def to_native(self, req: UnifiedRequest) -> dict:
        # 关键差异 1：system 必须抽出来单独放
        system_prompt = ""
        native_messages = []
        for m in req.messages:
            if m.role == "system":
                system_prompt += m.content + "\n"
                continue
            # 关键差异 2：tool 结果要转成 tool_result content block
            if m.role == "tool":
                native_messages.append({
                    "role": "user",
                    "content": [{
                        "type": "tool_result",
                        "tool_use_id": m.tool_call_id,
                        "content": m.content,
                    }],
                })
                continue
            native_messages.append({"role": m.role, "content": m.content})

        body = {
            "model": self.model_map[req.model],
            "messages": native_messages,
            "max_tokens": req.max_tokens or 4096,   # Anthropic 强制要求
            "temperature": req.temperature,
        }
        if system_prompt:
            body["system"] = system_prompt.strip()
        if req.tools:
            # 关键差异 3：工具 schema 结构也不同
            body["tools"] = [{
                "name": t["function"]["name"],
                "description": t["function"].get("description", ""),
                "input_schema": t["function"]["parameters"],
            } for t in req.tools]
        return body

    def from_native(self, raw: dict) -> UnifiedResponse:
        text_parts, tool_calls = [], []
        for block in raw.get("content", []):
            if block["type"] == "text":
                text_parts.append(block["text"])
            elif block["type"] == "tool_use":
                # 翻回 OpenAI 的 tool_calls 结构
                tool_calls.append({
                    "id": block["id"],
                    "type": "function",
                    "function": {
                        "name": block["name"],
                        "arguments": __import__("json").dumps(block["input"]),
                    },
                })
        usage = raw.get("usage", {})
        return UnifiedResponse(
            content="".join(text_parts),
            tool_calls=tool_calls,
            prompt_tokens=usage.get("input_tokens", 0),
            completion_tokens=usage.get("output_tokens", 0),
            model=raw.get("model", ""),
            finish_reason=raw.get("stop_reason", "stop"),
        )

    async def invoke(self, req: UnifiedRequest) -> UnifiedResponse:
        body = self.to_native(req)
        resp = await self.client.post(
            "/messages",
            json=body,
            headers={
                "x-api-key": self.api_key,
                "anthropic-version": "2023-06-01",
            },
        )
        resp.raise_for_status()
        return self.from_native(resp.json())

    async def stream(self, req: UnifiedRequest):
        # 流式部分见第七章，此处略
        raise NotImplementedError

看到这段你就懂了："268 个提供商"的本质，就是 268 个这样的适配器。OmniRoute 的工程量大头在这里——把每家的怪癖都吸收进适配器，对外只留一套干净接口。这也是为什么自己从零接入多模型极其痛苦，而用网关一劳永逸。

五、路由决策链：责任链模式的优雅落地

路由层要回答"这个请求发给谁"。新手容易写成一坨 if-else，但真实场景的决策优先级是层层递进的，责任链模式才是正解。

5.1 决策优先级设计

一个成熟网关的决策顺序通常是：

自定义路由（插件）
      ▼ 未命中
显式指定（请求里点名了模型）
      ▼ 未命中
会话粘性（同一会话尽量用同一模型，利于 KV Cache）
      ▼ 未命中
智能分类（用小模型给任务打分，分派难度）
      ▼ 未命中
默认兜底

每一层是独立策略，命中就返回，未命中传给下一层。

5.2 代码实现

from typing import Protocol, Optional

@dataclass
class RouteDecision:
    provider: str          # 目标提供商
    model: str             # 逻辑模型名
    fallbacks: list[tuple[str, str]]   # 回退链 [(provider, model), ...]
    reason: str            # 决策原因，用于日志排查

class RouteStrategy(Protocol):
    def decide(self, req: UnifiedRequest) -> Optional[RouteDecision]:
        ...

class ExplicitStrategy:
    """请求里显式点名了模型就直接用"""
    def __init__(self, registry: dict[str, str]):
        self.registry = registry   # model -> provider

    def decide(self, req):
        if req.model in self.registry:
            provider = self.registry[req.model]
            return RouteDecision(
                provider=provider, model=req.model,
                fallbacks=[], reason="explicit",
            )
        return None

class StickySessionStrategy:
    """同一会话粘同一模型，提升 KV Cache 命中率"""
    def __init__(self, cache):
        self.cache = cache   # 一个 TTL 字典/Redis

    def decide(self, req):
        sid = req.metadata.get("session_id")
        if not sid:
            return None
        cached = self.cache.get(f"sticky:{sid}")
        if cached:
            provider, model = cached.split("|")
            return RouteDecision(
                provider=provider, model=model,
                fallbacks=[], reason="sticky",
            )
        return None

class ClassificationStrategy:
    """用便宜小模型给任务难度分类，再分派"""
    def __init__(self, classifier_adapter):
        self.classifier = classifier_adapter

    async def decide_async(self, req):
        last_user = next(
            (m.content for m in reversed(req.messages) if m.role == "user"),
            "",
        )
        difficulty = await self._classify(last_user)
        if difficulty == "simple":
            return RouteDecision(
                provider="deepseek", model="deepseek-v3",
                fallbacks=[("groq", "llama-3.3-70b")],
                reason="classified:simple",
            )
        return RouteDecision(
            provider="anthropic", model="claude-3.7-sonnet",
            fallbacks=[("openai", "gpt-4o"), ("deepseek", "deepseek-v3")],
            reason="classified:complex",
        )

    async def _classify(self, text: str) -> str:
        # 用一个极小极快的模型判断，成本几乎可忽略
        prompt = f"判断任务难度，只回 simple 或 complex：\n{text[:500]}"
        # ... 调用小模型 ...
        return "complex"

class DefaultStrategy:
    def decide(self, req):
        return RouteDecision(
            provider="openai", model="gpt-4o-mini",
            fallbacks=[("deepseek", "deepseek-v3")],
            reason="default",
        )

5.3 责任链编排

class Router:
    def __init__(self, strategies: list):
        self.strategies = strategies

    async def route(self, req: UnifiedRequest) -> RouteDecision:
        for strat in self.strategies:
            # 支持同步与异步策略
            if hasattr(strat, "decide_async"):
                decision = await strat.decide_async(req)
            else:
                decision = strat.decide(req)
            if decision is not None:
                return decision
        raise RuntimeError("no route matched")   # 理论上 Default 会兜底

# 装配：顺序即优先级
router = Router([
    ExplicitStrategy(registry),
    StickySessionStrategy(cache),
    ClassificationStrategy(classifier),
    DefaultStrategy(),
])

这套设计的威力在于扩展零成本：想加一个"VIP 租户强制走最强模型"的策略？写个 VipStrategy 插到链首即可，其他代码一行不动。这就是 OmniRoute 能持续接入新策略新提供商而架构不腐化的根本。

六、多层回退与熔断：让网关"永不掉线"

路由决定了主路径和回退链，执行层要负责把这条链跑起来，遇错自动降级。

6.1 带回退的执行器

import asyncio
import logging

log = logging.getLogger("gateway.executor")

RETRYABLE = (httpx.TimeoutException, httpx.ConnectError)

class Executor:
    def __init__(self, adapters: dict[str, ProviderAdapter], breaker):
        self.adapters = adapters
        self.breaker = breaker

    async def execute(self, req: UnifiedRequest, decision: RouteDecision):
        chain = [(decision.provider, decision.model), *decision.fallbacks]
        last_err = None

        for provider, model in chain:
            # 熔断器打开则直接跳过这家
            if self.breaker.is_open(provider):
                log.warning("breaker open, skip %s", provider)
                continue

            adapter = self.adapters[provider]
            attempt_req = replace_model(req, model)
            try:
                resp = await self._call_with_retry(adapter, attempt_req)
                self.breaker.record_success(provider)
                resp.model = f"{provider}/{model}"
                return resp
            except Exception as e:
                last_err = e
                self.breaker.record_failure(provider)
                log.warning("provider %s failed: %s, falling back", provider, e)
                continue

        raise RuntimeError(f"all providers exhausted, last: {last_err}")

    async def _call_with_retry(self, adapter, req, max_retry=2):
        for i in range(max_retry + 1):
            try:
                return await adapter.invoke(req)
            except RETRYABLE as e:
                if i == max_retry:
                    raise
                # 指数退避 + 抖动
                await asyncio.sleep(0.5 * (2 ** i) + 0.1 * i)
            except httpx.HTTPStatusError as e:
                # 429/5xx 可重试，4xx 直接抛
                if e.response.status_code in (429, 500, 502, 503, 529):
                    if i == max_retry:
                        raise
                    retry_after = float(
                        e.response.headers.get("retry-after", 0.5 * (2 ** i))
                    )
                    await asyncio.sleep(retry_after)
                else:
                    raise

注意这里对 429 和 retry-after 头的处理——尊重提供商的限流信号是网关的基本礼貌，盲目重试只会让你被封得更快。

6.2 熔断器：别在死马身上浪费时间

import time

class CircuitBreaker:
    """三态熔断：closed -> open -> half-open"""
    def __init__(self, threshold=5, cooldown=30.0):
        self.threshold = threshold      # 连续失败多少次触发熔断
        self.cooldown = cooldown        # 熔断持续秒数
        self.state: dict[str, dict] = {}

    def _s(self, provider):
        return self.state.setdefault(
            provider, {"fails": 0, "open_until": 0.0}
        )

    def is_open(self, provider) -> bool:
        s = self._s(provider)
        if s["open_until"] > time.time():
            return True
        # 冷却期已过，进入 half-open，放行一次探测
        return False

    def record_success(self, provider):
        s = self._s(provider)
        s["fails"] = 0
        s["open_until"] = 0.0

    def record_failure(self, provider):
        s = self._s(provider)
        s["fails"] += 1
        if s["fails"] >= self.threshold:
            s["open_until"] = time.time() + self.cooldown
            log.error("circuit OPEN for %s (%.0fs)", provider, self.cooldown)

熔断器和回退链配合，效果就是：某家提供商连续挂 5 次，网关直接把它拉黑 30 秒，期间所有请求秒切下一家，30 秒后放一次探测请求试水。这就是 OmniRoute 宣传的"永不掉线"背后的朴素工程真相——没有魔法，只有把每个失败路径都认真处理。

七、流式响应：SSE 的归一化陷阱

流式是网关最容易翻车的地方，因为你要在字节流层面做协议翻译，还不能破坏 SSE 的实时性。

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def openai_sse_wrap(delta_text: str, model: str) -> str:
    """把一段文本包装成 OpenAI 流式 chunk 格式"""
    chunk = {
        "id": "chatcmpl-gw",
        "object": "chat.completion.chunk",
        "model": model,
        "choices": [{
            "index": 0,
            "delta": {"content": delta_text},
            "finish_reason": None,
        }],
    }
    return f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"

class AnthropicStreamAdapter:
    """把 Anthropic 的 event-stream 翻译成 OpenAI 的 delta 流"""
    async def stream(self, req, client, api_key, model_map):
        body = {**self.to_native(req), "stream": True}
        async with client.stream(
            "POST", "/messages", json=body,
            headers={"x-api-key": api_key, "anthropic-version": "2023-06-01"},
        ) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                event = json.loads(line[6:])
                # Anthropic 的增量在 content_block_delta 事件里
                if event.get("type") == "content_block_delta":
                    text = event["delta"].get("text", "")
                    if text:
                        yield await openai_sse_wrap(text, model_map[req.model])
            yield "data: [DONE]\n\n"

@app.post("/v1/chat/completions")
async def chat_completions(payload: dict):
    req = parse_openai_request(payload)          # 入向归一化
    req = await compressor.compress(req)         # 上下文压缩（下一章）
    decision = await router.route(req)

    if req.stream:
        adapter = stream_adapters[decision.provider]
        return StreamingResponse(
            adapter.stream(req, ...),
            media_type="text/event-stream",
        )
    else:
        resp = await executor.execute(req, decision)
        return to_openai_response(resp)          # 出向归一化

流式场景下回退会更复杂：一旦已经吐出了几个 token 再失败，就不能简单切换了（客户端已收到部分内容）。实践中的处理是——首字节前允许回退，首字节后只能重试同一家或直接报错。这个细节很多简易网关会忽略，导致流式回退时内容错乱。

八、上下文压缩：真正把成本打下来的核心

前面都是"能用"，这一章是"省钱"。据行业数据，合理的上下文压缩能让企业 AI 成本最高下降近 97%。网关在这一层的价值最被低估。

8.1 压缩的四种手段

class ContextCompressor:
    def __init__(self, summarizer_adapter, max_context_tokens=8000):
        self.summarizer = summarizer_adapter
        self.max_ctx = max_context_tokens

    async def compress(self, req: UnifiedRequest) -> UnifiedRequest:
        msgs = req.messages
        total = self._count_tokens(msgs)
        if total <= self.max_ctx:
            return req   # 没超预算，不动

        # 手段 1：去重——RAG 场景常有重复片段
        msgs = self._dedup(msgs)

        # 手段 2：滑动窗口——保留 system + 最近 N 轮
        msgs = self._sliding_window(msgs, keep_recent=6)

        # 手段 3：中段摘要——把中间的老对话压成一条摘要
        if self._count_tokens(msgs) > self.max_ctx:
            msgs = await self._summarize_middle(msgs)

        # 手段 4：工具结果裁剪——超长 tool 输出截断保留头尾
        msgs = self._truncate_tool_results(msgs, max_len=2000)

        return replace_messages(req, msgs)

    def _dedup(self, msgs):
        seen, out = set(), []
        for m in msgs:
            key = (m.role, hash(m.content))
            if key in seen and m.role in ("user", "tool"):
                continue
            seen.add(key)
            out.append(m)
        return out

    def _sliding_window(self, msgs, keep_recent):
        system = [m for m in msgs if m.role == "system"]
        rest = [m for m in msgs if m.role != "system"]
        return system + rest[-keep_recent:]

    async def _summarize_middle(self, msgs):
        system = [m for m in msgs if m.role == "system"]
        body = [m for m in msgs if m.role != "system"]
        head, tail = body[:2], body[-4:]
        middle = body[2:-4]
        if not middle:
            return msgs
        text = "\n".join(f"{m.role}: {m.content}" for m in middle)
        summary = await self.summarizer.invoke(UnifiedRequest(
            model="deepseek-v3",
            messages=[Message("user", f"用三句话概括对话要点：\n{text}")],
        ))
        summary_msg = Message("system", f"[早期对话摘要] {summary.content}")
        return system + head + [summary_msg] + tail

    def _truncate_tool_results(self, msgs, max_len):
        for m in msgs:
            if m.role == "tool" and len(m.content) > max_len:
                half = max_len // 2
                m.content = m.content[:half] + "\n...[已截断]...\n" + m.content[-half:]
        return msgs

    def _count_tokens(self, msgs) -> int:
        # 生产环境用 tiktoken；这里粗估
        return sum(len(m.content) for m in msgs) // 3

8.2 压缩的度量：省了多少，损了多少

压缩不能瞎压，得有指标护栏。一个负责任的网关会记录压缩前后的 token 数和"信息保留度"（可用下游任务成功率近似）：

@dataclass
class CompressionMetric:
    before_tokens: int
    after_tokens: int
    strategy_used: list[str]

    @property
    def saved_ratio(self) -> float:
        if self.before_tokens == 0:
            return 0.0
        return 1 - self.after_tokens / self.before_tokens

实战经验：摘要压缩要慎用，它引入了额外的模型调用和信息损失；优先用去重和滑动窗口这种无损/低损手段，只有超长会话才动摘要。OmniRoute 强调的"业界领先的上下文压缩"，核心就是把这套策略调优到"用户几乎感知不到质量下降，但账单肉眼可见地降"。

九、性能优化：网关不能成为新瓶颈

加了一层代理，最怕的就是网关自己变慢。几个关键优化点：

9.1 连接池复用

# 每个 provider 一个长连接客户端，全局复用，避免每请求握手
_clients: dict[str, httpx.AsyncClient] = {}

def get_client(provider: str, base_url: str) -> httpx.AsyncClient:
    if provider not in _clients:
        _clients[provider] = httpx.AsyncClient(
            base_url=base_url,
            timeout=httpx.Timeout(60.0, connect=3.0),
            limits=httpx.Limits(
                max_connections=200,
                max_keepalive_connections=50,
                keepalive_expiry=30.0,
            ),
            http2=True,   # 多路复用，显著降低高并发下的延迟
        )
    return _clients[provider]

9.2 路由决策缓存

分类路由要调小模型，有延迟。对相似请求做决策缓存：

import hashlib

class RouteCache:
    def __init__(self, ttl=300):
        self.ttl = ttl
        self.store: dict[str, tuple[RouteDecision, float]] = {}

    def _key(self, req):
        last = next((m.content for m in reversed(req.messages)
                     if m.role == "user"), "")
        return hashlib.md5(last[:200].encode()).hexdigest()

    def get(self, req):
        k = self._key(req)
        if k in self.store:
            decision, exp = self.store[k]
            if exp > time.time():
                return decision
        return None

    def put(self, req, decision):
        self.store[self._key(req)] = (decision, time.time() + self.ttl)

9.3 压缩异步化与预算护栏

上下文压缩里的摘要调用是同步阻塞的大头。可以设一个"压缩时间预算"，超时就退回无损压缩，绝不让压缩本身拖垮首字延迟：

async def compress_with_budget(compressor, req, budget_s=0.8):
    try:
        return await asyncio.wait_for(compressor.compress(req), timeout=budget_s)
    except asyncio.TimeoutError:
        # 超预算，退回纯滑动窗口（无模型调用）
        return compressor.fast_compress(req)

9.4 关键指标

网关必须暴露这几个指标，否则线上出问题两眼一抹黑：

• 网关自身开销（gateway overhead）：请求进网关到发出上游的时间，应 < 5ms
• 首字节延迟（TTFB）：流式场景的体感核心
• 回退率：多少请求触发了 fallback，飙高说明某家在抖
• 压缩节省率：saved_ratio 的均值
• 各 provider 的 P99 延迟与错误率：路由决策的数据基础

十、自托管部署与可观测性

OmniRoute 主打本地/自托管，这对企业至关重要——密钥不出内网，数据不被第三方采集。给一份生产级的 Docker Compose：

version: "3.9"
services:
  gateway:
    build: .
    ports:
      - "8080:8080"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY}
      - REDIS_URL=redis://redis:6379/0
      - MAX_CONTEXT_TOKENS=8000
      - OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
    depends_on:
      - redis
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 512M
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/healthz"]
      interval: 15s
      timeout: 3s
      retries: 3

  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru

  otel-collector:
    image: otel/opentelemetry-collector-contrib:latest
    volumes:
      - ./otel-config.yaml:/etc/otel/config.yaml
    command: ["--config=/etc/otel/config.yaml"]

用 OpenTelemetry 埋点，把每次调用的关键属性打进 span：

from opentelemetry import trace

tracer = trace.get_tracer("ai-gateway")

async def traced_execute(req, decision):
    with tracer.start_as_current_span("llm.invoke") as span:
        span.set_attribute("gen_ai.request.model", decision.model)
        span.set_attribute("gen_ai.provider", decision.provider)
        span.set_attribute("gateway.route.reason", decision.reason)
        resp = await executor.execute(req, decision)
        span.set_attribute("gen_ai.usage.prompt_tokens", resp.prompt_tokens)
        span.set_attribute("gen_ai.usage.completion_tokens", resp.completion_tokens)
        span.set_attribute("gen_ai.response.model", resp.model)
        return resp

客户端接入零成本，官方 SDK 只改 base_url：

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8080/v1",   # 指向你的网关
    api_key="your-gateway-key",            # 网关自己的鉴权 key
)

resp = client.chat.completions.create(
    model="gpt-4o",                        # 逻辑名，网关自行决定真实落点
    messages=[{"role": "user", "content": "解释一下责任链模式"}],
    extra_body={"scenario": "code-explain"},  # 传场景标签给路由层
)
print(resp.choices[0].message.content)

这就是网关最爽的地方：你的业务代码完全不知道背后接了多少家，切换、加模型、调路由，全在网关侧完成，应用零改动。

十一、选型与取舍：什么时候该上网关

聊了这么多实现，最后回到工程决策：不是所有项目都需要 AI 网关。

适合上网关的信号：

• 你接了 2 家以上模型提供商，且切换频繁
• 线上被提供商限流/宕机坑过，需要高可用
• 多租户/多业务线，需要统一计费和治理
• 上下文很长、token 账单肉痛，压缩收益明显
• 有合规要求，密钥和数据必须自托管

暂时不需要的信号：

• 单模型、小流量的原型项目——直接调 SDK 更简单
• 团队没有运维网关的精力——多一层就多一个故障点

对于决定要上的团队，选型上 OmniRoute 这类自托管 + OpenAI 兼容 + 内置压缩的方案是当下性价比最高的路线：它把上一章我们手写的所有东西都做好了，还维护着 268 家适配器——这个适配器的量，是任何团队都不想自己扛的苦役。

如果你有极致定制需求（比如自研路由算法、特殊合规审计），也完全可以基于本文的骨架自己撸一个精简版：一个 FastAPI + 责任链路由 + 熔断回退 + 压缩器，两三百行就能跑起最小可用版本，再按需长肉。

十二、总结与展望

我们从"深夜切模型"的痛点出发，把 AI 网关这层新基础设施彻底拆开了：

1. 协议归一是地基——268 个提供商本质是 268 个适配器，把厂商怪癖全吸收进去
2. 路由决策链用责任链模式做到优先级清晰、扩展零成本
3. 多层回退 + 熔断是"永不掉线"的朴素真相——没有魔法，只有认真处理每个失败路径
4. 上下文压缩是真正省钱的核心，但要用指标护栏防止过度压缩伤质量
5. 性能优化确保网关不成为新瓶颈：连接池、决策缓存、压缩预算
6. 自托管 + 可观测满足企业的合规与治理刚需

站在 2026 年往前看，AI 网关正在从"工具型产品"跃迁为"基础设施级组件"，就像微服务时代的 API 网关一样成为标配。接下来的演进方向已经很清晰：语义级缓存（相同语义的请求直接命中缓存，连模型都不调）、基于实时质量反馈的动态路由（根据各家近期真实表现自动调权重）、跨请求的 KV Cache 复用、以及成本-质量帕累托前沿的自动寻优。

OmniRoute 们打的这一仗，本质是在帮整个行业把"多模型时代的复杂性"收敛成一个 base_url。而作为工程师，理解它内部这套责任链、熔断、压缩的设计，不只是为了会用一个工具——这些模式在任何高可用代理系统里都通用。学会了，你手里就多了一把能反复用的锤子。

最后送一句实在话：别等线上炸了才想起加网关。 在你接第二家模型的那一刻，就是引入网关的最佳时机。