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

2026/6/25 23:41:41

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

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

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

以一个客服 Agent 为例：用户在第 3 轮提供了订单号，第 15 轮又问物流状态，Agent 却要求重新提供订单号——因为中间 12 轮的对话已经把原始信息挤出了有效注意力范围。更严重的是，当上下文逼近模型窗口上限时，API 调用的 Token 费用线性增长，但回复质量反而下降，ROI 急剧恶化。

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

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

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

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

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

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

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

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

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

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

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

from dataclasses import dataclass, field from typing import Optional import tiktoken import time @dataclass class Message: role: str # user / assistant / tool content: str token_count: int = 0 timestamp: 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_name self.max_total_tokens = max_total_tokens self.short_term_rounds = short_term_rounds self.summary_max_tokens = summary_max_tokens self.llm_client = llm_client self.encoding = tiktoken.encoding_for_model(model_name) # 短期记忆：完整对话消息列表 self.short_term: list[Message] = [] # 中期记忆：历史摘要 self.summaries: list[str] = [] # 系统提示（固定占用，不参与淘汰） self.system_prompt: Optional[str] = None def count_tokens(self, text: str) -> int: """精确计算 Token 数，而非估算""" return len(self.encoding.encode(text)) def set_system_prompt(self, prompt: str) -> None: self.system_prompt = prompt def 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 字\n\n" 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.content except 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 = 0 if 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_count return total def build_messages(self) -> list[dict]: """构建发送给 LLM 的完整消息列表 组装顺序：系统提示 → 历史摘要 → 短期对话 摘要放在系统提示之后，确保模型优先关注近期对话 """ messages = [] if self.system_prompt: messages.append({"role": "system", "content": self.system_prompt}) # 将所有摘要合并为一条系统消息注入 if self.summaries: combined = "\n\n".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，才值得引入分层记忆的额外复杂度。