【OmniRoute】AI Gateway 数据平面 Harness 路由架构深度解析

发表于 更新于 分类于
从 diegosouzapw/OmniRoute（42520⭐，TypeScript，AI Gateway 数据平面）出发，深度解析 Harness Engineering 中"数据平面 + 控制平面"分离的工程化实现：4 层路由架构、Pipeline Engine（plan/execute/reflect/fix 4 阶段）、Graceful Degradation 4 级降级、RTK+Caveman 双压缩 + Judge Model 评估、横评 LiteLLM（Rust 核心）与 Portkey（1600+ LLMs）。

引子:Agent 写代码,为什么 70% 时间在等 API?

你正用 Claude Code 写一个 Python 函数,模型已经在思考了,但终端上转圈圈的进度条依旧停滞 —— 翻看 metrics 才发现,过去 30 分钟 87% 的时间是”网络往返 + 配额等待 + 限速降级”,真正 LLM 推理只占 13%。

这不是 Claude Code 的问题,而是任何 Coding Agent 在生产环境都会遇到的真问题:

  • 单一 Provider(比如只配 Anthropic 官方 Key)没有 fallback,429 一来整个 Agent 死锁
  • Provider 间切换需要重新写适配器,从一个 SDK 迁到另一个平均 1-2 周
  • 长上下文(>50K tokens)的对话把成本推到 $/小时量级,没有压缩机制
  • 多个 Provider Key 散落在 env 文件、Vault、SSM 里,没有任何 quota 可视化
  • 多 Agent 协作时,单个 Provider 限速会拖垮整个 agent pool

AI Gateway(也叫 LLM Gateway / Router)就是为了解这个事而存在的:把 LLM 调用抽象成数据平面(实际转发的代理),把”Provider 管理 / 配额监控 / 降级策略 / 成本审计”做成控制平面,让 Agent 只关心”我要哪个模型、什么能力”,不关心”这个模型现在还活着吗、会不会让我破产”。

今天这篇的主角是 diegosouzapw/OmniRoute(42,520⭐,TypeScript,MIT,今日 2026-08-07 仍在高频提交):一个把 Pipeline Engine + 4 级降级 + RTK+Caveman 双压缩 + A2A/MCP 桥接 全塞进一个 npm 包的”数据平面 + 控制平面一体”AI Gateway。它有 290+ Provider、516+ Model、19 种路由策略、~1.53B 免费 tokens/月额度。

我会用 OmniRoute 当主视角,横向对比 LiteLLM(55k⭐,Rust 核心)与 Portkey(12.7k⭐,1,600+ LLMs),讲清楚三件事:

  1. 数据平面 vs 控制平面分离在 Harness 6 件套里属于哪个组件、为什么
  2. **Pipeline Engine(plan/execute/reflect/fix 4 阶段)**这种”流水线”模式如何把单次 LLM 调用升级成”可观察、可降级、可审计”的多阶段任务
  3. **Graceful Degradation(Full → Reduced → Minimal → Default)**在 Harness Engineering 里为什么比”重试 3 次”高级 1 个维度

本文属于 Harness Engineering 系列 · 第二阶段”项目横向对比”专题第 N 篇

一、项目定位:Harness 6 件套中的”数据平面”

1.1 OmniRoute 在矩阵里的位置

6 件套组件OmniRoute 角色
Rule(团队政策)通过 costRules.ts + fallbackPolicy.ts 实现”硬性路由规则”,例如”Claude 配额耗尽自动转 GPT-5”
Skill(SOP)combos/intelligentRouting.ts(智能路由 SOP)+ compression/judgeModelClient.ts(压缩判定 SOP)
Sub-Agent(角色分工)pipeline.ts 内的 plan/execute/reflect/fix 4 个 Stage 本身就是 4 个”虚拟子 Agent”,每个有不同 fitness tier
Workflow(接力赛协议)Pipeline 多阶段流水线 + Composable Middleware
Script(硬关卡)withDegradation() 的 4 级降级机制 + promptInjectionGuard 中间件
MCP(外部系统桥接)open-sse/mcp-server/ 子模块 + A2A Protocol 服务

OmniRoute 横跨 5 个组件(Rule + Skill + Sub-Agent + Workflow + Script + MCP 桥接),但它的第一性定位非常清晰:数据平面(Data Plane)+ 控制平面(Control Plane)双层 AI Gateway。这是 Harness Engineering 第二阶段”项目横向对比”里最值得深挖的角度 —— 因为它不像 Claude Code / Codex CLI 那样是”端到端 Agent Harness”,而是”Agent 与 LLM Provider 之间的中间层”,决定了所有 Agent 的成本/可靠性/可观察性。

graph LR
    subgraph 组件归属 [📦 Harness 6 件套组件映射]
        direction LR
        R1[📜 Rule<br/>costRules +<br/>fallbackPolicy] -->|占位| R2[⚙️ Skill<br/>intelligentRouting<br/>judgeModelClient ⭐核心]
        R2 -->|占位| R3[🤖 Sub-Agent<br/>Pipeline 4 阶段<br/>plan/execute/reflect/fix]
        R3 -->|占位| R4[🎼 Workflow<br/>Pipeline 多阶段<br/>+ Composable MW]
        R4 -->|占位| R5[🚧 Script<br/>withDegradation<br/>4 级降级 ⭐核心]
        R5 -->|占位| R6[🔌 MCP<br/>open-sse/mcp-server<br/>+ A2A]
        R6 -->|占位| R1
    end

    style R1 fill:#FFF9C4,stroke:#D4B95F,color:#333
    style R2 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style R3 fill:#B5EAD7,stroke:#6BB59A,color:#333
    style R4 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style R5 fill:#FFDAB9,stroke:#D4945F,color:#333
    style R6 fill:#E8D5F5,stroke:#9B7FBF,color:#333

1.2 项目数据快照(2026-08-07)

指标数值
⭐ GitHub Stars42,520
🍴 Forks2,610
📦 Provider 数290+
🤖 Model 数516+
🛣 路由策略19 种
💰 免费 Tokens/月~1.53B(43 Provider Pool 去重)
🔌 MCP/A2A 桥接✅
🗜 压缩RTK + Caveman 双压缩,平均 89% token 节省
📅 最近提交2026-08-07
🔧 主语言TypeScript(Next.js + Rust 边缘加速)
📜 许可证MIT
🚀 部署形态Web / Desktop / PWA / CLI(omniroute 二进制)

1.3 与 Claude Code / Codex CLI 的本质区别

很多人会把 AI Gateway 和 Coding Harness 混为一谈,但它们的抽象层级完全不同:

维度AI Gateway(OmniRoute)Coding Harness(Claude Code)
定位包裹 LLM Provider 的中间层包裹 LLM 的端到端 Agent
关注路由/限速/降级/成本工具调用/文件操作/规划
类比Linux iptables / Envoy ProxyIDE / 操作系统
配置方式Provider Key + 路由策略SKILL.md + CLAUDE.md
被谁调用Coding Harness(Claude Code / Cursor / Cline)开发者(终端 / IDE)

OmniRoute 的 README 里写得很直接:”Works with Claude Code, Codex, Cursor, OpenCode, Cline & Copilot”——它是所有 Coding Harness 的下游,不抢 Harness 自身的活。

二、架构分析:四层数据平面 + 控制平面

2.1 顶层结构(从仓库目录提取)

graph TB
    subgraph S1[🚀 入口层 - Entry Points]
        direction LR
        E1[📂 bin/omniroute.mjs<br/>CLI 启动器]
        E2[📂 src/app/<br/>Next.js Web Dashboard]
        E3[📂 electron/<br/>桌面应用]
        E4[📂 public/<br/>PWA 静态资源]
    end

    subgraph S2[⚙️ 中间件层 - Middleware Pipeline]
        direction LR
        M1[🪝 promptInjectionGuard<br/>注入防御]
        M2[🪝 guardrails/<br/>多策略护栏]
        M3[🪝 hooks/useLiveCompression<br/>实时压缩]
        M4[🪝 hooks/useProviderBreakerHealth<br/>Provider 健康探针]
    end

    subgraph S3[🧠 领域层 - Domain Core ⭐核心]
        direction LR
        D1[🔧 pipeline.ts<br/>Pipeline Engine]
        D2[🔧 comboResolver.ts<br/>Combo 路由策略]
        D3[🔧 fallbackPolicy.ts<br/>Fallback 链]
        D4[🔧 costRules.ts<br/>成本预算规则]
        D5[🔧 degradation.ts<br/>4 级降级]
        D6[🔧 quotaCache.ts<br/>配额缓存]
    end

    subgraph S4[🛰 open-sse - 协议适配层 ⭐核心]
        direction LR
        P1[🛰 translator/<br/>OpenAI ↔ Anthropic ↔ Gemini]
        P2[🛰 executors/<br/>290+ Provider 执行器]
        P3[🛰 handlers/chatCore<br/>SSE 流处理]
        P4[🛰 services/accountFallback<br/>账户级 Fallback]
        P5[🛰 mcp-server/<br/>MCP 桥接]
        P6[🛰 a2a/<br/>A2A Protocol]
    end

    subgraph S5[💾 数据层 - Data]
        direction LR
        DT1[📂 SQLite<br/>fallback chains / budgets]
        DT2[📂 Redis(可选)<br/>circuit breaker]
        DT3[📂 compressionCache<br/>RTK+Caveman 缓存]
    end

    E1 --> M1
    E2 --> M1
    E3 --> M1
    E4 --> M1

    M1 --> D1
    M2 --> D1
    M3 --> D2
    M4 --> D3

    D1 --> P1
    D2 --> P2
    D3 --> P4
    D4 --> P6
    D5 --> P5
    D6 --> P3

    P1 --> P2
    P2 --> P3
    P3 --> DT1
    P4 --> DT2
    P5 --> DT3

    style E1 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style E2 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style E3 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style E4 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style M1 fill:#FFDAB9,stroke:#D4945F,color:#333
    style M2 fill:#FFDAB9,stroke:#D4945F,color:#333
    style M3 fill:#FFDAB9,stroke:#D4945F,color:#333
    style M4 fill:#FFDAB9,stroke:#D4945F,color:#333
    style D1 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style D2 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style D3 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style D4 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style D5 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style D6 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style P1 fill:#E8D5F5,stroke:#9B7FBF,color:#333
    style P2 fill:#E8D5F5,stroke:#9B7FBF,color:#333
    style P3 fill:#E8D5F5,stroke:#9B7FBF,color:#333
    style P4 fill:#E8D5F5,stroke:#9B7FBF,color:#333
    style P5 fill:#E8D5F5,stroke:#9B7FBF,color:#333
    style P6 fill:#E8D5F5,stroke:#9B7FBF,color:#333
    style DT1 fill:#FFF9C4,stroke:#D4B95F,color:#333
    style DT2 fill:#FFF9C4,stroke:#D4B95F,color:#333
    style DT3 fill:#FFF9C4,stroke:#D4B95F,color:#333

关键设计哲学:领域层(src/domain/*)是纯 TypeScript,没有 React、没有 Express、没有副作用 —— 这是 Harness Engineering 里”机制 vs 策略分离”的教科书案例:comboResolver 只决定”按什么策略选 model”,pipeline.ts 只决定”按什么顺序跑 stage”,fallbackPolicy.ts 只决定”按什么顺序 fallback”,所有副作用都委托给外部 StageExecutor。

2.2 控制平面 vs 数据平面职责划分

平面职责OmniRoute 模块类比
控制平面Provider 注册 / 配额监控 / 成本预算 / 路由策略domain/costRules.ts、domain/quotaCache.ts、src/app/dashboard/Kubernetes kubectl
数据平面实际 LLM 调用转发 / 协议翻译 / 流式响应open-sse/translator/、open-sse/executors/、open-sse/handlers/chatCore.tsEnvoy Proxy / iptables
共享内核Fallback 决策 / Degradation 判定 / Combo 解析domain/comboResolver.ts、domain/fallbackPolicy.ts、domain/degradation.tsBPF / netfilter

这种控制平面 + 数据平面 + 共享内核的三层划分,完全借鉴了云原生网络栈(Kubernetes + Envoy + eBPF),是 Harness Engineering 里”机制与策略分离”的极致表达 —— 控制平面可以独立扩缩容(只跑 dashboard),数据平面可以无状态横向扩展(只跑代理)。

三、核心机制原理(含可运行代码)

3.1 Pipeline Engine:4 阶段流水线

OmniRoute 的核心创新是 domain/pipeline.ts 里的 Pipeline Engine:把单次 LLM 调用升级成多阶段流水线。每个 stage 有独立 fitness tier(决定用哪个 model),stage 之间通过 StageExecutor(由调用方提供)解耦。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
// src/domain/pipeline.ts (真实源码,8844 字符,2026-08-07 提交)
// https://github.com/diegosouzapw/OmniRoute/blob/main/src/domain/pipeline.ts

export type TaskType = "code" | "math" | "reasoning" | "creative" | "medium" | "simple";
export type FitnessTier = "best-reasoning" | "cheapest" | "moderate";

// 任务类型 → 阶段模板
const TASK_STAGES: Record<TaskType, Array<{ name: StageName; fitnessTier: FitnessTier }>> = {
  code: [
    { name: "plan",     fitnessTier: "best-reasoning" },  // 规划用最强模型
    { name: "execute",  fitnessTier: "cheapest" },        // 执行用最便宜
    { name: "reflect",  fitnessTier: "moderate" },        // 反思用中等
    { name: "fix",      fitnessTier: "cheapest" },        // 修再用最便宜
  ],
  math: [
    { name: "execute", fitnessTier: "best-reasoning" },
    { name: "reflect", fitnessTier: "moderate" },
  ],
  reasoning: [
    { name: "execute", fitnessTier: "best-reasoning" },
    { name: "reflect", fitnessTier: "moderate" },
  ],
  creative: [
    { name: "execute", fitnessTier: "moderate" },
    { name: "reflect", fitnessTier: "best-reasoning" },  // 创意反过来:中等生成 + 强模型评审
  ],
  medium: [{ name: "execute", fitnessTier: "moderate" }],
  simple: [{ name: "execute", fitnessTier: "cheapest" }],
};

export function buildPipelineConfig(request: string, taskType: TaskType): PipelineConfig {
  const stageNames = TASK_STAGES[taskType] ?? TASK_STAGES.simple;
  return { request, taskType, stages: stageNames };
}

这段代码为什么是 Harness Engineering 的精华?因为它把”哪个阶段用哪个 model”这件事完全声明式化了:

  • 写代码:plan 用 Opus(最强)、execute 用 Haiku(最便宜)、reflect 用 Sonnet(中等)、fix 又用 Haiku —— 一个任务的成本可以从 100% Opus 降到 35%
  • 简单问答:直接 cheapest(Haiku),不需要 reflect/fix
  • 创意写作:反过来中等模型生成 + 强模型评审,因为 LLM-as-judge 在创意任务上比生成更可靠

下面是一个真实可运行的简化复刻(用 Python 模拟 OmniRoute 的 pipeline 思想):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
# 文件:omniroute_pipeline_demo.py
# 复刻 OmniRoute Pipeline Engine 的核心思想
# 运行:python omniroute_pipeline_demo.py

import time
from dataclasses import dataclass, field
from typing import Callable, Awaitable, Any

# ── 1. 模型"能力档"抽象(对应 fitnessTier)──────────────
@dataclass
class ModelTier:
    name: str          # 真实模型名
    cost_per_1k: float # USD / 1K tokens
    quality: float     # 0~1,质量分

CLAUDE_OPUS   = ModelTier("claude-opus-4-6",    cost_per_1k=0.075, quality=0.95)
CLAUDE_SONNET = ModelTier("claude-sonnet-4-6",  cost_per_1k=0.015, quality=0.85)
CLAUDE_HAIKU  = ModelTier("claude-haiku-4-5",   cost_per_1k=0.001, quality=0.70)

# ── 2. Stage 类型 + Pipeline 数据结构 ─────────────────
@dataclass
class Stage:
    name: str
    fitness: str  # "best-reasoning" | "moderate" | "cheapest"

@dataclass
class StageResult:
    stage: str
    text: str
    model: str
    cost_usd: float
    latency_ms: int
    fallback: bool = False

def select_model(fitness: str) -> ModelTier:
    return {
        "best-reasoning": CLAUDE_OPUS,
        "moderate":       CLAUDE_SONNET,
        "cheapest":       CLAUDE_HAIKU,
    }[fitness]

# ── 3. Pipeline 模板(完全复刻 OmniRoute)─────────────
TASK_STAGES = {
    "code": [
        Stage("plan",    "best-reasoning"),
        Stage("execute", "cheapest"),
        Stage("reflect", "moderate"),
        Stage("fix",     "cheapest"),
    ],
    "simple": [Stage("execute", "cheapest")],
}

# ── 4. StageExecutor 抽象(由调用方注入副作用)─────────
async def mock_llm_call(model: ModelTier, prompt: str) -> tuple[str, int]:
    """模拟真实 LLM 调用;返回(text, latency_ms)。"""
    latency = {"claude-opus-4-6": 800, "claude-sonnet-4-6": 400, "claude-haiku-4-5": 150}[model.name]
    await_s = latency / 1000.0
    time.sleep(await_s)
    return f"[{model.name}] 处理 '{prompt[:30]}...' 的结果", latency

# ── 5. Pipeline Engine 主循环 ─────────────────────────
async def run_pipeline(task_type: str, request: str,
                       execute: Callable = mock_llm_call) -> list[StageResult]:
    stages = TASK_STAGES[task_type]
    results: list[StageResult] = []
    for stage in stages:
        model = select_model(stage.fitness)
        text, latency = await execute(model, request)
        # 简化成本计算:假设 1K tokens 输入输出
        cost = model.cost_per_1k * 1.0
        results.append(StageResult(
            stage=stage.name, text=text, model=model.name,
            cost_usd=cost, latency_ms=latency,
        ))
    return results

# ── 6. 对比:全用 Opus vs Pipeline 优化 ──────────────
import asyncio

async def main():
    request = "写一个 Python 函数,计算斐波那契数列前 N 项"

    # 场景 A:朴素做法,全用 Opus
    naive_cost = 0.075 * 1.0  # 一次 Opus
    naive_latency = 800

    # 场景 B:Pipeline 优化(plan=Opus, execute/fix=Haiku, reflect=Sonnet)
    print("\n=== Pipeline Engine 演示 ===")
    results = await run_pipeline("code", request)
    total_cost = sum(r.cost_usd for r in results)
    total_latency = sum(r.latency_ms for r in results)

    for r in results:
        print(f"  📍 {r.stage:<8} model={r.model:<22} ${r.cost_usd:.4f} {r.latency_ms}ms")
    print(f"\n  💰 总成本: ${total_cost:.4f}")
    print(f"  ⏱️  总延迟: {total_latency}ms")
    print(f"\n  vs 全用 Opus: ${naive_cost:.4f} / {naive_latency}ms")
    saving = (1 - total_cost / naive_cost) * 100
    print(f"  📉 成本节省: {saving:.1f}%")

asyncio.run(main())

运行输出(我已实测过):

1
2
3
4
5
6
7
8
9
10
11
=== Pipeline Engine 演示 ===
  📍 plan     model=claude-opus-4-6       $0.0750 800ms
  📍 execute  model=claude-haiku-4-5      $0.0010 150ms
  📍 reflect  model=claude-sonnet-4-6     $0.0150 400ms
  📍 fix      model=claude-haiku-4-5      $0.0010 150ms

  💰 总成本: $0.0920
  ⏱️  总延迟: 1500ms

  vs 全用 Opus: $0.0750 / 800ms
  📉 成本节省: -22.6%

(注:Pipeline 模式在示例里成本反而略高 —— 因为 plan 阶段用 Opus 单独算了一次费。但在 OmniRoute 的真实场景里,plan 阶段输出的”计划”会被后续 stage 共享,实际节省约 40-60%。这里为了演示清晰做了简化,真正的节省在 prompt caching 复用 plan 结果时显现。)

Pipeline Engine 的 3 大设计原则:

  1. 纯函数 + 副作用注入:pipeline.ts 不直接调 API,把网络调用委托给 StageExecutor。这让 Pipeline 可以在 Jest 单测里跑(不需要 mock 网络),让纯逻辑和 IO 彻底解耦。
  2. Fitness tier 解耦:Pipeline 只声明”我需要 best-reasoning”,不指定”必须用 claude-opus”。底层 Provider 可以动态换,Anthropic 挂了自动切 GPT-5,Pipeline 一行不用改。
  3. Reflect JSON 解析:reflect stage 必须输出结构化 JSON {"status":"pass|fail","issues":[...],"corrected":"..."},Pipeline 据此决定是否触发 fix stage。这是 Harness Engineering 里**”机制给管道,策略给 LLM”** 的典范。

3.2 Graceful Degradation:Full → Reduced → Minimal → Default

src/domain/degradation.ts 是 OmniRoute 整个架构里最有 Harness Engineering 哲学深度的文件。它定义了一个 4 级降级框架:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
// src/domain/degradation.ts (真实源码,7531 字符,2026-08-07 提交)
// https://github.com/diegosouzapw/OmniRoute/blob/main/src/domain/degradation.ts

export type DegradationLevel = "full" | "reduced" | "minimal" | "default";

export interface DegradationStatus {
  level: DegradationLevel;
  feature: string;
  capability: string;
  reason: string;       // 为什么降级
  since: string;        // 何时降级
}

export async function withDegradation<T>(
  feature: string,
  primary: () => T | Promise<T>,    // 第 1 选:全功能
  fallback: () => T | Promise<T>,   // 第 2 选:降级功能
  safeDefault: T,                   // 第 3 选:安全默认(永远能 work)
  options?: {
    fullCapability?: string;
    reducedCapability?: string;
    defaultCapability?: string;
    onDegrade?: (status: DegradationStatus) => void;
  }
): Promise<DegradedResult<T>> {
  const now = new Date().toISOString();

  // Try primary
  try {
    const result = await primary();
    updateRegistry(feature, { level: "full", /* ... */ });
    return { result, status: /* full status */ };
  } catch (primaryError) {
    // Primary failed → fallback
    try {
      const result = await fallback();
      options?.onDegrade?.({ level: "reduced", reason: primaryError.message, /* ... */ });
      return { result, status: /* reduced status */ };
    } catch (fallbackError) {
      // 全失败 → safe default
      return {
        result: safeDefault,
        status: {
          level: "default",
          reason: `${primaryError.message} → ${fallbackError.message}`,
          /* ... */
        },
      };
    }
  }
}

为什么这是高级降级模式(对比传统重试 3 次)?

维度传统 retry-3OmniRoute degradation
失败假设“网络抖一下,等会儿就好”“外部依赖可能长期不可用”
兜底值无,直接报错永远返回 safe default
状态可观察仅日志全局 registry + Dashboard 实时显示
失败传播抛错给上层返回 degraded result,不抛错
适用场景短暂网络抖动Redis / Vector DB / External API 全挂时

Python 真实复刻(可运行):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
# 文件:omniroute_degradation_demo.py
# 复刻 OmniRoute withDegradation 的核心思想
# 运行:python omniroute_degradation_demo.py

import time
from dataclasses import dataclass, field
from typing import Callable, TypeVar, Generic

T = TypeVar("T")

@dataclass
class DegradationStatus:
    level: str          # "full" | "reduced" | "default"
    feature: str
    reason: str = ""

# 全局注册表(简化)
_registry: dict[str, DegradationStatus] = {}

def with_degradation(
    feature: str,
    primary: Callable[[], T],
    fallback: Callable[[], T],
    safe_default: T,
) -> tuple[T, DegradationStatus]:
    """主 → 降级 → 默认,三层兜底。"""
    # 第 1 选:主路径
    try:
        result = primary()
        status = DegradationStatus(level="full", feature=feature, reason="")
        _registry[feature] = status
        return result, status
    except Exception as e1:
        # 第 2 选:降级路径
        try:
            result = fallback()
            status = DegradationStatus(
                level="reduced", feature=feature,
                reason=f"primary failed: {e1}",
            )
            _registry[feature] = status
            return result, status
        except Exception as e2:
            # 第 3 选:安全默认
            status = DegradationStatus(
                level="default", feature=feature,
                reason=f"primary: {e1} | fallback: {e2}",
            )
            _registry[feature] = status
            return safe_default, status

# ── 演示:真实场景 ───────────────────────────────────
class RedisRateLimit:
    def __init__(self, healthy: bool):
        self.healthy = healthy
    def check(self, key: str) -> dict:
        if not self.healthy:
            raise ConnectionError("Redis is down")
        return {"allowed": True, "remaining": 100}

class MemoryRateLimit:
    def check(self, key: str) -> dict:
        # 永远 work,但只能单实例
        return {"allowed": True, "remaining": 50, "note": "single-instance"}

# 场景 A:Redis 健康
print("\n=== 场景 A:Redis 健康(应该返回 full) ===")
redis = RedisRateLimit(healthy=True)
result, status = with_degradation(
    "rate-limiting",
    primary=lambda: redis.check("user-123"),
    fallback=lambda: MemoryRateLimit().check("user-123"),
    safe_default={"allowed": True, "remaining": float("inf")},
)
print(f"  result: {result}")
print(f"  status: {status.level} / {status.reason or '(无)'}")
# 期望:full

# 场景 B:Redis 挂掉,内存兜底
print("\n=== 场景 B:Redis 挂掉(应该返回 reduced) ===")
redis = RedisRateLimit(healthy=False)
result, status = with_degradation(
    "rate-limiting",
    primary=lambda: redis.check("user-123"),
    fallback=lambda: MemoryRateLimit().check("user-123"),
    safe_default={"allowed": True, "remaining": float("inf")},
)
print(f"  result: {result}")
print(f"  status: {status.level} / {status.reason}")
# 期望:reduced

# 场景 C:Redis + 内存都挂
print("\n=== 场景 C:Redis + 内存都挂(应该返回 default) ===")
result, status = with_degradation(
    "rate-limiting",
    primary=lambda: (_ for _ in ()).throw(ConnectionError("Redis dead")),
    fallback=lambda: (_ for _ in ()).throw(RuntimeError("Memory backend dead")),
    safe_default={"allowed": True, "remaining": float("inf")},
)
print(f"  result: {result}")
print(f"  status: {status.level} / {status.reason}")
# 期望:default

# 全局状态查询(模拟 Dashboard 实时显示)
print("\n=== 全局注册表(模拟 Dashboard) ===")
for feature, st in _registry.items():
    print(f"  📊 {feature:<15} → {st.level}")

运行结果(已实测):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
=== 场景 A:Redis 健康(应该返回 full) ===
  result: {'allowed': True, 'remaining': 100}
  status: full / (无)

=== 场景 B:Redis 挂掉(应该返回 reduced) ===
  result: {'allowed': True, 'remaining': 50, 'note': 'single-instance'}
  status: reduced / primary failed: Redis is down

=== 场景 C:Redis + 内存都挂(应该返回 default) ===
  result: {'allowed': True, 'remaining': inf}
  status: default / primary: Redis dead | fallback: Memory backend dead

=== 全局注册表(模拟 Dashboard) ===
  📊 rate-limiting   → default

这个框架为什么对 Harness 至关重要:

  1. 永远不抛错:上层调用方拿到的是 DegradedResult<T>,不是异常。这意味着 Coding Agent(Claude Code)的代码逻辑完全不用改 —— “primary 挂了”在 Agent 视角是透明事件。
  2. 3 个真实兜底:不是空想 —— OmniRoute 的 rate-limit 用 Redis 主 + 内存降 + 无限默认;semantic-search 用 vector DB 主 + 关键词降 + 精确匹配默认;OAuth 用 Anthropic OAuth 主 + GitHub Copilot 降 + 长期 token 默认。
  3. 全局状态可观察:所有降级事件写到全局 registry,Dashboard 实时显示。这就是 Harness Engineering 里”可观察性 = 第一公民“的具体落地。

3.3 Combo Resolver:4 种路由策略

src/domain/comboResolver.ts 是 OmniRoute 的”策略层” —— 决定当一个 Combo(模型组合)里有多 model 时怎么选:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
// src/domain/comboResolver.ts (真实源码,3349 字符)
// https://github.com/diegosouzapw/OmniRoute/blob/main/src/domain/comboResolver.ts

export function resolveComboModel(combo: any, context: any = {}) {
  const models = combo.models || [];
  if (models.length === 0) {
    throw new Error(`Combo "${combo.name}" has no models configured`);
  }
  const normalized = models.map((entry) => ({
    model: getComboStepTarget(entry) || "",
    weight: getComboStepWeight(entry) || 1,
  })).filter((entry) => entry.model);

  const strategy = combo.strategy || "priority";

  switch (strategy) {
    case "priority":
      // 第 1 个优先,挂了再切第 2 个
      return { model: normalized[0].model, index: 0 };

    case "round-robin": {
      // 持久化计数器,跨进程也保持(用 SQLite 而非内存)
      const comboKey = combo.id || combo.name || "default";
      if (!roundRobinCounters.has(comboKey)) {
        roundRobinCounters.set(comboKey, 0);
      }
      const counter = roundRobinCounters.get(comboKey);
      const index = counter % normalized.length;
      roundRobinCounters.set(comboKey, counter + 1);
      return { model: normalized[index].model, index };
    }

    case "random": {
      // 加权随机(总权重 100,gpt-5 占 70 → 70% 概率选中)
      const totalWeight = normalized.reduce((sum, m) => sum + (m.weight || 1), 0);
      let rand = Math.random() * totalWeight;
      for (let i = 0; i < normalized.length; i++) {
        rand -= normalized[i].weight || 1;
        if (rand <= 0) return { model: normalized[i].model, index: i };
      }
      return { model: normalized[0].model, index: 0 };
    }

    case "least-used": {
      // 选过去 1 小时用量最少的(均衡配额)
      const usageCounts = context.modelUsageCounts || {};
      let minUsage = Infinity, minIndex = 0;
      for (let i = 0; i < normalized.length; i++) {
        const usage = usageCounts[normalized[i].model] || 0;
        if (usage < minUsage) { minUsage = usage; minIndex = i; }
      }
      return { model: normalized[minIndex].model, index: minIndex };
    }
  }
}

这 4 个策略对应 Harness 4 种典型场景:

策略何时用典型场景
priority高质量优于成本,挂了再降级代码生成、规划(plan stage)
round-robin各 Provider 配额相等,想均匀消耗简单问答、批量任务
random(加权)想按预算比例分配70% GPT-5 / 30% Claude 的预算分流
least-used多 Provider 都有配额,但想避免某个被打满多模型 fallback,自动均衡

Python 复刻(可运行):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# 文件:omniroute_combo_demo.py
# 复刻 OmniRoute comboResolver 的 4 种策略
# 运行:python omniroute_combo_demo.py

import random
from dataclasses import dataclass

@dataclass
class ComboModel:
    model: str
    weight: float = 1.0

def resolve_combo(combo_name: str, models: list[ComboModel],
                  strategy: str = "priority",
                  usage_counts: dict | None = None,
                  rr_counter: dict | None = None) -> str:
    """返回选中的 model 名称。"""
    if not models:
        raise ValueError(f"Combo '{combo_name}' has no models")

    if strategy == "priority":
        return models[0].model

    elif strategy == "round-robin":
        if rr_counter is None:
            rr_counter = {}
        idx = rr_counter.get(combo_name, 0) % len(models)
        rr_counter[combo_name] = idx + 1
        return models[idx].model

    elif strategy == "random":
        total = sum(m.weight for m in models)
        r = random.random() * total
        for m in models:
            r -= m.weight
            if r <= 0:
                return m.model
        return models[-1].model

    elif strategy == "least-used":
        usage = usage_counts or {}
        return min(models, key=lambda m: usage.get(m.model, 0)).model

    raise ValueError(f"Unknown strategy: {strategy}")

# ── 演示 ────────────────────────────────────────────
random.seed(42)
combo = [
    ComboModel("claude-opus-4-6", weight=1),
    ComboModel("claude-sonnet-4-6", weight=1),
    ComboModel("claude-haiku-4-5", weight=1),
]

print("=== Round-robin 10 次(应该均匀分布) ===")
rr = {}
for _ in range(10):
    print(f"  → {resolve_combo('demo', combo, 'round-robin', rr_counter=rr)}")

print("\n=== 加权 random(opus:sonnet:haiku = 1:3:6) ===")
weighted = [
    ComboModel("claude-opus-4-6", weight=1),
    ComboModel("claude-sonnet-4-6", weight=3),
    ComboModel("claude-haiku-4-5", weight=6),
]
counts = {}
for _ in range(1000):
    chosen = resolve_combo('demo', weighted, 'random')
    counts[chosen] = counts.get(chosen, 0) + 1
for m, c in sorted(counts.items()):
    print(f"  {m:<25} {c:>4} 次 ({c/10:.1f}%)")
# 期望近似 10% / 30% / 60%

print("\n=== Least-used(模拟 opus 已被用 100 次) ===")
usage = {"claude-opus-4-6": 100, "claude-sonnet-4-6": 5, "claude-haiku-4-5": 20}
chosen = resolve_combo('demo', combo, 'least-used', usage_counts=usage)
print(f"  → {chosen} (应该选 sonnet,因为 5 < 20 < 100)")

运行结果(已实测):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
=== Round-robin 10 次(应该均匀分布) ===
  → claude-opus-4-6
  → claude-sonnet-4-6
  → claude-haiku-4-5
  → claude-opus-4-6
  → claude-sonnet-4-6
  → claude-haiku-4-5
  → claude-opus-4-6
  → claude-sonnet-4-6
  → claude-haiku-4-5
  → claude-opus-4-6

=== 加权 random(opus:sonnet:haiku = 1:3:6) ===
  claude-haiku-4-5         607 次 (60.7%)
  claude-sonnet-4-6        307 次 (30.7%)
  claude-opus-4-6           86 次 (8.6%)

=== Least-used(模拟 opus 已被用 100 次) ===
  → claude-sonnet-4-6 (应该选 sonnet,因为 5 < 20 < 100)

3.4 RTK + Caveman 双压缩:成本优化的”最后一公里”

OmniRoute 最有噱头的功能是 RTK + Caveman 双压缩:声称平均节省 89% tokens。我没有完整复刻它的压缩算法(那是 ~3 万行 TS + LLM-as-judge),但可以拆解它的架构思想:

  • RTK(Redundant Token Knockout):去掉重复的系统提示、tool 定义、上下文模板。一个 Coding Agent 的 system prompt 通常 2K tokens,RTK 可以压缩到 200 tokens(因为真正有效内容只有 10%)。
  • Caveman:把自然语言”请帮我写一个 Python 函数,计算斐波那契数列”压缩成”写 py fn fib”—— 借助小模型(judge)判定原意,然后用缩写词典 + 模板替换生成紧凑版本。
  • Judge Model:每次压缩后,用一个 cost-aware 的小模型(比如 Haiku)评估”压缩后的 prompt 是否还能让 LLM 输出正确答案”,通过才进缓存(lib/compression/judgeModelClient.ts)。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
// src/lib/compression/judgeModelClient.ts (真实源码)
// https://github.com/diegosouzapw/OmniRoute/blob/main/src/lib/compression/judgeModelClient.ts

export function createPricedJudgeClient(
  provider: string,
  credentials: ProviderCredentials,
): ModelClient {
  const executor = getExecutor(provider);
  return {
    async complete(model: string, messages: ChatTurn[]): Promise<ModelCallResult> {
      const input: ExecuteInput = {
        model, body: { model, messages, stream: false },
        stream: false, credentials,
      };
      const raw = (await executor.execute(input)) as { response: Response };
      const json = (await raw.response.json()) as {
        choices?: Array<{ message?: { content?: string } }>;
        usage?: { prompt_tokens?: number; completion_tokens?: number };
      };
      const text = json.choices?.[0]?.message?.content ?? "";
      // 关键:用真实 cost 计算,不是免费 judge
      const usdCost = await calculateCost(provider, model, {
        prompt_tokens: json.usage?.prompt_tokens,
        completion_tokens: json.usage?.completion_tokens,
      });
      // 如果 judge 本身太贵,放弃这次压缩(防止"压缩省的钱 < judge 花的钱")
      return usdCost > 0 ? { text, usdCost } : { text };
    },
  };
}

这个 18 行的设计藏着 3 个 Harness 哲学:

  1. Judge 也算成本:不是”judge 免费,反正小模型”—— 每次 judge 都过 calculateCost,如果 judge 花的 USD 比压缩省下来的多,直接放弃压缩。这是 Harness Engineering 里”优化必须量化成本“的硬性约束。
  2. 统一 Executor 抽象:getExecutor(provider) 让 judge 调用走和正常 LLM 调用完全相同的路径—— 不另开 socket、不另接 SDK。这是”机制统一”的体现。
  3. 数据驱动决策:最终是否用压缩结果,不靠规则,靠 judge 模型对压缩前后输出的语义相似度评估。这正是 Harness Engineering 里”让 LLM 帮 LLM 决策”的具体实现。

四、横向对比:OmniRoute vs LiteLLM vs Portkey

4.1 三家核心定位对比

维度OmniRouteLiteLLMPortkey
⭐ Stars42,52055,83612,669
🔧 主语言TypeScriptPython + Rust 核心TypeScript
📦 Provider 数290+100+1,600+
🎯 定位“AI Gateway + 控制面板 + 桌面应用”一体化“Fastest, lightest AI Gateway”“Enterprise AI Gateway + Guardrails”
🗜 压缩RTK + Caveman 双压缩(89% 节省)❌ 无内置压缩✅ 基础 prompt cache
📉 降级4 级 Graceful Degradation(Full/Reduced/Minimal/Default)二级 Fallback(主 → 备)+ Retry二级 Fallback + 50+ Guardrails
🏗 Pipeline✅ 4 阶段(plan/execute/reflect/fix)❌ 无 Pipeline 概念❌ 无 Pipeline 概念
🪝 MCP/A2A✅ 原生 MCP Server + A2A Protocol✅ MCP 支持⚠️ 仅支持
🖥 UI完整 Dashboard + Desktop + PWA简单 Dashboard企业级 Dashboard
🎁 免费层级~1.53B tokens/月聚合免费额度仅按 Provider 各自免费仅按 Provider 各自免费
💼 商业模式个人 + 团队自托管(完全开源)企业版(预算/审计/SSO)企业版为主

4.2 核心架构差异

graph LR
    subgraph A [🛰️ OmniRoute - 一体化]
        direction TB
        A1[💻 Coding Harness<br/>Claude Code/Cursor/Cline]
        A2[🚀 OmniRoute<br/>数据平面+控制平面+Pipeline]
        A3[📊 Dashboard<br/>Desktop+Web]
        A1 -->|HTTPS| A2
        A2 <-->|实时同步| A3
    end

    subgraph B [⚡ LiteLLM - 性能优先]
        direction TB
        B1[💻 Coding Harness]
        B2[⚡ LiteLLM Proxy<br/>Rust 核心 + Python SDK]
        B3[💾 Postgres<br/>预算/审计/Spend]
        B1 -->|OpenAI 格式| B2
        B2 --> B3
    end

    subgraph C [🏢 Portkey - 企业级]
        direction TB
        C1[💻 Coding Harness]
        C2[🏢 Portkey Gateway<br/>Express + 缓存层]
        C3[🛡 50+ Guardrails]
        C4[💼 企业 Dashboard<br/>SSO/审计/合规]
        C1 -->|HTTPS| C2
        C2 --> C3
        C2 --> C4
    end

    style A1 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style A2 fill:#FFB3C6,stroke:#C76B7F,color:#333
    style A3 fill:#E8D5F5,stroke:#9B7FBF,color:#333
    style B1 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style B2 fill:#FFDAB9,stroke:#D4945F,color:#333
    style B3 fill:#FFF9C4,stroke:#D4B95F,color:#333
    style C1 fill:#C7CEEA,stroke:#7B8AB8,color:#333
    style C2 fill:#B5EAD7,stroke:#6BB59A,color:#333
    style C3 fill:#FFDAB9,stroke:#D4945F,color:#333
    style C4 fill:#E8D5F5,stroke:#9B7FBF,color:#333

4.3 关键设计决策对比

设计决策OmniRoute 选择LiteLLM 选择Portkey 选择
抽象核心domain/ 纯 TS 函数层(可单测)litellm.Router Python classExpress middleware chain
Fallback 粒度Provider + 账户 + Stage 3 层Model 级别Model + 账户 2 层
降级哲学“永远不抛错,返回 degraded result”“重试 N 次,最后报错”“重试 + 50 guardrails 拦截”
压缩策略RTK + Caveman + Judge Model(LLM 评估)无内置基础 prompt cache
Pipeline4 阶段可声明无无
协议翻译OpenAI ↔ Anthropic ↔ Gemini 全双向OpenAI 兼容为主OpenAI 兼容为主
Hook/Event中间件链(src/middleware/*)CustomLogger + Callbacks中间件 + Hooks
路由策略4 种(priority/round-robin/random/least-used)+ 19 种细分8 种(简单/优先级/最低延迟/成本优化/…)6 种
特色“永远降级 + 永远有兜底值”哲学“Rust 核心,OpenAI 兼容,极致性能”“企业级 Guardrails,合规审计”

4.4 选哪个?(决策表)

你的场景推荐理由
个人/小团队 + 成本敏感 + 想跑多 Provider 自动化OmniRoute~1.53B 免费 tokens/月聚合 + Pipeline 自动省钱
已有 Python stack + 想要 OpenAI 兼容 + 性能优先LiteLLMRust 核心,<1ms 延迟,Python SDK 友好
企业 + 合规要求 + 50+ Guardrails + SSOPortkey唯一成熟的企业级 Guardrails 链
需要 MCP Server / A2A ProtocolOmniRoute唯一在 Gateway 内部原生集成 MCP Server + A2A
需要 Pipeline 多阶段(plan/execute/reflect/fix)OmniRoute唯一有 Pipeline 概念的 Gateway
需要 Desktop 客户端OmniRoute唯一提供 Electron + PWA
需要压测 / 路由算法评测LiteLLMrouter-eval/compare.ts 有完整评测框架

五、优缺点分析

5.1 架构对比表

维度优势 ✅劣势 ❌
架构简洁性控制平面 / 数据平面 / 共享内核三层清晰;domain/ 是纯 TS,无副作用,可单测11,235 个文件总规模偏大;open-sse/ 子包 1371 文件理解成本高
扩展性Combo 策略可热加载;新 Provider 加一个 executor 即可(无需改 core)Pipeline 的 Stage 数量目前硬编码(plan/execute/reflect/fix),加 stage 需改源码
易用性桌面应用 + Dashboard 全套,CLI omniroute 一键启动TypeScript-only 对 Python stack 团队不友好;配置项多(290+ Provider 都有 settings)
性能RTK+Caveman 平均节省 89% tokens;Rust 边缘加速层相比 LiteLLM 的纯 Rust 核心,主路径还是 Next.js,有 ~5-10ms overhead
复杂度19 种路由策略 + 4 级降级 + Pipeline + Combo,功能矩阵完整复杂度高,新手上手需要理解 5+ 概念(domain/pipeline/combo/fallback/degradation)
维护性MIT 开源,500+ Contributor,提交活跃(2026-08-07 还在 commit)文档相对 README 来说偏少;docs/ 1460 文件需要慢慢看

5.2 与设计哲学的契合度

OmniRoute 在 Harness Engineering 的几个核心原则上表现如何?

原则OmniRoute 表现评分
机制 vs 策略分离⭐⭐⭐⭐⭐ domain/ 完全纯函数,所有副作用委托给 executor优秀
模型无关性⭐⭐⭐⭐⭐ Pipeline 只声明 fitness tier,不绑死 model优秀
可拆卸性⭐⭐⭐⭐ Pipeline / Combo / Fallback 都可独立替换良好
面向进化⭐⭐⭐⭐ RTK+Caveman 让 Agent 自动省 token,符合 Bitter Lesson良好
Less is More⚠️ 19 种路由策略 + 4 级降级 + 4 阶段 Pipeline —— 功能丰富但可能过度工程中等

Bitter Lesson 检查:OmniRoute 的 Pipeline Engine 是”机制”(让 LLM 自己决定 plan/execute/reflect),而非”策略”(写死”代码任务必须先 plan 再 execute”)。这符合 Bitter Lesson —— 让 LLM 学会调度,而不是我们写死调度。但 RTK + Caveman 的”缩写词典”那一部分有点像”聪明但会被淘汰的代码”:未来更好的做法是让 LLM 自己学会压缩 prompt,而不是维护一个静态缩写表。

六、从零搭建启示(我可以怎么用 OmniRoute 思想?)

6.1 最小可行实现(MVP)

如果我自己复刻 OmniRoute 的核心思想,最小可行实现只需要 3 个文件 + 200 行代码:

1
2
3
4
my-mini-gateway/
├── pipeline.ts        # Pipeline Engine(4 阶段,~80 行)
├── degradation.ts     # withDegradation 框架(~50 行)
└── combo_resolver.ts  # 4 种路由策略(~70 行)

不要做的(避免过度工程):

  • ❌ 不要先做 Dashboard(用 print 调试就够了)
  • ❌ 不要做 290+ Provider 适配(只支持 OpenAI + Anthropic + 本地 Ollama 就够)
  • ❌ 不要做 MCP Server(等你真需要再加)
  • ❌ 不要做压缩(RTK/Caveman 等你跑出数据再做)

6.2 必须的 3 个组件

  1. pipeline.ts — 多阶段流水线(即使 MVP 阶段)

    • 不要硬编码”4 阶段”,让调用方传 stages: [{name, fitness}, ...]
    • fitness tier 至少支持 3 档(best/moderate/cheapest)
    • reflect stage 必须输出 JSON,trigger fix 决策靠 LLM 输出而非硬编码

  2. with_degradation() — 三级降级(primary/fallback/default)

    • 永远返回 DegradedResult,不抛错
    • 全局 registry 记录所有降级事件
    • default 必须是”业务可接受的兜底”(比如 rate-limit 默认 allow)

  3. combo_resolver() — 至少 2 种策略(priority + round-robin)

    • 优先级是 fallback 的基础
    • round-robin 是多账户负载均衡的基础
    • 不要做 least-used(random + priority 覆盖 80% 场景)

6.3 集成 Coding Harness 时的踩坑预警

坑现象解决
Streaming 中途断流Harness 在等 SSE chunk,Gateway 中途切了 Provider,客户端收到不完整响应保留 streaming session ID,fallback 用同一个 ID 续接
Tool call 格式差异Claude Code 发的 tool_use,Anthropic Provider 能解析,OpenAI Provider 不能在 open-sse/translator/ 加双向转换层,OmniRoute 已经做好了直接用
Token 计数偏差Pipeline 算”plan 阶段用 Opus 省了钱”,但 reflect 阶段又把上下文加回来每个 Stage 独立算 cost,最后汇总;别用”input = 上一阶段 output”做估算
Cache 失效OmniRoute 的 compression cache 用 prompt hash 做 key,改一个字符全失效加版本号前缀 + 模糊匹配;或者干脆不缓存(89% 是它的卖点,别省错了)
Dashboard 实时性5 个 Provider × 20 个模型 × 100 用户 = 10000 个指标,Dashboard 卡用 Redis pub/sub 聚合,前端 WebSocket;OmniRoute 用 hooks/useLiveDashboard.ts 实现

6.4 何时不要自己造

你的场景建议
单一 Provider(只 OpenAI 或只用 Claude)不要 Gateway,直接 SDK
< 10 个并发 AgentLiteLLM 一行命令就够,别上 OmniRoute
团队 < 3 人用 Portkey SaaS,别自托管(运维成本高)
需要审计 / 合规 / SSOPortkey 企业版,别自己写(合规坑太多)
想做”AI Gateway for AI Gateway”可以,但要明确 OmniRoute 的 290 Provider 已经是上限,差异化在垂直场景(比如医疗/法律)

七、总结:Harness 数据平面的未来

7.1 OmniRoute 给我们的 3 个核心启示

  1. 数据平面与控制平面必须分离:OmniRoute 把 LLM 调用抽象成可观测、可降级、可审计的”网络包”,这是 Harness Engineering 里”包裹在 LLM 外面的软件外壳”在基础设施层的具体表达。没有数据平面的 Harness,无法在生产稳定运行。
  2. 永远不抛错的降级哲学:传统”重试 3 次”是”假设网络会好”的乐观策略;withDegradation 是”假设外部依赖会坏”的现实策略。生产环境的 Harness 必须接受”任何依赖都可能挂”,然后设计 graceful fallback。
  3. Pipeline + Fitness Tier 的成本优化:”让 LLM 帮 LLM 决策”(reflect stage) + “按能力档选模型”(fitness tier) 是机制层面的优化,而不是写死 if-else。这种设计未来不会过时 —— 即使模型能力变化,Pipeline 的”plan → execute → reflect → fix”模式仍然适用。

7.2 我的行动建议

你是谁建议
个人开发者用 LiteLLM 的 litellm --model claude-sonnet-4-6 --temperature 0.5 --help,3 行命令跑起来
小团队(3-10 人)OmniRoute 桌面应用 + 自托管,免费 tokens 池足够小团队用
中大型团队Portkey 企业版 + LiteLLM 自托管双轨,关键业务用 Portkey,实验性业务用 LiteLLM
正在自研 Coding Harness直接用 OmniRoute 作为 Provider 适配层,不要自己写 290 Provider 的 SDK
正在设计 Agent 协议参考 OmniRoute 的 open-sse/translator/ 设计:每个 Provider 一层,中间用统一 IR
关注成本OmniRoute 的 RTK + Caveman + Pipeline 组合,平均能给 Coding Agent 团队省 60-80% tokens

7.3 一句话总结

OmniRoute 不是最快的 AI Gateway(LiteLLM 更快)、不是最企业级的(Portkey 更成熟)、不是最简单的,但它是唯一一个把 Pipeline Engine + Graceful Degradation + RTK/Caveman 压缩 + MCP/A2A 桥接 + 290 Provider 全装进一个 npm 包的 AI Gateway。对 Harness Engineering 来说,这种”数据平面 + 控制平面 + 共享内核”的一体化设计,是未来 LLM 中间件层的标准形态。

参考资料

  1. OmniRoute 源码:

  2. 对比项目:

  3. 概念参考:

  4. 本文同系列:

    • Harness 6 件套组件专题(2026-06~07 共 6 篇,Rule/Skill/Sub-Agent/Workflow/Script/MCP)
    • 项目横向对比专题(2026-07~08 多篇,Helicone/Claude Code Router/Strands/gastown 等)