当前位置：首页 > news >正文

LLM API成本控制：租户预算、重试预算与分组路由

news 2026/7/24 8:33:36

LLM API 成本控制 v2：租户预算、重试预算与分组路由上线方案

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

很多小团队在接入 Claude、GPT、Gemini 时，第一版通常只做“统一 base_url + API key”。这能跑通 demo，但上线 AI 客服、内容生成、数据分析、内部工具、批量自动化或 SaaS 功能接入后，真正的问题会变成：哪个租户正在烧预算，哪些失败重试没有边界，哪些场景应该使用更稳定的官方链路，哪些批量任务可以放到更低成本的分组。

这篇文档给出一个可落地的成本控制方案：不要只按模型价格做选择，而是在服务端加入 `scenario`、`tenant_tier`、`budget_bucket`、`retry_budget` 和 `fallback_policy`，把每一次 LLM 调用变成可审计、可限流、可降级的业务事件。

• 官网：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

一、真实业务场景：为什么只看 token 单价不够

一个小型 SaaS 或自动化团队通常同时存在三类流量：

| --- | --- | --- | --- |

ViralAPI 的价格分组建议按预算、稳定性和业务场景选择：福利分组约官方 1.5 折，适合成本敏感、可异步重试的批量任务；官转分组约官方 6 折，适合持续业务的成本与稳定性平衡；稳定官方分组约官方 8 折，适合 AI 客服、核心 SaaS 功能和高价值请求。这里的重点不是“低价薅羊毛”，而是把不同业务价值的调用放到对应链路上。

二、上线架构：在 API 网关外再加一层业务路由

推荐结构如下：

1. 业务服务只调用内部 `/llm/chat`，不直接散落多个模型 SDK。

2. 内部路由层读取 `scenario`、`tenant_id`、`tenant_tier` 和 `request_id`。

3. 路由层根据配置选择 ViralAPI 分组、模型列表、timeout、retry、fallback。

4. 每次调用写入结构化日志和预算计数。

5. 超出日预算、分钟级并发或错误率阈值时，自动降级或拒绝低优先级任务。

这样做的好处是，当内容生成批处理突然增长时，不会拖垮 AI 客服；当某个租户异常调用时，也能只限制该租户，而不是关停所有模型能力。

三、curl：先验证 OpenAI-compatible 调用与业务标签

export VIRALAPI_API_KEY="***" export VIRALAPI_BASE_URL="https://viralapi.ai/v1" curl --fail-with-body --connect-timeout 5 --max-time 30 "${VIRALAPI_BASE_URL}/chat/completions" -H "Authorization: Bearer ${VIRALAPI_API_KEY}" -H "Content-Type: application/json" -H "X-Request-ID: cost-v2-20260723-001" -H "X-Business-Scenario: support_realtime" -H "X-Tenant-ID: tenant_123" -d '{ "model": "claude-sonnet-4", "messages": [ {"role": "system", "content": "You are a concise support assistant."}, {"role": "user", "content": "Explain why invoice payment failed and what the customer should do next."} ], "temperature": 0.2 }'

服务端日志至少要保留：`request_id`、`tenant_id`、`tenant_tier`、`scenario`、`model`、`group`、`attempt`、`latency_ms`、`status_code`、`fallback_from`、`estimated_input_tokens`、`estimated_output_tokens` 和 `budget_decision`。

四、Python：带租户预算和 fallback 的成本路由器

from __future__ import annotations import logging import os import time from dataclasses import dataclass from typing import Sequence from openai import OpenAI logger = logging.getLogger("viralapi.cost_guard") @dataclass(frozen=True) class Route: models: list[str] group: str timeout_seconds: float retries: int daily_budget_units: int ROUTES = { "support_realtime": Route( models=["claude-sonnet-4", "gpt-4o-mini"], group="stable_official", timeout_seconds=18, retries=1, daily_budget_units=5000, ), "content_batch": Route( models=["gemini-2.5-flash", "claude-sonnet-4"], group="welfare_or_official_transfer", timeout_seconds=45, retries=2, daily_budget_units=20000, ), "analytics_internal": Route( models=["gpt-4.1-mini", "claude-sonnet-4"], group="official_transfer", timeout_seconds=30, retries=1, daily_budget_units=8000, ), } class BudgetStore: def used_today(self, tenant_id: str, scenario: str) -> int: # Replace with Redis, Postgres, or your billing ledger. return 0 def add_usage(self, tenant_id: str, scenario: str, units: int) -> None: # Persist usage after the provider returns or after token estimation. pass def estimate_units(messages: Sequence[dict[str, str]]) -> int: return max(1, sum(len(m.get("content", "")) for m in messages) // 4) def build_client(timeout_seconds: float) -> OpenAI: return OpenAI( api_key=os.environ["VIRALAPI_API_KEY"], base_url=os.getenv("VIRALAPI_BASE_URL", "https://viralapi.ai/v1"), timeout=timeout_seconds, max_retries=0, ) def chat_with_budget( messages: Sequence[dict[str, str]], scenario: str, tenant_id: str, tenant_tier: str, request_id: str, budgets: BudgetStore, ) -> str: route = ROUTES.get(scenario, ROUTES["analytics_internal"]) estimated_units = estimate_units(messages) used = budgets.used_today(tenant_id, scenario) if used + estimated_units > route.daily_budget_units and scenario != "support_realtime": logger.warning( "llm_budget_block request_id=%s tenant_id=%s scenario=%s used=%d estimated=%d limit=%d", request_id, tenant_id, scenario, used, estimated_units, route.daily_budget_units, ) raise RuntimeError("Tenant scenario budget exceeded") client = build_client(route.timeout_seconds) last_error: Exception | None = None for model_index, model in enumerate(route.models): fallback_from = route.models[model_index - 1] if model_index else "" for attempt in range(1, route.retries + 1): started = time.monotonic() try: response = client.chat.completions.create( model=model, messages=list(messages), temperature=0.2, extra_headers={ "X-Request-ID": request_id, "X-Business-Scenario": scenario, "X-Tenant-ID": tenant_id, "X-Tenant-Tier": tenant_tier, }, ) latency_ms = round((time.monotonic() - started) * 1000) budgets.add_usage(tenant_id, scenario, estimated_units) logger.info( "llm_success request_id=%s tenant_id=%s tier=%s scenario=%s group=%s model=%s attempt=%d latency_ms=%d units=%d fallback_from=%s", request_id, tenant_id, tenant_tier, scenario, route.group, model, attempt, latency_ms, estimated_units, fallback_from, ) return response.choices[0].message.content or "" except Exception as exc: last_error = exc logger.warning( "llm_error request_id=%s tenant_id=%s scenario=%s group=%s model=%s attempt=%d error=%s", request_id, tenant_id, scenario, route.group, model, attempt, type(exc).__name__, ) raise RuntimeError(f"LLM route failed request_id={request_id}") from last_error