多模型路由

要点

一个 AI 应用往往不会只绑定一个模型供应商——OpenAI、Claude、Gemini、本地模型各有各的适用场景
多模型路由解决的是「同一个 API 入口，按规则把请求分发到不同模型」这个问题
模型注册表是所有路由策略的基础——把模型名称、能力、端点、密钥集中到一个数据结构里
路由策略至少三种：按模型名称直接映射、按任务类型自动选择、按负载权重分配
故障转移让系统在某个供应商不可用时自动切换到备用模型，业务代码无需感知
负载均衡可以分散到多个 API Key 或多个端点，降低单个 key 的限流压力
模型能力查询接口让前端动态获取可用模型列表，不用硬编码在前端

1. 从一个具体的问题开始

你的 AI 应用需要同时接 OpenAI、Claude、Gemini，还有一套本地部署的开源模型。业务层代码会怎么写？

最开始的做法，可能是每个路由各写各的：

// src/routes/summary.ts
app.post('/api/summary', async (c) => {
  const res = await fetch('https://api.openai.com/v1/chat/completions', {
    headers: { 'Authorization': `Bearer ${c.env.OPENAI_API_KEY}` },
    body: JSON.stringify({ model: 'gpt-4o', messages: [...] }),
  })
  return c.json(await res.json())
})
 
// src/routes/translate.ts
app.post('/api/translate', async (c) => {
  const res = await fetch('https://api.anthropic.com/v1/messages', {
    headers: { 'x-api-key': c.env.ANTHROPIC_API_KEY },
    body: JSON.stringify({ model: 'claude-sonnet-4-20250514', messages: [...] }),
  })
  return c.json(await res.json())
})

这样做能跑，但问题很快会出现：

产品经理说「摘要也试试 Claude」——得改摘要路由的代码
运营说「本地部署了 Llama 3，便宜的任务走本地」——得在每个路由里加判断
OpenAI 挂了，用户全部报错——没有备用方案

共同点是：模型选择和业务逻辑绑死了。把模型配置从代码里抽出来，放到一个「注册表」里，才是正解：

// 注册表集中管理所有模型
const models = [
  { id: 'smart-chat', provider: 'openai', task: 'summary' },
  { id: 'smart-chat', provider: 'anthropic', task: 'translate' },
  { id: 'fast-chat', provider: 'local', task: 'simple-qa' },
]

新增模型、切换模型、调整路由规则，都只改注册表，不动业务代码。

2. 什么是多模型路由

多模型路由要解决的核心问题：同一个 API 入口，请求应该交给哪个模型处理？

HTTP 路由是根据 URL 路径把请求分发到处理函数。多模型路由类似，分发的目标从处理函数变成了模型供应商的 API。它通常包含这几层：

模型注册表：所有可用模型的信息——名称、供应商、能力、端点、密钥引用
路由策略：根据什么规则选择模型——模型名称、任务类型、负载情况
动态分发：选定模型后，构造对应供应商格式的请求并发出
故障转移：目标模型不可用时，自动切换到备用模型
负载均衡：多个 key 或多个端点之间分配请求

3. 为什么需要多模型路由

不同模型有不同的特点：

模型	擅长	短板
GPT-4o	通用能力强、工具调用稳定	价格较高
Claude Sonnet	长文本理解、指令跟随	function calling 格式不同
Gemini	上下文窗口大（100 万 token）	部分地区访问受限
本地 Llama 3	免费、数据不出服务器	能力弱于商业模型

一个实际的应用会根据任务特点选择模型：

简单的 FAQ 问答 → 用便宜快速的模型（GPT-4o-mini 或本地模型）
复杂的内容审核 → 用理解力强的模型（Claude Sonnet）
长文本摘要 → 用上下文窗口大的模型（Gemini）
需要 function calling → 用工具调用支持好的模型（GPT-4o）

如果把这些逻辑分散在各个业务模块里，每次新增任务类型或调整模型选择都要改业务代码。多模型路由把这些决策集中到一层，业务代码只需要声明「我要做什么」，不用关心「谁来提供」。

4. 模型注册表设计

注册表是所有路由策略的基础。先看类型定义：

// src/types/model.ts
export type ModelProvider = 'openai' | 'anthropic' | 'google' | 'local'
 
export type ModelCapability = {
  maxContextTokens: number
  supportsStreaming: boolean
  supportsFunctionCalling: boolean
  supportsVision: boolean
}
 
export type ModelEntry = {
  id: string              // 对外暴露的模型 ID
  provider: ModelProvider
  providerModelId: string // 供应商侧的模型名称
  capability: ModelCapability
  priority: number        // 路由优先级，数字越大越优先
  enabled: boolean
  apiKeyEnv: string       // API Key 对应的环境变量名
  endpoint?: string       // 自定义端点（本地模型或代理用）
}

几个设计选择：

id 和 providerModelId 分开：对外暴露 smart-chat，实际路由到 gpt-4o。以后换成 Claude 只改注册表，客户端不用动
priority 用于故障转移和任务路由：同类模型按优先级排序，主模型挂了自动切到下一个
apiKeyEnv 指向环境变量名，不是直接存 key：不同环境绑不同的 key，注册表不用改

一份注册表实例：

// src/registry/model-registry.ts
export const MODEL_REGISTRY: ModelEntry[] = [
  {
    id: 'smart-chat',
    provider: 'openai',
    providerModelId: 'gpt-4o',
    capability: { maxContextTokens: 128000, supportsStreaming: true, supportsFunctionCalling: true, supportsVision: true },
    priority: 10,
    enabled: true,
    apiKeyEnv: 'OPENAI_API_KEY',
  },
  {
    id: 'smart-chat',
    provider: 'anthropic',
    providerModelId: 'claude-sonnet-4-20250514',
    capability: { maxContextTokens: 200000, supportsStreaming: true, supportsFunctionCalling: true, supportsVision: true },
    priority: 8,
    enabled: true,
    apiKeyEnv: 'ANTHROPIC_API_KEY',
  },
  {
    id: 'long-context',
    provider: 'google',
    providerModelId: 'gemini-1.5-pro',
    capability: { maxContextTokens: 1000000, supportsStreaming: true, supportsFunctionCalling: true, supportsVision: false },
    priority: 10,
    enabled: true,
    apiKeyEnv: 'GOOGLE_API_KEY',
  },
  {
    id: 'fast-chat',
    provider: 'local',
    providerModelId: 'llama-3-70b',
    capability: { maxContextTokens: 8192, supportsStreaming: true, supportsFunctionCalling: false, supportsVision: false },
    priority: 10,
    enabled: true,
    apiKeyEnv: 'LOCAL_MODEL_API_KEY',
    endpoint: 'http://internal-model-server:8080',
  },
]

同一个 id 对应多条记录——smart-chat 有 OpenAI 和 Claude 两个备选，路由器可以按优先级或负载均衡来选择。

5. 路由策略

注册表解决了「有哪些模型」，路由策略解决「用哪个模型」。

策略一：按模型名称直接映射

客户端在请求里指定 model 字段，路由器查注册表找到对应的供应商和端点。这是最简单的策略，也是 OpenAI 兼容 API 的标准做法。

// 客户端请求
{ "model": "smart-chat", "messages": [...] }
// 路由器：查注册表 -> 找 id = "smart-chat" -> 取 priority 最高的 -> 构造请求

策略二：按任务类型自动选择

客户端不指定模型，只告诉服务端任务类型，路由器根据配置自动分配：

// src/registry/task-routing.ts
export const TASK_MODEL_MAP: Record<string, string> = {
  'chat': 'smart-chat',
  'summary': 'smart-chat',
  'translate': 'smart-chat',
  'long-doc': 'long-context',
  'simple-qa': 'fast-chat',
}
 
// 任务类型到备选链的映射（用于故障转移）
export const TASK_FAILOVER_CHAIN: Record<string, ModelProvider[]> = {
  'chat': ['openai', 'anthropic', 'local'],
  'summary': ['anthropic', 'openai'],
  'long-doc': ['google', 'anthropic'],
  'simple-qa': ['local', 'openai'],
}

策略三：按负载权重分配

同一个任务有多个可用的模型实例，按权重分配请求。适合分散限流压力或做 A/B 测试：

// src/types/routing.ts
export type RoutingStrategy = 'model-name' | 'task-type' | 'weighted'
export type WeightedEntry = { modelId: string; weight: number }

三种策略不互斥。实际项目里通常组合使用：先用任务类型确定模型 ID，再用权重分配具体实例，最后用优先级做故障转移。

6. 路由器实现

把注册表和路由策略拼到一起：

// src/router/model-router.ts
import { MODEL_REGISTRY } from '../registry/model-registry'
import { TASK_MODEL_MAP } from '../registry/task-routing'
import type { ModelEntry } from '../types/model'
 
export class ModelRouter {
  private registry = MODEL_REGISTRY
  private isHealthy: (provider: string) => boolean
 
  constructor(isProviderHealthy?: (provider: string) => boolean) {
    this.isHealthy = isProviderHealthy ?? (() => true)
  }
 
  // 按模型 ID 查找，返回优先级最高且健康的条目
  resolveById(modelId: string): ModelEntry | null {
    const candidates = this.registry
      .filter((m) => m.id === modelId && m.enabled)
      .filter((m) => this.isHealthy(m.provider))
      .sort((a, b) => b.priority - a.priority)
    return candidates[0] ?? null
  }
 
  // 按任务类型查找
  resolveByTask(taskType: string): ModelEntry | null {
    const modelId = TASK_MODEL_MAP[taskType]
    if (!modelId) return null
    return this.resolveById(modelId)
  }
 
  // 统一入口
  resolve(request: { model?: string; taskType?: string }): ModelEntry | null {
    if (request.taskType) return this.resolveByTask(request.taskType)
    if (request.model) return this.resolveById(request.model)
    return this.resolveByTask('chat')
  }
 
  // 获取故障转移链：同一个 modelId 下，按优先级排序的所有备选
  getFailoverChain(modelId: string): ModelEntry[] {
    return this.registry
      .filter((m) => m.id === modelId && m.enabled)
      .filter((m) => this.isHealthy(m.provider))
      .sort((a, b) => b.priority - a.priority)
  }
}

resolve() 是核心方法。它接收请求中的 model 或 taskType，返回一个 ModelEntry。路由器不关心请求内容，只负责选模型。

7. 供应商适配层

不同供应商的 API 格式不一样。适配器把这些差异抹平：

// src/adapters/provider-adapter.ts
import type { ModelEntry } from '../types/model'
 
export class ProviderAdapter {
  async send(model: ModelEntry, body: Record<string, unknown>, apiKey: string) {
    const urlMap: Record<string, string> = {
      openai: 'https://api.openai.com/v1/chat/completions',
      anthropic: 'https://api.anthropic.com/v1/messages',
      google: `https://generativelanguage.googleapis.com/v1beta/models/${model.providerModelId}:generateContent`,
      local: model.endpoint ?? 'http://localhost:8080/v1/chat/completions',
    }
    const headersMap: Record<string, Record<string, string>> = {
      openai: { 'Authorization': `Bearer ${apiKey}`, 'Content-Type': 'application/json' },
      anthropic: { 'x-api-key': apiKey, 'Content-Type': 'application/json', 'anthropic-version': '2023-06-01' },
      google: { 'Content-Type': 'application/json' },
      local: { 'Content-Type': 'application/json' },
    }
 
    const requestBody = this.transformRequest(model.provider, body, model.providerModelId)
    const res = await fetch(urlMap[model.provider], {
      method: 'POST',
      headers: headersMap[model.provider],
      body: JSON.stringify(requestBody),
    })
    if (!res.ok) throw new Error(`Provider ${model.provider} returned ${res.status}`)
    return res
  }
 
  private transformRequest(provider: string, body: Record<string, unknown>, modelId: string) {
    switch (provider) {
      case 'anthropic':
        return { model: modelId, max_tokens: 4096, messages: body.messages }
      case 'google':
        return { contents: body.messages } // 简化：实际需要格式转换
      default:
        return { model: modelId, messages: body.messages, stream: body.stream ?? false }
    }
  }
}

适配器的职责很单一：接收统一的参数，输出供应商能理解的请求。它不关心路由逻辑，也不关心故障转移。

8. 故障转移

供应商不可能永远在线。故障转移的思路：主模型请求失败时，自动尝试备选模型。

ModelRouter.getFailoverChain() 已经给出了按优先级排序的备选链，故障转移逻辑依次尝试即可：

// src/router/failover.ts
import type { ModelEntry } from '../types/model'
import { ProviderAdapter } from '../adapters/provider-adapter'
 
export class FailoverHandler {
  private adapter: ProviderAdapter
  constructor(adapter: ProviderAdapter) {
    this.adapter = adapter
  }
 
  async sendWithFailover(
    chain: ModelEntry[],
    body: Record<string, unknown>,
    env: Record<string, string>,
  ): Promise<Response> {
    const errors: Array<{ provider: string; error: string }> = []
 
    for (const model of chain) {
      try {
        const apiKey = env[model.apiKeyEnv] ?? ''
        return await this.adapter.send(model, body, apiKey)
      } catch (err) {
        errors.push({ provider: model.provider, error: err instanceof Error ? err.message : String(err) })
        console.warn(`[failover] ${model.provider} failed, trying next`, err)
      }
    }
 
    throw new Error(`All providers failed: ${JSON.stringify(errors)}`)
  }
}

几个实现细节：

按优先级顺序尝试：chain 已经按 priority 降序排列
记录每次失败的原因：全部失败时，错误信息包含每个供应商的具体报错
不做无限重试：备选链有多长就试多少次，避免雪崩式请求堆积
健康检查前置：getFailoverChain() 已过滤掉已知不健康的供应商

9. 负载均衡

故障转移解决「某个供应商挂了怎么办」，负载均衡解决「多个可用的 key 或端点之间怎么分配请求」。

最常见场景：同一个供应商有多个 API Key，每个 key 有独立的限流额度，轮询分配可以最大化总吞吐量。

// src/router/load-balancer.ts
export class LoadBalancer {
  private counters: Map<string, number> = new Map()
 
  roundRobin<T>(key: string, candidates: T[]): T {
    if (candidates.length === 0) throw new Error('No candidates available')
    const current = this.counters.get(key) ?? 0
    const selected = candidates[current % candidates.length]
    this.counters.set(key, current + 1)
    return selected
  }
 
  weighted<T>(key: string, candidates: Array<{ item: T; weight: number }>): T {
    const totalWeight = candidates.reduce((sum, c) => sum + c.weight, 0)
    const current = this.counters.get(key) ?? 0
    let accumulated = 0
    const normalized = current % totalWeight
    for (const candidate of candidates) {
      accumulated += candidate.weight
      if (normalized < accumulated) {
        this.counters.set(key, current + 1)
        return candidate.item
      }
    }
    return candidates[candidates.length - 1].item
  }
}

在 Workers 环境里，计数器是内存态的，随实例回收而重置。如果需要跨实例的精确负载均衡，要把计数器存到 KV 或 D1 里。对于大多数场景，内存级的轮询已经够用。

10. 模型能力查询

前端需要知道有哪些模型可用。与其硬编码，不如提供查询接口：

// src/routes/models.ts
import { Hono } from 'hono'
import { MODEL_REGISTRY } from '../registry/model-registry'
import type { AppEnv } from '../types'
 
export const modelsApp = new Hono<AppEnv>()
 
modelsApp.get('/', (c) => {
  // 去重：同一个 id 的多条记录合并为一条，取优先级最高的
  const modelMap = new Map<string, (typeof MODEL_REGISTRY)[0]>()
  for (const entry of MODEL_REGISTRY) {
    if (!entry.enabled) continue
    const existing = modelMap.get(entry.id)
    if (!existing || entry.priority > existing.priority) {
      modelMap.set(entry.id, entry)
    }
  }
 
  const models = Array.from(modelMap.values()).map((m) => ({
    id: m.id,
    provider: m.provider,
    capability: m.capability,
  }))
 
  return c.json({ object: 'list', data: models })
})

前端拿到后可以：根据 capability.supportsVision 决定是否显示上传图片按钮；根据 capability.maxContextTokens 做输入长度提示；后端新增模型后，前端自动获取，不用发版。

11. 串起来

把路由器、适配器、故障转移组合到一条路由里：

// src/routes/chat.ts
import { Hono } from 'hono'
import type { AppEnv } from '../types'
import { ModelRouter } from '../router/model-router'
import { ProviderAdapter } from '../adapters/provider-adapter'
import { FailoverHandler } from '../router/failover'
 
export const chatApp = new Hono<AppEnv>()
 
const router = new ModelRouter()
const adapter = new ProviderAdapter()
const failover = new FailoverHandler(adapter)
 
chatApp.post('/v1/chat/completions', async (c) => {
  const body = await c.req.json()
  const { model: requestedModel, taskType, messages, stream } = body
 
  // 1. 路由：选择模型
  const selected = router.resolve({ model: requestedModel, taskType })
  if (!selected) return c.json({ error: 'No available model found' }, 400)
 
  // 2. 获取故障转移链并发送请求
  const chain = router.getFailoverChain(selected.id)
  try {
    const res = await failover.sendWithFailover(chain, { messages, stream }, c.env)
    return c.json(await res.json())
  } catch (err) {
    console.error('[chat] all providers failed:', err)
    return c.json({ error: 'All models temporarily unavailable' }, 503)
  }
})

客户端的请求方式不变——还是发 { "model": "smart-chat", "messages": [...] }。背后的路由、适配、故障转移全部在服务端完成。

总结

这篇介绍了多模型路由的设计思路：

模型注册表集中管理所有模型的信息，包括供应商、能力、优先级、端点
路由策略支持三种模式：按模型名称映射、按任务类型选择、按权重分配
路由器结合注册表和策略，对外暴露一个 resolve() 方法
供应商适配层抹平不同 API 格式的差异，提供统一的调用接口
故障转移在主模型失败时自动尝试备选链上的下一个模型
负载均衡在多个 key 或端点之间分配请求，降低单个 key 的限流压力
模型查询接口让前端动态获取可用模型，不用硬编码