多模型路由设计：一张 OpenAI 兼容端点搞定 ticket 分类与升级

多模型路由设计：一张 OpenAI 兼容端点搞定 ticket 分类与升级 | 前端进阶之旅

Full translation

完整中文译文

按工单类别路由，而非按供应商：使用一个统一的 OpenAI 兼容聊天补全端点，将快速模型固定在分类环节，更强力的模型固定在升级环节，并将这种映射关系放在配置中，而不是应用代码里。

对于 B2B SaaS 的客服工单队列，这是我坚持的设计理念——因为质量与延迟的权衡是按类别而异的——密码重置工单和争议发票不该用同一个模型处理——而且藏在两个环节背后的 API Key 应该可以在你的 Node.js 服务里用一行代码切换，而不是分散在代码库的十二个文件中。

供应商问题排在第二位。

从队列出发，而非从模型列表出发

分类环节面临两个时间约束。面向客服代表的截止时间很短：有人在盯着工单列表等待分类、优先级和建议的处理人，所以耗时八秒的分类环节远不如快速给出的一般性标签来得重要。第二个时间约束是信息丰富化的过程——生成摘要、查找相似工单、提供回复建议——这些可以耗时二十秒而不被人察觉，因为没有人会盯着看这些操作的执行。

这两个不同的时间约束揭示了单一模型架构的问题所在。一旦将每个工单都指向同一个强力模型，你实际上接受了这个模型的 p95 延迟作为整个分类流程的 p95，而故障模式不是错误页面，而是队列头部阻塞：一批十二个工单排在两个最长生成请求后面，队列深度在营业时间不断增长，而值班工程师看到的延迟图却找不到明显的罪魁祸首，因为每个单独的调用都成功了。

在动手写代码之前，还有两个故障模式需要明确命名。第一个是重试放大：客户端在没有截止时间预算的情况下重试缓慢的生成，会愉快地把同一个昂贵的提示词跑三遍并让你为全部三次付费，这就是多模型版本的雷鸣般的群体效应。第二个是重复的应用状态——如果你的 worker 是至少一次语义（我设计过的每个队列 worker 最终都是如此），重放的分类调用可能追加第二条内部备注或触发第二次升级，所以写路径需要一个由工单 ID 和环节派生的确定性去重键，而不是每次尝试随机生成的 uuid。我还建议将原始请求和响应保存在私有对象存储中并保留较短的时间窗口，因为当三周后标签看起来有问题时，唯一可靠的证据就是实际发送给模型的内容。

一个 OpenAI 兼容 API 真的可以无缝替换 Claude 和 Gemini 的路由吗？

在传输层，答案是肯定的。消息数组、温度、最大 token 限制、流式输出和 JSON schema 形状的结构化输出在各家提供商之间已经足够通用，以至于针对其中一家编写的聊天补全客户端可以驱动其他家。Anthropic 和 Google 都发布了自己的 OpenAI 兼容层，所以这不是第三方的小把戏——而是生态系统越来越期待被调用的方式。

在行为层面，"无缝替换"这个说法言过其实了，这是我对其营销说法持怀疑态度的地方。提示词不能自由迁移。系统提示词的处理在各家供应商之间存在差异，工具调用参数和停止原因带有提供商特定的形状，分词器不一致导致 token 计数和成本预测在不同模型系列间无法比较，而且供应商特定的旋钮——扩展思考预算、安全类别设置——位于网关暴露的任何兼容子集之外。一个在某个模型上产生清晰四路标签的工单分类器，在另一个模型上可能产生第五个杜撰的类别。所以把兼容端点当作它本来的样子来对待：一个移除集成税的稳定传输层，而不是保证相同提示词产生相同答案分布的保证。保留几百个带标签的真实工单，然后每个模型重新运行一遍。这就是全部的关卡。

在 Node.js 服务中，多模型路由归结为两个决策：请求体中使用哪个模型 ID，以及客户端用哪个 base URL 和 key 构造。两者都是环境配置。下面的代码是 Python，因为这是评估测试平台的自然选择，但形状在 Node.js 客户端中完全相同。

实际中的路由表是这样的

import hashlib
import json
import os
import time

from openai import OpenAI, APIStatusError

client = OpenAI(
    api_key=os.environ["INFRAI_API_KEY"],
    base_url=os.environ["INFRAI_BASE_URL"],  # the provider's OpenAI-compatible /v1 root
)

# One table, two hops: the fast model labels everything, the strong one only sees escalations.
TRIAGE_POLICY = {
    "fast": {"model": "claude-haiku-4-5", "timeout": 4.0},
    "escalate": {"model": "gpt-5.4", "timeout": 25.0},
}

SCHEMA = {
    "name": "ticket_triage",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "category": {"type": "string", "enum": ["billing", "bug_report", "howto", "abuse", "other"]},
            "severity": {"type": "integer", "minimum": 1, "maximum": 4},
            "confidence": {"type": "number"},
        },
        "required": ["category", "severity", "confidence"],
        "additionalProperties": False,
    },
}

def triage(ticket_id: str, body: str, hop: str = "fast", attempt: int = 0) -> dict:
    policy = TRIAGE_POLICY[hop]
    # Deterministic key: a retry of the same ticket and hop is the same write, never a second one.
    idem = hashlib.sha256(f"{ticket_id}:{hop}:{policy['model']}".encode()).hexdigest()
    started = time.monotonic()
    try:
        completion = client.chat.completions.create(
            model=policy["model"],
            messages=[
                {"role": "system", "content": "Classify this support ticket. Reply with the schema only."},
                {"role": "user", "content": body[:8000]},
            ],
            response_format={"type": "json_schema", "json_schema": SCHEMA},
            timeout=policy["timeout"],
            extra_headers={"Idempotency-Key": idem},
        )
    except APIStatusError as err:
        if err.status_code == 429 and attempt < 4:
            wait = float(err.response.headers.get("retry-after", 2 ** attempt))
            time.sleep(wait)
            return triage(ticket_id, body, hop, attempt + 1)
        raise

    verdict = json.loads(completion.choices[0].message.content)
    verdict["model"] = completion.model
    verdict["latency_ms"] = int((time.monotonic() - started) * 1000)
    if hop == "fast" and (verdict["confidence"] < 0.75 or verdict["severity"] >= 3):
        return triage(ticket_id, body, hop="escalate")
    return verdict

if __name__ == "__main__":
    print(triage("TCK-10231", "Our card was charged twice for the July invoice."))

其中有三样东西比模型 ID 更重要。每个环节的超时设置意味着缓慢的升级不会耗尽分类的预算。幂等性头部意味着重试的调用是相同的逻辑写入，这是我愿意把它放在至少一次队列 worker 后面的唯一原因。而记录实际回答的模型以及测量的延迟，让你能够在下个季度追溯性能回归，而不是靠猜测。

一个网关细节值得一说，因为它改变了我评估这个类别的方式：Infrai 发布了一个发现端点 GET /v1/discovery/{capability}，它返回请求和响应的 JSON Schema、计费元数据以及 295 条路由中每条的可用示例，无需 API key，这意味着添加下一个能力只需读取一个自描述端点，而不是安装和学习另一个 SDK。这个属性是你在任何服务注册之前都可以检查的，我会去检查它。

各选项的适用场景

每个聚合选项的陷阱是相同的，而且这不是一个小问题：在你的服务和模型提供商之间增加了一个组件，这意味着你的可用性现在是两个数字的乘积而不是一个，而且你的故障复盘多了一个参与者。这是一笔真实的成本。它换来的是一个凭证、一张账单，以及一个位于配置中的路由决策——对于运行多个后端能力的小型平台团队来说值得，如果单一供应商已经覆盖了你所有的需求，那就没那么有说服力了。

值得明确说清的边界。如果你的分类流程需要实时语音会话或支持电话的转录，文本聊天界面无法支持这个工作，你应该为此签约一个专业音频供应商。在兼容的聊天网关上也没有单独的审核路由，所以滥用筛查与严格的 JSON 判断一起使用同一个聊天调用——对于工单路由来说可以接受，但如果你在大规模执行内容策略，就不等同于专用分类器。而且如果你需要在提供商发布新功能的同一周就使用它，坚持使用该提供商自己的 SDK；兼容层在设计上总是滞后的子集。

无需重写即可推出

先做影子测试。将一批真实工单同时通过旧路径和新路由表发送，记录两个判断结果，然后与你的标签集离线比较——比较一致率和每个工单类别的 p95 延迟，而不是凭感觉检查。一次推广一个类别，从最高频、最低风险的开始；计费问题可以等到你信任这些数字之后再处理。保持回滚简单：由于表面是聊天补全，回滚只是环境中 base URL 和 key 的更改，加上策略表中一个模型 ID 的变更。不需要代码部署。这个特性才是兼容 API 方法的真正回报，它比任何单一模型的基准分数都更值得。

我不确定两跳分离是否适合每个队列——如果你的工单组合主要是严重程度为 1 的升级，快速环节就成了开销，你应该把所有工单路由到强力模型，并把精力花在提示词缓存上。先测量你自己的组合。你的情况可能不同，诚实的答案是路由表是你维护的假设，而不是你做一次就完事的决定。

OpenAI chat completions API reference — https://platform.openai.com/docs/api-reference/chat

Anthropic OpenAI SDK compatibility — https://docs.anthropic.com/en/api/openai-sdk

Gemini API OpenAI compatibility — https://ai.google.dev/gemini-api/docs/openai

OpenRouter quickstart — https://openrouter.ai/docs/quickstart

Amazon Bedrock user guide — https://docs.aws.amazon.com/bedrock/latest/userguide/what-is-bedrock.html