1. 这篇文章真正要解决的问题
过去半年,明显感觉到一个现象:越来越多团队开始用 AI 智能体跑真实业务,但真正把智能体推到生产环境的人,几乎都遇到过同一个尴尬时刻——任务执行到一半,预算没了。
这里的“预算”不一定指钱。
它可能是 API 的 token 配额,可能是上下文窗口的长度上限,可能是单次任务的耗时上限,也可能是用户给智能体设定的“最多调用几次工具”。无论哪种形式,结果都类似:智能体在任务进行到最关键一步时被硬生生掐断,用户拿到的不是最终答案,而是一堆中间产物、一个报错,或者干脆是一段沉默。
更麻烦的是,很多智能体在预算耗尽前根本不会提前规划。它不知道自己还能用多少 token,不知道当前任务还有多长,不知道哪些步骤可以放弃、哪些步骤必须保住。于是它只能凭感觉继续往前跑,直到触到限制,然后“牺牲”——中途退出、返回截断内容、循环重试,或者给出一个质量明显下降的结果。
这个困境的本质,不是模型能力不够,而是工程预算约束没有被纳入智能体的设计逻辑。一个 AI 智能体如果只想着“完成任务”,不考虑“剩余资源够不够完成”,在生产环境里必然出问题。
从实践看,这个领域真正值得关注的不是某个具体框架,而是一套“预算感知”的工程设计思路:任务怎么拆、资源怎么分、中途怎么检查、快耗尽时怎么降级、最后怎么兜底。这篇文章就围绕这套思路展开,读完你可以直接照着做,把你的智能体从“跑着跑着就牺牲”改造成“预算再紧也能善终”。
2. 智能体“牺牲”的四种典型形态
先梳理一下智能体在预算耗尽前最常见的四种失败形态。只有先能准确识别“它正在走向牺牲”,才谈得上干预。
第一种:上下文窗口溢出。现在主流大模型的 context window 虽然越来越大,但并不是无限大。智能体在执行长任务时,会把历史对话、工具返回结果、中间文件内容不断塞进上下文。当 token 总数超过窗口上限,模型直接报错,之前的所有工作全部作废。这种情况在“读长文档 + 多轮工具调用 + 生成最终报告”这类任务中尤其常见。
第二种:成本预算超支。很多智能体部署在线上服务里,按 token 计费。如果任务没有事先拆分成本预算,一个任务可能调用几十次外部 API,每次调用回传的内容又很大,最后账单远远超出预期。这种失败不像上下文溢出那样有明确的报错,往往是在月底看账单时才被发现,但伤害更大。
第三种:执行循环与死锁。智能体在某个子任务上反复试错,比如同一个工具失败后重试、重试后再失败、再换参数重试。每轮循环都在消耗 token 和调用次数,但进度为零。如果没有明确的循环上限和轮次限制,它会在预算耗尽之前一直打转。
第四种:质量静默下降。这种最隐蔽。有的智能体在接近预算上限时,不会直接失败,而是开始“敷衍”——摘要越来越短、关键细节丢失、跳过核验步骤、直接假设某个外部调用成功。结果接口返回 200,但答案质量已经严重缩水,线上用户根本不会知道。
这四种形态都有一个共同点:问题不是出在“模型不会做”,而是出在“系统没让它在有限资源下学会取舍”。传统软件里有超时时间、有熔断、有降级策略,到了 AI 智能体这里,很多团队反而把这些工程常识忘了,直接把所有决策权交给模型。
所以下面要讲的,就是把传统分布式系统的“预算 - 熔断 - 降级 - 兜底”思路,迁移到智能体架构里。
3. 核心概念:预算感知智能体
在动手写代码之前,需要先统一几个术语。这些词后面会反复出现,如果概念不清楚,看代码时容易卡住。
Token 预算(Token Budget):一次任务允许消耗的最大 token 数量。它可能是钱(API 计费),也可能是长度(上下文窗口限制),在工程上统一建模为一个数值即可。
上下文窗口(Context Window):模型单次处理文本的最大长度。所有历史消息、工具结果、系统提示词都要放在这个窗口里。预算管理必须考虑“窗口剩余空间”,即使你的费用预算还很充足。
任务分解(Task Decomposition):把一个复杂任务拆成若干子任务,每个子任务有独立的资源预算。拆分之后,单个子任务的失败不会拖垮整个任务。
检查点(Checkpoint):智能体在某个子任务完成后,主动检查当前预算消耗和任务进度,决定继续、压缩、降级还是终止。
优雅降级(Graceful Degradation):当预算不足时,智能体主动放弃非必要环节,只保证核心目标完成。比如不生成详细图表,只给结论;不重跑验证,只标记“未验证”。
兜底输出(Fallback Output):当所有策略都无法让任务完整完成时,至少要返回一个可用的中间结果,而不是报错或空白。
把这些概念串起来,就是一套“预算感知型智能体”的架构:任务开始前规划预算 → 执行中定期检查 → 接近上限时触发压缩或降级 → 彻底不够时给出兜底结果。
下面进入实操,用一个最小示例把这套机制跑通。
4. 环境准备与最小代码骨架
本文示例使用 Python 3.9+,重点演示思想,不绑定具体框架。你可以用 LangChain、Dify、Coze 或自研封装,核心逻辑是一样的。
准备以下环境:
- Python 3.9 及以上版本
- 任意国产或国际大模型的 API 访问权限(本文用
openai风格的接口做演示,实际请替换为你的服务商 SDK) - 一个支持异步的任务队列(不强制,但推荐,便于做超时控制)
先建立一个最小工程目录:
agent-budget-demo/ ├── main.py # 入口,演示完整流程 ├── agent.py # 智能体执行器 ├── budget.py # 预算管理器 ├── summarize.py # 上下文摘要工具 └── requirements.txt # 依赖第一步先写预算管理器。它负责记录“总预算、已消耗、剩余量”,并提供检查接口。
# 文件路径:agent-budget-demo/budget.py from dataclasses import dataclass @dataclass class Budget: total_tokens: int used_tokens: int = 0 @property def remaining(self) -> int: return self.total_tokens - self.used_tokens @property def used_ratio(self) -> float: return self.used_tokens / self.total_tokens def consume(self, tokens: int) -> None: self.used_tokens += tokens def can_continue(self) -> bool: return self.remaining > 0 class BudgetManager: """ 预算管理器:维护总预算、子任务预算和全局检查逻辑。 当剩余比例低于阈值时,通过回调通知执行器做降级。 """ def __init__(self, total_tokens: int): self.budget = Budget(total_tokens=total_tokens) self.history = [] def record(self, stage: str, tokens: int) -> None: self.budget.consume(tokens) self.history.append( { "stage": stage, "tokens": tokens, "remaining": self.budget.remaining, } ) def check(self) -> dict: ratio = self.budget.used_ratio if ratio >= 0.8: return {"status": "critical", "ratio": ratio} if ratio >= 0.5: return {"status": "warning", "ratio": ratio} return {"status": "normal", "ratio": ratio}这个类本身很简单,但它决定了后面所有降级动作的触发时机。比较关键的设计是:预算消耗不是模型返回后统一算,而是每个阶段都做记录。这样能精确定位“哪个子任务吃掉了最多 token”。
5. 任务分解与上下文瘦身
第二步是任务分解和上下文压缩工具。任务分解解决“任务太大不能一口气做完”的问题,上下文压缩解决“历史信息太多塞不进去”的问题。
这里用一段伪代码演示典型的任务分解流程:
# 文件路径:agent-budget-demo/main.py(节选) def plan_subtasks(goal: str) -> list[dict]: """ 把一个复杂任务拆成子任务列表。 每个子任务包含:任务内容、预计消耗级别、是否可降级。 """ return [ { "id": "read_core_doc", "prompt": "阅读并理解核心文档,提取关键要点", "level": "high", "degradable": False, }, { "id": "load_supplement", "prompt": "加载补充材料,补充背景信息", "level": "medium", "degradable": True, }, { "id": "search_latest", "prompt": "搜索最新动态,更新数据", "level": "medium", "degradable": True, }, { "id": "write_report", "prompt": "输出最终报告", "level": "high", "degradable": False, }, ]核心原则:哪些步骤是不可舍弃的,哪些步骤可以在预算紧张时跳过。比如“搜索最新动态”可以降级为“基于已有知识生成”,但“输出最终报告”不能省。
然后写上下文摘要工具。它的作用是当历史内容太多时,把前面的长文本压缩成更短的摘要,保留关键信息,释放上下文空间。
# 文件路径:agent-budget-demo/summarize.py def summarize_messages(client, messages: list[dict], target_tokens: int) -> list[dict]: """ 将历史消息压缩成一段摘要,替换原消息列表。 注意:这是一个简化示例,实际场景需根据消息结构做更细粒度的处理。 """ text = "\n".join(msg["content"] for msg in messages) prompt = ( "请把下面这段对话历史压缩为不超过300字的技术摘要。" "保留任务目标、已完成的步骤、尚未完成的关键点、关键数据与结论。" "不要添加新内容,不要漏掉核心信息。\n\n" f"{text}" ) resp = client.chat.completions.create( model="your-model-name", messages=[{"role": "user", "content": prompt}], max_tokens=target_tokens, ) summary = resp.choices[0].message.content return [{"role": "system", "content": f"历史摘要:{summary}"}]这个函数的本质,是用一次额外的模型调用来“买”上下文空间。它本身会消耗 token,所以触发时机必须谨慎:只有在下一次任务需要更长上下文、且当前剩余空间不足时才调用。否则压缩行为本身反而会加速预算耗尽。
6. 智能体执行器:预算监控与优雅降级
第三步是核心执行器。它把前面所有模块串起来,在每一步执行后调用预算管理器检查状态,根据状态决定下一步动作。
# 文件路径:agent-budget-demo/agent.py import time class BudgetAwareAgent: def __init__(self, client, budget_manager: BudgetManager): self.client = client self.budget_manager = budget_manager self.history = [] self.results = {} def run(self, goal: str, subtasks: list[dict]) -> dict: # 这里的 MAX_LOOP 和 DEFAULT_RETRY 应该从配置读取 MAX_LOOP = 3 for sub in subtasks: status = self.budget_manager.check() if status["status"] == "critical": if sub["degradable"]: print(f"[降级] 跳过可降级任务: {sub['id']}") self.results[sub["id"]] = {"skipped": True} continue else: print(f"[关键] 任务不可跳过,尝试压缩后继续: {sub['id']}") # 简化示例:每次任务调用模型 token_cost = self._execute_subtask(sub["prompt"], MAX_LOOP) # 记录预算消耗 self.budget_manager.record(sub["id"], token_cost) # 每次执行后检查 current = self.budget_manager.check() print(f"[预算] 任务 {sub['id']} 消耗 {token_cost} token," f"剩余 {self.budget_manager.budget.remaining},状态 {current['status']}") # 如果已经临界,先尝试压缩历史 if current["status"] == "critical": self._safe_compress() return self._finalize() def _execute_subtask(self, prompt: str, max_loop: int) -> int: """执行单个子任务,返回消耗的 token 数。""" # 这里是实际的模型调用逻辑 # 注意设置单次调用的 max_tokens 上限,避免单次调用把预算吃光 ... return 123 # 示例返回值 def _safe_compress(self): """安全压缩上下文。这里需要判断当前剩余空间是否还能支持一次压缩调用。""" if self.budget_manager.budget.remaining < 500: print("[警示] 剩余预算不足以支撑上下文压缩,放弃压缩操作") return print("[压缩] 触发上下文摘要压缩") def _finalize(self) -> dict: """兜底输出:无论任务完成度如何,都返回一个结构化结果。""" return { "status": "completed", "results": self.results, "budget_history": self.budget_manager.history, }这个执行器里最关键的是_safe_compress里的判断:压缩本身也要花 token,如果预算剩余连压缩调用都支撑不起,就不能做压缩,而是应该直接进入输出阶段。很多智能体项目在预算优化时反而把预算耗尽,就是因为忽略了“优化动作本身有成本”。
再看一下_execute_subtask应该如何处理单次调用的超时和重试。AI 智能体最常见的预算杀手不是正常调用,而是失败后的无脑重试:
# 文件路径:agent-budget-demo/agent.py(补充方法) def _call_with_retry(self, messages: list[dict], max_tokens: int, retry_times: int = 2) -> str: """ 带重试的模型调用。每一次重试都会消耗 token, 所以重试上限必须显式配置,不能依赖模型自行判断。 """ attempt = 0 while attempt <= retry_times: try: resp = self.client.chat.completions.create( model="your-model-name", messages=messages, max_tokens=max_tokens, timeout=30, ) return resp.choices[0].message.content except Exception as e: attempt += 1 if attempt > retry_times: raise RuntimeError(f"模型调用失败,已重试 {retry_times} 次: {e}") # 退避等待 time.sleep(1 * attempt) raise RuntimeError("unreachable")重试次数、单次调用的max_tokens、超时时间,这三项是预算管理的底层基础设施。如果这三项没有显式配置,上层再多的预算策略都是空谈。
7. 完整示例:一个长报告生成任务
下面用一个完整场景把所有模块跑通。任务:让智能体基于一份内部技术文档,搜索补充资料,生成一份 2000 字的调研报告。
完整主程序:
# 文件路径:agent-budget-demo/main.py from budget import BudgetManager from agent import BudgetAwareAgent def main(): # 1. 初始化预算。假设总预算只有 5000 token,方便模拟临界场景。 budget_manager = BudgetManager(total_tokens=5000) # 2. 初始化智能体 client = create_client() # 替换为真实的 client 初始化 agent = BudgetAwareAgent(client, budget_manager) # 3. 定义任务 goal = "基于内部文档,撰写一份关于容器化部署最佳实践的调研报告" subtasks = [ {"id": "read_core_doc", "prompt": "阅读理解内部核心文档", "level": "high", "degradable": False}, {"id": "load_supplement", "prompt": "加载补充材料,提取背景信息", "level": "medium", "degradable": True}, {"id": "search_latest", "prompt": "搜索社区最新实践动态", "level": "medium", "degradable": True}, {"id": "write_report", "prompt": "生成最终调研报告", "level": "high", "degradable": False}, ] result = agent.run(goal, subtasks) print("执行完成,最终结果结构:") print(result["status"]) for stage in result["budget_history"]: print(stage) def create_client(): # 这里替换为你的模型客户端初始化 # 例如 openai.OpenAI(api_key=..., base_url=...) return None if __name__ == "__main__": main()运行这个示例,会看到每一阶段的预算消耗情况。当预算进入 critical 状态后,load_supplement和search_latest这类可降级任务会被跳过,而read_core_doc和write_report会保留。最终输出的budget_history能清楚看到每一步花了多少 token,这对于定位智能体的预算黑洞非常有用。
8. 运行结果与效果验证
这个示例不执行真实模型调用,所以不会看到真实的 token 消耗数字。但它的价值在于提供了一个可观测、可调整的预算管理骨架。
要验证这个架构是否真的有效,建议做三组实验:
实验一:基准测试。不启用预算管理,让智能体直接跑同一个任务,记录总耗时、总 token 消耗、任务完成度。
实验二:预算充足测试。把预算设成 20000 token,启用预算管理,跑同一个任务。理想结果是任务完整完成,验证预算管理不会在预算充足时误触发降级。
实验三:预算紧张测试。把预算压到 5000 token,再跑同一个任务。核心验证点是:智能体是否做到了“保核心、舍外围”,最终是否返回了报告,而不是中途报错。
判断标准主要有三条:
- 预算紧张时,任务是否仍然返回结构化结果,即使报告精度下降。
- 预算充足时,是否没有出现不必要的降级操作。
- 预算历史记录是否完整,能定位每一步的 token 消耗。
如果实验三出现了“智能体在 write_report 阶段直接报错”的情况,说明预算分配策略有问题:核心任务没有预留足够的预算。这时应该调整各子任务的预算权重,让不可降级的任务优先获得预算。更稳妥的做法是在任务规划阶段就预分配预算:
def plan_subtasks_with_budget(subtasks: list[dict], total_budget: int) -> list[dict]: """ 按优先级给子任务预分配预算。 不可降级任务分配 60% 预算,可降级任务共享剩余 40%。 """ hard_budget = int(total_budget * 0.6) soft_budget = total_budget - hard_budget for sub in subtasks: if not sub["degradable"]: sub["allocated_budget"] = hard_budget // sum( 1 for s in subtasks if not s["degradable"] ) else: sub["allocated_budget"] = soft_budget // sum( 1 for s in subtasks if s["degradable"] ) return subtasks这种预先分配的方式,比执行过程中实时判断更可控。它把“不确定性”从运行时转移到了设计时,这正是我们在工程上更希望看到的状态。
9. 常见问题与排查方法
预算管理相关的智能体问题有个特点:表面现象五花八门,根因往往集中在几个地方。下面列出高频问题。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 智能体运行到中后期突然报“context length exceeded” | 没有提前做上下文压缩,历史消息堆积过多 | 查看预算历史中的 token 增长曲线,重点看每次工具返回的原始内容大小 | 在子任务完成后立即对结果做摘要;限制单次工具返回的最大长度 |
| 任务没完成但预算还剩很多,智能体却不肯继续 | 循环重试次数过多,或单次任务 max_tokens 设置过小导致多次往返 | 查看历史记录中的重试日志和每次调用的完成原因 | 合理设置单次调用的 max_tokens;增加单次任务的 token 上限 |
| 降级后核心输出质量严重下降,用户无法接受 | 降级策略只做了“跳过”,没有做“简化” | 检查降级分支的具体逻辑 | 为不可跳过的任务提供“简化模式”:只输出结论、不写详细论证 |
| 压缩历史后,模型遗忘关键背景 | 摘要提示词没有明确要求保留关键细节 | 检查摘要工具的 prompt | 在摘要提示词中明确“必须保留具体数字、日期、决策结论、未完成事项” |
| 预算检查频率太低,发现时已经来不及 | 检查点只在子任务结束时触发 | 在模型调用前、收到响应后都增加检查点 | 把预算检查做成 decorator 或中间件,强制每次调用前检查 |
| 多子任务并行执行时,总预算失控 | 并行任务各自计算消耗,没有共享预算 | 检查是否使用全局 BudgetManager 实例 | 用集中式预算管理器,或使用 Redis 之类的外部存储做共享计数 |
真正容易踩坑的是第一行。很多团队做智能体时只关注 prompt 写得对不对,忽略了“工具返回结果”的大小。实际上,一次搜索返回的 20 条网页摘要,可能就吃掉 3000 token。如果不限制工具返回内容的长度,上下文窗口再大也撑不住。
10. 最佳实践与工程建议
预算管理这件事,越早设计越好。等智能体已经写了很多提示词和工具调用逻辑后再加预算管理,改动成本会高很多。以下几点是从生产环境实践中提炼的建议。
第一,把预算设计成可配置项,而不是写死在代码里。预算值、降级阈值、压缩触发比例,应该放在配置中心或环境变量中。不同场景的任务,预算需求差异很大:生成一份周报可能 3000 token 就够了,做一份行业调研可能需要 30000 token。把预算参数化,才能做到“同一套代码,适配不同任务”。
配置示例:
agent: budget: default_total_tokens: 10000 critical_ratio: 0.8 warning_ratio: 0.5 context: max_history_tokens: 4000 compress_ratio: 0.5 retry: max_attempts: 2 timeout_seconds: 30 degradable: enabled: true第二,所有 AI 调用都要有超时和重试上限,且重试次数必须小于等于 2。很多智能体“预算耗尽”的真相是:同一个调用失败了 5 次,每次都消耗完整的输入 token。设置重试上限是最便宜的预算保护措施。
第三,对每一次模型调用做 token 审计。记录 prompt 的 token 数、completion 的 token 数、模型名称、耗时、重试次数。有了这些数据,才能回答“预算花到哪里去了”。推荐把审计日志输出到标准 JSON 格式,方便后续接入监控系统:
{ "timestamp": "2025-06-01T10:00:00Z", "agent": "report-agent", "subtask": "write_report", "model": "your-model-name", "prompt_tokens": 3500, "completion_tokens": 1200, "total_tokens": 4700, "cache_hit": false, "elapsed_ms": 8420 }第四,给每个任务定义“最低可用输出”。也就是说,不管预算多紧张,这个任务至少要交付什么。比如调研报告任务的最低可用输出是“标题 + 三个核心结论 + 风险提示”,而不是“完整 2000 字报告”。把这个最低输出定义成单独的 prompt,在预算临界时调用。这个设计思路可以保证智能体即使“牺牲”,也是带着最小成果牺牲,而不是空手而归。
第五,安全与权限边界不能因为预算紧张而放开。预算不足时,智能体可能会尝试绕过某些校验以节省 token,比如跳过敏感操作确认、直接使用预设凭证。这在生产环境是绝对禁止的。预算管理只能影响“做不做外围任务”,不能影响“是否遵守权限和审批约束”。涉及删除、写入、资金操作时,无论预算剩余多少,都必须走完审批流程。
11. 总结与下一步实践方向
“AI 智能体在预算耗尽前的牺牲困境”不是一个技术噱头,而是每个真正把智能体落到生产环境的团队都会遇到的工程问题。
回到核心判断:智能体不应该是一个只会“闷头往前跑”的执行器,而应该是一个“时刻知道资源还有多少、知道哪些可以放弃、知道放弃之后怎么兜底”的预算感知系统。这不是某个框架的功能,而是你可以在自己的代码里实现的工程机制。
下一步建议你从三个方向入手:
- 如果你是刚接触智能体开发,先不要急着接复杂框架。按照本文的最小骨架,实现一个带预算管理的智能体跑通一个真实任务,体会“预算检查点”对任务质量的直接影响。
- 如果你已经在用 Dify、Coze、LangGraph 这类平台,请检查它们提供的预算限制、超时控制、断点恢复功能,把本文的降级思路映射到平台的对应能力上。
- 如果你已经在生产环境跑智能体,优先做两件事:给所有模型调用加上 token 审计日志;给每个任务定义最低可用输出。这两件事投入小、收益大。
把每个智能体都当作要上生产环境的服务来设计,而不仅仅是一个“能回答问题”的脚本。预算不是限制,而是需求的一部分。学会在有限预算下做取舍,你的智能体才能真正从实验室走进业务线。