1. 从“无限对话”到“有效对话”为什么我们需要对话压缩最近在折腾一个基于大语言模型的对话Agent项目遇到了一个典型问题随着对话轮次增加上下文窗口很快就被塞满了。这不仅仅是“窗口满了”那么简单它直接导致了几个连锁反应模型响应速度变慢、API调用成本飙升更关键的是模型开始“失忆”——它无法从冗长的历史中找到真正相关的信息回答质量断崖式下跌。这让我意识到构建一个能长期运行的对话Agent核心挑战不在于如何发起对话而在于如何管理对话。我们需要的不是一个只会“听”和“说”的机器而是一个具备“记忆管理”能力的智能体。它必须能像人类一样在漫长的交流中主动提炼重点、遗忘无关细节、并保留对当前任务至关重要的核心信息。这就是“对话压缩”要解决的根本问题。LangGraph作为LangChain生态中用于构建有状态、多环节应用的工作流编排框架为我们实现这种具备“记忆管理”能力的Agent提供了绝佳的工具箱。它不再将对话视为一个线性的、不断增长的列表而是将其建模为一个可以动态更新和修剪的“图”。在这个图里节点是处理步骤边是状态流转而“对话历史”本身就是一个可以被节点操作的核心状态。所以当我们谈论“带对话压缩的对话Agent”时我们本质上是在构建一个具备长期记忆与短期工作记忆协同的系统。短期工作记忆即压缩后的、高度相关的上下文用于指导模型生成下一轮响应长期记忆或完整的原始历史则作为知识库在需要深度回溯时被检索和提取。这种架构才是让Agent真正“可持续”对话的关键。2. LangGraph核心概念与对话状态建模在动手写代码之前我们必须先理解LangGraph是如何看待一个对话系统的。如果你用过LangChain的ConversationChain你可能习惯把历史记录当作一个不断追加的字符串或消息列表。但在LangGraph中一切围绕“状态”展开。2.1 理解StateGraph与状态流转LangGraph的核心是StateGraph。你需要定义一个State类它本质上是一个字典TypedDict规定了你的工作流中流动的数据结构。对于对话Agent这个State至少需要包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 完整的对话历史用于长期存储和可能的检索 full_history: Annotated[List, add_messages] # 经过压缩后准备送入LLM的当前上下文 compressed_context: str # 用户的最新输入 human_input: str # Agent思考后的输出 agent_output: str这里的关键是Annotated[List, add_messages]。add_messages是一个归约器它定义了当多个节点都想修改full_history时如何合并这些修改。默认的add_messages会将新的消息列表与旧列表简单相加这符合对话历史追加的直觉。工作流中的每个节点一个函数都接收这个完整的State并返回一个字典其中包含它想要更新的State字段。LangGraph的运行时负责将这些更新应用到全局状态上。这种“全局状态局部更新”的模式使得我们可以清晰地分离关注点一个节点负责理解用户意图一个节点负责压缩历史另一个节点负责调用LLM生成回复。2.2 设计对话压缩的工作流节点一个基础的、带压缩的对话Agent工作流通常包含以下几个关键节点接收输入节点将用户的最新输入存入state[‘human_input’]。对话压缩节点这是核心。它读取state[‘full_history’]和state[‘human_input’]判断是否需要压缩以及如何压缩最终输出更新后的state[‘compressed_context’]。调用LLM节点将state[‘compressed_context’]和state[‘human_input’]组合成提示词调用大语言模型得到state[‘agent_output’]。更新历史节点将本轮完整的交互用户输入和Agent输出作为消息追加到state[‘full_history’]中。这些节点通过边连接起来。一个最简单的线性流程是接收输入 - 压缩对话 - 调用LLM - 更新历史。但LangGraph的强大之处在于你可以基于State的内容决定下一步走向哪个节点实现条件逻辑。例如在压缩节点中如果判断历史很短无需压缩可以直接跳转到调用LLM节点节省一次LLM调用用于压缩本身。3. 对话压缩策略的深度剖析与实现对话压缩不是简单地把长文本变短而是有策略地提炼信息。不同的策略适用于不同的场景也带来了不同的复杂度和效果。3.1 策略一增量式摘要Incremental Summarization这是最直观的方法。每次对话轮次后或者历史达到一定长度时触发一个压缩节点。该节点调用LLM对full_history生成一个简洁的摘要并用这个摘要替换或代表之前的所有历史。实现要点from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI def summarize_conversation(state: AgentState): full_messages state[‘full_history’] if len(full_messages) 5: # 设置触发压缩的阈值 # 历史太短无需压缩返回原始历史作为上下文 return {“compressed_context”: format_messages(full_messages)} # 构建摘要提示词 summary_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个高效的对话摘要助手。请将以下的对话历史浓缩成一个简洁的段落摘要保留所有关于事实、决定和待办事项的关键信息。摘要语言应与原始对话语言一致。”), (“user”, “{conversation}”) ]) llm ChatOpenAI(model“gpt-4o-mini”, temperature0) chain summary_prompt | llm # 将消息列表格式化成字符串 conversation_text format_messages(full_messages) summary chain.invoke({“conversation”: conversation_text}) # 关键决策压缩后的上下文是什么 # 方案A只用摘要。优点是极简但可能丢失细节。 # compressed_context summary.content # 方案B摘要 最近几轮原始对话。在简洁和细节间折衷更推荐。 last_k_turns format_messages(full_messages[-4:]) # 保留最近2轮对话 compressed_context f“””对话摘要 {summary.content} 最近对话 {last_k_turns} “”” return {“compressed_context”: compressed_context}注意摘要的“提示词工程”至关重要。你需要明确告诉LLM摘要的用途例如“用于让AI助手继续对话”并指定需要保留的信息类型事实、用户偏好、任务状态、待办事项等。一个模糊的“请总结一下”指令会导致摘要无法用于后续对话。优缺点分析优点概念简单能显著减少令牌数对于话题集中的长对话效果不错。缺点信息损耗摘要必然丢失细节可能丢失后续对话需要的关键信息。累积误差多次摘要叠加可能导致信息扭曲或偏离原意。成本每次压缩都需要调用一次LLM增加了成本和延迟。3.2 策略二基于向量检索的上下文提取Retrieval-Based Compression这种方法将完整的对话历史视为一个“知识库”。当新用户输入到来时不是压缩全部历史而是从这个历史库中检索与当前输入最相关的片段将其作为上下文。实现要点from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter # 初始化向量存储在工作流外持久化 embeddings OpenAIEmbeddings(model“text-embedding-3-small”) vectorstore Chroma(embedding_functionembeddings, persist_directory“./chat_history_db”) text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) def retrieve_relevant_context(state: AgentState): human_input state[‘human_input’] full_history_text format_messages(state[‘full_history’]) # 1. 将完整历史分割并存入向量库如果是第一次或历史有更新 # 注意实际项目中需要维护一个标志位避免每次对话都重复存储相同历史。 # 这里为简化假设每次都会更新。 if state[‘full_history’]: chunks text_splitter.split_text(full_history_text) # 为每个块添加元数据如时间戳或轮次便于追踪 vectorstore.add_texts(chunks) # 2. 基于用户当前输入进行检索 docs vectorstore.similarity_search(human_input, k3) # 检索最相关的3个片段 retrieved_context “\n\n”.join([doc.page_content for doc in docs]) # 3. 组合上下文检索到的相关内容 最近的1-2轮对话保证连贯性 last_turn format_messages(state[‘full_history’][-2:]) if len(state[‘full_history’]) 2 else “” compressed_context f“””根据你的问题从历史对话中找到的相关信息 {retrieved_context} 最近对话 {last_turn} “”” return {“compressed_context”: compressed_context}优缺点分析优点精准提供的上下文与当前问题高度相关极大提升了回复的准确性和相关性。无损原始历史被完整保存没有信息损耗。可扩展历史库可以非常大突破了LLM上下文窗口的限制。缺点架构复杂需要引入向量数据库和嵌入模型。可能遗漏全局信息如果检索失败或用户问题需要联系多个分散的历史点可能会提供不完整的上下文。无法理解对话流纯粹的语义检索可能忽略对话的时间顺序和逻辑演进。3.3 策略三混合策略与智能触发机制在实际项目中单一策略往往不够。一个健壮的方案需要结合多种策略并智能决定何时压缩、如何压缩。智能触发机制示例def should_compress(state: AgentState) - dict: 判断节点决定是否需要进行压缩以及采用何种策略。 full_history state[‘full_history’] human_input state[‘human_input’] # 规则1基于令牌数触发 estimated_tokens estimate_token_count(format_messages(full_history)) if estimated_tokens 3000: # 假设模型窗口为4k留出1k空间 return {“next”: “summarize”} # 跳转到摘要节点 # 规则2基于话题漂移触发简易版 # 计算当前输入与历史最后几句的相似度可通过嵌入模型计算余弦相似度 # 如果相似度低于阈值说明话题可能已切换触发检索式压缩 if topic_shift_detected(human_input, full_history[-3:]): return {“next”: “retrieve”} # 跳转到检索节点 # 规则3无需压缩直接进入生成环节 return {“next”: “generate”}在LangGraph中你可以将这个判断节点的输出{“next”: “node_name”}用于配置条件边从而动态路由工作流。混合上下文组装最终的compressed_context可以是一个精心设计的模板compressed_context_template “”” # 系统角色设定 你是一个有帮助的助手。 # 对话背景摘要来自增量摘要策略 {conversation_summary} # 与当前问题相关的历史片段来自检索策略 {relevant_history_snippets} # 确保对话连贯性的最近交流原始消息 {last_few_messages} # 当前用户问题 用户{human_input} 请回答 “””这种混合方式结合了摘要的全局观、检索的精准性和原始对话的连贯性是目前构建高质量长对话Agent的较佳实践。4. 在LangGraph中构建完整Agent代码实战与避坑指南理论说完了我们动手把一个完整的Agent搭起来。这里我们实现一个混合策略的Agent默认使用检索式压缩当历史过长时自动触发一次摘要来重置检索库。4.1 定义状态与工作流图import operator from typing import TypedDict, Annotated, List, Literal from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter # 1. 定义状态 class State(TypedDict): messages: Annotated[List, add_messages] # 完整历史 human_input: str compressed_context: str next_step: Literal[“retrieve”, “summarize”, “generate”] # 用于控制流程 # 2. 初始化核心组件在实际应用中这些应被注入或全局管理 llm ChatOpenAI(model“gpt-4o-mini”, temperature0.7) embeddings OpenAIEmbeddings() vectorstore Chroma(embedding_functionembeddings, persist_directory“./hist_vec_db”, collection_name“conversation”) text_splitter RecursiveCharacterTextSplitter(chunk_size1000, chunk_overlap100) summary_llm ChatOpenAI(model“gpt-4o-mini”, temperature0) # 用于摘要的LLM可配置更低temperature # 3. 定义节点函数 def receive_input(state: State): 节点接收用户输入 # 在实际场景中输入可能来自API请求 return {“human_input”: state[“human_input”], “next_step”: “retrieve”} def route_by_state(state: State): 路由节点根据状态决定下一步 # 简易逻辑如果消息太多则总结否则检索 if len(state[“messages”]) 10: # 假设10轮后触发摘要 return {“next_step”: “summarize”} return {“next_step”: “retrieve”} def summarize_history(state: State): 节点执行摘要压缩 if not state[“messages”]: return {“compressed_context”: “”, “next_step”: “generate”} # 生成摘要 summary_prompt ChatPromptTemplate.from_messages([ (“system”, “生成一个非常简洁的段落总结对话中的关键事实、用户目标和已做出的决定。”), (“user”, “{history}”) ]) history_text “\n”.join([f“{m.type}: {m.content}” for m in state[“messages”]]) chain summary_prompt | summary_llm summary chain.invoke({“history”: history_text}) # 关键步骤摘要后清空或更新向量库并用摘要作为新的“基础记忆” # 这里我们选择1. 清空旧向量库。2. 将摘要作为新文档存入。 global vectorstore # 注意实际生产环境需要更精细的向量库管理这里仅为演示。 vectorstore.delete_collection() # 删除旧集合 vectorstore Chroma.from_texts( [summary.content], embeddingembeddings, persist_directory“./hist_vec_db”, collection_name“conversation” ) # 压缩上下文 摘要 最近一轮对话保证连贯 last_msg state[“messages”][-1] if state[“messages”] else “” compressed_ctx f“对话背景摘要{summary.content}\n\n最近交流{last_msg}” return {“compressed_context”: compressed_ctx, “next_step”: “generate”} def retrieve_context(state: State): 节点执行检索压缩 query state[“human_input”] # 从向量库检索 docs vectorstore.similarity_search(query, k2) retrieved “\n”.join([d.page_content for d in docs]) # 同时将最近的对话也作为上下文的一部分 recent_msgs state[“messages”][-4:] if len(state[“messages”]) 4 else state[“messages”] recent_text “\n”.join([f“{m.type}: {m.content}” for m in recent_msgs]) compressed_ctx f“””相关历史信息 {retrieved if retrieved else ‘无相关历史’} 最近对话 {recent_text if recent_text else ‘无’} “”” return {“compressed_context”: compressed_ctx, “next_step”: “generate”} def generate_response(state: State): 节点生成回复 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个基于以下对话历史进行回复的助手。请保持回答友好、专业且相关。\n\n{context}”), (“user”, “{input}”) ]) chain prompt | llm response chain.invoke({ “context”: state[“compressed_context”], “input”: state[“human_input”] }) # 准备更新完整历史的消息 new_messages [ HumanMessage(contentstate[“human_input”]), AIMessage(contentresponse.content) ] return {“messages”: new_messages} # add_messages归约器会自动追加 # 4. 构建图 workflow StateGraph(State) # 添加节点 workflow.add_node(“receive_input”, receive_input) workflow.add_node(“router”, route_by_state) workflow.add_node(“summarize”, summarize_history) workflow.add_node(“retrieve”, retrieve_context) workflow.add_node(“generate”, generate_response) # 设置边 workflow.set_entry_point(“receive_input”) workflow.add_edge(“receive_input”, “router”) # 条件边根据router节点的结果决定下一步 workflow.add_conditional_edges( “router”, lambda state: state[“next_step”], { “summarize”: “summarize”, “retrieve”: “retrieve”, } ) workflow.add_edge(“summarize”, “generate”) workflow.add_edge(“retrieve”, “generate”) workflow.add_edge(“generate”, END) # 编译图 app workflow.compile()4.2 运行Agent与结果分析现在我们可以运行这个Agent进行多轮对话。# 初始化状态 initial_state {“messages”: [], “human_input”: “”, “compressed_context”: “”, “next_step”: “retrieve”} # 第一轮对话 config {“configurable”: {“thread_id”: “test_thread_1”}} result1 app.invoke({**initial_state, “human_input”: “你好我想计划一次去日本的旅行。”}, config) print(“AI:”, result1[“messages”][-1].content) # 打印AI的回复 # 此时vectorstore中会存入关于日本旅行的对话片段。 # 第二轮对话 result2 app.invoke({“human_input”: “我对京都的文化古迹特别感兴趣。”}, config) print(“AI:”, result2[“messages”][-1].content) # retrieve节点会搜索vectorstore中与“京都文化古迹”相关的历史片段并将其作为上下文。 # 模拟多轮对话后... for i in range(10): app.invoke({“human_input”: f“测试消息{i}”}, config) # 当消息轮次超过10轮我们在router节点中设定的阈值下一次调用会触发summarize节点。 # summarize节点会生成整个对话的摘要清空旧的vectorstore并将摘要作为新文档存入。 # 之后的对话检索将基于这个摘要进行实现了“记忆重置”。4.3 实战避坑指南与经验心得在实现过程中我踩过不少坑这里分享几个关键点向量库的管理是最大挑战上面的示例为了简洁在summarize节点中直接delete_collection()是危险操作。在生产环境中你需要更精细的策略为每个对话线程thread_id创建独立的向量集合避免不同用户对话相互干扰。增量更新而非全量重建每次新对话产生后只将新的消息块添加到向量库而不是每次都从头构建。Chroma的add_texts方法支持增量添加。摘要后如何更新向量库一个稳妥的做法是保留两个向量库一个存原始历史片段只增不删一个存摘要文档。检索时可以同时从两者中查询并合并结果。压缩触发条件的设计不要只依赖简单的轮次计数。结合令牌数估算和语义变化检测会更可靠。例如使用tiktoken库精确计算上下文令牌数或计算当前用户输入与历史平均嵌入向量的余弦距离来判断话题是否已切换。压缩带来的信息丢失与幻觉风险LLM生成的摘要可能引入错误幻觉。在关键任务场景如医疗、法律咨询压缩策略需要更保守。可以考虑保留原始事实的“锚点”在摘要中强制要求保留日期、数字、具体名称等关键实体。提供“查看完整历史”的备用路径当Agent对某个历史细节不确定时可以设计一个“回溯”节点从持久化存储中调取原始对话记录进行核实。性能与成本的权衡检索式压缩需要计算嵌入向量和相似度搜索虽然比调用LLM摘要便宜但仍有开销。对于实时性要求极高的场景可以设置一个缓存层将常见问题及其相关上下文缓存起来。测试测试再测试对话压缩系统的行为比普通聊天机器人更复杂。必须设计全面的测试用例长上下文依赖测试在对话第50轮询问第3轮提到的细节看Agent能否正确回答。话题切换测试 abruptly改变话题观察压缩策略是否能快速适应不再提供无关的历史信息。压力测试模拟极长的对话观察内存、响应时间的变化。5. 进阶思考超越压缩的对话记忆架构对话压缩是解决上下文长度限制的利器但它本质上是一种“损失性”的优化。对于需要完美记忆的复杂任务我们可以探索更高级的架构。分层记忆系统这是更接近人类记忆的模型。感官记忆/工作记忆对应compressed_context容量小、速度快存放当前任务直接相关的信息。短期记忆对应完整的、未压缩的近期对话如最近20轮存储在快速访问的数据库如Redis中。长期记忆对应所有对话的原始记录、提取出的关键事实结构化数据、以及通过摘要形成的“高维概念”存储在向量数据库和关系型数据库中。 Agent可以根据需要通过检索、推理等方式从长期记忆中提取信息加载到工作记忆中。将对话历史视为知识图谱另一种思路是不压缩文本而是将对话中提到的实体、关系、事件提取出来构建成一个知识图谱。当用户提问时在图谱上进行查询和推理。这种方式信息结构化程度高查询精准但构建图谱的复杂度和成本也更高。反思与递归修正让Agent具备“反思”能力。在生成回复后增加一个节点让另一个LLM或同一LLM的不同提示检查回复是否与完整的对话历史一致。如果发现矛盾或信息缺失可以触发一个修正流程重新检索更完整的历史并生成新的回复。这为压缩系统增加了一层质量保证。实现这些进阶架构LangGraph依然是优秀的编排框架。你可以定义更复杂的State包含working_memory、long_term_memory_facts、knowledge_graph等字段并设计专门的节点来维护和查询这些不同的记忆模块。图的边也可以变得更加复杂支持基于记忆查询结果的循环和递归。最终选择哪种策略取决于你的具体应用场景、对信息完整性的要求以及可接受的成本。对于大多数应用从“检索摘要”的混合策略开始是一个务实且有效的起点。它平衡了能力、复杂度和成本能让你的对话Agent真正具备进行深度、持久且有价值对话的能力。