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_factorylist) # 关键实体 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( rolerole, contentcontent, timestamptime.time() ) # 写入 Redis 短期记忆 key fconv:{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 fconv:{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_tokensself.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, f0-{len(overflow)} ) # 将摘要向量化存入长期记忆 embedding await self.vector.embed(summary.content) await self.vector.upsert( collectionconversation_memory, idhashlib.md5(f{session_id}:{len(overflow)}.encode()).hexdigest(), vectorembedding, 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 fconv:{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( collectionconversation_memory, querycurrent_input, top_k2, 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_tokens100 ) 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 预算的合理分配以及向量召回的噪声过滤。对于短对话场景不建议引入三层架构的额外复杂度。