本节摘要:分摊让成本可见,预算让它可管。本节先立预算的四层层级:公司 → 团队 → 功能 → 会话/用户——上面三层管"计划",最底下一层管"事故",单个失控会话最多烧掉多少美元必须是设计值而非意外。然后给告警线的设置方法:静态总额线之外,重点讲消耗速率法(日消耗达到日预算的多少、发生在当天几点之前,就说明今天不正常)与三线模型——提醒线(人看一眼)、暂停线(自动降级或暂停非核心功能)、升级线(找负责人做决策);再配动态基线防"温水煮青蛙"。代码给出纯标准库的预算检查器与会话级熔断器。最后一条提醒:预算不是越低越好,砍预算前先看每成功会话成本——成本与质量是一根绳上的两个结。
公司级预算 ──── 月/季度,对齐采购与折扣谈判(财务视角) └─ 团队预算 ── 月,责任归属与 showback/chargeback(第 3.2 节) └─ 功能预算 ── 周/日,产品决策:定价、限额、灰度放量 └─ 会话/用户上限 ── 实时,事故止损:单会话 token 熔断
上面三层是计划工具(周期长、粒度粗),最底下一层是事故工具(实时、单次生效)。多数团队的问题恰恰是只有上面、没有底下:预算按月盯,而一个失控 agent 循环在 40 分钟里就能烧掉一个月预算的可观份额(社区常见事故形态)。
⚠️ 预算的单位建议直接用 token/天与美元/天双轨(社区实践):token 数不受价格调整影响,便于定技术上限;美元数对齐财务。两套数都来自第 3.1 节账本,不必另建。
静态总额线("月预算用了 80% 告警")是最常见也最迟钝的线:月初第 3 天用掉 80% 和第 27 天用掉 80%,是完全不同的危机。改进有两招:
第一招:消耗速率法。 把月预算折成日预算,看"到此刻为止的消耗是否符合时间进度":
(文字流程图) 早 10 点已消耗日预算的 70% ──▶ 速率异常:按此速率全天将达日预算 ~2.5 倍 ──▶ 触发提醒线,先查再烧
第二招:三线模型。 一条线只做一件事,避免"要么不响、要么半夜叫醒所有人":
| 线 | 触发条件(示意) | 动作 | 通知谁 |
|---|---|---|---|
| 提醒线 | 日消耗速率达日预算 150%,或命中率跌破基线 | 人查账本 TopN | 值班群 |
| 暂停线 | 速率达 300%,或单会话超上限 | 自动降级(切小模型/关非核心功能/熔断会话) | 值班 + 功能负责人 |
| 升级线 | 月预算剩余 <10%,或暂停线一天内触发 3 次 | 决策会:加预算还是降规格 | 团队负责人 |
动态基线防温水煮青蛙:用过去 7~14 天同时段消耗做基线(社区常用做法),偏离基线带(如 ±40%,示意)即告警——它抓的是"每天贵一点点"的缓慢漂移,静态线永远抓不到这类问题(第 4 章漂移信号是同一思想在质量侧的应用)。
# budget_alert.py —— 预算检查(消耗速率+三线)与会话级熔断(纯标准库) from collections import defaultdict from dataclasses import dataclass @dataclass class Budget: name: str day_limit: float # 日预算(美元) month_limit: float # 月预算(美元) # 三线参数(示意,按业务调) WARN_RATE = 1.5 # 提醒线:今日消耗达日预算 150% PAUSE_RATE = 3.0 # 暂停线:达日预算 300% ESCALATE_LEFT = 0.1 # 升级线:月预算剩余不足 10% def check(b: Budget, spent_today: float, spent_month: float) -> list[str]: actions = [] rate = spent_today / b.day_limit if rate >= PAUSE_RATE: actions.append(f"[暂停] {b.name}: 今日已花 {spent_today:.2f} 美元" f"(日预算 {b.day_limit:.2f},速率 {rate:.1f} 倍),执行降级") elif rate >= WARN_RATE: actions.append(f"[提醒] {b.name}: 今日消耗速率 {rate:.1f} 倍于日预算,查 TopN") if spent_month >= b.month_limit * (1 - ESCALATE_LEFT): actions.append(f"[升级] {b.name}: 月预算剩余不足 10%,报负责人决策") return actions or [f"[正常] {b.name} 预算内"] class SessionCap: """会话级熔断:单会话 token 超上限直接拒绝,把最坏损失钉在设计值上。""" def __init__(self, cap_tokens: int = 200_000): self.cap = cap_tokens self.used: dict[str, int] = defaultdict(int) self.tripped: set[str] = set() def allow(self, session_id: str, planned_tokens: int) -> bool: if session_id in self.tripped: return False if self.used[session_id] + planned_tokens > self.cap: self.tripped.add(session_id) # 熔断后人工复位,复位流程见 runbook return False self.used[session_id] += planned_tokens return True if __name__ == "__main__": b = Budget("客服-机器人回复", day_limit=120.0, month_limit=3000.0) for line in check(b, spent_today=210.0, spent_month=2750.0): print(line) cap = SessionCap(cap_tokens=200) print("第一次 150:", cap.allow("s-1001", 150)) # True print("第二次 100:", cap.allow("s-1001", 100)) # False(触发熔断) print("复位前再试:", cap.allow("s-1001", 10)) # False(熔断保持)
输出(示意):
[提醒] 客服-机器人回复: 今日消耗速率 1.8 倍于日预算,查 TopN [升级] 客服-机器人回复: 月预算剩余不足 10%,报负责人决策 第一次 150: True 第二次 100: False(触发熔断) 复位前再试: False(熔断保持)
两个实现要点:planned_tokens 用"本次调用预计输入+max_tokens"做预扣(宁可误杀不可漏杀);熔断状态要落库并在观测面板可见——被熔断的会话数本身就是第 4 章的一个质量漂移信号(用户侧表现为"机器人不理我了")。
线设好只是前半程。后半程三件事决定它是不是摆设:
成本治理的终点不是把预算压到零。正确的联动关系:
(文字流程图) 预算吃紧 ──▶ 先看单位经济:每成功会话成本是否恶化? ├──▶ 恶化 ──▶ 是质量问题(循环/冗长输出),修质量就是省成本(第 4 章) └──▶ 持平 ──▶ 是用量增长,谈定价:限额、套餐、缓存工程 (本站《请求缓存与智能路由降低 token 成本》文集)
把预算线砍到影响质量(强制短输出、砍检索篇数),点踩率会先于账单告诉你错了。第 7.1 节会把这套联动制度化为"优化闭环"。
三支柱已立起两根。下一章补上第三根——质量:抽样裁判怎么抽、用户反馈怎么读、漂移信号怎么设,让"好不好"不再靠感觉;三根支柱合拢后,第 5 章的故障诊断就有了全部弹药。