★ FEATURED ARTICLE 下午三点半线上 Agent 任务跑到第 37 步API 突然甩回来一个 400 错误This models maximum context length is 1048576 tokens. However。任务就这么断了前 40 分钟的工作全白费重跑又是一轮时间和钱不重跑又拿不到这份客户要的结果。这不是我一个人踩过的坑——凡是把 Agent 放进生产环境跑过的人大概率都被“对话超长”引发的断流恶心过一回。今天想聊的不是“怎么保证不报错”这是骗子才敢承诺的事。更实际的是当任务已经跑了一半上下文眼看就要爆掉的时候怎么给 Agent 修一条紧急逃生通道让它能保住已完成的成果、从断点续跑实在不行也能体面地停下来而不是把整个任务连根扔进回收站。我把这条通道前后迭代了三个版本从最初的异常捕获打印堆栈到现在的分层逃生机制中间踩了不少坑。下面把完整思路、代码实现和排查经验一次梳理清楚。1. 对话超长的“案发现场”断流到底是怎么发生的1.1 不是用户话多是 Agent 的工作日志把上下文喂胖了很多人第一次遇到这个报错时会下意识觉得是不是用户把提示词写太长了实际情况恰恰相反Agent 任务里的上下文膨胀大部分是它自己一口一口吃出来的。一个典型的 Agent 执行循环是这样的模型收到任务开始推理决定调用工具工具返回结果模型看到结果再推理再调用下一个工具。每一轮循环里工具的参数、工具的返回值、模型自己的中间推理过程全都会追加进对话历史里成为下一次请求的输入。也就是说Agent 是一个会不断“自产自销”上下文的角色。拿我做过的一个批量资料整理任务举例Agent 每处理一个文件要调用检索工具拿资料、调用解析工具读文件、再调用写入工具落库。这三个工具每次返回的数据量少则几千 token多则几万 token。任务要处理 20 个文件整个对话历史轻轻松松滚到几十万 token。如果中间某个文件解析失败Agent 还会反复重试重试日志又是新一轮上下文开销。这里可以打一个生活化的比方普通多轮对话就像两个人聊天聊一个小时也就翻来覆去那几句话Agent 任务就像一个人一边干活一边把自己每句话、看到的每份文件全文背下来背到后面连自己最初要干嘛都快忘了。等到 API 真报 context length exceeded已经是这个“背书的人”彻底撑不住的那一刻。所以对话超长从来不是瞬间发生的而是一个缓慢累积、最后突然爆发的慢性病。治本的方向是任务设计层面做上下文治理但在没治好之前得先有急救方案。1.2 排查第一步区分“API 硬限制”与“本地累积”遇到断流先别急着改代码第一件事是把现场日志翻出来确认断流的类型。我一般分两类看第一类是 API 硬限制直接报错比如标题里这种 400 错误返回值里会写明 maximum context length 是多少超了多少。这类错误是模型服务端拒绝请求特征是突然出现、无前置征兆上一秒任务还正常下一秒直接终止。第二类是本地累积导致的质量崩溃特征不太一样任务不报错但 Agent 开始“失忆”比如忘记早前已经完成过某步骤、重复调用同一个工具、答非所问。这是因为上下文太长之后模型对早期信息的注意力明显下降行为开始漂移。很多团队把这种情况归为模型能力问题其实根源还是上下文管理。怎么快速区分看日志里每次请求的 prompt_tokens 数值就行。如果它是随着轮次平稳上涨那就是典型的本地累积如果某一次请求突然暴涨十几万 token去检查那一步的工具返回是不是把整个文件内容、整份查询结果全塞进上下文了这才是真正的元凶。我在实际排查中还遇到过一种更隐蔽的情况Agent 框架本身在拼装消息时会重复注入系统提示词和工具定义。表面上看历史才几十轮实际每次请求都额外带上完整工具 schema这部分开销很容易被忽略。排查时建议把“send 给 API 之前最终的 messages 数组”落到日志里不要只看模型返回的内容。2. 逃生通道整体设计救什么怎么救按什么顺序救2.1 设计目标把三种结局按优先级排好断流事故发生后理想的抢救顺序不是简单想成“让他继续跑”而是分出三个优先级递进的结局第一优先级把已经跑出来的成果保住。比如任务已经处理完了 13 个文件那这 13 个文件的结果要能确认落地不能因为第 14 个文件触发断流就让人觉得前 13 个都白干了。第二优先级让任务能从断点继续。这就意味着不能只保成果还要保住“干到哪一步了”的状态包括当前处理对象的索引、未执行的计划动作、必要的中间变量。第三优先级安全收尾。如果上下文已经彻底没法救那就不能硬来要停下来把已完成的、未完成的、执行报错的信息全部写入审计记录让人或者外部编排系统能做后续处理。我的逃生通道设计就是围绕这三个优先级展开的。目标很明确不追求让所有任务都一定跑通——这做不到也违背客观规律——而是让每一次断流都变成可控、可恢复、可追踪的事件。有个设计原则值得多说一句逃生通道不是给 Agent 用的是给你的系统用的。这句话什么意思Agent 自己是不会主动发现自己上下文快满的它只会傻乎乎地继续往火堆里添柴。救火这件事必须由外层运行时来做。所以整个通道要架在 Agent 之上以旁路的方式监控、干预而不是把判断逻辑交给模型本身。2.2 五层逃生机制从监控到人工接管我在项目里把逃生通道拆成了五层每一层有各自职责触发条件逐层收紧层级职责触发条件用量监控层每次调用前估算上下文用量记录预警连续调用时持续运行预警层当用量接近阈值时主动降载用量达到 max context 的 70%快照层保存完整断点状态到持久化存储用量达到 85%或捕获到异常时压缩层对早期对话做摘要压缩腾出空间用量达到 70% 后持续压缩85% 强制执行续跑层断点恢复、幂等校验、人工接入口任务恢复时或人手动触发这里有一个关键取舍为什么不等到报错了再统一处理因为报错意味着请求已经被服务端拒绝那一刻之前积累的上下文已经全部作废。如果能在 70%、85% 这种“还没爆但快爆”的时候介入就可以在请求之间做手脚——压缩历史也好、保存状态也好——不影响当前正在跑的这一步代价最小。有人可能觉得 70% 就开始干预会不会太保守我的经验是token 估算本身有误差加上摘要压缩这个动作本身也要消耗一次模型调用的上下文空间预留越充分越不容易在急救过程中二次炸锅。宁可在 70% 时多做几次压缩也好过在 96% 时手忙脚乱。2.3 为什么不直接调大 max_tokens 参数肯定会有人问既然上下文限制是 1048576 tokens那我把自己的任务设计得更“省”一点或者干脆选支持更大上下文的模型不就行了吗我刚开始也是这么想的后来被现实教育了。原因有三。一是成本和延迟受不了。上下文越长每次请求的 prefill 计算量越大响应延迟明显增加费用也在涨。Agent 动辄几十轮调用每一轮都拖着巨大的上下文跑最后账单会让你怀疑人生。二是模型性能会下降。业界通常叫“lost in the middle”上下文塞得太满时模型对中间部分信息的注意力明显变差工具调用参数容易抄错指令遵循也打折。我实测下来长上下文模型在文件量多的任务里后期错误率比中期高出一大截。三是最要命的调大限制只是把爆点往后推并没有解决增长本身。一个设计糟糕的 Agent 任务给 100 万 token 也一样会吃满只是从“跑 20 分钟爆”变成“跑 2 小时爆”。不治本的方案只是让事故来得更晚、损失更大。所以我后来坚决不把希望寄托在“更大的上下文”上而是把重点放在限制摆在那里怎么在里面精打细算怎么在爆之前保住身家性命。3. 逃生通道落地实现从预估到恢复的完整链路3.1 调用前先算账上下文用量的估算逻辑逃生通道的第一层是监控监控的前提是能估算出“现在到底用了多少 token”。每次请求前都要算一遍不能等 API 的 usage 返回了才知道——那已经是事后了。我用的估算函数长这样import json from typing import Any, Dict, List, Optional def estimate_prompt_tokens(messages: List[Dict[str, Any]]) - int: 粗略估算一组消息的 token 数不追求精确只求量级靠谱。 total 0 for msg in messages: # 每条消息本身有固定开销 total 4 content msg.get(content, ) if isinstance(content, str): total len(content) // 2 elif isinstance(content, list): for item in content: if item.get(type) text: total len(item.get(text, )) // 2 elif item.get(type) in (image_url, input_audio): total 1000 # 多媒体部分按经验估 # 工具调用参数的 JSON 序列化开销 if msg.get(tool_calls): total len(json.dumps(msg[tool_calls], ensure_asciiFalse)) // 2 # 工具返回内容 if msg.get(tool_call_id): total 8 total len(msg.get(role, )) // 2 return total我不会完全依赖这个估算结果因为它最大的作用是“量级判断”当前大概是 30 万还是 90 万这已经足够决定是否触发逃生动作。如果想要更精确可以在请求返回后读取 usage 里的 prompt_tokens然后用这个真实值去校准同一个 messages 数组的估算系数。实际上我在工程里对估算函数做过一次校准对比了估算值和真实 usage 的偏差发现中文内容多时偏差在 15% 上下英文内容多时偏差在 8% 上下。这个精度用来做阈值判断完全够了。关键是估算必须要在请求发出前完成这样才有时间在 70%、85% 阈值处做降载和快照。3.2 快照落地把断点状态写成不依赖内存的实体快照是整个逃生通道的核心资产。设计快照时我吃过亏——一开始只保存了消息历史结果恢复时发现任务状态丢了。Agent 任务的状态不止是对话还包括当前处理到第几个数据项、哪些步骤已完成、哪些子任务排队中、局部变量的值等等。这些如果只存在于进程内存里一旦断了就等于归零。完整快照的数据结构我用了统一格式方便序列化和恢复校验import time import uuid from typing import Any, Dict, List, Optional SNAPSHOT_DIR ./escape_snapshots def take_snapshot( task_id: str, conversation: List[Dict[str, Any]], task_state: Dict[str, Any], completed_steps: List[str], pending_actions: List[Dict[str, Any]], extra: Optional[Dict[str, Any]] None, ) - str: snapshot_id f{task_id}-{int(time.time()*1000)}-{uuid.uuid4().hex[:8]} payload { snapshot_id: snapshot_id, task_id: task_id, created_at: time.time(), conversation: conversation, task_state: task_state, completed_steps: completed_steps, pending_actions: pending_actions, extra: extra or {}, } path f{SNAPSHOT_DIR}/{snapshot_id}.json tmp_path f{SNAPSHOT_DIR}/.{snapshot_id}.tmp with open(tmp_path, w, encodingutf-8) as f: json.dump(payload, f, ensure_asciiFalse, indent2) # 先写临时文件再 rename避免半截文件被误读 import os os.replace(tmp_path, path) return snapshot_id这个实现里有几个细节我在意过。一是快照写入必须原子化。如果进程在写 JSON 写到一半时被 kill留下一个损坏的文件恢复阶段的挫败感比断流本身还强。先写临时文件再 rename 是我实测最稳的做法。二是快照一定包含 pending_actions也就是“接下来本来打算做什么”。恢复时不光要知道过去发生了什么还要知道将来的计划。很多现成的 Agent 框架只保存历史消息丢失 pending_actions 后恢复的 Agent 会陷入重新规划的迷茫甚至重复执行已完成的操作。三是extra字段要留出来。不同的任务都有自己特殊的运行时上下文比如一个分布式文件锁的路径、一个临时目录的位置、一个已获取但尚未使用的凭据。给一个兜底字段比每次遇到新情况都改公共结构要灵活得多。3.3 摘要压缩与降级续跑保住最近概括过去当上下文用量逼近危险线时正面硬扛空间不够唯一的出路是把早期对话压缩成摘要。摘要压缩的逻辑朴素但有效把最早的 N 轮消息交给模型让它输出一段结构化摘要然后在原来位置上替换掉那 N 轮只保留最近几轮的完整消息和全部工具响应。我这里放一个压缩函数但必须提前声明这个函数只是骨架真正要调好关键在于下面的 must_keep_keys 参数。from openai import OpenAI client OpenAI() # 按你自己的服务商配置 def compress_history( conversation: List[Dict[str, Any]], keep_last: int 6, must_keep_keys: Optional[List[str]] None, ) - List[Dict[str, Any]]: if len(conversation) keep_last 2: return conversation old_part conversation[:-keep_last] recent_part conversation[-keep_last:] prompt { role: system, content: ( 请将以下 Agent 对话历史压缩成摘要。 保留已经完成的步骤、关键结论、已使用过的资源标识、 容易导致重复操作的执行状态。\n 压缩后的内容仍作为 system 消息返回直接输出摘要文本。 ), } user_part {role: user, content: json.dumps(old_part, ensure_asciiFalse)} resp client.chat.completions.create( modelyour-model, messages[prompt, user_part], max_tokens3000, ) summary_text resp.choices[0].message.content # 关键把不能丢的硬信息强制注入摘要而不是完全相信模型 for key in (must_keep_keys or []): matches extract_values_by_key(old_part, key) if matches: summary_text f\n[must-keep] {key}: {json.dumps(matches, ensure_asciiFalse)} summary_msg {role: system, content: f历史摘要{summary_text}} return [summary_msg] recent_part这里要重点讲两个容易踩的坑都是我实际经历过的。第一个坑是压缩的时机不能太晚太晚会导致压缩本次调用的输入本身就已经超限。解决办法是上文的阈值设计在 70% 时做第一次压缩那时候空间还很富余模型处理得动真要拖到 92% 再压缩压缩用的这次请求可能先一步撞上硬限制。第二个坑是模型压缩摘要时会“丢东西”。尤其是一些看起来不起眼但恢复时必须的细节比如某个文件的原始路径、某个数据库查询用的条件参数。解决办法就是 must_keep_keys 硬注入机制——不依赖模型自觉而是从原消息的 JSON 结构里直接按 key 抽取字符串拼进摘要。这些信息宁可冗余留着也不能在恢复时才想起来缺了。3.4 断点恢复与幂等校验快照是逃生通道的“安全气囊”但气囊弹开后能不能把人救回来还要看恢复逻辑写得好不好。我的恢复函数包含四个阶段def resume_from_snapshot(snapshot_id: str) - Dict[str, Any]: # 1. 读取并校验快照文件 snapshot load_snapshot(snapshot_id) if not snapshot: raise RuntimeError(fsnapshot not found: {snapshot_id}) validate_snapshot_schema(snapshot) # 2. 幂等性预检 check_idempotency(snapshot) # 3. 重建外部资源 restore_external_resources(snapshot.get(extra, {})) # 4. 拼接可续跑的消息序列 resumed_messages build_resume_messages(snapshot[conversation]) return { task_state: snapshot[task_state], pending_actions: snapshot[pending_actions], resumed_messages: resumed_messages, }幂等校验这一段最容易被忽略。Agent 任务恢复常常发生在已经部分执行的中间状态比如某个文件已经写入数据库了但快照恰好保存在“写库前”。恢复后重新执行就会产生重复写入。我的做法是在快照里记录每个数据项的处理编号恢复后从编号推断哪些数据需要跳过哪些需要回滚重来。如果任务涉及外部系统最简单的方式是在每个数据项上记录一个唯一处理标记数据库里按标记做唯一约束这样就算重复写也不会产生脏数据。还有一种更粗暴但有效的方案恢复后不直接续跑而是把任务从最近的“检查点”重新 planning 一次。就是说把已完成的成果和未完成的目标一起交给模型让它基于当前真实状况制定新的执行计划而不是逼它沿着旧的历史一条道走到黑。这样容错性更高。我个人是“快照恢复 重新规划”双轨使用如果任务结构和状态完整直接续跑如果状态残缺就降级为重新规划。4. 实操阶段踩过的坑技术细节才是翻车重灾区4.1 token 估算的误差中英文与工具调用的隐蔽开销很多人做 token 估算时第一反应是用字符数除以 4 或者除以 2我一开始也这么干后来在清理日志时发现偏差比想象中大多了。最典型的是中英文差异。中文在常见 tokenizer 里大约 1 个汉字对应 1 到 2 个 token而英文大约 4 个字符对应 1 个 token。如果你的 Agent 任务大量处理中文文档用len(text) // 2估算会整体低估。我在一个中文客服摘要任务里对比过估算偏差最高到了 30%差点导致压缩动作没有及时触发。另一个隐蔽开销来自工具调用。模型生成 tool_calls 时参数是以 JSON 格式输出的里面的字段名、嵌套结构、转义符全都占 token。工具返回结果也是结构化 JSON字段名同样占 token。如果工具定义里塞了很长的描述、样例那这部分是每轮请求都要重复带上的固定开销累积起来非常可观。建议你自己在项目里做一次校准抓取真实的 prompt_tokens 和消息内容算一个配置化的校正系数。不要迷信某个固定的“字符转 token”公式因为 tokenizer 不同、内容语言不同系数都不一样。4.2 快照恢复丢状态外部资源不能被序列化我的第一个恢复版本只保存了 messages、task_state 这些内存里的东西恢复后才发现 Agent 还是跑不起来卡在了一个很蠢的地方它不知道之前建立的 HTTP 会话已经失效了。随着 Agent 任务越来越长它会慢慢积累一批外部资源和文件服务器之间的认证连接、数据库连接池里的句柄、一个上传任务返回的 media_id、甚至是一个临时创建的目录。这些东西活在进程外的世界里不可能被 JSON 序列化进快照。解决方法的思路是把“外部资源”的恢复简化成“外部资源的重建凭证”。快照里不存句柄而是存重建句柄所需的信息。比如数据库连接不存连接对象存连接参数和需要执行的初始化 SQL文件上传不存返回值里的临时引用存本地文件的路径和重试逻辑。我自己的恢复检查清单是这么写的每次恢复后按清单逐项验证数据源连接是否可访问文件系统路径是否存在且有权限认证 token 是否过期临时资源是否被旧进程占用数据库里已写入的记录是否与快照状态一致当时还没有完整自动化这套检查恢复失败的概率就会高出一截。后来把这些检查全部做成了解析函数在恢复阶段自动跑一遍能过的才放行不能过的直接截断重试。这一步让恢复成功率从 60% 左右提到了 95% 上下。4.3 摘要压缩吞掉关键 ID必须有一张保命清单这个坑我印象最深因为当时线上事故就是因为压缩丢了一个看起来很普通的值。任务流程是Agent 上传文件到对象存储拿到一个上传成功后的 file_id之后要把这个 file_id 和其他业务数据一起写入业务库。上下文膨胀触发压缩后模型把早期对话压成摘要时把 file_id 这个值给漏了。恢复后任务不知道上传步骤已经完成又去上传一遍结果产生一份重复文件业务库里的数据也对不上号。从那以后我在代码里专门维护了一份 must-keep 清单任务里每个环节的 ID 类字段都在清单里业务单据编号文件上传返回的标识数据库自增主键或唯一键定时任务的分片号已拿到但还未使用的票据类数据压缩函数在生成摘要之后必须扫描原始消息里所有包含这些 key 的值把命中项作为独立的 [must-keep] 段硬拼在摘要后面。这一步不经过模型理解直接字符串拼接确保不会丢。看起来有点暴力但恰恰是在生产中验证过的最稳方案。4.4 并发触发逃生通道快照写入的竞态问题当你只有一个任务实例时逃生通道很好写。但生产环境里往往同时跑着几十上百个 Agent 任务问题就来了。我第一次上线时碰到的情况是这样的同一次故障中两个任务几乎同时触发逃生通道都往同一个快照目录里写文件文件名按任务名拼接结果后写的覆盖了先写的恢复时加载出来的是另一个任务的状态。排查了半天才发现是文件命名冲突。解决办法很简单但值得强调快照文件名必须带上 task_id、时间戳和随机后缀三重组合保证唯一。另外写入时用临时文件 os.replace 的原子替换方案避免出现半截文件被恢复进程读到的脏场景。这两点合在一起基本能杜绝并发写入导致的竞态问题。如果任务量再往上走还应该考虑给每个任务一个独立子目录比如SNAPSHOT_DIR/{task_id}/既避免文件名碰撞又便于人工按任务维度清理过期快照。5. 把断流从事故变成流程逃生通道的运维视角5.1 关键指标每次调用的账都要算清楚逃生通道上线后我用一段时间的数据积累验证了一个很朴素但容易被忽略的观点断流不是一个“一次性事件”它是一个可以用指标刻画、预测、管理的业务风险。我建议在 Agent 运行的每个关键节点记录这几项指标指标说明参考作用prompt_tokens每次请求发送前估算值判断上下文增长趋势completion_tokens模型返回的 token 数看单轮输出是否失控context_usage_ratioprompt_tokens / max_context触发降载与快照的依据压缩触发次数每次逃生压缩的记录评估任务设计的上下文健康度快照恢复成功率恢复成功的快照占比验证逃生通道自身质量压缩后 token 降幅压缩前后差值判断压缩策略是否有效这些指标不一定要上复杂监控系统日志结构化输出就行。我是把每次调用的估算值、实际 usage、是否触发逃生动作、动作结果都打成了 JSON 日志后续用脚本统计即可。如果你们团队已经有 Prometheus 或开源可观测平台直接把这些数值上报成指标更方便还可以直接做趋势图。从这些数据里能读出很多有意思的规律。比如在高峰期并发任务会挤占同一个下游服务的配置同时触发频繁重试重试过程中上下文急剧膨胀。这时候逃生通道虽然能兜底更重要的是反过来发现任务并发设计里的问题。5.2 三级告警与人工接管人在回路里才有兜底逃生通道的设计我一直强调一个度自动化的动作要果断但该喊人的时候必须喊人。我把它做成了三级告警体系一级预警context_usage_ratio 达到 70%系统自动压缩只记录日志不通知人。二级危险达到 85%强制快照并降级续跑同时发一条告警消息到工作群。三级致命捕获到断流异常且自动恢复失败停止所有自动化重试挂起任务立即通知值班人员。三级告警的关键是“停止自动重试”。很多事故之所以雪上加霜是因为 Agent 框架自带的 retry 逻辑在 API 持续报错时反复重试每次重试还带着巨大的上下文浪费时间和成本。逃生通道接管后自动重试应当被取消换成等待人介入的挂起状态。我同时留了一个人工接管接口值班人员可以通过管理页面输入 task_id查看它的最新快照、上下文用量、失败原因然后选择“从最近快照恢复”“重建任务状态放弃旧上下文”或者“彻底终止并保留审计记录”三种动作。这个人工通道在自动化方案失控的那几次事故里救回了至少三个重要任务。5.3 逃生审计数据从被动救火到反推任务设计最后分享一个我认为最有长期价值的实践每一次逃生事件都要留下完整审计记录不只是为了追责更是为了反推任务设计。审计记录里我保留了这些字段触发原因、触发时的上下文用量、快照大小、压缩前 token 数、压缩后 token 数、恢复耗时、恢复后任务是否成功完成、整体耗时对比不触发的基线。每两周做一次简单汇总就能看出哪些任务类型特别容易撑爆上下文、哪个外部工具的返回体量过胖、哪个环节的工具调用设计可以精简。实际调整过一轮后我管理的 Agent 任务断流率下降了接近一半。记住逃生通道最重要的贡献不是“救火”而是让你积累起足够的数据知道哪里最容易起火提前把火源拆掉。这比任何花哨的自动化方案都划算。最后再聊一点我个人的体会。逃生通道上线头两周我自己最怵的就是那个三级致命告警它一响就说明上一轮的自动化急救失败了。但后来我慢慢接受了这个事实Agent 跑长任务就像开一辆老车上高速你能做的是把备胎、工具箱、救援电话都准备好而不是祈祷它永远不出毛病。真正成熟的系统不是从不失败的系统而是失败之后能快速定位、快速恢复、并且一次比一次稳定的系统。这条逃生通道就是我把这句话落到代码里的方式。
阅读完成 · 觉得有帮助？
