1. 项目概述为什么我们需要带记忆压缩的对话Agent最近在折腾一些需要长期记忆和复杂推理的AI应用比如智能客服、游戏NPC或者个人学习助手我发现一个普遍存在的痛点对话轮次一多模型就容易“失忆”或者“算力爆炸”。直接把所有历史对话都塞进上下文不仅会快速耗尽宝贵的Token限额导致成本飙升更关键的是无关的闲聊和冗余信息会严重干扰模型对当前任务核心的专注力让它抓不住重点。这就像让你在堆满杂物的房间里找一把钥匙东西越多效率越低。于是“对话压缩”这个技术点就变得至关重要。它不是简单地删除历史而是智能地提炼、总结和重构对话的核心状态只把对下一步决策最关键的信息保留下来。而LangGraph作为LangChain生态中用于构建有状态、多环节应用的新锐框架为实现这种带压缩能力的智能体Agent提供了极其优雅的解决方案。今天要聊的就是如何利用LangGraph快速搭建一个具备“对话记忆压缩”功能的简易对话Agent。这个Agent能自动判断何时该压缩记忆、如何压缩并在压缩后依然保持对话的连贯性与目标指向性。无论你是想做一个能深入聊天的机器人还是构建一个处理复杂多轮工作流的自动化工具这套思路都能直接拿来用。2. 核心设计思路LangGraph的状态管理与压缩触发器在动手写代码之前得先想明白两个核心问题状态记忆在LangGraph中如何流转以及何时、如何触发压缩这是整个设计的骨架。2.1 理解LangGraph的“图”与“状态”LangGraph的核心抽象是“图”Graph。图中的节点Node代表一个执行单元比如调用一次LLM或执行一个工具函数边Edge定义了节点之间的流转条件。而贯穿整个图执行过程的是一个共享的**状态State**对象。这个状态是一个字典包含了当前对话的所有信息。对于我们的对话Agent这个状态至少需要包含messages: 一个列表存放所有的对话消息用户输入、AI回复。compressed_history: 一个字符串存放压缩后的对话摘要。num_turns: 一个整数记录自上次压缩以来的对话轮次用于判断触发时机。LangGraph的执行过程就是带着这个状态字典按照图定义的路径在各个节点间“跑”一圈。每个节点都可以读取和修改这个状态。2.2 压缩策略的设计条件与时机压缩不能每轮都做那样成本太高也失去了意义。我们需要一个触发策略。常见的策略有轮次触发最简单直接。设定一个阈值比如5轮当未压缩的对话轮次达到这个数时自动触发压缩。Token长度触发更精细的方式。计算当前messages列表的总Token数当接近模型上下文窗口上限如80%时触发压缩。这需要接入模型的Tokenizer进行计算。关键点触发在检测到对话主题切换、任务阶段完成例如从“询问需求”切换到“提供方案”时触发。这需要更复杂的意图识别逻辑。在我们的简易实现中我们将采用“轮次触发”为主“Token长度检查”为辅的混合策略兼顾简单性与实用性。我们会设定一个轮次阈值同时也会检查历史消息的预估长度防止在轮次未到但单轮信息量巨大时导致上下文溢出。2.3 图的流程设计我们的Agent工作流将设计成以下几个核心节点它们会按条件流转路由节点Router判断用户输入是开启新话题还是延续当前对话。这决定了是使用压缩后的历史还是完整历史。压缩判断节点Should_Compress根据当前状态num_turns,messages长度判断是否需要执行压缩。压缩执行节点Compress_History如果需要压缩则调用LLM将冗长的messages提炼成一个简短的compressed_history摘要并清空或截断旧的messages。核心响应节点Call_LLM最终将处理后的历史可能是原始消息也可能是压缩摘要最新消息与用户当前问题一起发送给LLM生成回复。这些节点通过边连接起来形成一个闭环。每次用户提问Agent就带着状态在这个环里跑一遍输出回答并更新状态。3. 环境准备与核心工具选型工欲善其事必先利其器。实现这个Agent我们主要依赖两个核心库langgraph和langchain。当然还需要一个LLM的API接口。3.1 安装依赖首先创建一个新的Python环境然后安装必要的包。建议使用uv或poetry管理这里用pip示例pip install langgraph langchain-openai tiktokenlanggraph: 本篇的主角用于构建有状态的工作流图。langchain-openai: LangChain提供的OpenAI官方集成包方便我们调用GPT系列模型。tiktoken: OpenAI开源的Token计数库用于精确计算消息长度实现Token触发策略。如果你打算使用其他模型如Anthropic的Claude或开源的Ollama则需安装对应的langchain-anthropic或langchain-community包。3.2 初始化LLM与Token计数器接下来初始化我们的核心工具LLM客户端和Token计数器。import os from langchain_openai import ChatOpenAI import tiktoken # 1. 初始化LLM。我们使用gpt-3.5-turbo性价比高能力足够。 # 请将您的API Key设置在环境变量OPENAI_API_KEY中 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.2) # temperature调低使压缩和生成行为更稳定、更可预测。 # 2. 初始化Token计数器用于计算消息列表的Token长度。 # 注意编码需要与使用的模型匹配gpt-3.5-turbo使用cl100k_base encoder tiktoken.get_encoding(cl100k_base) def count_tokens_in_messages(messages): 计算一个LangChain消息列表的近似Token数。 total_tokens 0 for message in messages: # 消息内容 total_tokens len(encoder.encode(message.content)) # 粗略估计角色名等系统开销每个消息加5个Token total_tokens 5 return total_tokens关键选择解析为什么选gpt-3.5-turbo对于压缩和对话任务它已经足够出色且成本低廉。压缩任务不需要极强的创造力而需要可靠性和一致性。为什么单独用tiktoken虽然langchain也有get_num_tokens方法但tiktoken是官方推荐计数最准且不依赖特定模型类更灵活。temperature0.2压缩和任务型对话需要确定性更高的输出降低温度可以减少LLM的随机性让压缩摘要更聚焦对话回复更稳定。4. 构建状态模型与图节点有了工具我们开始构建LangGraph的核心组件。首先定义状态然后实现各个节点函数。4.1 定义状态结构我们使用TypedDict来明确状态的结构这有利于类型提示和代码可读性。from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage import operator class AgentState(TypedDict): Agent的完整状态定义。 # 完整的对话消息列表 messages: Annotated[Sequence[BaseMessage], operator.add] # 压缩后的历史摘要 compressed_history: str # 自上次压缩以来的对话轮次一问一答算一轮 turns_since_compression: int # 用户当前输入的问题 current_input: str注解说明Annotated[Sequence[BaseMessage], operator.add]这是LangGraph的一个魔法。它告诉框架messages这个字段在节点间传递时应该用操作符即operator.add来合并。这意味着每个节点都可以往messages列表里追加append新消息而框架会自动帮我们把所有节点追加的消息合并起来。这是实现对话记忆累积的关键。compressed_history是一个普通的字符串。turns_since_compression是一个计数器。current_input临时存放用户本轮输入用于在节点间传递。4.2 实现节点函数现在我们实现四个核心节点函数。4.2.1 节点1路由与输入处理这个节点负责接收用户原始输入并初始化状态。from langchain_core.messages import HumanMessage def process_input(state: AgentState): 接收用户输入并初始化状态中的消息列表。 user_input state[current_input] # 判断是否是全新对话如果压缩历史为空且消息列表为空则是新对话 is_new_chat not state[compressed_history] and len(state[messages]) 0 new_messages [] if is_new_chat: # 全新对话直接使用用户输入 new_messages [HumanMessage(contentuser_input)] else: # 延续对话将压缩历史作为系统提示的一部分与最新用户输入结合 # 这里我们选择将压缩历史放在系统消息中这是一种常见做法 # 实际上更优的做法是创建一个专用的“上下文”字段这里为简化我们将其预置到用户消息前。 # 注意我们并不直接修改messages而是将处理逻辑放在Call_LLM节点。 # 此节点主要做判断并可以设置一个标志位。但为了流程清晰我们简化设计 # 本节点只负责将用户输入转为HumanMessage并存入状态。 # 压缩历史的拼接将在Call_LLM节点进行。 new_messages [HumanMessage(contentuser_input)] # 更新状态追加消息轮次计数器1 return { messages: new_messages, turns_since_compression: state[turns_since_compression] 1, current_input: user_input # 保留一份后续节点可能用到 }实操心得在真实项目中is_new_chat的判断逻辑可能更复杂比如检查是否有session_id或对话超时。这里做了最大简化。我们没有在本节点进行历史拼接因为拼接方式如何将compressed_history和messages组合成LLM的prompt是LLM调用节点的职责。这样设计符合“单一职责原则”。4.2.2 节点2压缩判断节点这个节点根据策略决定是否要走压缩分支。def should_compress(state: AgentState) - str: 判断是否需要压缩历史。返回下一个节点的名称。 MAX_TURNS 3 # 演示用轮次阈值设小一点 MAX_TOKENS 2000 # 演示用Token阈值 current_turns state[turns_since_compression] current_messages state[messages] # 策略1轮次触发 if current_turns MAX_TURNS: print(f[Trigger] Turn count ({current_turns}) {MAX_TURNS}. Compressing.) return compress_history # 策略2Token长度触发估算 estimated_tokens count_tokens_in_messages(current_messages) if estimated_tokens MAX_TOKENS: print(f[Trigger] Token estimate ({estimated_tokens}) {MAX_TOKENS}. Compressing.) return compress_history # 不需要压缩继续走向LLM调用节点 print(f[Trigger] No compression needed. Turns: {current_turns}, Est.Tokens: {estimated_tokens}) return call_llm注意事项MAX_TURNS和MAX_TOKENS是超参数需要根据实际应用调整。对于客服场景MAX_TURNS可以设大些如10对于高密度信息交换场景则设小些如3-5。count_tokens_in_messages是我们之前定义的函数。这里只是估算因为严格计算需要包含可能的系统提示和格式Token。但对于触发判断而言这个估算值已经足够可靠。打印日志print在实际部署时应改为更规范的日志记录但对于调试和理解流程非常有用。4.2.3 节点3压缩执行节点这是压缩功能的核心它调用LLM来生成摘要。from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import SystemMessage def compress_history(state: AgentState): 压缩对话历史生成摘要并重置相关状态。 # 1. 准备压缩提示词 compression_prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个高效的对话摘要助手。你的任务是将一段对话历史压缩成一个简洁、连贯的段落摘要。摘要需要保留所有关键事实、用户意图和决策点忽略无关的寒暄和重复信息。摘要语言需为中文。), HumanMessage(contentf请压缩以下对话历史\n\n{state[messages]}\n\n生成一个简洁的摘要) ]) # 2. 调用LLM进行压缩 compression_chain compression_prompt | llm response compression_chain.invoke({}) # 提示词已包含所有内容输入为空dict compressed_summary response.content # 3. 更新状态 # - 将新生成的摘要存入compressed_history。注意是覆盖不是追加。 # - 清空messages列表因为其信息已被摘要捕获。 # - 重置turns_since_compression计数器。 # - 注意我们不清除compressed_history而是让它累积。也可以选择只保留最新一份取决于业务逻辑。 # 这里采用累积方式将旧摘要和新摘要合并形成更长的历史脉络。 old_summary state.get(compressed_history, ) if old_summary: new_compressed_history old_summary \n---\n compressed_summary else: new_compressed_history compressed_summary print(f[Compression] Generated summary: {compressed_summary[:100]}...) # 打印前100字符 return { compressed_history: new_compressed_history, messages: [], # 清空原始消息列表 turns_since_compression: 0 # 重置计数器 }关键细节与避坑指南提示词工程压缩提示词的质量直接决定摘要效果。务必强调“保留关键事实、意图和决策”并指定输出语言。你可以根据你的对话领域微调这个提示词例如对于技术支持对话可以要求“保留报错信息、操作步骤和解决方案”。摘要合并策略上述代码采用了“累积摘要”策略。优点是能保留很长的历史脉络缺点是摘要会越来越长最终可能又需要被压缩。另一种策略是“替换策略”即new_compressed_history compressed_summary只保留最新一份摘要。后者更节省空间但可能丢失远期上下文。你需要根据对话的长时期依赖性做选择。清空消息压缩后messages被清空。这意味着后续的call_llm节点将看不到原始的对话消息只能看到compressed_history和最新的用户输入。这是压缩的核心目的——用简短的摘要替代冗长的原文。成本考虑压缩本身需要调用一次LLM产生额外成本。因此触发策略的阈值设置需要在“上下文节省的收益”和“压缩操作的成本”之间取得平衡。4.2.4 节点4调用LLM生成回复这是最终生成答案的节点。def call_llm(state: AgentState): 根据当前状态压缩历史最新消息调用LLM生成回复。 # 1. 构建最终的对话上下文 all_context_parts [] # 如果有压缩历史将其作为系统消息或首个用户消息的一部分加入 if state[compressed_history]: # 方案A作为独立的系统消息更清晰LLM更容易区分 # all_context_parts.append(SystemMessage(contentf以下是此前的对话摘要\n{state[compressed_history]})) # 方案B作为第一个HumanMessage的上下文更节省消息数本示例采用 # 我们将摘要和最新用户问题合并到一个HumanMessage中 enhanced_input f【历史对话摘要】\n{state[compressed_history]}\n\n【当前问题】\n{state[current_input]} current_message HumanMessage(contentenhanced_input) else: # 没有压缩历史直接使用最新的用户消息通常只在对话开始时发生 current_message state[messages][-1] if state[messages] else HumanMessage(contentstate[current_input]) # 2. 构建消息列表。如果压缩后messages被清空则列表里只有当前处理过的消息。 # 如果未触发压缩messages里包含多轮历史我们直接使用它。 messages_for_llm [] if state[messages]: # 存在未压缩的原始消息 messages_for_llm list(state[messages]) # 使用完整的原始消息历史 else: # 消息被清空了使用我们刚构建的包含摘要的当前消息 messages_for_llm [current_message] # 3. 可以添加一个系统提示定义Agent的角色 system_msg SystemMessage(content你是一个有帮助的AI助手。请根据对话历史专业、准确地回应用户的问题。) final_messages [system_msg] messages_for_llm # 4. 调用LLM response llm.invoke(final_messages) # 5. 将AI的回复也追加到状态中的messages里以便在未压缩时保留完整对话。 # 注意如果messages被清空过这里追加的就是[HumanMessage(enhanced_input), AIMessage(response)]。 # 这保证了下一轮开始时state[“messages”]里至少包含最近的一轮完整对话。 ai_message response return { messages: [ai_message] # 使用operator.add所以这是追加 }逻辑梳理与技巧上下文组装逻辑这是最易出错的部分。代码清晰地处理了两种路径路径A刚压缩过state[‘messages’]为空state[‘compressed_history’]有值。我们构建一个包含摘要和当前问题的enhanced_input作为HumanMessage。路径B未压缩state[‘messages’]包含完整的原始对话历史。我们直接使用这个历史忽略compressed_history因为它可能包含更早的、已被当前原始历史覆盖的摘要。系统消息始终在消息列表开头添加一个系统消息用来稳定AI的行为角色。这是一个好习惯。状态更新只将AI的回复追加到messages。用户的输入已经在process_input节点被追加过了。通过operator.add的注解这些追加操作会被LangGraph自动合并。5. 组装工作流图并运行测试节点都准备好了现在用LangGraph把它们组装起来。5.1 构建图from langgraph.graph import StateGraph, END # 1. 创建图构建器 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(process_input, process_input) workflow.add_node(should_compress, should_compress) # 这是一个判断节点返回下一个节点名 workflow.add_node(compress_history, compress_history) workflow.add_node(call_llm, call_llm) # 3. 设置入口点 workflow.set_entry_point(process_input) # 4. 添加边连接节点 workflow.add_edge(process_input, should_compress) # 根据should_compress节点的返回值动态决定下一步 workflow.add_conditional_edges( should_compress, # 这个函数接收节点的输出即一个字符串并返回下一个节点的名称 lambda x: x, # should_compress直接返回了compress_history或call_llm { compress_history: compress_history, call_llm: call_llm } ) workflow.add_edge(compress_history, call_llm) workflow.add_edge(call_llm, END) # 一轮对话结束 # 5. 编译图 app workflow.compile()图解 这个图形成了一个链式条件分支的结构process_input-should_compress- (条件分支) - 如果是”compress_history”-compress_history-call_llm- END。 - 如果是”call_llm”-call_llm- END。5.2 运行与测试让我们模拟一个多轮对话观察压缩如何被触发。# 初始化状态 initial_state { messages: [], compressed_history: , turns_since_compression: 0, current_input: } print( 第1轮对话 ) # 用户输入 result1 app.invoke({**initial_state, current_input: 你好我想了解一下你们公司的云计算产品。}) print(fAI回复: {result1[messages][-1].content[:50]}...) print(f状态 - 轮次: {result1[turns_since_compression]}, 压缩历史: {result1[compressed_history][:30] if result1[compressed_history] else 空}) print(\n 第2轮对话 ) result2 app.invoke({**result1, current_input: 具体来说存储服务有哪些类型}) print(fAI回复: {result2[messages][-1].content[:50]}...) print(f状态 - 轮次: {result2[turns_since_compression]}, 压缩历史: {result2[compressed_history][:30] if result2[compressed_history] else 空}) print(\n 第3轮对话 ) result3 app.invoke({**result2, current_input: 每种类型适合什么场景能举个例子吗}) print(fAI回复: {result3[messages][-1].content[:50]}...) print(f状态 - 轮次: {result3[turns_since_compression]}, 压缩历史: {result3[compressed_history][:30] if result3[compressed_history] else 空}) # 注意我们设的MAX_TURNS3所以这一轮之后turns_since_compression3下一轮应该触发压缩。 print(\n 第4轮对话 (应触发压缩) ) result4 app.invoke({**result3, current_input: 对象存储的价格怎么计算}) # 观察控制台应该会打印 [Trigger] Turn count (3) 3. Compressing. print(fAI回复: {result4[messages][-1].content[:50]}...) print(f状态 - 轮次: {result4[turns_since_compression]} (应重置为0), 压缩历史: {result4[compressed_history][:100]}...) print(f状态 - 原始消息数: {len(result4[messages])} (压缩后应显著减少))运行这段代码你将在控制台看到触发压缩的日志并且能观察到第4轮对话时turns_since_compression被重置为0compressed_history字段出现了摘要文本而messages列表的长度也变短了可能只包含最近一轮的对话。6. 高级优化与常见问题排查基础版本跑通了但在生产环境中你可能会遇到以下问题。这里分享一些优化思路和排查技巧。6.1 压缩质量不佳导致信息丢失问题LLM生成的摘要丢失了关键细节导致后续回答跑偏。解决方案优化提示词在压缩提示词中更具体地要求。例如compression_prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个对话历史压缩专家。请严格遵循以下规则生成摘要 1. 保留所有涉及的具体实体如产品名、型号、数字、时间、地点。 2. 保留用户明确表达的需求、偏好和痛点。 3. 保留已达成的一致结论或待办事项。 4. 用客观、简洁的第三人称叙述。 5. 输出为中文。), HumanMessage(contentf对话历史\n{history}\n\n请生成摘要) ])分层次压缩不一次性压缩全部历史。可以先让LLM提取关键QA对或事实列表再合成段落。这增加了成本但提升了保真度。保留关键消息不要清空所有messages。可以保留最近1-2轮原始消息与摘要结合使用。这确保了最新上下文绝对准确。6.2 压缩后对话连贯性断裂问题压缩后AI的回复感觉像是开始了新对话对之前提到的细节引用生硬或错误。解决方案在系统消息中明确指示在call_llm节点的系统消息里明确告诉AI如何利用摘要。例如“以下是之前对话的摘要它概括了到目前为止的核心内容。请基于此摘要和当前最新问题来回答注意保持信息的连贯性。”使用更强大的模型进行压缩如果使用gpt-3.5-turbo压缩效果不好可以尝试用gpt-4进行压缩。虽然单次成本高但压缩后的摘要质量更好可能减少后续多轮对话的Token消耗总体成本或许更优。人工校验模板设计一个固定的摘要模板让LLM填充。例如“用户咨询了[主题]。已了解其需求是[需求]。已提供的建议包括[建议]。待解决的问题是[待解决]。” 这能保证摘要结构稳定信息点齐全。6.3 性能与成本优化问题压缩判断和压缩操作本身带来额外延迟和成本。优化方案异步压缩压缩操作不需要阻塞用户得到本轮回复。可以在call_llm节点生成回复后异步触发一个压缩任务为下一轮对话做准备。LangGraph支持异步节点但这会稍微增加状态管理的复杂度。更经济的触发策略结合对话内容语义判断。例如使用一个轻量级文本分类模型或提示LLM判断当前对话是否处于“段落边界”如用户说“好的那我们接下来聊聊...”在边界处触发压缩比固定轮次更智能。分级压缩不是所有历史都需要用LLM压缩。对于非常早期的历史可以用更简单的方法如提取关键词、首句进行高度压缩只对近期历史使用LLM精细压缩。6.4 常见错误排查表问题现象可能原因排查步骤压缩从未触发MAX_TURNS或MAX_TOKENS设置过大should_compress逻辑错误状态turns_since_compression未正确更新。1. 检查should_compress函数中的阈值和打印日志。2. 确认process_input节点中turns_since_compression是否在递增。3. 检查compress_history节点是否将计数器重置为0。压缩后AI回复忽略历史call_llm节点中上下文组装逻辑有误。压缩后messages为空但未将compressed_history放入发送给LLM的提示中。1. 在call_llm节点打印final_messages的内容确认压缩历史是否被正确包含。2. 检查if state[“messages”]:和else:分支的逻辑是否正确。状态更新混乱Annotated注解使用错误节点返回的状态字典键名错误。1. 确认AgentState中messages字段是否使用了operator.add。2. 检查每个节点返回的字典键名是否与AgentState定义完全一致。Token数计算不准导致上下文溢出或压缩过早count_tokens_in_messages函数估算不准未考虑模型特定的格式Token。1. 对于OpenAI模型使用tiktoken的精确编码。2. 可以调用OpenAI的API如ChatOpenAI().get_num_tokens_from_messages(...)获取更精确计数但会增加延迟。3. 设置一个安全边际如预留300个Token。6.5 将Agent扩展为支持工具的智能体目前的Agent只能对话。LangGraph真正的威力在于轻松集成工具函数调用。只需稍作修改在状态中增加tool_calls和tool_results字段。在call_llm节点使用llm.bind_tools([...])来让LLM具备调用工具的能力。新增一个execute_tools节点用于解析LLM的响应执行工具并将结果返回状态。修改图流程在call_llm后根据响应是否包含工具调用来决定是走向execute_tools节点执行工具还是直接结束。带工具调用的图通常是循环的LLM - (可能调用工具) - 执行工具 - 将结果返回状态 - 再次流向LLM直到LLM给出最终答案。LangGraph处理这种循环流程非常自然。