Agent对话超长断流实战：Token预算管理、摘要压缩与断点续跑方案

做Agent开发最怕什么？不是模型能力不够，而是任务跑一半，API突然报“对话超长”，然后整个流程断流。我前阵子给一个长期运行的Agent项目打补丁，专门解决这个“对话超长”导致的断流问题。今天把整套方案拆开讲：从Token预算管理、自动摘要压缩，到断流后的断点续跑和降级重试，全都是实际跑过、验证过、踩过坑之后沉淀下来的东西。适合正在写Agent框架、做多轮工具调用的朋友参考，也适合被API错误码折磨的初学者拿来应急。

先说明一下，我这里说的Agent，不是套壳聊天机器人，而是那种会自己调工具、跑循环、做多步骤任务的智能体。这类项目对上下文的消耗极快，也是最容易撞上“对话超长”的。下面直接进入正题。

1. 先把现场复现一遍：对话超长断流是怎么发生的

1.1 一个典型的翻车现场

假设你用Python写了一个Agent，让它去完成“跨平台数据调研+整理报告”的任务。任务逻辑大概是这样：接收用户指令，调用搜索API获取信息，针对每个来源做内容提取，汇总成结构化笔记，最后生成报告。

每一步，Agent都需要把全部历史对话发给大模型API，让模型决定下一步调什么工具。前几步都很正常，到了二十多轮的时候，请求直接返回一个400错误，错误信息里有这么一句话：

api error: 400 this model's maximum context length is 1048576 tokens. However your request had ...

有的服务商不会返回这么明确的错误，而是直接断开连接，日志里只有connection reset by peer，这就是大家说的“断流”。任务已经做了一半，结果最后几步全没了，前面几十次API调用全部白费。

我遇到这个问题时，第一反应是“模型上下文不够大”，换了一个更大的模型继续跑。结果跑了更久，还是断。后来才意识到，问题不在模型大小，而在Agent自身的上下文管理是失控的。

1.2 为什么Token会不受控地膨胀

这里要理解Agent的请求模式。大多数Agent框架的实现，是把system prompt、历史消息、工具返回内容全部拼进一个JSON数组，每一次请求完，把模型回复追加进去，再调用工具，又把工具结果追加进去。这是一个典型的“上下文累积”模型。

具体来说，有四个地方会让Token不受控：

• 系统提示词，几百到几千token不等，这部分相对固定还好；
• 每一轮用户/助手对话，虽然单条不大，但累积到上百轮就很可观；
• 每调用一次工具，工具返回的内容可能很大，比如网页抓取返回值几万token；
• Agent循环次数多了以后，旧的对话不会删。

如果任务里有一个for循环要对100个网页做提取，每一次工具返回平均5000 token，20次就是100000 token，加上推理历史，很容易超过模型限制。更隐蔽的问题是，有些Agent框架会把工具结果放在一个不可见的内部消息里，日志看不出来，但token确实被算进去了，于是你根本不知道它为什么超。

1.3 报错和断流其实是同一件事的两张脸

表面上看，“400对话超长”和“连接被断开”是两种现象，但根子可能是同一个：请求体超过了服务端能处理的最大长度。服务端在解析HTTP请求的时候发现Body太大，或者模型在prefill阶段发现输入超过max context length，就会给出400；如果Web网关在反代层先断了连接，客户端看到的就是断流。

所以修“逃生通道”不能只盯着错误码，要把“请求体体积”当作一个可观测指标来管。下面我讲的所有方案，本质都是在“发送请求之前、之中、之后”各设一道闸门。

2. 逃生通道设计：三层安全网，拦住对话超长

2.1 第一层：Token预算监控与预警

先明确一个概念：max context length是硬上限，预算是你自己设置的软上限。我习惯把安全阈值设为模型最大上下文长度的80%，剩余20%留给模型输出。比如模型支持128k，那你的输入+历史消息+工具返回的总和最好不超过102k。就算是DeepSeek这种支持1M上下文的新模型，也不要满打满算，因为服务端还有系统预留和最大输出配额。

实践里的做法是，在每一次构造请求之前计算当前messages的总token数。最精确的做法是用官方tokenizer，比如tiktoken或者模型服务商提供的计算接口。但在生产环境里，每轮都跑完整tokenizer成本太高，我一般用字符长度估算：中文场景下，1个汉字约合1.5到2个token，英文和代码大概4个字符一个token。

我写了一个简单估算函数，误差在20%以内，做预算监控足够了：

def estimate_tokens(text: str) -> int: # 是否支持中文按1.5字符/token估算；其他字符按4字符/token cn_chars = sum(1 for ch in text if '\u4e00' <= ch <= '\u9fff') other_chars = len(text) - cn_chars return int(cn_chars * 1.5 + other_chars / 4) + 10 # 加10是给消息格式留余量

这样我们在每次add_message之后就能算出当前列表的总体积。一旦超过软阈值，就触发下一层。

2.2 第二层：内容压缩与摘要替换

当预算监控发现要超限时，不是直接拒绝任务，而是启动“内容压缩”。核心思路是：把最旧的一部分对话压缩成一段摘要，替换掉原始多条消息，只保留最近几轮完整对话和一轮完整的system prompt。这样模型既有全局记忆，又不会把历史原始文本全部塞进来。

摘要本身也是一次API调用，要注意成本，不能每个循环都触发。我的做法是设置一个“触发水位”，比如达到85%时才执行一次摘要，摘要完成后把消息列表的token量直接降回40%左右。这样可以保证一个长任务最多压缩两三次，不会变成“摘要链套娃”。

举例来说，假设你有60轮历史消息，总token量已经到110k。压缩时，只把最早50轮做摘要，保留最近10轮原始对话。摘要后，这50轮可能只占5k，整个消息列表降到15k左右。模型依然知道之前做过什么，但不需要背负50轮原始文本。

这里要注意一个优先级：system prompt必须保留，工具返回中与当前目标无关的历史结果可以优先压缩，最近几轮的工具结果尽量保留原文，因为模型下一步决策大概率依赖最近的工具返回。

2.3 第三层：降级与逃生出口

即便做了预算和压缩，还是可能遇到服务端不稳定、API Key没配好、模型临时不可用等情况。所以Agent必须有一个“逃生出口”：当主模型连续失败N次后，不再继续重试，而是切换策略。

我的降级顺序是：

• 第一次降级：改用更短的提示词版本，去掉所有示例和中英文双语说明；
• 第二次降级：只保留最近2轮对话继续跑，把前面所有历史全部摘要成一句话；
• 第三次降级：切换备用模型，比如主用DeepSeek时，备用一个轻量模型，哪怕回答质量略低，也要保住任务流程不中断；
• 最终兜底：将当前已完成的部分结果持久化，告知用户“任务部分完成，需要恢复”。

这个降级开关平时不触发，但它是整个逃生通道里最重要的兜底。我见过太多项目只顾着把上下文塞满，完全没有考虑“塞不下怎么办”，一旦断了，前面所有工作全丢。

3. 核心实现：一个可复用的上下文管家

3.1 上下文管家的基本结构

既然问题核心是“消息列表失控”，那最直接的解决方案是写一个ContextManager类，让所有消息读写都经过它。Agent不再直接操作原始消息数组，而是调用它提供的几个方法。

这个类需要做四件事：

• 维护system_prompt和历史消息；
• 在每次添加消息后估算token总量；
• 超过阈值时自动触发压缩；
• 对外提供build_request()，返回一个可以提交给API的完整消息数组。

这样设计的好处是，Agent业务代码里不需要到处写压缩逻辑，只要所有消息都走add_message()，逃生通道就是自动生效的。

3.2 关键代码与参数选择

下面是我实际用过的一个简化版本：

import json class ContextManager: def __init__(self, system_prompt: str, max_context: int = 128000, safety_ratio: float = 0.8): self.system_prompt = {"role": "system", "content": system_prompt} self.history = [] self.max_context = max_context self.trigger_tokens = int(max_context * safety_ratio) def _estimate_tokens(self, text: str) -> int: cn_chars = sum(1 for ch in text if '\u4e00' <= ch <= '\u9fff') other_chars = len(text) - cn_chars return int(cn_chars * 1.5 + other_chars / 4) + 10 def _total_tokens(self) -> int: total = self._estimate_tokens(self.system_prompt["content"]) total += sum(self._estimate_tokens(m["content"]) for m in self.history) return total def add_message(self, role: str, content: str): self.history.append({"role": role, "content": content}) if self._total_tokens() > self.trigger_tokens: self.compress() def compress(self): # 至少保留最近的4条完整消息，历史更早的部分做摘要 if len(self.history) <= 6: return keep_count = 4 to_summarize = self.history[:-keep_count] recent = self.history[-keep_count:] summary_text = self._call_summary(to_summarize) self.history = [ {"role": "system", "content": f"[早期对话摘要] {summary_text}"}, *recent ] def _call_summary(self, messages) -> str: # 这里用轻量模型或主模型做摘要，摘要内容需要限制长度 # 演示起见，简化为直接拼接前500字 content = "\n".join(m["content"] for m in messages) return content[:500] def build_request(self): return [self.system_prompt, *self.history]

参数怎么选？

• safety_ratio建议设为0.8。设太低会频繁触发压缩，增加摘要API调用成本；设太高则留给模型输出的空间不足，可能因为单次回复过长再次超限。
• keep_count建议4到6。保留太少，模型可能丢掉近期的关键工具返回；保留太多，压缩效果不明显。
• 上面的_call_summary是简写，生产环境请接入实际摘要API，并指定摘要目标长度，比如“用不超过500字概括所有历史内容”。

3.3 工具返回内容也要管

另一个容易忽略的点是工具返回。很多人把工具结果原样塞进messages，比如搜索引擎返回的整个HTML，这是token膨胀的主要来源。我觉得应该对工具结果做一层“先截断再使用”：只保留核心字段、摘要或前N个字符。

我写了一个通用截断函数：

def trim_tool_result(raw: str, max_len: int = 2000) -> str: if len(raw) <= max_len: return raw # 优先保留头和尾，因为头和尾往往有结构化信息 head_len = int(max_len * 0.6) tail_len = max_len - head_len return raw[:head_len] + "\n...[已截断]...\n" + raw[-tail_len:]

对于结构化JSON，不要直接截断，而是先解析，只保留关键键值，比如标题、链接、摘要，删除正文和HTML标签。对于HTML，先用正则或者BeautifulSoup提取纯文本，再截断。这一层属于低成本高收益，通常加上之后，上下文体积能下降一半以上。

4. 断流之后怎么救：重试、幂等与断点续跑

4.1 先把断流分类，再决定重试方式

不是所有断流都要重试。我一般把断流分成三类，处理方式完全不同：

• 上下文超长类（400/413）：这是请求体超限，重试没有意义，必须先压缩。
• 服务端超时类（504、读超时）：往往是瞬时过载，可以重试。
• 连接被重置（connection reset）：可能是网关或网络问题，也可能是服务端主动断开，需要检查服务商状态页和日志。

盲目重试是大忌。我见过一个项目在上下文超长后，用通用重试函数直接重发原请求，结果连续重试5次全部失败，不仅浪费时间，还可能触发服务商限流，导致后面几分钟内所有请求都被拒绝。

所以正确的重试前检查是：如果错误类型属于超长类，先调用ContextManager.compress()，压缩完成后再发新请求；如果压缩后仍超长，继续检查是否有工具返回未截断、是否有重复消息堆积。

4.2 指数退避与请求幂等

对于真正值得重试的瞬时故障，重试必须带退避策略。我常用的规则是：第一次失败等1秒，第二次等2秒，第三次等4秒，以此类推，最大等待时间不超过30秒，并且每次加上随机抖动。抖动很关键，否则多个请求同时重试，会在服务端形成新的波峰。示例逻辑：

import random import time def retry_with_backoff(attempt: int, max_wait: int = 30) -> float: wait = min(2 ** attempt, max_wait) wait += random.uniform(0, 1) return wait

幂等是另一个容易被忽略的问题。Agent任务里如果调用的是扣费API，比如短信发送、文件写入、支付接口，重试前必须判断上一次是否真的执行成功。上一轮工具调用可能已经完成，但响应在回传时断流，这时候重试会导致重复扣费。

我的做法是给工具调用加request_id，服务端做去重；如果服务端没有全局去重能力，就在状态快照里记录“该步骤已经完成”，重试时直接跳过已完成步骤。

4.3 状态快照：让任务从断点续跑

当Agent执行复杂任务时，每完成一个“关键子任务”，就把当前状态序列化到一个JSON文件或Redis里。状态内容包括：

{ "task_id": "task_20250101_123456", "plan": ["调研", "提取", "汇总", "生成报告"], "current_step": 2, "context_messages": [...], "tool_cache": {"search_1": "结果摘要"}, "retry_count": 3 }

保存和加载快照的代码很简单：

import os import json def save_checkpoint(state: dict, path: str): with open(path, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) def load_checkpoint(path: str): if not os.path.exists(path): return None with open(path, "r", encoding="utf-8") as f: return json.load(f)

一旦断流导致进程退出，就可以加载快照，恢复ContextManager里面的history，从断点继续，不需要从头跑。快照不仅用于断流恢复，也方便人工接管：你可以看到任务已完成一半的结果，决定是继续还是改指令。

5. 踩坑记录与排查速查表

5.1 我踩过的四个坑

坑一：摘要时把系统提示词也丢了。Agent执行摘要后，行为规范全部失效，后续任务偏离指令。后来修复为：system prompt单独保存，不参与摘要，压缩时只处理历史对话和工具返回。

坑二：每次循环都做token统计，导致性能下降。尤其在高频调用工具的场景下，每轮多一次遍历成本不小。后来改成每3步统计一次，并且只在调用工具之后统计，因为工具返回往往是体积变化最大的点。

坑三：重试风暴。上下文超长后，我用通用重试函数直接重发原请求，结果连续重试5次全部失败，还触发了服务商限流。后来在重试前加了need_compress()判断，超长错误先压缩再重试。

坑四：没有做工具结果截断。这是最典型的“延迟爆炸”：前10轮的小对话没问题，第11次工具返回3万token，直接超限。现象就是任务前期很稳，越往后越容易断，日志里什么规律都找不到。后来把所有出入Agent的工具返回统一走trim_tool_result，问题消失。

5.2 排查速查表

5.3 什么时候该换更大上下文的模型

现在不少模型已经支持1M token的上下文，但不要以为用了大上下文就一劳永逸。大上下文意味着更长的prefill时间、更高的成本和潜在的延迟。如果任务上下文只是单纯“累积了太多垃圾信息”，换更大模型反而掩盖了设计缺陷。

我的经验是：如果任务需要的信息密度高、交互轮次超过30轮、且工具返回天然大，那优先做“结构化记忆”而不是硬塞上下文。只有当业务无法拆分、必须要求模型记住大量原文时，才考虑超大上下文模型。比如需要模型在很长的代码仓库里做全局推理，这种情况大上下文有优势；但如果是“搜索网页并总结要点”，完全可以把每次搜索结果压缩成摘要，没必要保留原始HTML。

最后说点我个人的体会。给Agent加逃生通道，本质上不是某一个函数的事，而是要把“上下文超长”当作战术上必然会发生的正常运行时状态来对待，而不是意外。我从一开始只修报错，到后来把预算监控、摘要压缩、断点续跑全链路打通，最大的感受是：一个Agent项目的稳定性和可维护性，取决于你对失败的设计，而绝不仅仅是对成功的追求。如果你现在也被对话超长断流折腾，不妨先把我这个上下文管家塞进项目里，跑一个长任务看看。很多问题会提前暴露，也就能提前救回来。