Agent系统的上下文溢出通常表现为两种形态：一是请求时由于token总数超过模型上限而被直接拒绝，二是对话累积到一定程度后响应延迟急剧升高，甚至模型开始重复回答或遗漏指令。滑动窗口和摘要压缩是当前应对这一问题的两条主流路线，两者都把控制发送给模型的上下文长度作为目标，但设计逻辑和适用的业务场景差异很大。本文从上下文构成的底层原理出发，分别剖析这两种方案的实现细节，并给出可落地的混合工程策略。
一、上下文溢出的根因与可观测性
要解决溢出问题，先要理解上下文从哪里来。一次Agent调用发送给模型的完整内容通常由系统提示、用户本次指令、历史对话消息、工具调用结果、检索到的参考文档五部分组成。其中历史消息和工具结果增长最快，每一轮工具返回的JSON或文本都可能长达数百甚至上千token，多轮累积下来就会迅速逼近模型的最大输入限制。
需要特别强调的是，即便历史消息没有达到硬性上限，上下文过长也会带来两个隐性成本。第一个是注意力计算量随序列长度平方增长，推理延迟明显上升；第二个是模型对早期信息的关注度会被大量中间内容稀释，导致较早在系统提示或用户第一轮对话中提出的约束条件被忽略。后者在实测中往往比显式报错更难排查，属于质量层面的溢出。
因此，生产环境里应当对上下文做主动观测。建议在调用模型前用tokenizer统计消息列表的token数，并记录每轮请求的响应延迟。当token数达到模型上限的70%至80%时，就应当触发压缩策略，而不是等到报错才处理。以下代码展示了基于token阈值检查上下文是否接近溢出的思路：
def check_context_usage(messages, model_max_tokens=128000, warn_ratio=0.8):
"""检查当前上下文的token占用情况"""
total_tokens = 0
for msg in messages:
total_tokens += estimate_tokens(msg.get("content", ""))
usage_ratio = total_tokens / model_max_tokens
status = "ok" if usage_ratio <= warn_ratio else "need_compress"
return {"total_tokens": total_tokens, "usage_ratio": usage_ratio, "status": status}
这里的estimate_tokens可以是tokenizer编码后的长度，也可以根据字符数粗略估算。一旦状态返回need_compress，就说明当前上下文已经触及安全红线，应当立即对历史消息做处理。
二、滑动窗口：简单直接的短期记忆方案
滑动窗口的核心思路是只保留最近N轮对话，把更早的消息丢弃。N可以由轮数决定，也可以由token总数决定。该方案实现成本低，不需要额外调用大模型，因此对响应延迟几乎没有影响，特别适合高频低延迟的交互场景，比如客服机器人和命令行辅助工具。
实现滑动窗口时，最常见的做法是使用固定长度的双端队列。新消息从尾部进入，超出容量上限的旧消息自动从头部退出。系统提示通常被排除在窗口之外，以便保留任务指令。需要小心的是，如果窗口过小，用户早期表达的关键偏好就会被挤掉，导致Agent在长任务中丢失原始目标。例如一个数据分析任务，用户在第一轮指定了统计口径，几轮问答之后该条件被移出窗口，Agent后续给出的结果就可能与用户预期不一致。
from collections import deque
class SlidingWindowMemory:
def __init__(self, max_messages=20, system_prompt=""):
self.messages = deque(maxlen=max_messages)
self.system_prompt = system_prompt
def add_message(self, role, content):
self.messages.append({"role": role, "content": content})
def build_request(self):
full_messages = []
if self.system_prompt:
full_messages.append({"role": "system", "content": self.system_prompt})
full_messages.extend(self.messages)
return full_messages
def trim_to_token_limit(self, max_tokens=8000):
while self.messages and estimate_tokens(self.messages[0]["content"]) > max_tokens:
self.messages.popleft()
按消息条数滑动实现简单，但不同消息的长度差异往往很大，极端情况下一条工具返回就能占满窗口。更精细的做法是按token数滑动：估算每条消息的token数，从尾部向前累加，直到达到预设上限，然后丢弃更早的消息。这种方式能更稳定地控制请求体积，代价是需要额外计算token长度。
三、摘要压缩：用语义概括换取长期记忆
摘要压缩的思路是把早期的大量历史消息交给摘要模型，加工成一段精炼的总结，然后以这条总结替换原始消息，压缩率通常在70%到90%之间。它解决了滑动窗口完全遗忘早期信息的问题，让Agent能够记住跨多轮的关键结论，例如用户偏好、已经执行过的操作步骤和尚未完成的事项。
实现摘要压缩需要注意触发时机。常见的做法是设置一个高水位和一个低水位：当上下文token数超过高水位时启动压缩，压缩完成后判断是否降到了低水位以下，如果仍然偏高则继续压缩更早的片段。这样做可以避免频繁调用摘要模型带来的额外开销和时间成本。摘要本身也要保留结构化信息，包括目标、已完成步骤、未完成步骤、关键约束和最近上下文概要，这样后续对话才能准确使用这些记忆。
async def compress_old_messages(history, recent_n=6, summary_model="gpt-4o-mini"):
"""把历史消息中较老的部分压缩成摘要"""
if len(history) <= recent_n:
return history
old_messages = history[:-recent_n]
recent_messages = history[-recent_n:]
prompt = (
"请为以下对话生成结构化摘要，包含：用户目标、已完成步骤、"
"未完成步骤、关键数据、重要约束。不要添加原文没有的信息。\n\n"
+ json.dumps(old_messages, ensure_ascii=False)
)
summary_resp = await call_llm(prompt, model=summary_model)
summary_text = summary_resp["choices"][0]["message"]["content"]
return [{"role": "system", "content": "对话摘要：" + summary_text}] + recent_messages
上述实现把摘要放在整个消息列表的最前面，替代被压缩掉的旧消息，同时保留最近几轮对话用于处理即时任务。这里的recent_n决定了保留多少条原始信息，通常取6到8条比较合适。摘要内容会随着对话推进被反复更新，每次都在上一轮摘要的基础上合并新内容，这实际上形成了一种递归压缩。
摘要压缩也存在明显风险：多轮摘要叠加后会出现信息衰减，摘要模型可能遗漏某个数字或某条约束，且这种遗漏会被后续轮次承袭和放大。缓解办法包括将摘要写入向量数据库，按时间维度分片保存，以及在压缩时对关键数据做二次校验。
四、滑动窗口与摘要压缩的混合策略
滑动窗口和摘要压缩不是二选一的关系，生产环境中更合理的方式是组合使用。滑动窗口负责短期记忆，保证最近几轮对话完整无损；摘要机制负责长期记忆，在窗口越界后将较早的内容沉淀为结构化的总结。两者结合之后，系统提示、对话摘要和最近对话三者共同构成发送给模型的上下文。
具体策略可以这样设计：首先把系统提示固定在上下文头部；其次维护一个可滑动的近期对话缓冲区，缓冲区满足时优先滑动丢弃；当滑动后仍超出token上限，就把缓冲区中最老的部分交给摘要模型，生成新的摘要替换原消息。摘要的生成频率可以通过低水位控制，避免每轮都触发。
class HybridMemory:
def __init__(self, max_recent_tokens=6000, high_watermark=8000, low_watermark=5000):
self.max_recent_tokens = max_recent_tokens
self.high_watermark = high_watermark
self.low_watermark = low_watermark
self.summary = ""
self.recent_messages = []
async def add_and_compact(self, message):
self.recent_messages.append(message)
await self.compact_if_needed()
async def compact_if_needed(self):
current_tokens = estimate_tokens(self.summary + json.dumps(self.recent_messages))
if current_tokens <= self.high_watermark:
return
self.summary = await generate_summary(self.summary, self.recent_messages[: -4])
self.recent_messages = self.recent_messages[-4:]
if estimate_tokens(self.summary + json.dumps(self.recent_messages)) > self.low_watermark:
self.summary = await generate_summary(self.summary, self.recent_messages[: -2])
self.recent_messages = self.recent_messages[-2:]
def build_request(self):
messages = [{"role": "system", "content": self.summary}]
messages.extend(self.recent_messages)
return messages
从工程角度还有几个建议。一是将摘要结果持久化，即使Agent实例重启也能恢复长期记忆；二是为摘要设置版本号或时间戳，方便在出现关键信息丢失时回溯原始对话；三是在连续任务场景中，把用户的核心目标单独存放在一个独立的槽位，不参与滑动窗口的淘汰，确保方向性指令永远不被压缩掉。
最后强调一下评估方法。在切换压缩策略后，用一组覆盖长任务、多轮工具调用和用户中途改需求的测试用例来回归验证，比较压缩前后的任务完成率和关键信息召回率。不要仅仅用token消耗量来衡量方案优劣，稳定完整地完成任务才是终极目标，这也是滑动窗口与摘要压缩需要被组合使用、不断调优的根本原因。