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 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_factorytime.time) metadata: dict field(default_factorydict) 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( rolerole, contentcontent, token_countself.count_tokens(content), metadatametadata 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( modelself.model_name, messages[{role: user, content: prompt}], max_tokens500, timeout10, ) 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[历史对话摘要 {i1}]\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才值得引入分层记忆的额外复杂度。