构建多模型 AI 网关的错误分类体系

大多数 AI 网关故障并非无迹可寻。它们之所以代价高昂，是因为每个提供商对它们的命名各异、各 SDK 重试逻辑不同，而且每个团队总是在故障发生后才去争论，而不是在分派之前就先做好分类。

一个跨 DeepSeek、Qwen、GLM、Kimi 及其它 OpenAI 兼容接口进行路由的多模型网关，需要的不仅仅是"稍后重试"。它需要一套共享的错误分类体系，用以告知 SDK：是该重试、该切换路由、该截断输出、该请求用户操作，还是该立即停止。它还需要附加带日期的定价元数据，因为重试策略本质上也是成本策略。

本文展示如何将这套分类体系构建为一个小型生产级原语。目标不是掩盖提供商的行为，而是将其充分规范化，让工程师、支持团队和财务人员无需逐条读取原始提供商响应就能解释发生了什么。

以下所有源代码和定价核查均于 2026 年 8 月 22 日刷新。AIWave 的公开定价页面目前列出 DeepSeek V4 Flash 为每 1M tokens $0.638 输入、$1.914 输出、$0.0203 缓存命中输入；DeepSeek V4 Pro 为每 1M tokens $1.914 输入、$5.742 输出、$0.0638 缓存命中输入。AIWave 在其可预测定价页面上将这些全天候行与官方 DeepSeek 峰值和非峰值行分开，因此该主张应该是关于运营可预测性和统一路由的，而非全网最低价承诺。QwenCloud 的定价文档描述了按量计费、上下文缓存、Batch API 行为、工具调用费用和失败呼叫计费行为。Z.AI 列出了 GLM 行及其独立的缓存输入定价，Kimi K3 公开资料列出了长上下文定价，包含缓存命中和缓存未命中的输入行。

为什么错误名称还不够

如果你的网关只存储 HTTP 状态码，信息会很快丢失。429 可能意味着账户级并发、租户级限流、提供商容量，或内部工作队列的突发波动。400 可能意味着消息格式错误、不支持的工具 schema、上下文溢出，或模型不支持所请求的模式。超时可能意味着提供商延迟、流中断、用户中止请求，或本地网络路径问题。

这些情况不应该有相同的行为。有些值得快速重试。有些应该切换到另一个模型家族。有些应该停止，因为重复调用只会增加延迟并可能增加成本。有些应该成为面向用户的错误，因为请求需要被修改。

分类体系为每种故障赋予一个稳定的内部含义：

重要的是分类名称本身，而是它们数量少、稳定且可操作。

每次重试背后的三条策略

重试看起来像可靠性逻辑，但它们同时也是产品和财务逻辑。重试可以提高成功率，也可能制造重复工作、更长的等待时间和可避免的 token 消耗。因此，一套有用的分类体系需要将三条策略分开。

第一，对错误进行分类。这回答了以网关术语来说发生了什么。第二，决定请求是否可重试。这取决于幂等性、任务类型、流状态和提供商信号。第三，决定重试预算是否还允许再调用一次。该预算应使用带日期的费率卡行，而不是猜测。

例如，一个非流式的 JSON 提取调用在提供商返回 5xx 后可能安全地重试一次。而一个已经发出部分指令的流式编码助手，除非客户端能够丢弃部分响应，否则可能不安全地重放。一个工具调用规划请求可能只在策略表明对该任务类别来说质量比成本更重要时，才能从 Flash 模型安全地切换到 Pro 模型。

重试决策应返回结构化元数据：

{
  "error_class": "capacity_limit",
  "retryable": true,
  "reroute_allowed": true,
  "retry_after_ms": 1800,
  "max_attempts": 2,
  "cost_budget_usd": 0.02,
  "policy_version": "error-taxonomy-2026-08-22",
  "rate_card_source_date": "2026-08-22"
}

该元数据可以写在最终使用记录旁边。之后，当客户问为什么请求花了 11 秒或为什么使用了回退模型时，你就有答案了。

费率卡属于错误处理的一部分

不要让重试层将每个模型视为平等。对短输出 Flash 路由的重试和对长上下文路由的重试是不同的预算事件。当提供商为新鲜输入、缓存输入、输出、工具调用、上下文层级或失败呼叫行为暴露单独字段时，这一点变得尤为重要。

至少为失败的路由存储以下字段：

failed_call_policy 字段很容易被忽略，但事后很难重建。如果提供商文档说明某些失败呼叫可能影响计费、配额或工具费用，你的重试预算应在触发另一个请求之前就知道这一点。如果提供商没有暴露足够的细节，就存储 unknown 并保持重试预算保守。

Python 示例：分类、预算、重试

以下 Python 示例展示了一个紧凑的网关层。它对提供商错误进行分类，使用带日期的 AIWave DeepSeek 行估算重试成本，并返回调用方可以记录的决定。

import os
import random
import time
from dataclasses import dataclass, asdict
from typing import Literal

from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("AIWAVE_API_KEY", "YOUR_API_KEY_HERE"),
    base_url="https://aiwave.live/v1",
)

ErrorClass = Literal[
    "auth_config",
    "capability_mismatch",
    "context_overflow",
    "capacity_limit",
    "provider_unavailable",
    "stream_interrupted",
    "tool_schema_error",
    "commercial_block",
    "unknown",
]

@dataclass(frozen=True)
class RateCard:
    model: str
    input_per_m: float
    cached_input_per_m: float
    output_per_m: float
    source_url: str
    source_date: str

@dataclass(frozen=True)
class ErrorDecision:
    error_class: ErrorClass
    retryable: bool
    reroute_allowed: bool
    max_attempts: int
    retry_after_ms: int
    estimated_retry_usd: float
    policy_version: str
    reason: str

AIWAVE_FLASH = RateCard(
    model="deepseek-v4-flash",
    input_per_m=0.638,
    cached_input_per_m=0.0203,
    output_per_m=1.914,
    source_url="https://aiwave.live/pricing",
    source_date="2026-08-22",
)

def classify_error(status_code: int | None, body: str) -> ErrorClass:
    text = body.lower()
    if status_code in {401, 403}:
        return "auth_config"
    if status_code == 402 or "quota" in text or "balance" in text:
        return "commercial_block"
    if status_code == 404 or "model" in text and "not found" in text:
        return "capability_mismatch"
    if "context" in text and ("length" in text or "window" in text):
        return "context_overflow"
    if status_code == 429 or "rate limit" in text or "concurrency" in text:
        return "capacity_limit"
    if status_code and 500 <= status_code < 600:
        return "provider_unavailable"
    if "tool" in text and "schema" in text:
        return "tool_schema_error"
    return "unknown"

def estimate_retry_cost(
    row: RateCard,
    input_tokens: int,
    cached_input_tokens: int,
    output_cap: int,
) -> float:
    fresh_input_tokens = max(input_tokens - cached_input_tokens, 0)
    cost = (
        fresh_input_tokens / 1_000_000 * row.input_per_m
        + cached_input_tokens / 1_000_000 * row.cached_input_per_m
        + output_cap / 1_000_000 * row.output_per_m
    )
    return round(cost, 6)

def decide_error_policy(
    error_class: ErrorClass,
    row: RateCard,
    input_tokens: int,
    cached_input_tokens: int,
    output_cap: int,
    already_streamed: bool,
) -> ErrorDecision:
    estimated_retry_usd = estimate_retry_cost(
        row, input_tokens, cached_input_tokens, output_cap
    )

    if error_class in {"auth_config", "commercial_block", "tool_schema_error"}:
        return ErrorDecision(
            error_class, False, False, 0, 0, 0.0,
            "error-taxonomy-2026-08-22",
            "caller_or_account_action_required",
        )

    if error_class == "context_overflow":
        return ErrorDecision(
            error_class, False, True, 0, 0, 0.0,
            "error-taxonomy-2026-08-22",
            "summarize_or_route_to_approved_long_context_model",
        )

    if already_streamed:
        return ErrorDecision(
            error_class, False, False, 0, 0, estimated_retry_usd,
            "error-taxonomy-2026-08-22",
            "partial_output_already_released",
        )

    if error_class in {"capacity_limit", "provider_unavailable"}:
        return ErrorDecision(
            error_class, estimated_retry_usd <= 0.02, True, 2,
            random.randint(1200, 3200), estimated_retry_usd,
            "error-taxonomy-2026-08-22",
            "bounded_retry_with_jitter",
        )

    return ErrorDecision(
        error_class, False, False, 0, 0, estimated_retry_usd,
        "error-taxonomy-2026-08-22",
        "unclassified_errors_fail_closed",
    )

def chat_with_error_policy(prompt: str):
    input_tokens = max(1, len(prompt) // 4)
    output_cap = 1200
    attempts = 0

    while True:
        attempts += 1
        try:

这不是一个完整的网关。它是最小可用的循环：分类、估算、决策、记录，然后仅在策略允许时重试。

在生产环境中，用网关使用的 tokenizer 替换粗略的 token 估算。在调用后记录实际使用量。添加提供商特定的适配器，在私人日志中保留原始状态、原始响应体、请求 ID 和响应头，同时只向产品代码暴露规范化后的分类。

切换路由不是盲目的回退

切换路由常被当作自动可靠性功能来对待。它应该更加审慎。如果 DeepSeek Flash 执行步骤遇到容量问题，切换到另一个具备执行能力的模型可能没问题。但如果 Pro 规划步骤在一个长提示后失败，切换到不同家族可能会改变质量、延迟、隐私姿态和单位经济效益。

使用明确的切换路由规则：

切换路由规则应该附加在原始请求上，而不是在错误之后才发明。这使行为可重现。它也防止一次事故演变成隐藏的模型迁移。

向客户展示什么

默认不要暴露原始提供商错误。它们可能包含内部标识符、令人困惑的提供商名称，或误导用户采取错误操作的消息。而是将规范化分类映射到产品安全的响应。

对于 context_overflow，解释请求需要更少的上下文或一个摘要步骤。对于 capacity_limit，说明路由暂时受限，请求可以重试。对于 tool_schema_error，返回一个面向开发者的消息，包含被拒绝的字段和支持的 schema 结构。对于 commercial_block，要求账户所有者检查计费或计划状态。

支持人员应该有更丰富的视图：规范化分类、原始提供商状态、请求 ID、选择的模型、回退模型、费率卡的来源日期、重试次数、估算的重试成本、实际使用量和策略版本。这些足以在大多数工单中回答问题，而不暴露提示内容。

重要的可观测性字段

将以下字段添加到每个网关跨度中：

这些字段让你能够构建仪表板，显示按模型分类的失败组合、按租户分类的重试支出、按提供商家族分类的容量事件，以及按版本分类的策略变化。它们也让你能够区分提供商不稳定性和客户端误用。SDK 发布后 tool_schema_error 的峰值与 provider_unavailable 的峰值不是同一个问题。

实施步骤

以影子模式启动。对每个错误进行分类并记录你本会做出的决定，但暂时不改变重试行为。在 Tier 1 和 Tier 2 生产流量上运行一周，特别关注美国、英国、德国、日本、新加坡等开发者密集的市场，这些地方对可靠性的期望很高。

接下来，为一个任务类别启用该分类体系。短提取或非流式支持草稿是很好的候选者，因为部分输出更容易丢弃。保持 max attempts 较低。当每日重试支出超过一个小阈值时添加警报。

然后为选定的模型家族添加切换路由规则。要求每个切换路由对进行重放测试。重放测试应包括成功调用、上下文溢出、schema 不匹配、提供商 5xx、容量限制和流中断。将策略版本与每个重放结果一起存储。

最后，将分类体系作为发布审查的一部分。任何新模型、提供商、SDK 功能或定价行在上线前都应回答四个问题：

实施用的来源链接

如果在 AIWave 上构建，请从 Chat Completions 文档、模型目录、定价页面和可预测定价计算器开始。模型目录和定价页面提供了错误策略所需的路由和费率卡上下文。

对于外部来源检查，请保持指向官方 DeepSeek 定价文档、DeepSeek 速率限制文档、QwenCloud 定价文档、Z.AI 定价文档和 Kimi K3 页面的最新链接。在更改重试预算或回退策略之前，重新检查这些页面。

结语

在为你多模型 AI 网关上线错误分类体系之前，确保它能做到以下事情：

真正有价值的转变既是文化上的，也是技术上的。错误处理不是 SDK 边缘的事后想法。它是一个路由面。一旦故障被分类、被预算化、被版本化，网关就能在压力下做出无聊的决策：在安全时重试、在策略允许时切换路由、在调用方必须采取行动时停止，并将每个决定与带日期的来源关联起来。