LLM 多轮对话状态管理：从上下文窗口优化到会话持久化
一、上下文窗口的"挤牙膏"困境：长对话场景的核心瓶颈
大模型应用在多轮对话场景中面临一个根本性矛盾：对话历史越长，模型理解越准确，但 Token 消耗也越大。当对话轮次超过 20 轮时，仅历史消息就可能占满 128K 的上下文窗口，留给系统提示和工具描述的空间所剩无几。更棘手的是，用户可能在第 15 轮突然回到第 3 轮的话题，如果中间轮次的上下文已被截断，模型将完全丢失之前的约定和约束。
传统方案是简单的滑动窗口截断——保留最近 N 轮对话，丢弃更早的历史。这种做法虽然控制了 Token 开销，但牺牲了长程依赖能力。在客服、法律咨询、代码审查等需要持续引用早期约定的场景中，信息丢失直接导致回答质量下降。
二、多轮对话状态管理的三层架构
解决上述问题需要从"全量保留"转向"有损压缩+按需召回"的架构模式。
graph TB A[用户输入] --> B[会话管理层] B --> C[短期记忆：最近 N 轮原文] B --> D[中期记忆：摘要压缩] B --> E[长期记忆：向量检索] C --> F[LLM 上下文组装] D --> F E -->|按需召回| F F --> G[模型输出] G --> B subgraph 状态持久化 H[Redis: 短期会话] I[PostgreSQL: 摘要存储] J[向量数据库: 长期索引] end C -.-> H D -.-> I E -.-> J短期记忆保留最近 5-8 轮对话原文，确保当前语境的精确性。中期记忆对更早的对话进行摘要压缩，将每 5 轮对话压缩为一段 100-200 字的摘要，保留关键决策和约定。长期记忆将摘要和关键实体存入向量数据库，当用户提及历史话题时，通过语义检索召回相关片段。
三、生产级代码实现
3.1 会话状态管理器
# conversation_manager.py # 多轮对话状态管理器，实现三层记忆架构 from dataclasses import dataclass, field from typing import Optional import hashlib import json @dataclass class ConversationTurn: role: str content: str timestamp: float summary: Optional[str] = None # 延迟生成的摘要 entities: list[str] = field(default_factory=list) # 关键实体 class ConversationManager: SHORT_TERM_LIMIT = 8 # 短期记忆保留轮数 SUMMARY_INTERVAL = 5 # 每 N 轮生成一次摘要 SUMMARY_MAX_TOKENS = 150 # 摘要最大 Token 数 def __init__(self, redis_client, pg_client, vector_client, llm_client): self.redis = redis_client self.pg = pg_client self.vector = vector_client self.llm = llm_client async def add_turn(self, session_id: str, role: str, content: str) -> None: """添加一轮对话到短期记忆""" turn = ConversationTurn( role=role, content=content, timestamp=time.time() ) # 写入 Redis 短期记忆 key = f"conv:{session_id}:short" await self.redis.rpush(key, json.dumps({ "role": turn.role, "content": turn.content, "timestamp": turn.timestamp })) # 检查是否需要生成摘要 turn_count = await self.redis.llen(key) if turn_count % self.SUMMARY_INTERVAL == 0: await self._generate_summary(session_id) # 裁剪短期记忆，保留最近 N 轮 if turn_count > self.SHORT_TERM_LIMIT: await self.redis.ltrim(key, -self.SHORT_TERM_LIMIT, -1) async def _generate_summary(self, session_id: str) -> None: """将即将移出短期记忆的对话压缩为摘要""" key = f"conv:{session_id}:short" all_turns = await self.redis.lrange(key, 0, -1) # 取出需要摘要的轮次（超出短期记忆的部分） overflow = all_turns[:len(all_turns) - self.SHORT_TERM_LIMIT] if not overflow: return # 调用 LLM 生成摘要 summary_prompt = ( "请将以下对话压缩为一段简洁摘要，保留关键决策、约定和实体信息：\n" + "\n".join(t["content"] for t in overflow) ) summary = await self.llm.chat( messages=[{"role": "user", "content": summary_prompt}], max_tokens=self.SUMMARY_MAX_TOKENS ) # 提取关键实体用于长期记忆索引 entities = await self._extract_entities(overflow) # 持久化摘要到 PostgreSQL await self.pg.execute( """INSERT INTO conversation_summaries (session_id, summary, entities, turn_range, created_at) VALUES ($1, $2, $3, $4, NOW())""", session_id, summary.content, entities, f"0-{len(overflow)}" ) # 将摘要向量化存入长期记忆 embedding = await self.vector.embed(summary.content) await self.vector.upsert( collection="conversation_memory", id=hashlib.md5(f"{session_id}:{len(overflow)}".encode()).hexdigest(), vector=embedding, metadata={"session_id": session_id, "summary": summary.content} ) async def get_context(self, session_id: str, current_input: str) -> list[dict]: """组装 LLM 上下文：短期原文 + 摘要 + 按需召回""" messages = [] # 1. 加载最近摘要（中期记忆） summaries = await self.pg.fetch( """SELECT summary FROM conversation_summaries WHERE session_id = $1 ORDER BY created_at DESC LIMIT 3""", session_id ) if summaries: summary_text = "之前的对话摘要：\n" + "\n".join( s["summary"] for s in summaries ) messages.append({"role": "system", "content": summary_text}) # 2. 加载短期记忆原文 short_key = f"conv:{session_id}:short" turns = await self.redis.lrange(short_key, 0, -1) for t in turns: messages.append({"role": t["role"], "content": t["content"]}) # 3. 按需召回长期记忆 if current_input: relevant = await self.vector.search( collection="conversation_memory", query=current_input, top_k=2, filter={"session_id": session_id} ) if relevant: recall_text = "相关的历史信息：\n" + "\n".join( r["metadata"]["summary"] for r in relevant ) # 插入到系统消息之后、短期记忆之前 messages.insert(1, {"role": "system", "content": recall_text}) return messages async def _extract_entities(self, turns: list[dict]) -> list[str]: """从对话中提取关键实体（人名、项目名、约束条件等）""" prompt = "从以下对话中提取关键实体（人名、项目名、技术术语、约束条件），每行一个：\n" prompt += "\n".join(t["content"] for t in turns) result = await self.llm.chat( messages=[{"role": "user", "content": prompt}], max_tokens=100 ) return [line.strip() for line in result.content.split("\n") if line.strip()]四、架构权衡与适用边界
摘要质量与延迟的权衡。摘要生成需要调用 LLM，每次摘要的延迟约 500ms-2s。如果用户高频发送消息，摘要生成可能跟不上对话节奏。解决方案是将摘要生成异步化，先写入短期记忆，后台任务定期批量压缩。
Token 预算分配策略。假设总上下文窗口为 128K Token，建议按 10%:20%:70% 分配给系统提示、摘要+召回、短期原文。当短期原文超过 70% 预算时，触发强制摘要压缩。
向量召回的噪声问题。语义检索可能召回与当前话题相似但实际无关的历史片段，引入噪声反而降低回答质量。建议在召回结果后增加一轮相关性判断，由 LLM 决定是否采纳。
适用边界：三层记忆架构适用于对话轮次超过 20 轮、需要引用历史约定的长对话场景。对于短对话（10 轮以内）场景，简单的滑动窗口即可满足需求，引入三层架构会增加不必要的复杂度。
五、总结
多轮对话状态管理的核心是从"全量保留"转向"分层压缩+按需召回"。短期记忆保留最近轮次原文，中期记忆通过 LLM 摘要压缩历史对话，长期记忆通过向量索引实现语义检索。在工程实践中，需要关注摘要生成的异步化处理、Token 预算的合理分配，以及向量召回的噪声过滤。对于短对话场景，不建议引入三层架构的额外复杂度。
