客服机器人月账单暴涨？用 tiktoken 计量、提示缓存与超时熔断控住大模型调用成本

我喜欢就喜欢

50人浏览 · 2026-11-06 20:36:53

我喜欢就喜欢 · 2026-11-06 20:36:53 发布

摘要：线上 LLM 调用最容易失控的是「看不见的 token」。本文用 Python 给出生产向最小闭环：tiktoken 预计量、按价粗算费用、按前缀稳定原则吃提示缓存折扣、硬超时 + 有限重试与简易熔断，并把每次调用的费用写进日志，方便对账与限流。

把大模型接到客服分流、摘要、翻译、工单归类这类日活接口后，成本往往不是「单次贵」，而是「不知道贵在哪」。常见现象是：system 提示越写越长、上下文越塞越多、超时后 SDK 默默重试三遍、高峰期同一故障请求把下游打穿。账单月底才报警，再回查代码往往已经改不动——谁加的长提示、哪条链路在重试、缓存命中率是多少，日志里全没有。

本文不讨论模型选型玄学，只落地四件事：调用前能量 token、调用中能吃缓存、失败时有熔断、事后能对账。示例以 OpenAI 兼容接口与 tiktoken 为主；提示缓存与超时策略写成厂商无关的高层模式，便于迁到各家云 API。目标读者是要把 LLM 从 demo 推进生产的后端同学：你要的不是概念图，而是能进代码评审的计量、封装与日志字段。

1. 先计量：tiktoken 数清楚再开口

计费单位是 token，不是字符，更不是「大概几句话」。中文、代码、混合标点在不同分词器下差异很大；对 OpenAI 系模型，本地用 tiktoken 与账单最接近。其它厂商请优先用其官方 count / tokenize API，tiktoken 只做粗估，别拿粗估去跟财务对账。

from dataclasses import dataclass
import tiktoken

@dataclass(frozen=True)
class PriceTable:
    input_per_mtok: float   # 美元 / 百万 input token
    output_per_mtok: float
    cached_input_per_mtok: float  # 缓存命中价，无缓存则等于 input

# 示例价目，上线前换成你们合同价或公开价目表（建议配置中心版本化）
PRICES = {
    "gpt-4o-mini": PriceTable(0.15, 0.60, 0.075),
}

def encoding_for(model: str):
    try:
        return tiktoken.encoding_for_model(model)
    except KeyError:
        # 未知模型名时用较新的通用编码做粗估，并在日志打 warning
        return tiktoken.get_encoding("o200k_base")

def count_tokens(text: str, model: str) -> int:
    return len(encoding_for(model).encode(text or ""))

def estimate_cost(
    model: str,
    prompt_tokens: int,
    max_output_tokens: int,
    cached_prompt_tokens: int = 0,
) -> dict:
    p = PRICES[model]
    billable_input = max(prompt_tokens - cached_prompt_tokens, 0)
    input_cost = billable_input / 1e6 * p.input_per_mtok
    cached_cost = cached_prompt_tokens / 1e6 * p.cached_input_per_mtok
    # 粗算按「可能打满 max_tokens」给上限，真实费用用响应 usage 回填
    output_cost_cap = max_output_tokens / 1e6 * p.output_per_mtok
    return {
        "prompt_tokens": prompt_tokens,
        "cached_prompt_tokens": cached_prompt_tokens,
        "output_cap": max_output_tokens,
        "est_usd_cap": round(input_cost + cached_cost + output_cost_cap, 6),
    }

上线前对「最长 system + 典型用户输入」跑一遍 estimate_cost，把 est_usd_cap × QPS × 86400 写进容量评审。比事后对着发票猜有效得多。实践里再加两条硬规矩：一是接口层设 max_tokens 上限，禁止业务侧无限放开；二是预计量超日预算时直接拒绝或降级到规则引擎，而不是「先调完再看」。

2. 提示缓存：稳定前缀在前，变量内容在后

各家「提示缓存 / prompt cache」细节不同，但共性几乎一致：缓存按精确前缀匹配。前缀里随便改一个时间戳、用户名、请求 ID 或工具定义，后面整段都会 miss，等于白写长提示。高层写法只有一条主线：

固定块靠前：长 system、产品知识、字段说明、安全约束。
可变块靠后：本轮用户问题、会话摘要、动态检索片段。
前缀禁动态字段：当前时间、trace id、随机盐一律不要塞进 system 头部。
短提示可能根本不入缓存；流量太稀疏时 TTL 也会过期，别把「缓存折扣」写进对外承诺。

def build_messages(static_system: str, user_question: str) -> list[dict]:
    # 错误：f"用户{uid}于{now}提问：...\n\n{STATIC}" —— 前缀每次都变，缓存必 miss
    return [
        {"role": "system", "content": static_system},  # 稳定、可复用
        {"role": "user", "content": user_question},    # 变化放最后
    ]

厂商侧：有的自动对超长前缀做缓存，并在 usage 里回传 cached_tokens；有的要在内容块上显式打 cache_control。封装层统一读「缓存命中 token 数」，再喂给价目表即可，不要把厂商字段散落在业务代码里。观测上至少盯两个数：缓存命中率与因前缀抖动导致的 miss 占比；后者往往来自某次「顺手」在 system 里拼了日期。

3. 超时、有限重试与简易熔断

成本失控的另一半来自失败路径：默认重试叠 SDK 重试、超时设得过长、故障时每个请求都干等到超时。原则很简单，也最容易在评审里被漏掉：

客户端只设一层重试；把 SDK 的 max_retries 关掉或置 0，避免「业务重试 × SDK 重试」爆炸。
硬超时（连接 + 读）写死，并单独算「最坏总等待 = 超时 × 尝试次数 + backoff」，确保小于上游 HTTP/网关 deadline。
只对瞬时错误重试（超时、5xx、限流）；参数错、鉴权错、内容违规等 4xx 直接失败，重试只会烧钱。
连续失败打开熔断：冷却期内短路，保护线程池与账单；冷却后半开探测一次再决定是否恢复。

import time
import logging
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout

log = logging.getLogger("llm.cost")

class CircuitOpen(Exception):
    pass

class SimpleBreaker:
    def __init__(self, fail_threshold: int = 3, cool_down_sec: float = 30.0):
        self.fail_threshold = fail_threshold
        self.cool_down_sec = cool_down_sec
        self.failures = 0
        self.open_until = 0.0

    def before_call(self):
        if time.monotonic() < self.open_until:
            raise CircuitOpen("circuit open, skip LLM call")

    def on_success(self):
        self.failures = 0
        self.open_until = 0.0

    def on_failure(self):
        self.failures += 1
        if self.failures >= self.fail_threshold:
            self.open_until = time.monotonic() + self.cool_down_sec
            self.failures = 0

def actual_usd(model: str, in_tok: int, out_tok: int, cached: int) -> float:
    p = PRICES[model]
    return (
        max(in_tok - cached, 0) / 1e6 * p.input_per_mtok
        + cached / 1e6 * p.cached_input_per_mtok
        + out_tok / 1e6 * p.output_per_mtok
    )

def call_with_budget(
    fn,
    *,
    model: str,
    prompt_text: str,
    max_output_tokens: int,
    timeout_sec: float = 20.0,
    max_attempts: int = 2,
    breaker: SimpleBreaker,
):
    breaker.before_call()
    prompt_tokens = count_tokens(prompt_text, model)
    pre = estimate_cost(model, prompt_tokens, max_output_tokens)
    log.info(
        "precheck model=%s tokens=%s est_usd_cap=%s",
        model, prompt_tokens, pre["est_usd_cap"],
    )

    last_err = None
    for attempt in range(1, max_attempts + 1):
        t0 = time.monotonic()
        try:
            with ThreadPoolExecutor(max_workers=1) as pool:
                fut = pool.submit(fn)  # fn 内部关掉 SDK 自动重试
                result = fut.result(timeout=timeout_sec)
            usage = result.get("usage", {})
            in_tok = usage.get("prompt_tokens", prompt_tokens)
            out_tok = usage.get("completion_tokens", 0)
            cached = usage.get("cached_tokens", 0)
            usd = actual_usd(model, in_tok, out_tok, cached)
            log.info(
                "ok model=%s attempt=%s latency_ms=%d in=%d out=%d cached=%d usd=%.6f",
                model, attempt, int((time.monotonic() - t0) * 1000),
                in_tok, out_tok, cached, usd,
            )
            breaker.on_success()
            return result
        except (FuturesTimeout, TimeoutError, OSError) as e:
            last_err = e
            log.warning("retryable fail attempt=%s err=%s", attempt, type(e).__name__)
            time.sleep(0.2 * attempt)  # 极短退避，避免雪崩式叠加重试
        except Exception:
            breaker.on_failure()
            log.exception("non-retryable fail")
            raise
    breaker.on_failure()
    raise RuntimeError(f"LLM call failed after {max_attempts} attempts") from last_err

把「每次调用的 model / token / cached / usd / latency / attempt」打成结构化日志（或写 metrics），按接口维度做日预算告警。很多团队缺的不是更便宜的模型，而是看不见单次调用花了多少；一旦看得见，砍长提示、降 max_tokens、合并重复调用才会变成可验收的优化项。

4. 落地清单（可直接贴进评审）

检查项	建议
预计量	发版前对最长提示跑 tiktoken，估算日费用上限
价目表	版本化配置，价格变更可回放历史账单
提示顺序	静态前缀固定，变量垫底，前缀禁动态字段
超时	单次硬超时；总等待写入 SLA
重试	一层、有限次、只对瞬时错
熔断	连续失败冷却，保护线程与账单
对账	每调用记 usd；日汇总对发票

5. 几个高频踩坑（比换模型更管用）

把「估算 token」当成账单真相。 多模态输入、工具描述、图片 token 往往不在本地粗估里；以响应 usage 为准回填，估算只负责事前拦截。
system 每次拼接日期或用户画像。 缓存命中率会断崖下跌，长提示等于按全价反复付。
超时与重试叠两层。 业务层 2 次 × SDK 默认 2～3 次，一次故障可变成六次计费尝试。
只记成功调用的费用。 超时与熔断前的半截请求、限流重试同样产生成本，失败路径也要打 usd 字段（可为 0 但要标 attempt）。
价目表写死在代码常量里且不版本化。 厂商调价后历史日志无法复算，对账只能靠发票总额。

把上述坑写进 code review checklist，通常比再调一次 temperature 更能稳住月账单。

综合来看，客服机器人的「月账单暴涨」通常能从「月底惊吓」变成「白天可熔断、可降级、可解释」。成本治理不是一次性优化，而是把计量、缓存、超时与日志变成和业务代码同级的基础设施；评审里缺任何一项，都建议打回补齐再上线。

你们线上目前是按「预计量拦截超预算请求」，还是只在账单出来后做事后归因？