调用 LLM API 的成本纪律：四个组合拳

做 LLM 应用的早期，很容易陷入一个默认假设：把所有请求都打给最强的模型，反正「质量最重要」。这个假设在原型阶段无害，但一旦上了批量处理或者生产流量，账单的增速往往比你预期的快一个量级。

问题不是「用了贵的模型」，而是对每一次调用缺乏意识——不知道这次花了多少，为什么花，下次怎么少花一点。

我沉淀出来的做法是把四个独立的模式组合进同一条 pipeline，每个模式只做一件事，但组合起来之后，成本就从黑箱变成了可观测的对象。

模型路由：让复杂度决定价格

同一厂商的不同模型，价格差距可以达到 4 倍甚至更多。Haiku 级别的模型处理分类、摘要、格式转换类任务完全够用；Sonnet 级别才值得用在推理链较长、上下文量大的任务上。

路由逻辑不必复杂，两个维度往往就够了：输入文本长度和需要处理的条目数量。

def select_model(text_length: int, item_count: int) -> str:
    if text_length >= 10_000 or item_count >= 30:
        return MODEL_SONNET   # 复杂任务
    return MODEL_HAIKU        # 简单任务，便宜 3-4 倍

关键原则：从最便宜的开始，只在明确不够用时升档。「这个任务感觉挺难的」不是升档的理由；「低档模型给出了有推理缺口的输出」才是。阈值是否合理，要靠实际日志来校准，不要凭感觉定死。

预算熔断：失败要快，不要慢慢烧钱

批量任务有一个典型陷阱：任务跑了一半发现效果不对，但这时候已经花掉了一半的预算。如果能在超出预算时立即停止，至少损失可控。

用不可变数据结构跟踪累计花销是个好主意——每次调用返回一个新的 tracker 而不是修改共享状态，调试时可以完整回溯调用链：

@dataclass(frozen=True, slots=True)
class CostTracker:
    budget_limit: float = 1.00
    records: tuple[CostRecord, ...] = ()
 
    def add(self, record: CostRecord) -> "CostTracker":
        return CostTracker(budget_limit=self.budget_limit, records=(*self.records, record))
 
    @property
    def over_budget(self) -> bool:
        return sum(r.cost_usd for r in self.records) > self.budget_limit

每次调用前先检查 tracker.over_budget，超了就抛异常，不继续往下打。预算限额在任务开始前设定，不是在任务跑完之后对账。

预算是前置约束，不是事后账单

在批量任务启动前设置明确的 budget_limit，让超预算的情况在运行时可见、可停止，而不是等账单邮件到了才发现问题。

窄口重试：只对临时错误重试

不加区分地对所有错误重试，是一种常见的成本陷阱。认证失败重试 3 次，只是多花了 3 倍的时间；prompt 格式有问题重试 10 次，是用 10 倍的成本证明了 prompt 依然有问题。

正确的做法是只对临时性错误重试：网络连接失败、限流（429）、服务端 5xx。对认证错误、请求格式错误这类永久性错误，立即失败，不浪费 token。

_RETRYABLE = (APIConnectionError, RateLimitError, InternalServerError)
 
def call_with_retry(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except _RETRYABLE:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # 指数退避
        # AuthenticationError / BadRequestError 直接往上抛

重试间隔用指数退避，避免在限流期间持续轰炸同一个接口。

提示词缓存：重复的 system prompt 只付一次

如果你的 system prompt 超过 1024 tokens，每次请求都把它完整地发一遍，是一种不必要的浪费。Claude 支持对消息内容打上缓存标记，缓存命中时输入成本会大幅降低，同时首 token 延迟也会缩短。

messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": system_prompt,
                "cache_control": {"type": "ephemeral"},  # 缓存这部分
            },
            {
                "type": "text",
                "text": user_input,  # 每次变化的部分不缓存
            },
        ],
    }
]

只有静态的、每次请求都相同的内容值得缓存；用户输入这类每次都变化的内容不应该加缓存标记。对于批量处理场景，共享同一份 system prompt 加上不同的 user_input，缓存的收益会非常显著。

四个模式组合起来

这四个模式的关系不是非此即彼，而是正交可叠加的：路由决定用哪个模型，缓存降低单次输入成本，熔断控制总量，重试处理抖动。一个完整的 pipeline 函数把它们串起来，签名大概是：

def process(text: str, tracker: CostTracker) -> tuple[Result, CostTracker]:
    model = select_model(len(text), ...)
    if tracker.over_budget:
        raise BudgetExceededError(...)
    response = call_with_retry(lambda: client.messages.create(
        model=model,
        messages=build_cached_messages(system_prompt, text),
    ))
    record = CostRecord(model=model, ...)
    return parse_result(response), tracker.add(record)

返回新的 tracker 而不是修改旧的，让调用方可以完整地观测整个批次的消费分布，而不只是一个最终总数。

一句话心法

成本控制不是省钱，是把每次 LLM 调用变成可观测、可追溯、可干预的动作——知道花在哪里，才能决定哪里值得多花。