1. 事故现场：一个协程跑掉 3000 万 Token 的账单是怎么来的
先说结论：这不是被攻击，也不是密钥泄露，而是一个后台异步 Agent 协程在处理一篇格式错乱的 PDF 时，陷入了递归推理，在无人值守的情况下独自跑了 8 个小时，单次 Session 消耗了 3000 万 Token，隔夜账单直接飙到上千美金。财务早上发邮件质问的时候，我们才意识到——代码里给 CPU 和内存都设了 limit，唯独没给 Token 设上限。
如果你正在写 Agent、写异步任务、写后台批处理，并且调用的是按 Token 计费的 LLM API，那这篇文章就是写给你的。它不聊虚的架构图，只交付一套可以复制粘贴的「预算闸门」骨架：单协程 Token 上限、超限熔断、告警触发，以及怎么验证拦截真的生效。适合谁？适合所有把 LLM 调用放进循环、放进递归、放进 while True 的人——也就是几乎所有 Agent 开发者。
事故的根因其实很朴素。传统后端里，一个失控的循环最多把 CPU 打满，K8s 会 OOMKill 它，成本可控。但 LLM 调用不一样：每一次调用都是一次真实的、按量计费的网络请求，循环不会崩，它只会安静地、一笔一笔地花钱。更麻烦的是 LLM 输出具有非确定性，模型可能因为一段乱码 PDF 反复「再想想」，Agent 的 planner 看到「还没完成」就再调一次，于是递归停不下来。没有预算闸门，就等于给这段代码开了一张没有额度上限的信用卡。
我试过最直接的止血方式：在调用 LLM 的那一层包一个计数器，超过阈值就抛错。听起来简单，但要做得线程安全、要区分 prompt 和 completion 的 Token、要在流式返回时也能实时打断，还是有几个坑。下面从接入层开始，一步步把这套闸门搭起来。
2. 前置准备：用 TaoToken 统一接入，让预算闸门有地方挂
预算闸门要生效，前提是你能在一个统一的入口观测和拦截所有 LLM 调用。如果团队里每个 Agent 各写各的 SDK、各配各的 key，那闸门只能挂在某一个协程里，漏掉的调用照样烧钱。所以第一步是把调用收敛到一个兼容 OpenAI 协议的入口。
TaoToken 在这里的角色是统一接入层：它提供 OpenAI 兼容的 API，你现有的openaiSDK 或httpx调用几乎不用改，只需要换base_url和api_key。官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 端点是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数，直接用它做 base_url 即可。
为什么统一入口对预算治理这么关键？因为闸门需要拿到每次调用的 usage 字段（prompt_tokens / completion_tokens）。如果调用散落在十几个地方，你没法保证每个地方都上报。收敛到一个 base_url 之后，你可以在自己的封装层里统一读取 usage，统一扣减预算，统一触发熔断。这跟「把数据库连接收敛到连接池」是一个道理——治理的前提是收口。
拿到 Key 的路径是：登录后进入控制台，在 API Keys 页面创建一个新 key。控制台入口是 https://taotoken.net/console ，创建 key 的页面是 https://taotoken.net/api-keys 。建议给后台 Agent 单独建一个 key，不要和线上核心业务共用，这样即使某个 Agent 失控，你也能在控制台按 key 维度看到它的消耗曲线，快速定位是哪个协程在烧钱。
如果你只是想先验证模型通不通、usage 字段长什么样，可以直接用模型对话页面手动发一条请求看看返回结构： https://taotoken.net/models 。这一步不用写代码，先确认返回体里有 usage 对象，后面闸门才有数据可扣。
3. 可复制配置：线程安全的 Token 预算闸门骨架
下面这套代码是 Go 版本，核心是一个线程安全的计数器加一个熔断判断。为什么用atomic而不是mutex？因为 Agent 协程可能并发调用，锁竞争会拖慢推理链路，而预算扣减本身是「加一个数、比一个数」的简单操作，原子操作足够且更快。
package budget import ( "context" "errors" "fmt" "sync/atomic" ) // ErrBudgetExceeded 预算超限错误，调用方据此熔断 var ErrBudgetExceeded = errors.New("session token budget exceeded") // Gatekeeper 单协程/单 Session 的 Token 预算闸门 type Gatekeeper struct { maxTokens int64 // 硬上限 usedTokens int64 // 已消耗，原子读写 warnTokens int64 // 预警线，触发告警 warned int32 // 是否已告警，避免重复推送 } func NewGatekeeper(maxTokens, warnTokens int64) *Gatekeeper { return &Gatekeeper{ maxTokens: maxTokens, warnTokens: warnTokens, } } // Consume 扣减预算，超额立即返回错误触发熔断 func (g *Gatekeeper) Consume(tokens int64) error { newTotal := atomic.AddInt64(&g.usedTokens, tokens) if newTotal > g.maxTokens { return fmt.Errorf("%w: used=%d limit=%d", ErrBudgetExceeded, newTotal, g.maxTokens) } // 预警线只触发一次 if newTotal > g.warnTokens && atomic.CompareAndSwapInt32(&g.warned, 0, 1) { go sendAlert(fmt.Sprintf("session token warning: used=%d", newTotal)) } return nil } func (g *Gatekeeper) Used() int64 { return atomic.LoadInt64(&g.usedTokens) }关键点有三个。第一，Consume在调用前扣 prompt 预估、调用后扣实际 completion，两次都判断，避免「先花后算」的窗口期。第二，预警线用CompareAndSwap保证只告警一次，否则一个失控协程会刷爆你的告警群。第三，maxTokens是硬上限，一旦超过直接返回错误，调用方必须break或return，不能吞掉错误继续循环。
接下来是带预算拦截的 LLM 调用包装。这里用 OpenAI 兼容的 HTTP 调用演示，你可以换成任意 SDK：
package budget import ( "bytes" "context" "encoding/json" "net/http" ) type chatReq struct { Model string `json:"model"` Messages []message `json:"messages"` } type message struct { Role string `json:"role"` Content string `json:"content"` } type chatResp struct { Choices []struct { Message message `json:"message"` } `json:"choices"` Usage struct { PromptTokens int64 `json:"prompt_tokens"` CompletionTokens int64 `json:"completion_tokens"` TotalTokens int64 `json:"total_tokens"` } `json:"usage"` } // CallLLMWithBudget 带预算闸门的调用 func CallLLMWithBudget(ctx context.Context, g *Gatekeeper, prompt string) (string, error) { // 1. 调用前按 prompt 长度预估，先扣一笔，防止并发穿透 estimatedPrompt := int64(len(prompt) / 3) // 粗略估算，中文约 1 token/字 if err := g.Consume(estimatedPrompt); err != nil { return "", err } body, _ := json.Marshal(chatReq{ Model: "gpt-4o-mini", Messages: []message{{Role: "user", Content: prompt}}, }) req, _ := http.NewRequestWithContext(ctx, "POST", "https://taotoken.net/api/v1/chat/completions", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+apiKey) resp, err := http.DefaultClient.Do(req) if err != nil { return "", err } defer resp.Body.Close() var out chatResp if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { return "", err } // 2. 调用后按真实 usage 补扣，修正预估误差 if err := g.Consume(out.Usage.TotalTokens); err != nil { return "", err } return out.Choices[0].Message.Content, nil }注意base_url用的是https://taotoken.net/api/v1/chat/completions，这是 OpenAI 兼容路径。如果你用官方openai-goSDK，把BaseURL设成https://taotoken.net/api/v1即可，其余代码不用动。
然后是主循环，也就是事故里那个「递归推理」的位置。给它套上闸门：
func RunAgentSession(ctx context.Context, task string) error { // 单 Session 硬上限 100k，预警线 50k g := NewGatekeeper(100_000, 50_000) for step := 1; step <= 100; step++ { out, err := CallLLMWithBudget(ctx, g, task+" 继续推理下一步") if err != nil { if errors.Is(err, ErrBudgetExceeded) { // 熔断：记录现场，退出循环，绝不重试 log.Printf("[BUDGET CUTOFF] step=%d used=%d", step, g.Used()) return err } return err } _ = out } return nil }这段代码里最重要的不是计数器，而是errors.Is(err, ErrBudgetExceeded)之后直接 return，不重试。事故之所以烧到 3000 万，就是因为外层有个retry逻辑，把「模型没返回有效结果」当成可重试错误，于是无限重试。预算熔断必须被识别为「不可重试错误」，这是纪律问题，不是技术问题。
4. 验证请求：怎么确认闸门真的拦住了
写完代码不验证，等于没写。验证分三步，从单元测试到真实请求。
第一步，单元测试直接打满预算，确认熔断触发：
func TestGatekeeperCutoff(t *testing.T) { g := NewGatekeeper(2000, 1500) // 模拟三轮调用，每轮 1200 token for i := 1; i <= 3; i++ { err := g.Consume(1200) if err != nil { t.Logf("第 %d 轮被拦截: %v, used=%d", i, err, g.Used()) if !errors.Is(err, ErrBudgetExceeded) { t.Fatalf("期望预算超限错误，实际: %v", err) } return } } t.Fatal("三轮后仍未触发熔断，闸门失效") }预期输出是第 2 轮就被拦截（1200+1200=2400 > 2000），used停在 2400。如果三轮都过了，说明你的maxTokens设太大或者判断写反了。
第二步，发一条真实请求，确认 usage 字段能读到。用 curl 直接打：
curl https://taotoken.net/api/v1/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -d '{ "model": "gpt-4o-mini", "messages": [{"role": "user", "content": "用一句话解释什么是 Token 预算闸门"}] }'返回体里会带usage.total_tokens，把这个值打印出来，确认它和你代码里扣减的数字一致。如果 usage 是 0 或者缺失，检查是不是用了流式（stream）模式——流式返回默认不带 usage，需要额外参数或在最后一个 chunk 里取。
第三步，做一次「失控模拟」。故意写一个不设退出条件的循环，把maxTokens设成 5000，跑起来看它是不是在消耗到 5000 左右就抛错退出。这一步是端到端验证，能同时验证计数器、熔断、日志三件事。跑通之后，你会看到类似这样的日志：
[BUDGET CUTOFF] step=4 used=5120看到这行，闸门就算装好了。
5. 本篇常见错排查
错误一：ErrBudgetExceeded被外层 retry 吞掉。现象是闸门明明触发了，账单还在涨。排查方法：在熔断分支打日志，看它有没有被retry包住。修复方式是给错误加类型判断，预算超限直接return，不进入重试队列。
错误二：并发协程共享一个 Gatekeeper，但用了普通 int 而非 atomic。现象是used数字对不上，偶尔超过上限。原因是usedTokens += tokens在并发下会丢更新。修复：所有读写走atomic.AddInt64/atomic.LoadInt64，不要用mutex包一层再读裸变量。
错误三：只扣 completion 不扣 prompt。现象是实际消耗比闸门统计高 30% 以上。prompt 在长上下文 Agent 里往往比 completion 还大，必须两次都扣。修复：调用前按 prompt 预估扣一次，调用后按 usage 补扣。
错误四：流式返回时闸门失效。现象是开了 stream 之后，usage拿不到，扣减为 0。修复：流式场景下按 chunk 数量或字符数实时估算扣减，或者请求时带上stream_options: {"include_usage": true}，在最后一个 chunk 取 usage。
错误五：预警线告警刷屏。现象是失控协程每秒推一条告警，运维群被淹没。修复：用CompareAndSwap保证预警只触发一次，或者加时间窗口限流。
错误六：base_url 写错导致请求 404。现象是调用直接失败，闸门根本没机会工作。检查点：OpenAI 兼容路径是https://taotoken.net/api/v1，不要漏掉/v1，也不要带 UTM 参数。接入文档在 https://taotoken.net/doc ，里面有完整的路径说明和示例。
6. 把闸门装进你的 Agent：下一步做什么
预算闸门不是一次性工程，它需要跟着你的 Agent 一起演进。单协程上限只是第一层，往上还有单用户单日、单任务类型、系统全局三层。建议的阈值骨架是：单次调用 8k、单 Session 100k、单用户单日 2M、系统单日按预算倒推。每一层都用同样的Consume模式，只是作用域不同。
如果你还在频繁调模型、反复试 prompt，可以先用模型对话页面手动验证返回结构和 usage： https://taotoken.net/models 。等 Agent 逻辑稳定了，需要长期跑编码类任务或常驻 Agent，可以看 Coding Plan： https://taotoken.net/coding-plan ，它更适合有持续调用需求的场景。所有调用的 Key 统一在 https://taotoken.net/api-keys 管理，接入细节查 https://taotoken.net/doc 。
最后留一个我踩过的坑：闸门的阈值不要拍脑袋定，先用一周的真实数据跑出 P99 消耗，再乘以 1.5 作为上限。定太低会误杀正常任务，定太高等于没装。装好之后，你会发现自己终于敢让 Agent 在夜里自己跑了——因为你知道，最坏情况下它也只能花掉你预设的那点预算。
