AI Agent 长对话管理：上下文窗口溢出的工程解法

AI Agent 长对话管理：上下文窗口溢出的工程解法

一、对话越长越笨：Agent 上下文管理的真实困境

大模型 Agent 在短对话场景下表现尚可，但当对话轮次超过 20 轮、上下文逼近 Token 上限时，问题集中爆发：模型开始遗忘早期约定、重复执行已完成的工具调用、甚至产生与上下文矛盾的回复。这不是模型能力不足，而是上下文管理策略的缺失。

拿客服 Agent 来说，这类问题其实很典型：用户在第 3 轮已经给过订单号，到了第 15 轮再追问物流状态，Agent 却还得让用户重新报一遍订单号。原因不复杂——中间那 12 轮对话，已经把最初那条关键信息挤出了有效注意力范围。更麻烦的是，一旦上下文快碰到模型窗口上限，API 调用的 Token 成本会线性往上走，可回复质量不但没提升，反而开始下滑，ROI 也会迅速变差。

长对话管理的核心矛盾是：可用上下文窗口是有限的，但业务对话的信息量是无限增长的。必须有一套工程化的策略，在有限窗口内保留高价值信息、淘汰低价值信息，同时控制 Token 成本。

二、上下文窗口的运作机制与溢出策略

理解上下文管理，先要理解大模型如何消费上下文窗口。每次 API 调用时，模型接收的是完整的消息列表——包括系统提示、历史对话、工具调用结果。这个列表的总 Token 数不能超过模型上下文窗口（如 GPT-4o 的 128K、Claude 的 200K）。

graph LRsubgraph 上下文窗口组成A[系统提示 System Prompt] --> B[对话历史 Conversation History]B --> C[工具调用结果 Tool Results]C --> D[当前用户输入 Current Input]endsubgraph 溢出处理策略E[滑动窗口截断] --> F[摘要压缩]F --> G[向量检索补充]G --> H[分层记忆架构]endD -->|Token 超限| E

四种主流的溢出处理策略各有适用场景：

滑动窗口截断是最简单的策略——保留最近 N 轮对话，丢弃更早的。优点是实现简单，缺点是丢失关键信息。适合闲聊型 Agent，不适合需要长期记忆的任务型 Agent。

摘要压缩是用 LLM 对早期对话生成摘要，用摘要替代原文。保留语义的同时大幅减少 Token 占用。但摘要本身有信息损失，且额外的 LLM 调用增加了延迟和成本。

向量检索补充是将对话历史存入向量数据库，每次对话时检索最相关的片段注入上下文。适合知识密集型场景，但检索质量依赖 Embedding 模型，且增加了系统复杂度。

分层记忆架构是综合方案——短期记忆保留最近对话，中期记忆存储摘要，长期记忆使用向量检索。三层协同，在 Token 预算内最大化信息密度。

三、生产级分层记忆系统实现

以下实现基于分层记忆架构，包含 Token 预算管理和自动摘要压缩。

from dataclasses import dataclass, fieldfrom typing import Optionalimport tiktokenimport time@dataclassclass Message:role: str# user / assistant / toolcontent: strtoken_count: int = 0timestamp: float = field(default_factory=time.time)metadata: dict = field(default_factory=dict)class ConversationMemory:"""分层对话记忆管理器设计思路：- 短期记忆：保留最近 N 轮完整对话，保证即时上下文连贯- 中期记忆：对超出短期窗口的对话自动生成摘要- Token 预算：硬性约束，任何情况下不超限"""def __init__(self,model_name: str = "gpt-4o",max_total_tokens: int = 120000,# 留 8K 给输出short_term_rounds: int = 10,# 短期保留最近 10 轮summary_max_tokens: int = 2000, # 摘要最大 Token 数llm_client = None,# LLM 客户端，用于生成摘要):self.model_name = model_nameself.max_total_tokens = max_total_tokensself.short_term_rounds = short_term_roundsself.summary_max_tokens = summary_max_tokensself.llm_client = llm_clientself.encoding = tiktoken.encoding_for_model(model_name)# 短期记忆：完整对话消息列表self.short_term: list[Message] = []# 中期记忆：历史摘要self.summaries: list[str] = []# 系统提示（固定占用，不参与淘汰）self.system_prompt: Optional[str] = Nonedef count_tokens(self, text: str) -> int:"""精确计算 Token 数，而非估算"""return len(self.encoding.encode(text))def set_system_prompt(self, prompt: str) -> None:self.system_prompt = promptdef add_message(self, role: str, content: str, metadata: dict = None) -> None:"""添加消息并触发记忆管理"""msg = Message(role=role,content=content,token_count=self.count_tokens(content),metadata=metadata or {},)self.short_term.append(msg)self._manage_memory()def _manage_memory(self) -> None:"""核心记忆管理逻辑：超限时压缩早期对话"""# 计算当前总 Token 占用total = self._calculate_total_tokens()if total <= self.max_total_tokens:return# 策略：将超出短期窗口的早期对话压缩为摘要while len(self.short_term) > self.short_term_rounds and total > self.max_total_tokens:# 取出最早的一轮对话（user + assistant）overflow_messages = []while len(self.short_term) > self.short_term_rounds:overflow_messages.append(self.short_term.pop(0))if overflow_messages:summary = self._generate_summary(overflow_messages)self.summaries.append(summary)total = self._calculate_total_tokens()def _generate_summary(self, messages: list[Message]) -> str:"""使用 LLM 对早期对话生成摘要摘要提示词的关键：要求保留实体名、数值、决策结论，丢弃寒暄、重复确认等低信息密度内容"""if not self.llm_client:# 降级策略：无 LLM 客户端时，拼接关键信息return self._fallback_summary(messages)conversation_text = "n".join(f"{m.role}: {m.content}" for m in messages)prompt = ("请将以下对话压缩为一段摘要。要求：n""1. 保留所有实体名称、数值、日期、订单号等关键信息n""2. 保留用户的核心诉求和已做出的决策n""3. 丢弃寒暄、重复确认、格式化输出等低价值内容n""4. 摘要不超过 300 字nn"f"对话内容：n{conversation_text}")# 调用 LLM 生成摘要，设置超时防止挂起try:response = self.llm_client.chat.completions.create(model=self.model_name,messages=[{"role": "user", "content": prompt}],max_tokens=500,timeout=10,)return response.choices[0].message.contentexcept Exception as e:# LLM 调用失败时降级为基础拼接return self._fallback_summary(messages)def _fallback_summary(self, messages: list[Message]) -> str:"""降级摘要：提取用户消息的关键片段"""user_msgs = [m.content[:100] for m in messages if m.role == "user"]return " | ".join(user_msgs) if user_msgs else "[对话已压缩]"def _calculate_total_tokens(self) -> int:"""计算当前上下文的总 Token 占用"""total = 0if self.system_prompt:total += self.count_tokens(self.system_prompt)for summary in self.summaries:total += self.count_tokens(summary)for msg in self.short_term:total += msg.token_countreturn totaldef build_messages(self) -> list[dict]:"""构建发送给 LLM 的完整消息列表组装顺序：系统提示 → 历史摘要 → 短期对话摘要放在系统提示之后，确保模型优先关注近期对话"""messages = []if self.system_prompt:messages.append({"role": "system", "content": self.system_prompt})# 将所有摘要合并为一条系统消息注入if self.summaries:combined = "nn".join(f"[历史对话摘要 {i+1}]n{s}"for i, s in enumerate(self.summaries))messages.append({"role": "system","content": f"以下是之前对话的摘要：n{combined}",})# 短期记忆：完整对话for msg in self.short_term:messages.append({"role": msg.role, "content": msg.content})return messages

关键设计决策说明：

Token 计算使用 tiktoken 而非估算。 字符数与 Token 数的映射因模型而异，中文场景下 1 个汉字约 1.5-2 个 Token。估算误差会导致实际调用时超限或预算浪费。

摘要生成有降级策略。 LLM 调用可能因网络、限流等原因失败，此时退回到基础拼接而非抛出异常，保证对话不中断。

摘要注入位置在系统提示之后、短期对话之前。 这个位置确保模型在处理当前对话时，历史摘要作为背景知识存在，但不会干扰对近期上下文的理解。

四、分层记忆的代价与适用边界

分层记忆架构并非没有代价，需要在多个维度做权衡：

摘要的信息损失不可逆。 一旦对话被压缩为摘要，原始措辞、语气、隐含语义都会丢失。如果后续对话需要精确引用早期对话的原文，摘要无法满足。对策是：对关键实体（订单号、金额、日期）做结构化提取，存入独立的键值存储，不依赖摘要。

额外的 LLM 调用增加延迟和成本。 每次摘要生成是一次额外的 API 调用，增加 500ms-2s 的延迟。在对话密集场景下，可以异步生成摘要——先截断对话保证当前请求正常响应，后台异步压缩。

Token 预算分配需要调优。 系统提示、摘要、短期对话各占多少 Token，没有通用最优解。系统提示过长会挤压对话空间，摘要过多会稀释近期上下文的注意力权重。建议按 1:2:7 的比例分配，并根据实际效果调整。

适用场景： 任务型 Agent（客服、工单处理、技术支持），对话轮次通常超过 10 轮且需要跨轮次引用信息。不适用场景： 单轮问答、创意写作等短对话场景，分层记忆的复杂度不值得。

五、总结

说白了，AI Agent 的长对话管理，归根结底是在有限的 Token 预算里，如何把信息密度拉到最大的工程难题。分层记忆架构之所以有效，就在于它构建了一套“短期完整对话 + 中期摘要 + 长期检索”的三层体系，即便上下文窗口面临溢出风险，也能确保关键信息不丢失。在生产环境中，真正考验功力的是那些细节：Token 的精确计算、摘要降级策略以及预算分配比例。至于架构选型，其实不必过度设计：短对话场景下，滑动窗口足以应付；只有当对话轮次持续拉长、且任务型 Agent 必须依赖跨轮次记忆时，引入分层记忆带来的额外复杂度才是值得的。