你给 Claude Code 设置了 200 美元月度预算，某个重构任务跑了三小时后突然停止，日志显示「Usage limit reached」——但任务才完成 60%，生成的代码处于半成品状态。或者在 LangGraph 里，一个爬取 + 分析流水线在第 47 个子任务时触发了 OpenAI 的 RateLimitError 兜底逻辑，整个 workflow 回滚，之前 46 步的结果全部丢失。预算耗尽有两个层面的危害：一是直接的经济损失，二是中间状态不可恢复带来的重复计算成本。
常见原因
1. 没有设置每次调用的 max_tokens 上限
模型默认会生成到自然结束或达到模型最大输出长度。如果 Agent 被要求「写完整个模块」，它可能生成数千 token 的代码，而你的预算是按「几百 token 的短回复」估算的。一个任务的实际消耗可以是预估的 10 倍。
怎么判断：查看计费后台的「单次调用 token 分布」，如果 p99 的 completion token 数远高于 p50，说明长尾调用在吃掉大部分预算。
2. 重试风暴放大了消耗
工具调用失败后，Agent 框架自动重试 3 次，每次重试都带上完整的对话历史（随着轮数增加，历史越来越长），导致重试的 token 消耗比第一次调用高出 2-5 倍。
怎么判断：统计 Trace 里同一个 tool call 的重试次数，以及重试时 prompt token 的增长幅度。
3. 子 Agent 数量超出预期
CrewAI 或 AutoGen 的动态任务分解会根据任务复杂度自动增加子 Agent 数量。如果原始任务描述模糊（如「分析这个代码库」），编排器可能启动 20 个子 Agent，而你预期的是 3 个。
怎么判断：在 Trace 里统计实际启动的 Agent 实例数，与 workflow 配置里的 max_workers 或 max_agents 对比。
4. 中间结果没有持久化，失败后全量重跑
任务在第 80% 处失败，但因为没有 checkpoint，框架从头重跑，触发更多 API 调用，进一步消耗预算，形成恶性循环。
怎么判断：检查 workflow 代码里是否有 checkpoint 或 resume 逻辑。如果失败后日志显示从 step 1 重新开始，就是这个问题。
5. 上下文压缩策略缺失
长对话没有使用 prompt caching 或滑动窗口压缩，每轮都把全量历史发给模型，导致 prompt token 随轮数线性增长。到第 50 轮时，每次调用的 prompt 可能已经有 100K token。
怎么判断：用 tiktoken 统计各轮的 prompt token 数，如果它随轮数单调递增，就缺少压缩。
6. 预算阈值设置在了账单层而不是代码层
只在 OpenAI 或 Anthropic 控制台设置了月度硬限额，没有在代码里实现软限额和提前预警。当硬限额触发时，任务直接报错终止，没有机会做优雅降级。
怎么判断：代码里搜索 budget / cost_limit / token_limit 关键词，如果只有硬编码的错误处理而没有主动检查，就是这个问题。
最短修复路径
Step 1：为每次 LLM 调用设置显式 max_tokens
# Anthropic SDK
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048, # 单次最大输出
messages=[{"role": "user", "content": prompt}]
)
# LangChain
llm = ChatAnthropic(model="claude-sonnet-4-6", max_tokens=2048)Step 2：实现代码层软限额，超过 80% 时提前预警
class BudgetTracker:
def __init__(self, max_usd: float):
self.max_usd = max_usd
self.spent = 0.0
def record(self, input_tokens: int, output_tokens: int, model: str):
cost = self._calc_cost(input_tokens, output_tokens, model)
self.spent += cost
ratio = self.spent / self.max_usd
if ratio >= 0.8:
raise BudgetWarning(f"已用 {self.spent:.2f} USD，超过 80% 阈值")
if ratio >= 1.0:
raise BudgetExhausted(f"预算耗尽：{self.spent:.2f} / {self.max_usd:.2f} USD")
def _calc_cost(self, in_tok: int, out_tok: int, model: str) -> float:
# claude-sonnet-4-6: 3 USD/M input, 15 USD/M output
rates = {"claude-sonnet-4-6": (3e-6, 15e-6)}
in_rate, out_rate = rates.get(model, (1e-5, 3e-5))
return in_tok * in_rate + out_tok * out_rateStep 3：启用 checkpoint，支持断点续传
# LangGraph 示例
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
graph = workflow.compile(checkpointer=checkpointer)
# 运行时传入 thread_id，失败后用相同 thread_id 恢复
config = {"configurable": {"thread_id": "task-20260525-001"}}
try:
result = graph.invoke(input_data, config=config)
except BudgetExhausted:
print("预算耗尽，下次用相同 thread_id 从断点继续")Step 4：启用 prompt caching 减少重复 token
# Anthropic prompt caching：对稳定的系统提示加 cache_control
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=[{
"type": "text",
"text": long_system_prompt,
"cache_control": {"type": "ephemeral"} # 缓存系统提示
}],
messages=conversation_history
)Step 5：对子 Agent 数量加硬限制
# CrewAI 限制并发 agent 数
crew = Crew(
agents=agents,
tasks=tasks,
max_rpm=10, # 每分钟最多 10 次 API 调用
max_iter=5, # 每个 task 最多 5 轮迭代
)
# AutoGen 限制对话轮数
groupchat = autogen.GroupChat(
agents=agents,
messages=[],
max_round=20 # 整个 group chat 最多 20 轮
)预防建议
- 在所有 LLM 调用处设置
max_tokens，不要依赖模型默认值 - 用 Langfuse 或 LangSmith 的成本仪表盘监控每条 workflow 的实际消耗，对超出 p95 的任务触发告警
- 为所有长任务启用 checkpoint，保证失败后可以从中断点恢复，而不是全量重跑
- 在任务启动前做「预算预估」：根据任务复杂度和历史数据估算 token 消耗，超出预算直接拒绝启动
- 对递归或动态展开的子任务设置
max_depth和max_agents，防止任务树爆炸 - 启用 Anthropic prompt caching，对稳定的 system prompt 和工具定义加
cache_control，可降低 60-90% 的 prompt token 成本 - 区分「任务预算」和「月度预算」：每个 workflow 实例有独立的 token 预算，超出后优雅降级而不是强制中止
常见问答 (FAQ)
Q: Claude Code 的 Usage Limit 和 API 账单限额有什么区别？ A: Claude Code 的 Usage Limit 是 Anthropic 在订阅层面设置的软限额（如 Pro 计划每 5 小时的 token 配额），超过后需要等待重置或升级计划。API 账单限额是你在 Anthropic Console 里手动设置的月度硬限额，超过后 API 调用直接返回 429。两者都需要在代码层捕获并优雅处理。
Q: 使用 prompt caching 后 checkpoint 还有必要吗？ A: 有。Prompt caching 只降低重复调用的成本，不能恢复任务的执行状态。如果任务在第 30 步失败，没有 checkpoint 就必须从第 1 步重跑（即使每步的 token 成本更低，总成本依然很高）。两者解决不同的问题，应该同时启用。
Q: 任务预算耗尽后，已经生成的中间结果还能用吗？ A: 取决于是否有 checkpoint。有 checkpoint 的框架（LangGraph、Temporal）会持久化每个节点的输出，可以用已完成部分的结果；没有 checkpoint 的框架（如裸 AutoGen、纯 LangChain chain）则全部丢失。这是部署长任务前必须解决的架构问题。
Q: 如何估算一个新任务的 token 消耗？ A: 先用小规模输入（10% 数据量）跑一次，记录实际 token 消耗，然后线性外推。注意：对话式 Agent 的 token 消耗不是线性的——随着轮数增加，历史上下文会累积，总消耗通常是 O(n²) 而不是 O(n)。要在估算中加入这个系数。