SaaS 聊天机器人 API 容量规划实战指南

对于 SaaS 聊天机器人而言，单一 API 加备用模型的设计只有在备用模型有足够配额、延迟预算和成本空间来承接被限流的主模型流量时才算站得住脚。

简短回答：对于应用内 SaaS 聊天机器人，选择一个能在一个 key 背后暴露多个模型选项的聊天 API，在路由前先发现可用模型，并在应用中保持一个小而明确的备用策略。只有当特定提供商的行为比通用契约更重要时才使用独立提供商集成；只有当这种控制值得在值班轮换中再增加一个服务时才自建网关。

这不是在追求最长的模型目录。这是一个关于谁负责提供商集成、资格检查、重试行为和紧急停机的买与造的决策。OpenAI、Claude 和 Gemini 可能都会出现在产品规划中，但可靠的运行时决策始于凭证当前可用的模型，而不是采购幻灯片。

SaaS 聊天机器人 API 应该如何跨备用模型工作？

从服务目标出发逆向推演。一个有用的聊天机器人目标可能是：每次交互在应用deadline内返回可接受的响应，或以 UI 能处理的形式失败。它不应该默默承诺不同模型会产生等效答案。传输可用性和回答质量是两个独立信号，在流量迁移前两者都需要产品自有验收标准。

第一个操作步骤是发现。查询 GET /v1/models，从 key 可用的模型中构建一个白名单，并按受控计划刷新。然后用 POST /v1/chat/completions 执行实际交互。在团队能回答更基本的问题之前——哪些候选模型当前是合格的——构建自定义路由器的价值很小。

交接策略要窄。HTTP 429 是容量信号，可以作为切换到其他合格模型的理由，前提是原始请求仍有足够时间。一个在发现阶段不可用的模型不得进入候选池。产品级质量拒绝只有在应用对该拒绝有具体规则时才能触发第二次尝试。不要把每个非成功状态都变成级联；两三次各自合理的重试可能在最糟糕的时刻消耗掉整个延迟预算并放大负载。

一次交互，一个deadline。

备用列表在发布前也需要成本估算。提示词长度、输出限制和备用频率都很重要，所以用平均请求作为容量规划单位是很弱的。我会限制输出、测试实际的对话长度分布，并与财务预算一起设置重试预算。我不确定是否存在通用的最佳模型顺序，因为提供的证据没有建立模型级质量、配额、延迟或定价；使用应用自身的对话进行评估才能解决这种不确定性。

买与造的边界

我会拿到平台路线图评审桌上的表格是刻意简朴的。它问的是集成和值班工作在哪个层面落地，而不是这个月哪个logo有最亮眼的基准测试。

托管行对于一个简单表象背后的广度很有吸引力。这确实是这里的实际优势：平台团队可以在添加后端能力时保持一致的 HTTP 契约，而不是积累特定提供商的 SDK 和凭证流程。它不能消除应用策略、模型评估或回滚所有权。

问题是具体的。没有专用的审核端点，因此文本或图像审核需要带 json_schema 回退的聊天模型，并且必须根据应用的风险要求来判断。不支持生产级 ASR，实时语音会话不是通用多区域能力，upscale 仅限 Lanczos。对于语音识别，开源项目 Whisper 等专业选项是更值得评估的路径。当供应商特定功能控制产品质量时，坚持使用直接提供商；当自托管是刻意的平台责任而非偶然的选择时，选择 LiteLLM。

一个 Go 语言的有界实现

以下程序刻意做得很小。它发现模型，检查配置的 primary 和 fallback ID 是否出现在返回的目录中，并在同一个deadline下最多发起两次完成尝试。模型 ID 来自环境变量，因为本文没有确定哪个特定 ID 是此工作负载的正确选择。目录响应作为 JSON 文本搜索而不是解码成虚构的响应schema；生产代码应该用其选定运行时使用的文档化schema替换那个保守的检查。

它也将 429 与其他失败区别对待，当服务器提供完整秒数时遵循 Retry-After，否则使用指数退避，并在备用后停止。没有紧凑循环。如果完成可以授权状态变更工具，该工具必须接收稳定的操作 ID 并在写入边界去重；重试聊天请求绝不能重复已提交的业务操作。

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

type message struct {
    Role    string `json:"role"`
    Content string `json:"content"`
}

type chatRequest struct {
    Model    string    `json:"model"`
    Messages []message `json:"messages"`
}

func request(ctx context.Context, client *http.Client, method, url, key string, body []byte) (*http.Response, []byte, error) {
    var reader io.Reader
    if body != nil {
        reader = bytes.NewReader(body)
    }
    req, err := http.NewRequestWithContext(ctx, method, url, reader)
    if err != nil {
        return nil, nil, err
    }
    req.Header.Set("Authorization", "Bearer "+key)
    if body != nil {
        req.Header.Set("Content-Type", "application/json")
    }
    resp, err := client.Do(req)
    if err != nil {
        return nil, nil, err
    }
    data, readErr := io.ReadAll(resp.Body)
    resp.Body.Close()
    return resp, data, readErr
}

func retryDelay(resp *http.Response, attempt int) time.Duration {
    if seconds, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && seconds >= 0 {
        return time.Duration(seconds) * time.Second
    }
    return time.Duration(1<<attempt) * 250 * time.Millisecond
}

func complete(ctx context.Context, client *http.Client, baseURL, key string, models []string) ([]byte, error) {
    prompt := []message{{Role: "user", Content: "Explain our trial cancellation policy in two sentences."}}
    for attempt, model := range models {
        payload, err := json.Marshal(chatRequest{Model: model, Messages: prompt})
        if err != nil {
            return nil, err
        }
        resp, data, err := request(ctx, client, http.MethodPost, baseURL+"/v1/chat/completions", key, payload)
        if err != nil {
            return nil, err
        }
        if resp.StatusCode >= 200 && resp.StatusCode < 300 {
            return data, nil
        }
        if resp.StatusCode != http.StatusTooManyRequests || attempt == len(models)-1 {
            return nil, fmt.Errorf("chat completion status %d: %s", resp.StatusCode, data)
        }
        timer := time.NewTimer(retryDelay(resp, attempt))
        select {
        case <-ctx.Done():
            timer.Stop()
            return nil, ctx.Err()
        case <-timer.C:
        }
    }
    return nil, fmt.Errorf("retry budget exhausted")
}

func main() {
    baseURL := strings.TrimRight(os.Getenv("CHAT_API_BASE_URL"), "/")
    key := os.Getenv("CHAT_API_KEY")
    models := []string{os.Getenv("PRIMARY_MODEL"), os.Getenv("FALLBACK_MODEL")}
    if baseURL == "" || key == "" || models[0] == "" || models[1] == "" {
        panic("CHAT_API_BASE_URL, CHAT_API_KEY, PRIMARY_MODEL, and FALLBACK_MODEL are required")
    }

    ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second)
    defer cancel()
    client := &http.Client{}

    resp, catalog, err := request(ctx, client, http.MethodGet, baseURL+"/v1/models", key, nil)
    if err != nil {
        panic(err)
    }
    if resp.StatusCode < 200 || resp.StatusCode >= 300 {
        panic(fmt.Sprintf("model discovery status %d: %s", resp.StatusCode, catalog))
    }
    for _, model := range models {
        encoded, _ := json.Marshal(model)
        if !bytes.Contains(catalog, encoded) {
            panic("configured model is absent from discovery: " + model)
        }
    }

    result, err := complete(ctx, client, baseURL, key, models)
    if err != nil {
        panic(err)
    }
    fmt.Println(string(result))
}

此示例刻意只在 429 时回退，因为现有证据支持将限流作为切换的理由，而没有为聊天完成定义更广泛的错误分类。团队可以在其选定运行时文档化额外触发条件后采用更多触发条件，但猜测哪些失败可重试不是可靠性工程。它是穿着runbook徽章的负载放大器。

金丝雀在上生产流量前应该证明什么？

验证必须执行策略而不是仅仅证明两个配置的模型名称都能返回响应。首先运行发现，拒绝候选缺失的配置。然后对有限队列进行金丝雀测试，同时记录主尝试次数、429 响应、备用尝试次数、备用成功次数、耗尽的重试预算、总deadline错失次数和应用级回答拒绝作为独立信号。单张成功率图会掩盖哪个层正在消耗错误预算。

在受控测试中强制执行支持的交接。确认 429 等待 Retry-After，确认缺失header激活退避，确认原始 context 取消发现和完成工作，确认备用模型只接收重试预算允许的流量。检查错误体但不记录凭证。具体的金丝雀百分比和告警阈值取决于流量形态和服务目标；捏造通用数字会让本指南看起来精确但容量问题仍未解答。

质量需要自己的关卡。不同模型可以满足 HTTP 契约但仍可能改变拒绝行为、格式化、工具选择或回答的有用性。在获得资格前评估代表性的应用内对话，限制输出，并在金丝雀期间审查样本。你的情况可能不同——特别是在长对话中——因此发布记录应该标识提示集和策略版本，而不是声称永久的模型等效。

容量是团队倾向于敷衍的部分。如果主模型通常承载 90% 的聊天流量，限流可以将突发流量重定向而不是平滑平均值；备用池需要为这种转移留出余量，而应用在无法做到时仍需要硬并发和重试上限。网关可以规范化调用表面，但无法制造配额或延长用户deadline。

在错误预算耗尽前回滚

回滚应该是策略变更，而不是代码部署。保持最后接受的模型顺序，使候选资格可配置，并将策略版本和尝试顺序附加到请求记录。如果金丝雀违反延迟、质量、容量或成本边界，则移除新的备用候选并恢复之前的顺序。在事件期间不要增加重试；那会消耗更多受限资源并使恢复信号更难读取。

让紧急制动杠杆变得平淡无奇。

最终所有权测试很简单：值班工程师应该能够说出当前活跃的候选模型，解释为什么某次交互发生了切换，看到还剩多少deadline，以及在不触碰应用代码的情况下禁用切换。如果托管通用契约使这些任务更容易，就买它。如果特定提供商控制是产品，就接受独立集成。如果网关控制是战略性的且团队有运营能力，就自建。一个 key 减少集成表面；它不外包可靠性决策。

LiteLLM，自托管 LLM 网关：https://github.com/BerriAI/litellm

OpenAI Whisper，开源语音识别：https://github.com/openai/whisper