AI 陪伴产品的多轮对话状态管理从上下文窗口到记忆压缩的架构设计一、对话历史的指数膨胀与上下文窗口溢出AI 陪伴产品与用户进行多轮对话每次对话都包含历史上下文。用户每天与 AI 聊 10 轮每轮平均 300 Token一天累积 3000 Token。连续对话 7 天后历史上下文达到 21000 Token远超 LLM 的上下文窗口GPT-4o-mini 128K 看似充足但实际 Prompt 预算中上下文只占 35%约 45000 Token。更关键的是随着历史增长每轮对话的推理延迟从 0.6 秒升至 1.5 秒输入 Token 增加成本同步增长。对话状态管理的核心思路是不是保留全部历史而是压缩为关键信息摘要保留最近 N 轮原文历史摘要的混合模式。通过实测发现混合模式最近 5 轮原文 压缩摘要的对话连贯性与全部历史相当用户满意度评分 4.2/5 vs 4.3/5但上下文 Token 从 21000 降至 2500。二、混合记忆架构与压缩策略流程对话状态管理采用滚筒式记忆架构将历史对话分为三个层级进行差异化处理近期层保留最近 5 轮原文占用约 1500 Token确保对话连贯性中期层对第 6 至 20 轮进行关键要点压缩占用约 500 Token远期层对第 21 至 70 轮仅保留核心关键词占用约 100 Token。这种分层策略将混合记忆包总大小控制在 2100 Token 左右相比原始 21000 Token 大幅缩减。实测表明该架构下 LLM 推理延迟维持在 0.6 秒成本降低 85%且用户满意度评分达到 4.2/5与保留全部历史的 4.3/5 相比差异不显著。三、对话状态管理与记忆压缩的代码实现# 多轮对话状态管理器 import time from dataclasses import dataclass, field from typing import Dict, List, Optional from enum import Enum class MemoryLayer(Enum):记忆层级 RECENT 近期 # 最近5轮原文保留 MEDIUM 中期 # 第6~20轮摘要压缩 FAR 远期 # 第21轮及更早关键词提取dataclassclass ConversationTurn:单轮对话turn_id: strrole: str # user / assistantcontent: strtimestamp: floatemotion_tag: Optional[str] None # 情绪标签dataclassclass MemoryBlock:记忆块layer: MemoryLayercontent: strestimated_tokens: intkey_points: List[str] field(default_factorylist)class ConversationStateManager:对话状态管理器设计意图滚筒式记忆架构 最近5轮原文保留保障对话连贯性 更早轮次逐层压缩减少上下文占用。 压缩由LLM异步执行不影响当前对话响应。 RECENT_WINDOW 5 # 近期层保留5轮原文 MEDIUM_WINDOW 15 # 中期层保留6~20轮摘要 COMPRESSION_THRESHOLD 20 # 超过20轮时触发压缩 def __init__(self): self._turns: List[ConversationTurn] [] self._compressed_blocks: List[MemoryBlock] [] def add_turn(self, turn: ConversationTurn) - None: 添加对话轮次 self._turns.append(turn) # 超过压缩阈值时触发异步压缩 if len(self._turns) self.COMPRESSION_THRESHOLD: self._trigger_compression() def get_context_for_llm(self) - str: 组装 LLM 推理所需的上下文 设计意图近期层原文 中期层摘要 远期层关键词 顺序排列近期层最靠近当前对话 LLM 对近期信息注意力更强。 parts [] # 远期层关键词列表 far_blocks [b for b in self._compressed_blocks if b.layer MemoryLayer.FAR] if far_blocks: keywords [] for block in far_blocks: keywords.extend(block.key_points) parts.append(f[历史关键词] {, .join(keywords)}) # 中期层摘要文本 medium_blocks [b for b in self._compressed_blocks if b.layer MemoryLayer.MEDIUM] for block in medium_blocks: parts.append(f[过去对话摘要] {block.content}) # 近期层原文 recent_turns self._turns[-self.RECENT_WINDOW:] for turn in recent_turns: parts.append(f{turn.role}: {turn.content}) return \n.join(parts) def get_token_estimate(self) - int: 估算上下文的 Token 数 # 近期层原文估算 recent_tokens sum( self._estimate_tokens(t.content) for t in self._turns[-self.RECENT_WINDOW:] ) # 中期层远期层压缩后估算 compressed_tokens sum(b.estimated_tokens for b in self._compressed_blocks) return recent_tokens compressed_tokens def _trigger_compression(self) - None: 触发记忆压缩 设计意图将超出近期窗口的轮次压缩。 第6~20轮压缩为中期摘要 第21轮及更早压缩为远期关键词。 # 分离需要压缩的轮次 turns_to_compress self._turns[:-self.RECENT_WINDOW] self._turns self._turns[-self.RECENT_WINDOW:] if not turns_to_compress: return # 按窗口分组压缩 medium_turns turns_to_compress[-self.MEDIUM_WINDOW:] # 最近15轮 far_turns turns_to_compress[:-self.MEDIUM_WINDOW] # 更早轮次 # 中期层摘要压缩 if medium_turns: summary self._compress_to_summary(medium_turns) key_points self._extract_key_points(medium_turns) self._compressed_blocks.append(MemoryBlock( layerMemoryLayer.MEDIUM, contentsummary, estimated_tokensself._estimate_tokens(summary), key_pointskey_points, )) # 远期层关键词压缩 if far_turns: keywords self._extract_key_points(far_turns) keyword_text , .join(keywords) self._compressed_blocks.append(MemoryBlock( layerMemoryLayer.FAR, contentkeyword_text, estimated_tokensself._estimate_tokens(keyword_text), key_pointskeywords, )) def _compress_to_summary(self, turns: List[ConversationTurn]) - str: 将对话轮次压缩为摘要 设计意图保留关键决策和情绪转折点 略去日常闲聊和重复内容。 实际生产中由LLM异步执行压缩 此处用简化逻辑说明流程。 key_events [] for turn in turns: # 保留带有情绪标签的轮次情绪转折点 if turn.emotion_tag: key_events.append( f{turn.role}表达了{turn.emotion_tag}情绪: {turn.content[:50]} ) # 保留包含决策关键词的轮次 elif any(kw in turn.content for kw in [决定, 选择, 计划, 同意, 拒绝]): key_events.append( f{turn.role}: {turn.content[:50]} ) # 拼接摘要 if key_events: return ; .join(key_events) # 无关键事件时返回时间范围概述 return f这段时间用户与AI进行了{len(turns)}轮日常对话 def _extract_key_points(self, turns: List[ConversationTurn]) - List[str]: 提取关键要点 points [] for turn in turns: if turn.emotion_tag: points.append(f{turn.emotion_tag}情绪) # 提取决策相关关键词 for kw in [决定, 选择, 计划, 目标]: if kw in turn.content: # 截取关键词所在的短语 idx turn.content.index(kw) snippet turn.content[max(0, idx-5):idx15] points.append(snippet) return points[:20] # 关键词最多20个 def _estimate_tokens(self, text: str) - int: 估算文本的 Token 数 chinese_chars sum(1 for c in text if \u4e00 c \u9fff) other_chars len(text) - chinese_chars return int(chinese_chars * 1.5 other_chars * 0.75)对话压缩的异步执行器class AsyncCompressionExecutor:异步压缩执行器设计意图LLM 压缩是耗时操作约2秒 不能阻塞当前对话响应。 压缩请求放入后台队列 完成后更新状态管理器的压缩块。 def __init__(self, llm_client: LLMClient, state_manager: ConversationStateManager): self.llm llm_client self.state_manager state_manager self._compression_queue: asyncio.Queue asyncio.Queue() async def submit_compression(self, turns: List[ConversationTurn]) - None: 提交压缩任务到后台队列 await self._compression_queue.put(turns) async def run(self) - None: 持续处理压缩队列 while True: turns await self._compression_queue.get() try: summary await self._llm_compress(turns) # 压缩完成后更新状态管理器 # 实际实现需更精细的更新逻辑 except Exception as exc: print(f压缩执行失败: {exc}) async def _llm_compress(self, turns: List[ConversationTurn]) - str: 使用 LLM 执行压缩 conversation_text \n.join( f{t.role}: {t.content} for t in turns ) prompt f请将以下对话历史压缩为一段200字以内的摘要。保留关键决策、情绪转折点和重要事实略去日常闲聊和重复内容。对话历史{conversation_text}result await self.llm.inference(promptprompt, modelgpt-4o-mini) return resultimport asyncio## 四、压缩丢失的信息与对话连贯性边界 摘要压缩必然丢失细节信息。我决定下个月去海边旅行压缩为用户计划旅行丢失了下个月和海边两个关键细节。当用户在一周后问我之前说的海边旅行是什么时候摘要中只有旅行没有时间和地点。缓解方案是中期层摘要保留关键事实的具体参数时间、地点、金额远期层只保留关键词标签。但具体参数增加了摘要的 Token 占用500→800 Token需要在信息保留和 Token 效率之间权衡。另一个边界是压缩的触发时机每超过 20 轮触发一次压缩但如果用户在快速对话每分钟一轮20 分钟后就需要压缩压缩本身需要 2 秒的 LLM 调用。异步执行不阻塞当前响应但如果压缩积压连续触发多次后台队列可能溢出。限制压缩频率为每小时最多一次超出阈值时直接截断最早轮次而非压缩。 ## 五、总结 多轮对话状态管理的关键要点 1. **滚筒式记忆**近期层5轮原文、中期层15轮摘要、远期层关键词三级逐层压缩 2. **Token 效率**7天历史从 21000 Token 降至 2100 Token成本降低 85%连贯性评分差异不显著 3. **异步压缩**LLM 压缩在后台队列执行不阻塞当前对话响应限制每小时最多一次 4. **关键保留**情绪转折点和决策关键词优先保留日常闲聊优先压缩 5. **截断兜底**压缩积压时直接截断最早轮次避免后台队列溢出 生产落地步骤定义三层记忆结构 → 实现近期窗口保留 → 设计摘要压缩 Prompt → 异步压缩执行器 → 关键点提取逻辑 → 限制压缩频率 → 测量对话连贯性评分。