LangGraph 多轮对话记忆管理：从基础到高级的完整指南-程序员充电站

引言：为什么对话记忆管理如此重要？

在构建智能对话系统时，多轮对话的记忆管理是决定系统能否进行连贯、有上下文交互的关键因素。想象一下这样的场景：用户询问“北京的天气如何？”，系统回答后，用户接着问“那上海呢？”。如果系统忘记了之前的对话上下文，就无法理解这个“那”指的是什么。

LangGraph 作为构建复杂对话工作流的强大工具，提供了多种灵活的记忆管理策略。本文将深入探讨这些策略，帮助您构建更智能、更具人性的对话系统。

一、LangGraph 记忆管理基础

1.1 理解 LangGraph 的记忆机制

LangGraph 中的记忆管理核心是StateGraph的概念。每个对话状态都封装在一个可扩展的状态对象中，允许我们在对话流程中持久化和更新信息。

在 LangGraph 中，通常使用TypedDict来定义状态，并使用Annotated结合operator.add来指定如何将新值与旧值合并，这对于累积历史记录（如消息列表）非常有用。

from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END import operator # 定义对话状态结构 class ConversationState(TypedDict): # 完整的对话历史，使用 operator.add 累积 messages: Annotated[List[str], operator.add] # 当前用户输入 user_input: str # 系统回复 system_response: str # 其他自定义记忆字段 user_name: str conversation_topic: str interaction_count: int

1.2 最简单的记忆：无状态对话

对于简单的交互，我们可以选择不保留历史记录，每次对话都是独立的。

from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage class StatelessConversation: def __init__(self): # 假设已配置 OpenAI API Key self.llm = ChatOpenAI(model="gpt-3.5-turbo") def respond(self, user_input: str) -> str: """每次对话都是全新的，不保留历史""" messages = [HumanMessage(content=user_input)] response = self.llm.invoke(messages) return response.content # 使用示例 conversation = StatelessConversation() print(conversation.respond("你好，我是小明")) print(conversation.respond("我的名字是什么？")) # 系统会忘记之前的对话

二、四种核心记忆管理策略

2.1 策略一：完整历史记录 (Full History)

保留所有对话历史，适用于需要完整上下文的理解场景。

from langgraph.graph import StateGraph from typing import TypedDict, Annotated, List import operator from datetime import datetime from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI # 假设已安装 class FullHistoryState(TypedDict): """完整历史记录的状态定义""" messages: Annotated[List[dict], operator.add] current_input: str def add_to_history(state: FullHistoryState): """将当前对话添加到历史记录""" new_message = { "role": "user", "content": state["current_input"], "timestamp": datetime.now().isoformat() } # 使用列表包装确保 operator.add 可以正确累积 return {"messages": [new_message]} def generate_response(state: FullHistoryState): """基于完整历史生成回复""" llm = ChatOpenAI(model="gpt-3.5-turbo") # 构建包含所有历史的消息 formatted_messages = [] # 限制长度避免 token 超限 (例如：只取最近 10 条) for msg in state["messages"][-10:]: if msg["role"] == "user": formatted_messages.append(HumanMessage(content=msg["content"])) else: # 这里的逻辑需要确保 AIMessage 的内容是从前一次迭代中得到的 # 实际 LangGraph 中，状态会包含完整的用户和 AI 消息历史 pass # 简化处理，假设 state["messages"] 已经包含了用户和助手的消息 # 添加当前用户输入 formatted_messages.append(HumanMessage(content=state["current_input"])) response = llm.invoke(formatted_messages) # 准备将系统回复加入历史 ai_message = { "role": "assistant", "content": response.content, "timestamp": datetime.now().isoformat() } return { "messages": [ai_message], # 将助手回复也加入历史 "system_response": response.content } # 构建完整历史记录的对话图 # (LangGraph 编译和运行示例略去，专注于节点逻辑)

优点与缺点分析：

属性	描述
优点	保留完整上下文，理解能力强。
缺点	Token 消耗大，历史越长响应越慢，容易触及 LLM 上下文限制。
适用场景	复杂任务处理、深度分析对话、短期深度交流。

2.2 策略二：滑动窗口记忆 (Sliding Window)

只保留最近 N 轮对话，平衡记忆与效率。

from collections import deque from datetime import datetime class SlidingWindowMemory: def __init__(self, window_size: int = 5): self.window_size = window_size self.conversation_history = deque(maxlen=window_size) self.summary = "" # 窗口外对话的摘要 def add_interaction(self, user_input: str, ai_response: str): """添加新的对话交互""" interaction = { "user": user_input, "assistant": ai_response, "timestamp": datetime.now().isoformat() } # 如果窗口已满，将最旧的对话转为摘要 (简化，实际应调用 LLM) if len(self.conversation_history) == self.window_size: self._summarize_oldest() self.conversation_history.append(interaction) def _summarize_oldest(self): """摘要化最旧的对话 (简化实现)""" oldest = self.conversation_history[0] summary_text = f"用户提到：{oldest['user'][:50]}..." if not self.summary: self.summary = summary_text else: self.summary += f"；{summary_text}" # 保持摘要长度 if len(self.summary) > 200: self.summary = self.summary[:197] + "..." def get_context(self) -> str: """获取当前对话上下文""" context_parts = [] # 添加摘要 if self.summary: context_parts.append(f"先前对话摘要：{self.summary}") # 添加窗口内对话 for i, interaction in enumerate(self.conversation_history): context_parts.append( f"用户：{interaction['user']}\n" f"助手：{interaction['assistant']}" ) return "\n\n".join(context_parts)

滑动窗口优化技巧：

可以实现自适应滑动窗口，根据对话的重要性（例如提及姓名、地址等关键信息）临时扩大窗口，以保留重要信息。

2.3 策略三：基于摘要的记忆压缩 (Summarization)

将长对话历史压缩为摘要，节省 token 并保留关键信息。这种方法需要一个额外的 LLM 调用（通常是更小的模型）来执行摘要任务。

class SummarizedMemory: def __init__(self): self.detailed_history = [] # 最近几轮详细对话 self.conversation_summary = "" self.key_facts = {} # 关键事实存储 self.summary_threshold = 3 # 对话轮数阈值 # (add_interaction, _extract_key_facts, _update_summary, get_context_prompt # 逻辑与您提供的代码一致，这里省略重复代码以保持文章简洁) def _extract_key_facts(self, user_input: str, ai_response: str): """从对话中提取关键事实 (简化规则提取)""" personal_info_keywords = { "名字": ["我叫", "我的名字是"], "地点": ["住在", "来自"], } for fact_type, triggers in personal_info_keywords.items(): for trigger in triggers: if trigger in user_input: # 提取具体信息... # (提取逻辑略) pass def _update_summary(self): """更新对话摘要 (实际应调用 LLM)""" # (调用 LLM 生成摘要的逻辑略) print("--- 正在更新对话摘要 ---") self.conversation_summary = "新生成的摘要内容..." self.detailed_history = self.detailed_history[-2:] # 保留最近2轮详细对话 def get_context_prompt(self) -> str: """构建用于生成回复的上下文 prompt""" context_parts = [] # 添加对话摘要 if self.conversation_summary: context_parts.append(f"对话摘要：{self.conversation_summary}") # 添加关键事实 if self.key_facts: facts_text = "，".join([f"{k}：{v}" for k, v in self.key_facts.items()]) context_parts.append(f"已知信息：{facts_text}") return "\n".join(context_parts) # LangGraph 集成示例 def process_with_summary(state: dict): """使用摘要记忆处理对话""" # ... (处理逻辑略，核心是调用 memory.get_context_prompt()) pass

2.4 策略四：结构化记忆存储 (Vector/Retrieval-Augmented)

使用向量数据库等结构化方式存储和检索记忆，实现长期、语义相关的记忆。

from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter class VectorMemorySystem: def __init__(self, persistence_path: str = "./memory_db"): # 假设已配置 OpenAI Embeddings self.embeddings = OpenAIEmbeddings() self.vector_store = None self.persistence_path = persistence_path # (初始化和加载逻辑略) def add_conversation(self, user_input: str, ai_response: str, metadata: dict = None): """将对话添加到向量记忆""" conversation_text = f"用户：{user_input}\n助手：{ai_response}" # 分割文本为块，添加到向量存储，并持久化 # ... (添加和保存逻辑略) pass def retrieve_relevant_memories(self, query: str, k: int = 5) -> list: """检索与查询相关的记忆""" if not self.vector_store: return [] # 基于相似性检索 # docs = self.vector_store.similarity_search(query, k=k) # 格式化检索结果... # ... (检索和格式化逻辑略) return [] # LangGraph 中的使用 def retrieve_memories(state: dict): """检索相关记忆节点""" memory = state["vector_memory"] # 检索相关上下文 relevant_memories = memory.retrieve_relevant_memories( state["user_input"], k=5 ) context_text = "\n\n".join([ f"相关对话：{mem['content'][:200]}..." for mem in relevant_memories[:3] # 取最相关的3条 ]) return { "retrieved_context": relevant_memories, "context_text": context_text }

三、高级记忆管理技巧

3.1 混合记忆策略 (Hybrid Memory)

结合多种记忆策略的优势，实现多层次的记忆管理，这是现代对话系统的主流方案。

class HybridMemoryManager: """混合记忆管理器：短期（Sliding Window）、长期（Vector Store）、事实（Key Value）""" def __init__(self): # 1. 短期记忆：用于连贯性 self.short_term = SlidingWindowMemory(window_size=5) # 2. 长期记忆：用于深度检索 self.long_term = VectorMemorySystem() # 3. 事实记忆：用于关键信息快速提取 self.important_facts = {} def process_interaction(self, user_input: str, ai_response: str): """处理对话交互，使用多层记忆""" self.short_term.add_interaction(user_input, ai_response) # 检测重要信息并存储到长期记忆/事实记忆 importance = self._assess_importance(user_input, ai_response) if importance > 0.7: self.long_term.add_conversation(user_input, ai_response, metadata={"importance": importance}) facts = self._extract_facts(user_input) self.important_facts.update(facts) def get_context_for_generation(self, user_input) -> dict: """获取多层次的对话上下文""" return { "short_term": self.short_term.get_context(), "important_facts": self.important_facts, "retrieved_memories": self.long_term.retrieve_relevant_memories(user_input) } def _assess_importance(self, user_input: str, ai_response: str) -> float: """评估对话重要性 (简化启发式方法)""" # ... (评估逻辑略) return 0.5 def _extract_facts(self, user_input: str) -> dict: """从输入中提取关键事实 (简化)""" # ... (提取逻辑略) return {}

3.2 记忆的持久化和恢复

对于需要跨会话保留的用户数据，持久化是必需的。

import json import pickle from datetime import datetime import os class PersistentMemoryManager: """支持持久化的记忆管理器 (基于用户 ID)""" def __init__(self, user_id: str, storage_path: str = "./user_memories"): self.user_id = user_id self.storage_path = storage_path self.memory = {} self.load_memory() def save_memory(self): """保存记忆到磁盘 (JSON 和 Pickle)""" os.makedirs(self.storage_path, exist_ok=True) # ... (保存逻辑略，关键是使用 pickle 保存完整的对象状态) print(f"记忆已保存到 {self.storage_path}/{self.user_id}_memory.json") def load_memory(self): """从磁盘加载记忆""" json_path = f"{self.storage_path}/{self.user_id}_memory.json" try: with open(json_path, 'r', encoding='utf-8') as f: data = json.load(f) self.memory = data.get("memory", {}) print(f"已加载用户 {self.user_id} 的记忆") except FileNotFoundError: print(f"未找到用户 {self.user_id} 的记忆文件，创建新记忆") self.memory = { "conversation_history": [], "user_profile": {}, "important_facts": {} }

四、实战：构建完整的 LangGraph 对话系统

构建一个完整的 LangGraph 工作流，将检索、生成和更新三个节点串联起来。

from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END import operator from datetime import datetime from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI # 假设已安装 class CompleteConversationState(TypedDict): """完整的对话状态定义""" user_input: str short_term_memory: list # 短期对话历史 long_term_memory: dict # 提取的关键事实 conversation_summary: str # 摘要 retrieved_context: list # 检索到的记忆 ai_response: str should_store: bool def create_conversation_workflow(): """创建完整的对话工作流""" def retrieve_memories(state: CompleteConversationState): """节点 1: 记忆检索（结合短期和长期事实）""" relevant_memories = [] # 模拟从短期和长期记忆中检索 recent_memories = state.get("short_term_memory", [])[-3:] relevant_memories.extend(recent_memories) context_text = "\n".join([f"- {mem}" for mem in relevant_memories]) return { "retrieved_context": relevant_memories, "enriched_input": f"{state['user_input']} [上下文：{len(relevant_memories)}条相关记忆]" } def generate_response(state: CompleteConversationState): """节点 2: 响应生成（基于检索到的上下文）""" llm = ChatOpenAI(model="gpt-4") context_parts = [] if state["conversation_summary"]: context_parts.append(f"对话摘要：{state['conversation_summary']}") if state["retrieved_context"]: context_parts.append("相关历史已检索。") context_text = "\n".join(context_parts) prompt = f"""{context_text} 当前用户输入：{state['user_input']} 请基于以上对话历史和上下文，给出恰当回复：""" response = llm.invoke([HumanMessage(content=prompt)]) # 假设通过某种机制判断是否需要长期存储 should_store = len(state['user_input']) > 50 return { "ai_response": response.content, "should_store": should_store } def update_memories(state: CompleteConversationState): """节点 3: 记忆更新（写入短期和长期存储）""" # 1. 更新短期记忆 new_short_term = state.get("short_term_memory", []) new_short_term.append({ "user": state["user_input"], "assistant": state["ai_response"], "timestamp": datetime.now().isoformat() }) new_short_term = new_short_term[-10:] # 2. 更新长期记忆/摘要 updates = {} if state.get("should_store", False): # 模拟摘要更新和关键信息提取 updates["conversation_summary"] = "（新摘要）" updates["long_term_memory"] = {"last_key_fact": "..."} return { "short_term_memory": new_short_term, "long_term_memory": {**state.get("long_term_memory", {}), **updates.get("long_term_memory", {})}, "conversation_summary": updates.get("conversation_summary", state.get("conversation_summary", "")) } # 构建工作流图 workflow = StateGraph(CompleteConversationState) workflow.add_node("retrieve", retrieve_memories) workflow.add_node("generate", generate_response) workflow.add_node("update_memory", update_memories) workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "generate") workflow.add_edge("generate", "update_memory") workflow.add_edge("update_memory", END) return workflow.compile()

五、最佳实践和性能优化

5.1 记忆管理的最佳实践

分层记忆策略：采用类似人脑的记忆架构是最高效的。

记忆层	目的	存储形式	适用工具
工作记忆 (Working)	当前对话的连贯性	滑动窗口/列表	LangGraph State
情景记忆 (Episodic)	重要的对话事件/经历	向量数据库 (RAG)	FAISS, Chroma
语义记忆 (Semantic)	提取的事实和知识	Key-Value Store/知识图谱	Redis, Neo4j, LangGraph State (Facts)