1. 为什么你的 LLM 项目总在月底被账单“背刺”做智能体开发的朋友大概率都经历过这种时刻功能跑通了Demo 演示也顺利结果月底一看 API 账单整个人都不好了。尤其是用 LangChain 搭多步 Chain 或者 Agent 的时候一次用户请求背后可能触发五六次 LLM 调用每次调用都在烧 Token而你根本不知道钱具体花在了哪一步。这个问题的根源在于LLM 调用是黑盒的。你调用chain.invoke()它内部可能先做意图识别、再检索、再总结、再格式化输出每一步都是一次独立的模型请求。如果没有埋点你只能看到最终结果看不到中间过程。等到发现成本失控钱已经花出去了。LangChain 的回调机制Callback System就是为解决这个问题设计的。它允许你在 Chain、LLM、Tool、Retriever 的各个生命周期节点插入钩子函数实时捕获 Token 使用量、调用耗时、模型名称等关键数据。基于这些数据你可以做三件事实时监控知道钱花在哪、阈值告警快超支时收到通知、自动降级超预算时切换到便宜模型或直接拦截。这篇文章面向的是已经用 LangChain 搭过 Chain 或 Agent、但对成本控制还没有系统方案的开发者。我会从回调埋点开始一步步给出可复制的配置代码演示如何设置 Token 预算阈值、如何在超预算时触发降级、以及如何验证拦截是否生效。全程用 OpenAI 兼容接口举例你可以直接替换成自己的模型服务地址。先说结论成本控制的核心不是“省”而是“可控”。你需要知道每一分钱花在哪、什么时候会超、超了之后怎么办。下面进入具体操作。2. 用 TaoToken 统一接入多模型让回调数据有处可查在讲回调配置之前先解决一个前置问题多模型接入的统一性。很多团队同时用 GPT-4、GPT-3.5、Claude 等不同模型每个模型的 Token 计费方式、返回字段格式都不一样。如果回调里要针对每个模型写不同的解析逻辑维护成本很高。我试过用 TaoToken 做统一接入层它提供 OpenAI 兼容的 API 格式所有模型走同一个 Base URL返回结构一致。这样回调里的 Token 解析逻辑只需要写一套不用为每个模型单独适配。具体来说TaoToken 的核心价值在于统一 Base URL所有模型请求都发到https://taotoken.net/api不用在代码里维护多个 endpoint。回调里拿到的LLMResult结构统一token_usage字段格式一致。模型 ID 标准化通过统一的 Model ID 指定模型回调里serialized参数拿到的模型名称是标准化的方便做成本映射。Key 管理集中一个 API Key 管理所有模型调用回调里做用户级配额时不用关心 Key 的归属问题。如果你还没配置过可以按下面三步操作第一步访问 TaoToken 控制台 创建一个 API Key。建议为不同环境开发/测试/生产创建不同的 Key方便后续做项目级配额。第二步在 API Keys 页面 复制你的 Key保存到环境变量里。不要硬编码在代码中。第三步在 LangChain 里配置 ChatOpenAI 时把base_url指向 TaoToken 的 API 地址import os from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-3.5-turbo, # 或 gpt-4、claude-3-sonnet 等 base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], temperature0.3, )配置好之后你的所有 LLM 调用都会经过 TaoToken 转发回调里拿到的 Token 数据是统一的。这样后面写成本监控和预算拦截时只需要处理一种数据格式。如果你用的是 Claude Code 做开发辅助TaoToken 也支持 Anthropic 格式的接入具体可以参考 Claude Code 接入文档。不过本文主要聚焦 LangChain 回调Claude Code 的配置不展开。有一点需要注意TaoToken 是 API 接入层不是模型本身。它帮你统一了接口格式和 Key 管理但 Token 计费还是按实际调用的模型来算。所以回调里的成本计算逻辑还是要根据你实际使用的模型来配置单价。3. 可复制的回调配置Token 预算拦截器完整代码这一节给出完整的可复制配置。核心思路是写一个自定义的AsyncCallbackHandler在on_llm_end里捕获 Token 使用量累计到当前请求的预算上下文中当累计值超过阈值时在on_llm_start里抛出异常阻止后续 LLM 调用。先看配置文件。我习惯把预算策略放在一个独立的 JSON 文件里方便不同环境切换{ budget_policy: { per_request_limit: 8000, per_request_soft_limit: 6000, daily_limit: 200000, hourly_limit: 30000, alert_threshold: 0.8, degradation_model: gpt-3.5-turbo, premium_model: gpt-4-turbo, cost_per_1k_tokens: { gpt-4-turbo: 0.02, gpt-3.5-turbo: 0.001 } } }把这个文件保存为budget_config.json放在项目根目录。下面是对应的回调实现import json import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from langchain_core.callbacks import AsyncCallbackHandler from langchain_core.outputs import LLMResult dataclass class BudgetContext: 单次请求的预算上下文 request_id: str total_tokens: int 0 total_cost: float 0.0 call_count: int 0 soft_limit_hit: bool False hard_limit_hit: bool False model_usage: Dict[str, int] field(default_factorydict) class TokenBudgetCallback(AsyncCallbackHandler): Token 预算拦截回调 def __init__(self, config_path: str budget_config.json): with open(config_path, r) as f: cfg json.load(f)[budget_policy] self.per_request_limit cfg[per_request_limit] self.soft_limit cfg[per_request_soft_limit] self.alert_threshold cfg[alert_threshold] self.cost_map cfg[cost_per_1k_tokens] self.degradation_model cfg[degradation_model] self.context BudgetContext(request_idfreq-{int(time.time())}) async def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any, ) - None: LLM 调用前检查预算 if self.context.hard_limit_hit: raise BudgetExceededError( f请求 {self.context.request_id} 已超硬限制 f({self.context.total_tokens}/{self.per_request_limit})拒绝继续调用 ) # 预估本次 Prompt 的 Token粗略估算1 token ≈ 4 字符 estimated sum(len(p) // 4 for p in prompts) if self.context.total_tokens estimated self.per_request_limit: self.context.hard_limit_hit True raise BudgetExceededError( f预估 Token {estimated} 将导致超限当前已用 f{self.context.total_tokens}限制 {self.per_request_limit} ) async def on_llm_end(self, response: LLMResult, **kwargs: Any) - None: LLM 调用后累计 Token 和成本 for gen_list in response.generations: for gen in gen_list: info gen.generation_info or {} usage info.get(token_usage, {}) tokens usage.get(total_tokens, 0) model (response.llm_output or {}).get(model_name, unknown) self.context.total_tokens tokens self.context.call_count 1 self.context.model_usage[model] ( self.context.model_usage.get(model, 0) tokens ) # 计算成本 price self.cost_map.get(model, 0.001) self.context.total_cost (tokens / 1000) * price # 软限制检查 if ( not self.context.soft_limit_hit and self.context.total_tokens self.soft_limit ): self.context.soft_limit_hit True print( f[BUDGET WARNING] 请求 {self.context.request_id} f已达软限制 {self.soft_limit}当前 {self.context.total_tokens} tokens ) def get_summary(self) - Dict[str, Any]: return { request_id: self.context.request_id, total_tokens: self.context.total_tokens, total_cost: round(self.context.total_cost, 6), call_count: self.context.call_count, model_usage: self.context.model_usage, soft_limit_hit: self.context.soft_limit_hit, hard_limit_hit: self.context.hard_limit_hit, } class BudgetExceededError(Exception): 预算超限异常 pass这段代码的关键设计点预检在on_llm_start在 LLM 调用真正发生之前先估算 Prompt 的 Token 数如果加上已用量会超限直接抛异常。这样能避免“钱已经花了才发现超支”的问题。累计在on_llm_end每次 LLM 调用结束后从generation_info里提取真实的token_usage累加到上下文。不同模型的返回字段可能略有差异这里做了兼容处理。软硬双阈值软限制6000触发告警但不拦截硬限制8000直接抛异常。这样给开发者一个缓冲区间可以在软限制触发时做降级处理。成本实时计算每次调用后立刻算出累计成本方便后续做成本维度的配额管理。把这个回调注入到 Chain 里from langchain_core.prompts import ChatPromptTemplate callback TokenBudgetCallback(budget_config.json) llm ChatOpenAI( modelgpt-3.5-turbo, base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], callbacks[callback], ) prompt ChatPromptTemplate.from_template(请详细分析以下问题{question}) chain prompt | llm try: result chain.invoke( {question: 解释一下 Transformer 的注意力机制}, config{callbacks: [callback]}, ) print(callback.get_summary()) except BudgetExceededError as e: print(f预算拦截{e})注意config{callbacks: [callback]}这一行很关键。LangChain 的 Chain 和 LLM 各自有独立的回调配置如果只在 LLM 上设置回调Chain 级别的生命周期事件如on_chain_start不会触发。两边都设置才能完整覆盖。4. 验证请求一次超预算拦截的完整演示配置写好了怎么验证它真的生效这一节用一个具体的超预算场景来演示。先构造一个会触发硬限制的请求。假设per_request_limit设为 8000我们故意传一个超长文本import asyncio from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate async def test_budget_interception(): callback TokenBudgetCallback(budget_config.json) llm ChatOpenAI( modelgpt-3.5-turbo, base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], callbacks[callback], ) prompt ChatPromptTemplate.from_template( 请逐段分析以下文本每段给出详细评论\n\n{text} ) chain prompt | llm # 构造一个超长文本约 40000 字符 ≈ 10000 tokens long_text 这是一段测试文本用于验证 Token 预算拦截机制。 * 1000 try: result await chain.ainvoke( {text: long_text}, config{callbacks: [callback]}, ) print(调用成功未触发拦截) print(callback.get_summary()) except BudgetExceededError as e: print(f拦截成功{e}) print(f拦截时状态{callback.get_summary()}) asyncio.run(test_budget_interception())运行这段代码你会看到类似这样的输出拦截成功预估 Token 10000 将导致超限当前已用 0限制 8000 拦截时状态{request_id: req-1712345678, total_tokens: 0, total_cost: 0.0, call_count: 0, model_usage: {}, soft_limit_hit: False, hard_limit_hit: True}注意total_tokens是 0因为拦截发生在on_llm_start阶段LLM 调用还没有真正执行。这就是预检的价值在花钱之前拦住。再测试一个软限制场景。把文本长度控制在 6000-8000 tokens 之间async def test_soft_limit(): callback TokenBudgetCallback(budget_config.json) llm ChatOpenAI( modelgpt-3.5-turbo, base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], callbacks[callback], ) prompt ChatPromptTemplate.from_template(总结{text}) chain prompt | llm # 约 7000 tokens 的文本 medium_text 这是一段中等长度的测试文本。 * 500 result await chain.ainvoke( {text: medium_text}, config{callbacks: [callback]}, ) summary callback.get_summary() print(f调用成功Token 使用{summary[total_tokens]}) print(f软限制触发{summary[soft_limit_hit]}) print(f成本${summary[total_cost]}) asyncio.run(test_soft_limit())输出会显示soft_limit_hit: True同时打印告警信息。这时候你可以选择在软限制触发后做降级处理比如把后续调用切换到gpt-3.5-turbo。如果你想在真实的多步 Chain 里验证可以搭一个简单的 Agentfrom langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.tools import tool tool def search(query: str) - str: 搜索工具返回模拟结果 return f关于 {query} 的搜索结果... tools [search] agent_prompt ChatPromptTemplate.from_template( 你是一个助手可以使用工具回答问题。\n\n问题{input} ) agent create_openai_tools_agent(llm, tools, agent_prompt) executor AgentExecutor(agentagent, toolstools, callbacks[callback]) result await executor.ainvoke( {input: 帮我搜索 LangChain 回调机制}, config{callbacks: [callback]}, ) print(callback.get_summary())Agent 场景下一次用户请求可能触发多次 LLM 调用思考→调工具→再思考→输出。回调会累计所有调用的 Token最终汇总在get_summary()里。如果中间某次调用触发了硬限制整个 Agent 执行会被中断已消耗的 Token 会记录在上下文里。验证通过后你可以把get_summary()的数据上报到监控系统。最简单的做法是打印到日志进阶做法是发送到 Prometheus 或自建的时序数据库。TaoToken 控制台也提供了用量统计可以作为对账参考。5. 常见报错排查401、local proxy failed、reading choices 怎么处理配置回调的过程中最容易遇到的报错集中在接入层和回调数据解析层。这一节列出几个高频问题和对应对策。报错一401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}这个报错说明 API Key 配置有问题。检查三个地方第一环境变量TAOTOKEN_API_KEY是否设置正确可以用echo $TAOTOKEN_API_KEY确认第二ChatOpenAI初始化时api_key参数是否传了正确的值第三Key 是否已过期或被禁用去 API Keys 页面 确认状态。如果用的是.env文件注意load_dotenv()的调用时机要在ChatOpenAI初始化之前。报错二local proxy failed / Connection erroropenai.APIConnectionError: Connection error.这个报错通常是网络层的问题。检查base_url是否写对应该是https://taotoken.net/api不要多加路径或斜杠。如果公司网络有出口限制确认能正常访问该地址。另外检查是否有环境变量HTTP_PROXY或HTTPS_PROXY干扰了请求可以临时 unset 掉再试。报错三reading choices / KeyError: choicesKeyError: choices这个报错说明回调里解析LLMResult时假设了返回结构里有choices字段但实际返回格式不同。在on_llm_end里不要直接访问response.llm_output[choices]而是用.get()做兼容async def on_llm_end(self, response: LLMResult, **kwargs): llm_output response.llm_output or {} model llm_output.get(model_name, unknown) # 不要写 llm_output[choices]用 get for gen_list in response.generations: for gen in gen_list: usage (gen.generation_info or {}).get(token_usage, {}) # 处理 usage报错四OAuth / token refresh failed如果你用的是需要 OAuth 的模型服务可能会遇到 token 刷新失败。LangChain 的ChatOpenAI默认用 API Key 认证不涉及 OAuth。如果你在用其他需要 OAuth 的集成检查 token 是否过期、refresh token 是否有效。对于 TaoToken 接入直接用 API Key 即可不需要 OAuth 流程。报错五回调没有触发 / Token 数据为 0回调配置了但get_summary()返回全 0通常是两个原因第一回调没有同时注入到 Chain 和 LLM只在一边设置了第二用的是同步invoke但回调是AsyncCallbackHandler需要改用ainvoke或者用同步的BaseCallbackHandler。检查方法在on_llm_end里加一行print(callback triggered)看是否真的被调用。如果没有打印说明回调没注入成功。报错六BudgetExceededError 没有抛出预检逻辑没生效检查on_llm_start里的估算逻辑。如果 Prompt 很短但实际输出很长预检可能不会触发但on_llm_end里的累计会触发软限制。硬限制的预检只检查 Prompt 部分输出部分的 Token 无法预知所以硬限制主要防的是“输入超长”场景。对于输出超长需要在on_llm_new_token里做流式拦截如果开启了 streaming。排查完这些常见问题你的回调应该能稳定运行了。如果还有异常可以去 接入文档 查一下接口返回格式的说明对照回调里的解析逻辑。6. 从监控到降级把预算控制接入你的日常开发流回调跑通之后下一步是把它变成日常开发的一部分。我的做法是在开发环境用宽松阈值只告警不拦截在预发环境用中等阈值软限制触发降级在生产环境用严格阈值硬限制直接拦截。降级策略可以这样实现在on_llm_end里检测到软限制触发后动态切换后续调用的模型。LangChain 支持在运行时通过config覆盖模型参数但更简单的做法是在 Chain 层面做路由async def smart_invoke(chain, inputs, callback): try: return await chain.ainvoke( inputs, config{callbacks: [callback]} ) except BudgetExceededError: # 降级到便宜模型重试 cheap_llm ChatOpenAI( modelgpt-3.5-turbo, base_urlhttps://taotoken.net/api, api_keyos.environ[TAOTOKEN_API_KEY], callbacks[callback], ) cheap_chain chain.prompt | cheap_llm return await cheap_chain.ainvoke( inputs, config{callbacks: [callback]} )对于长期运行的 Agent 服务建议把预算上下文持久化到 Redis这样跨请求的累计消耗也能追踪。TaoToken 的 Coding Plan 提供了包月套餐适合高频调用的场景配合回调的用量统计可以更精确地做容量规划。如果你还在选模型阶段可以先用 模型对话 测试不同模型的输出质量和 Token 消耗再决定生产环境用哪个。实测下来简单任务用 gpt-3.5-turbo 能省 90% 以上的成本复杂推理再切到 gpt-4-turbo。最后提醒一点回调里的成本计算用的是硬编码单价实际账单可能有差异。建议每周对一次账用 TaoToken 控制台的用量数据校准回调里的cost_map。这样你的预算拦截才会越来越准。