王中阳

主题

如何设计一个可替换模型供应商的 LLM Client？ ​

可替换 LLM Client 的核心不是把 URL 放进配置，而是用稳定的领域接口隔离消息、错误、流事件和能力差异。业务层只表达“生成什么”，Provider Adapter 负责“如何调用”，路由层再根据能力、健康度和预算选择实现。

先说结论 ​

一个可靠的 LLM Client 应拆成三层：

  1. 领域接口：定义业务真正需要的输入、输出、流事件和错误；
  2. Provider Adapter：转换供应商协议，不泄漏专有字段；
  3. Router/Policy：做模型选择、超时预算、重试、降级和观测。

“兼容某种请求格式”不等于可以无损替换。工具调用、结构化输出、停止原因、Token 统计和错误语义都可能不同；能力不一致时必须显式拒绝或降级，不能悄悄忽略参数。

适合谁，不适合谁 ​

本文适合需要接入多个模型、私有部署模型或预留迁移能力的 TypeScript 后端开发者。

如果系统只有一个短期验证页面、没有稳定业务接口，也没有切换需求，可以先使用简单封装，不必提前建设复杂路由平台。本文示例是接口设计，不对应任何供应商的特定 API、版本或价格。

先定义最小领域模型 ​

不要把供应商 SDK 类型传遍代码库。先保留业务稳定需要的字段：

ts
type Role = 'system' | 'user' | 'assistant'

interface Message {
  role: Role
  content: string
}

interface GenerateRequest {
  messages: Message[]
  maxOutputTokens: number
  temperature?: number
  responseSchema?: unknown
  signal: AbortSignal
}

interface Usage {
  inputTokens?: number
  outputTokens?: number
}

interface GenerateResult {
  text: string
  finishReason: 'stop' | 'length' | 'tool' | 'blocked' | 'unknown'
  usage: Usage
  providerRequestId?: string
}

type StreamEvent =
  | { type: 'delta'; text: string }
  | { type: 'usage'; usage: Usage }
  | { type: 'done'; finishReason: GenerateResult['finishReason'] }

interface ModelClient {
  generate(request: GenerateRequest): Promise<GenerateResult>
  stream(request: GenerateRequest): AsyncIterable<StreamEvent>
}

usage 字段允许缺失，是因为并非所有部署都会返回同样的统计。缺失时应标记为“未知”，不能填零；零会污染成本报表。

Adapter 如何实现 ​

Adapter 只做协议转换和错误归一化，不在这里写业务 Prompt：

ts
class ProviderError extends Error {
  constructor(
    message: string,
    readonly kind:
      | 'invalid_request'
      | 'rate_limited'
      | 'timeout'
      | 'unavailable'
      | 'content_blocked'
      | 'unknown',
    readonly retryable: boolean,
  ) {
    super(message)
  }
}

class OpenAICompatibleAdapter implements ModelClient {
  constructor(
    private readonly endpoint: string,
    private readonly apiKey: string,
  ) {}

  async generate(request: GenerateRequest): Promise<GenerateResult> {
    let response: Response
    try {
      response = await fetch(this.endpoint, {
        method: 'POST',
        headers: {
          authorization: `Bearer ${this.apiKey}`,
          'content-type': 'application/json',
        },
        body: JSON.stringify(this.toProviderRequest(request)),
        signal: request.signal,
      })
    } catch (error) {
      if (error instanceof Error && error.name === 'AbortError') {
        throw new ProviderError('provider request timed out', 'timeout', true)
      }
      throw new ProviderError('provider network request failed', 'unavailable', true)
    }

    if (!response.ok) throw await this.toProviderError(response)
    const raw: unknown = await response.json()
    return this.normalize(raw)
  }

  async *stream(_request: GenerateRequest): AsyncIterable<StreamEvent> {
    throw new ProviderError('streaming is not supported', 'invalid_request', false)
  }

  private async toProviderError(response: Response): Promise<ProviderError> {
    if (response.status === 429) {
      return new ProviderError('provider rate limited', 'rate_limited', true)
    }
    if (response.status === 408) {
      return new ProviderError('provider request timed out', 'timeout', true)
    }
    if (response.status >= 500) {
      return new ProviderError('provider unavailable', 'unavailable', true)
    }
    return new ProviderError('provider rejected request', 'invalid_request', false)
  }

  private toProviderRequest(request: GenerateRequest): Record<string, unknown> {
    const body: Record<string, unknown> = {
      messages: request.messages,
      max_output_tokens: request.maxOutputTokens,
      temperature: request.temperature,
    }
    if (request.responseSchema) {
      body.response_format = {
        type: 'json_schema',
        json_schema: request.responseSchema,
      }
    }
    return body
  }

  private normalize(_raw: unknown): GenerateResult {
    // 在真实 Adapter 中做运行时校验后再返回领域对象。
    throw new Error('implement provider-specific validation')
  }
}

示例使用 OpenAI-compatible 请求字段，因此显式把领域层的 responseSchema 映射为该协议的 response_format。接入其他供应商时必须在对应 Adapter 中改写这一步；若供应商不支持 Schema，应在能力检查阶段拒绝请求，不能静默丢弃。normalize 同样必须按所接供应商的文档实现，并用固定样例测试；未经校验的类型断言会把协议变化推迟到业务层爆炸。

抽象层决策表 ​

决策推荐做法原因边界
模型名称使用内部 model alias避免业务绑定供应商命名alias 变更要有发布记录
错误归一化为少量领域错误便于统一重试与告警保留原始状态码供诊断
能力建立 capability 声明路由前即可拒绝不支持请求不用“兼容”掩盖语义差异
流式统一为领域事件隔离 SSE/分块协议差异必须保留结束和错误事件
Token 用量使用可空字段不伪造缺失数据账单核对仍以供应商记录为准
专有参数放入受控扩展区或独立接口保持主接口稳定核心业务不应依赖任意透传

路由、超时、重试与降级 ​

路由先检查能力，再考虑策略：

ts
interface Candidate {
  name: string
  client: ModelClient
  supportsSchema: boolean
  supportsStreaming: boolean
}

function choose(
  candidates: Candidate[],
  request: GenerateRequest,
  needsStreaming: boolean,
): Candidate {
  const matched = candidates.filter(candidate =>
    (!request.responseSchema || candidate.supportsSchema) &&
    (!needsStreaming || candidate.supportsStreaming),
  )
  if (matched.length === 0) throw new Error('no model satisfies required capabilities')
  return matched[0]
}
  • 超时：路由层分配总预算，Adapter 接收同一个取消信号；切换 Provider 不能重新获得完整预算。
  • 重试：仅重试标记为 retryable 的错误，并限制总尝试次数；无效请求、内容拦截和能力缺失不重试。
  • 降级：备用模型必须满足任务的最低能力。结构化输出任务不能降级到不支持约束且无校验补偿的路径。
  • 观测：记录内部模型别名、实际 Provider、尝试序号、分阶段耗时、错误类别、结束原因和用量；密钥与完整敏感正文不得进入日志。

自动切换还要考虑幂等性：纯文本生成通常可重新请求，但“生成后立即调用工具”不是纯读取。应把生成与副作用提交拆开，并用业务幂等键保护。

常见误区与失败边界 ​

  1. 最低公分母接口：为了统一而删除工具调用、Schema 等关键能力，最终业务绕过抽象层。
  2. 任意参数透传：业务代码仍充满供应商字段，名义上抽象、实际上耦合。
  3. 遇错就换模型：输入错误会在每个模型上重复失败；输出风格变化也可能破坏用户体验。
  4. 隐藏模型切换：不记录实际路由，无法复现线上结果。
  5. 把 SDK 当领域层：SDK 升级会迫使大量业务代码同时修改。

Provider 抽象只能降低协议和迁移耦合，不能保证不同模型输出语义完全一致。迁移或新增路由时仍需运行评估集。

可执行检查清单 ​

  • [ ] 业务代码只依赖内部 ModelClient 和领域类型
  • [ ] 每个 Adapter 都有请求映射、响应校验和错误映射测试样例
  • [ ] 能力缺失会显式失败，不会静默忽略
  • [ ] 总超时能跨路由和重试传播
  • [ ] 重试只覆盖可恢复错误，并有次数与时间预算
  • [ ] 备用模型通过同一任务评估集
  • [ ] 生成与工具副作用分离，并使用幂等键
  • [ ] 日志能还原实际路由，但不泄漏密钥和敏感正文
  • [ ] 用量缺失记为未知，不记为零

下一步 ​

从 AI 工程化专题 可以查看该抽象在完整工程链路中的位置。Client 统一后，继续建设 结构化输出可靠性，用运行时校验和最小评估集验证模型替换不会破坏业务契约。

AI 应用开发训练营 · 现在开始

别只收藏 AI 文章，今天就做出第一个能上线的项目

文章解决认知，项目才证明能力。把 Agent、RAG、MCP 变成可运行、可展示、可面试表达的成果，现在就从一次岗位准备度自测开始。

立即开始 AI 岗位准备度自测 查看训练营路线
真实学员成果先看成果，再决定是否开始 →
  1. 01选方向
  2. 02做项目
  3. 03出成果
  4. 04拿面试