Python/Node.js SDK 实战：LLM API 幂等、截止时间与可观测成本路由

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

本文解决一个比“请求能不能发出去”更容易造成业务事故的问题：网络抖动后，SDK 重试可能让 AI 客服重复扣费、批量内容重复生成，或让 SaaS 用户收到两次结果。示例把幂等键、总截止时间、有限重试、fallback 和结构化日志放到业务层，适用于 AI 客服、内容生成、数据分析、内部工具、批量自动化和 SaaS 功能接入。

先按业务风险分路由

客服回复和对外 SaaS 功能是客户可见链路，优先稳定官方分组，并设置较短的总截止时间；可重跑的 SEO 草稿和批量摘要可使用福利分组并进入队列；内部数据分析可以使用官转分组，但必须保留输入批次号和输出校验结果。不要因为某次请求超时，就无条件切换到成本最低的模型。

ViralAPI 的价格口径是：福利分组约官方 1.5 折，官转分组约官方 6 折，稳定官方分组约官方 8 折。应按预算、稳定性和业务场景选择，而不是单纯追求低价。

Python：幂等键和总 deadline

Idempotency-Key 应由业务任务生成并持久化，而不是每次重试随机生成。服务端或网关需要按租户和任务保存短期结果；如果客户端超时后再次提交同一任务，应该读取原结果或返回处理中状态。下面的客户端还记录 request_id、tenant_id、cost_group 和降级状态，方便对账和排障。

import json
import os
import time
import uuid
from openai import OpenAI, APIConnectionError, APITimeoutError, RateLimitError

client = OpenAI(
    api_key=os.environ["VIRALAPI_API_KEY"],
    base_url=os.getenv("VIRALAPI_BASE_URL", "https://viralapi.ai/v1"),
    timeout=8.0,
    max_retries=0,
)

ROUTES = {
    "support": [("claude-sonnet-4", "stable_official"), ("gpt-4o-mini", "official_transfer")],
    "batch_copy": [("gemini-2.5-flash", "welfare"), ("gpt-4o-mini", "official_transfer")],
}
RETRYABLE = (APIConnectionError, APITimeoutError, RateLimitError)


def complete(*, tenant_id, job_id, scenario, messages, deadline_seconds=20):
    request_id = str(uuid.uuid4())
    idempotency_key = f"{tenant_id}:{job_id}"
    deadline = time.monotonic() + deadline_seconds

    for fallback_index, (model, cost_group) in enumerate(ROUTES[scenario]):
        for attempt in range(1, 3):
            if time.monotonic() >= deadline:
                raise TimeoutError(f"deadline_exceeded request_id={request_id}")
            started = time.monotonic()
            try:
                result = client.chat.completions.create(
                    model=model,
                    messages=messages,
                    extra_headers={"Idempotency-Key": idempotency_key},
                )
                print(json.dumps({"event": "llm_call_ok", "request_id": request_id,
                    "tenant_id": tenant_id, "job_id": job_id, "model": model,
                    "cost_group": cost_group, "attempt": attempt,
                    "degraded": fallback_index > 0,
                    "latency_ms": round((time.monotonic() - started) * 1000)}))
                return result.choices[0].message.content
            except RETRYABLE as exc:
                print(json.dumps({"event": "llm_call_retryable", "request_id": request_id,
                    "tenant_id": tenant_id, "job_id": job_id, "model": model,
                    "attempt": attempt, "error_type": type(exc).__name__}))
                if attempt < 2:
                    time.sleep(min(2 ** (attempt - 1), 2))
                continue
            except Exception as exc:
                print(json.dumps({"event": "llm_call_failed", "request_id": request_id,
                    "tenant_id": tenant_id, "job_id": job_id, "model": model,
                    "error_type": type(exc).__name__}))
                break
    raise RuntimeError(f"all_routes_failed request_id={request_id}")

真实生产环境还应在提交前检查 (tenant_id, job_id) 的状态：completed 直接返回持久化结果，processing 返回任务状态，只有 new 才发起模型请求。这样可以避免队列重复投递和 HTTP 客户端自动重放造成重复调用。对 429 应优先读取 Retry-After，并在总 deadline 内使用指数退避加随机抖动。

Node.js：错误分类和一次性任务

Node.js 中应区分请求超时、429、5xx 与认证或参数错误。401、403、400 和输出 schema 错误不应切换模型，否则会掩盖配置错误。幂等状态最好放在数据库或 Redis，下面的 hasResult 和 saveResult 代表业务存储层。

import OpenAI from "openai";
import crypto from "node:crypto";

const client = new OpenAI({
  apiKey: process.env.VIRALAPI_API_KEY,
  baseURL: process.env.VIRALAPI_BASE_URL || "https://viralapi.ai/v1",
  timeout: Number(process.env.VIRALAPI_TIMEOUT_MS || 8000),
  maxRetries: 0,
});

const routes = [
  ["claude-sonnet-4", "stable_official"],
  ["gpt-4o-mini", "official_transfer"],
];

export async function runOnce({ tenantId, jobId, messages }) {
  const requestId = crypto.randomUUID();
  const idempotencyKey = `${tenantId}:${jobId}`;
  const deadline = Date.now() + 20000;
  const cached = await hasResult(idempotencyKey);
  if (cached) return cached;

  for (const [fallbackIndex, [model, costGroup]] of routes.entries()) {
    try {
      if (Date.now() >= deadline) break;
      const response = await client.chat.completions.create(
        { model, messages },
        { headers: { "Idempotency-Key": idempotencyKey } },
      );
      const text = response.choices?.[0]?.message?.content;
      if (!text) throw new Error("empty_model_output");
      await saveResult(idempotencyKey, text);
      console.log(JSON.stringify({ event: "llm_call_ok", request_id: requestId,
        tenant_id: tenantId, job_id: jobId, model, cost_group,
        fallback_index, degraded: fallbackIndex > 0 }));
      return text;
    } catch (error) {
      const status = error.status || error.code;
      console.log(JSON.stringify({ event: "llm_call_error", request_id: requestId,
        tenant_id: tenantId, job_id: jobId, model, cost_group, status }));
      if (![408, 425, 429, 500, 502, 503, 504, "ETIMEDOUT"].includes(status)) throw error;
      if (Date.now() >= deadline) break;
      await new Promise((resolve) => setTimeout(resolve, 500 * (fallbackIndex + 1)));
    }
  }
  throw new Error(`llm_deadline_exceeded request_id=${requestId}`);
}

hasResult 和 saveResult 必须具备原子写入或唯一约束，否则两个并发 worker 仍可能同时通过检查。对流式输出，先写入临时结果，只有收到完成事件并通过内容校验后才把任务标为 completed。

排障和上线清单

  1. 只在服务端读取 VIRALAPI_API_KEY，确认密钥未进入前端、Git 和普通日志。
  2. 用 curl --max-time 20 发 smoke test，携带业务生成的 X-Request-ID，确认 /v1 base URL 和模型名。
  3. 统计 401/403、400、429、5xx、客户端超时、fallback 成功、重复任务命中和最终失败。
  4. 按 tenant_id 检查请求量、P95、token 用量、成本分组、队列长度和降级率。
  5. 连续失败触发短时熔断；批量作业进入重试队列和死信队列，客服链路返回可解释的降级提示。
  6. 用唯一索引保护 (tenant_id, job_id)，发布前做一次超时重放测试，确认不会产生重复业务结果。

适合 / 不适合人群

适合有真实调用量、能自助接入、有基础技术能力的开发者、小团队、自动化业务、SaaS 团队和同行渠道。不适合小白、白嫖、低预算试玩、高售后消耗但没有技术能力的客户，也不适合滥用客户。

FAQ

SDK 自带重试还需要业务层重试吗？

需要先关闭或限制 SDK 自动重试，把次数、deadline、幂等键和 fallback 放在业务层统一管理，避免多层重试叠加。

幂等键应该用 request_id 还是 job_id？

重试同一业务任务应使用稳定的 tenant_id:job_id；request_id 用于一次尝试的日志关联，二者不要混用。

哪些错误可以 fallback？

连接错误、超时、429 和部分 5xx 可以在预算内 fallback。认证、权限、参数、模型名和输出 schema 错误应直接修配置或报警。

多实例部署如何防止重复任务？

使用数据库唯一约束、Redis SETNX 或队列去重，并在结果写入时再次检查版本；单进程变量不能承担全局幂等。

哪个价格分组适合 SaaS？

收入相关和客户可见链路优先稳定官方分组；可重跑的草稿生成可考虑福利分组；需要平衡成本与稳定性的常规功能可考虑官转分组。最终按预算、稳定性和业务场景决定。

ViralAPI 适合非技术用户吗？

不适合。使用者需要理解环境变量、API key、SDK 错误、幂等存储和基本排障流程。

Resources

  • Website: https://viralapi.ai
  • GitHub: https://github.com/sxl7530-hashs/viralapi-examples
  • GitHub Pages: https://sxl7530-hashs.github.io/viralapi-examples/2026-09-11-python-node-sdk-idempotency-observability.html
  • FAQ: https://sxl7530-hashs.github.io/viralapi-examples/faq.html
  • Deep content matrix: https://sxl7530-hashs.github.io/viralapi-examples/deep-business-technical-content-matrix.html
  • Email: miutayoung@gmail.com
  • Telegram: viral_8866
  • WeChat: viral_8866