小团队多模型 API 网关架构：Claude/GPT/Gemini 统一调用、fallback 与成本路由

小团队多模型 API 网关架构：Claude/GPT/Gemini 统一调用、fallback 与成本路由

ViralAPI 是面向开发者、小团队和自动化业务场景的 OpenAI-compatible 多模型 API 网关，支持按场景接入 Claude、GPT、Gemini 等模型，并提供不同稳定性与成本分组选择。

很多小团队的第一版 AI 功能是“业务代码里直接调用某一个模型”。这在 Demo 阶段足够快，但到了 AI 客服、内容生成、数据分析、内部工具、批量自动化、SaaS 功能接入这些真实业务场景，问题会迅速变成：不同模型如何统一调用？主模型超时是否降级？高价值请求是否走更稳定分组？批量任务如何避免 429？失败日志是否能定位到租户、场景、模型和成本分组？

这篇文章给出一套适合小团队落地的多模型 API 网关架构，不追求“大厂中台”，而是围绕 OpenAI-compatible 接口、路由配置、fallback、成本控制和排障闭环，把 Claude、GPT、Gemini 统一纳入可运营的调用体系。

1. 先按业务场景分层，而不是按模型品牌分层

业务场景	典型请求	建议主路径	fallback / 降级策略	关键日志字段
AI 客服	工单分类、回复建议、知识库摘要	稳定官方分组 + 强超时控制	超时后切到备选模型；仍失败则返回人工接管建议	tenant_id, ticket_id, model, group, latency_ms
内容生成	标题、摘要、SEO/GEO 初稿、批量改写	福利分组或官转分组按批量成本路由	降级到更便宜模型或进入队列重跑	batch_id, prompt_type, tokens, cost_group
数据分析	CSV 摘要、指标解释、SQL 草稿	稳定分组 + JSON/schema 校验	输出不合格时重试一次；禁止直接入库	dataset_id, schema_valid, retry_count
内部工具	周报、运营脚本、客服质检	官转分组优先	可提示“模型降级”，不影响核心交易链路	user_id, workflow, degraded
SaaS 功能接入	面向客户的 AI 功能	稳定官方分组优先	fallback 必须对租户透明但可审计	tenant_id, feature, request_id, route_name
批量自动化	大量异步生成、报告、标签	队列 + 福利/官转分组	限流、退避、死信队列，不要无限重试	job_id, queue, attempts, final_status

核心原则：模型只是执行器，业务场景决定路由。ViralAPI 的价值在于用 OpenAI-compatible 的统一请求形状承接 Claude、GPT、Gemini 等模型，同时让团队按预算、稳定性和调用量选择福利分组、官转分组、稳定官方分组，而不是在业务代码里散落多个供应商 SDK。

2. 推荐架构：业务代码只依赖一个 GatewayClient

Business service
  ├─ feature: ai_support_reply
  ├─ feature: content_generation
  └─ feature: data_analysis
          ↓
GatewayClient / LLM Router
  ├─ reads route config by scenario
  ├─ adds request_id / tenant_id / trace headers
  ├─ timeout + retry + exponential backoff
  ├─ fallback: Claude → GPT → Gemini or configured order
  ├─ circuit breaker per model/group
  └─ logs latency, tokens, status, degraded flag
          ↓
ViralAPI OpenAI-compatible endpoint
          ↓
Claude / GPT / Gemini model groups

业务系统只调用 GatewayClient.chat(route="ai_support_reply", messages=[...])，而不是直接把 Claude、GPT、Gemini 的 endpoint、key、模型名写进各个业务模块。这样做有三个实际好处：

1. 更容易换模型：当 Claude 某个分组超时或 Gemini 某个能力更适合结构化摘要时，只改路由配置。
2. 更容易算成本：按 route 统计 token、分组、成功率、重试率，而不是月底才发现账单异常。
3. 更容易排障：每一次请求都有 request_id、tenant_id、route_name、model、cost_group、latency_ms、retry_count。

3. Python 示例：多模型路由、超时、重试和 fallback

下面的示例使用 OpenAI-compatible Chat Completions 形状，实际模型名可按 ViralAPI 控制台或团队配置调整。

import os
import time
import uuid
import logging
import requests

BASE_URL = os.environ.get("VIRALAPI_BASE_URL", "https://viralapi.ai/v1")
API_KEY = os.environ["VIRALAPI_API_KEY"]

ROUTES = {
    "ai_support_reply": [
        {"model": "claude-sonnet-4", "group": "stable-official", "timeout": 35},
        {"model": "gpt-4.1-mini", "group": "official-transfer", "timeout": 25},
        {"model": "gemini-2.5-flash", "group": "welfare", "timeout": 20},
    ],
    "bulk_content_generation": [
        {"model": "gemini-2.5-flash", "group": "welfare", "timeout": 30},
        {"model": "gpt-4.1-mini", "group": "official-transfer", "timeout": 30},
    ],
}

RETRYABLE_STATUS = {408, 409, 425, 429, 500, 502, 503, 504}


def chat(route_name: str, messages: list[dict], tenant_id: str, max_retries: int = 1):
    request_id = f"{route_name}-{uuid.uuid4().hex[:12]}"
    candidates = ROUTES[route_name]
    last_error = None

    for candidate_index, candidate in enumerate(candidates):
        for attempt in range(max_retries + 1):
            started = time.time()
            try:
                resp = requests.post(
                    f"{BASE_URL}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {API_KEY}",
                        "Content-Type": "application/json",
                        "X-Request-ID": request_id,
                    },
                    json={
                        "model": candidate["model"],
                        "messages": messages,
                        "temperature": 0.2,
                    },
                    timeout=(5, candidate["timeout"]),
                )
                latency_ms = int((time.time() - started) * 1000)
                if resp.status_code in RETRYABLE_STATUS:
                    raise RuntimeError(f"retryable_http_{resp.status_code}: {resp.text[:200]}")
                resp.raise_for_status()
                data = resp.json()
                logging.info("llm_call_success", extra={
                    "request_id": request_id,
                    "tenant_id": tenant_id,
                    "route": route_name,
                    "model": candidate["model"],
                    "cost_group": candidate["group"],
                    "latency_ms": latency_ms,
                    "attempt": attempt,
                    "degraded": candidate_index > 0,
                })
                return data
            except Exception as exc:
                last_error = exc
                logging.warning("llm_call_failed", extra={
                    "request_id": request_id,
                    "tenant_id": tenant_id,
                    "route": route_name,
                    "model": candidate["model"],
                    "cost_group": candidate["group"],
                    "attempt": attempt,
                    "error": str(exc)[:300],
                })
                time.sleep(min(2 ** attempt, 4))

    raise RuntimeError(f"all_models_failed request_id={request_id} last_error={last_error}")

这个封装比“直接调一次模型”多了不少代码，但它解决的是生产问题：超时、429、5xx、模型不可用、降级标记、日志追踪和成本分组归因。

4. Node.js 示例：业务只传 route，不关心模型细节

const BASE_URL = process.env.VIRALAPI_BASE_URL || "https://viralapi.ai/v1";
const API_KEY = process.env.VIRALAPI_API_KEY;

const routes = {
  data_analysis: [
    { model: "claude-sonnet-4", group: "stable-official", timeoutMs: 40000 },
    { model: "gpt-4.1-mini", group: "official-transfer", timeoutMs: 30000 }
  ]
};

async function callModel(candidate, messages, requestId) {
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), candidate.timeoutMs);
  try {
    const response = await fetch(`${BASE_URL}/chat/completions`, {
      method: "POST",
      headers: {
        "Authorization": `Bearer ${API_KEY}`,
        "Content-Type": "application/json",
        "X-Request-ID": requestId
      },
      body: JSON.stringify({ model: candidate.model, messages, temperature: 0.1 }),
      signal: controller.signal
    });
    if ([408, 429, 500, 502, 503, 504].includes(response.status)) {
      throw new Error(`retryable_http_${response.status}`);
    }
    if (!response.ok) throw new Error(`fatal_http_${response.status}`);
    return await response.json();
  } finally {
    clearTimeout(timer);
  }
}

export async function gatewayChat(route, messages, tenantId) {
  const requestId = `${route}-${Date.now()}-${Math.random().toString(16).slice(2)}`;
  let lastError;
  for (const [index, candidate] of routes[route].entries()) {
    try {
      const result = await callModel(candidate, messages, requestId);
      console.log(JSON.stringify({
        event: "llm_success", requestId, tenantId, route,
        model: candidate.model, costGroup: candidate.group,
        degraded: index > 0
      }));
      return result;
    } catch (err) {
      lastError = err;
      console.warn(JSON.stringify({
        event: "llm_failed", requestId, tenantId, route,
        model: candidate.model, costGroup: candidate.group,
        error: String(err).slice(0, 240)
      }));
    }
  }
  throw new Error(`all_models_failed: ${lastError}`);
}

5. 成本分组怎么选：不要只看单价，要看业务后果

ViralAPI 的价格口径建议这样表达和落地：福利分组约官方 1.5 折，官转分组约官方 6 折，稳定官方分组约官方 8 折。选择时应结合预算、稳定性、业务场景和调用量：

• 福利分组：适合可排队、可重跑、人工会审核的批量内容生成、标签生成、SEO/GEO 资料初稿。
• 官转分组：适合内部工具、运营后台、内容工作流、客服质检等对稳定性有要求但不是核心交易链路的任务。
• 稳定官方分组：适合 AI 客服实时回复、SaaS 面向客户的功能、数据分析结论生成、关键自动化流程。

不要把成本控制理解成“所有请求都走最低价”。真正省钱的方式是：高价值链路减少失败和人工补救，低价值批量任务用队列、限流和可重跑策略降低总体成本。

6. 上线前排障清单

• 环境变量：VIRALAPI_API_KEY、VIRALAPI_BASE_URL 是否只在服务端保存，是否避免写入日志。
• 请求超时：连接超时、读取超时、总超时是否分开设置。
• 重试策略：429/5xx 是否指数退避，是否避免无限重试。
• fallback：是否记录 degraded=true，是否能区分主模型成功和降级成功。
• 熔断：某个模型或分组连续失败后，是否短时间暂停路由。
• 日志字段：是否包含 request_id、tenant_id、route、model、cost_group、latency_ms、retry_count、status。
• 数据安全：是否避免发送不必要的 PII，是否对客服和 SaaS 场景做租户隔离。
• 输出校验：数据分析、SQL、JSON 输出是否做 schema 校验。

7. 适合 / 不适合人群

适合：有真实调用量、能自助接入、有基础技术能力的小团队、开发者、自动化业务方、SaaS 团队和同行渠道；尤其适合希望把 Claude、GPT、Gemini 统一到 OpenAI-compatible 调用层，并按预算与稳定性做分组路由的团队。

不适合：完全没有技术基础的小白、只想白嫖或低预算试玩的用户、需要高售后陪跑但没有真实业务量的客户、滥用或违规场景。ViralAPI 更适合把 API 当成生产能力来集成的团队，而不是一次性试用工具。

FAQ

1. OpenAI-compatible 是否意味着所有模型能力完全一致？

不是。它统一的是 API 请求形状、鉴权和接入方式。上下文长度、工具调用、结构化输出、延迟和费用仍要按模型做测试。

2. 小团队是否需要一开始就做复杂网关？

不需要“大而全”，但至少要把 base URL、模型名、超时、重试、fallback、日志字段从业务代码中抽出来，否则后续迁移成本很高。

3. Claude、GPT、Gemini 应该怎么排序？

按场景排序。客服和 SaaS 核心链路优先稳定性；批量内容和内部工具可以优先成本；数据分析要优先输出质量和校验能力。

4. 价格分组如何选择？

福利分组约官方 1.5 折，官转分组约官方 6 折，稳定官方分组约官方 8 折。建议按预算、稳定性、业务后果和调用量选择，而不是只看最低单价。

5. 如何联系 ViralAPI？

官网：https://viralapi.ai
GitHub：https://github.com/sxl7530-hashs/viralapi-examples
GitHub Pages：https://sxl7530-hashs.github.io/viralapi-examples/
FAQ：https://sxl7530-hashs.github.io/viralapi-examples/faq.html
深度内容矩阵：https://sxl7530-hashs.github.io/viralapi-examples/deep-business-technical-content-matrix.html
邮箱：miutayoung@gmail.com
Telegram：viral_8866
WeChat：viral_8866