1. 项目概述为什么我们需要带记忆压缩的对话Agent最近在折腾一些需要长期记忆和复杂推理的AI应用比如智能客服、游戏NPC或者个人学习助手发现一个普遍痛点传统的聊天机器人聊着聊着就“失忆”了。要么是上下文窗口有限聊到后面把前面的关键信息给忘了要么是记住了所有对话但把一堆无关紧要的闲聊也塞进上下文导致每次调用大模型LLM的成本飙升、速度变慢而且核心指令容易被淹没。这时候“对话压缩”就成了一个刚需。它不像简单的“记住最近N轮对话”那么粗暴而是能智能地提炼历史对话的精华把冗长的聊天记录压缩成一段精炼的“摘要”或“关键事实”然后只把这个摘要和最新问题一起喂给LLM。这样既保留了长期记忆又极大地节省了上下文窗口让Agent变得更聪明、更经济。而LangGraph正是构建这类有状态、可循环、带复杂逻辑的Agent的理想框架。它来自LangChain生态但设计理念更偏向于用“图”Graph来定义和控制Agent的工作流。节点Node代表一个个处理步骤边Edge代表步骤间的流转条件这让实现一个“先判断、再压缩、后回答”的智能对话流变得异常清晰。所以这个项目标题“LangGraph带对话压缩的对话Agent简易实现”瞄准的就是这个场景利用LangGraph的图计算能力构建一个能自动判断何时需要压缩记忆、并执行压缩的对话智能体。它适合任何想给LLM应用加上“持久化且高效记忆”的开发者无论是做产品原型还是深入研究Agent架构这个实现都是一个很好的起点。2. 核心架构与LangGraph工作流设计要理解这个Agent怎么工作得先吃透LangGraph的核心概念。你可以把它想象成一个流程图设计工具但每个步骤都能执行代码并且步骤之间的走向可以动态决定。2.1 图Graph结构设计我们的Agent工作流主要包含以下几个核心节点它们通过有向边连接形成一个循环路由节点route这是大脑的“前额叶”负责判断当前该做什么。它分析最新的用户输入和当前的对话历史或压缩后的记忆决定下一步是直接“回答”问题还是需要先“压缩”过长的历史。压缩节点compress这是记忆的“过滤器”。当路由节点判断历史对话太长或太杂乱时激活此节点。它会调用LLM将完整的对话历史总结成一段简洁、保留关键事实的摘要。回答节点respond这是主要的输出器官。在拥有合适的上下文可能是原始历史也可能是压缩后的摘要加最新几轮对话后调用LLM生成对用户当前问题的回复。状态State这是贯穿整个图的“工作记忆”。LangGraph使用一个共享的状态字典通常是一个TypedDict在各个节点间传递信息。我们的状态至少需要包含messages: 完整的对话消息列表包括用户和AI的。compressed_history: 压缩后的对话摘要文本。num_turns: 或许还有一个记录对话轮次的计数器用于触发压缩条件。边Edge定义了节点执行后的流向。例如route节点结束后可能指向compress或respond。compress节点完成后通常会指向respond。respond节点完成后流程会暂停等待下一次用户输入然后重新从route节点开始。2.2 状态管理对话记忆的存储与演化状态管理是LangGraph的精髓也是实现记忆压缩的关键。我们不会在每次交互后丢弃对话而是有策略地更新状态。初始化状态对话开始时messages列表为空compressed_history为空字符串。常规对话用户和AI的每一轮问答都以HumanMessage和AIMessage的形式追加到messages列表中。同时compressed_history保持不变。触发压缩当messages长度超过某个阈值比如10轮对话或者route节点通过LLM判断历史已不相关时触发压缩。执行压缩compress节点读取messages中的所有内容调用LLM生成摘要。然后关键操作来了我们用这个新的摘要替换掉compressed_history并且可以选择性地清空或只保留最近几轮的messages。这意味着压缩后的摘要成为了新的、凝练的“长期记忆基底”后续对话将基于这个基底和新的短时记忆进行。响应生成在respond节点我们不会把上百条messages全塞给LLM。而是构造一个聪明的提示compressed_historymessages中最近3-5轮对话。这样LLM既能把握整个对话脉络又只处理最相关的少量信息。注意压缩策略需要谨慎设计。不能无脑压缩否则可能丢失重要细节。一种常见策略是“窗口压缩”即始终保留最新的N条原始消息将更早的消息压缩成摘要。另一种是“条件压缩”由LLM判断是否需要进行压缩。2.3 路由逻辑智能判断何时压缩路由逻辑是这个Agent智能与否的开关。最简单的实现是基于轮次计数def route(state: State) - Literal[compress, respond]: if len(state[messages]) TURN_THRESHOLD: return compress else: return respond但更高级的做法是让一个轻量级的LLM比如gpt-3.5-turbo或一个经过微调的分类器来决策。我们可以向这个路由LLM提问“给定当前对话历史和最新问题为了更准确地回答下一个问题我们需要压缩历史对话以提取关键信息吗”。让模型根据对话的连贯性、主题漂移程度来做判断这样更加灵活和精准。3. 分步实现与核心代码解析理论讲完了我们动手实现。这里以OpenAI的模型为例使用langgraph和langchain-openai库。3.1 环境准备与依赖安装首先确保你的Python环境建议3.10并安装必要库pip install langgraph langchain-openai langchain-core python-dotenv创建一个.env文件来安全存储你的OpenAI API密钥OPENAI_API_KEY你的密钥3.2 定义状态与图结构from typing import TypedDict, Annotated, Literal, Sequence from langgraph.graph import StateGraph, END from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage import operator from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate import os from dotenv import load_dotenv load_dotenv() # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] # 自动追加消息 compressed_history: str # 压缩后的历史摘要 turn_count: int # 对话轮次计数器 # 2. 初始化模型和关键提示词 llm ChatOpenAI(modelgpt-4o-mini, api_keyos.getenv(OPENAI_API_KEY)) # 压缩提示词模板 COMPRESS_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个高效的对话摘要助手。请将以下对话历史压缩成一段简洁的摘要保留所有关于事实、用户偏好、决策和关键承诺的信息。摘要语言需简洁、客观用于后续对话的上下文。当前已有一个旧摘要{old_summary}), (user, 请压缩以下对话\n\n{full_history}) ]) # 路由判断提示词模板高级版 ROUTE_PROMPT ChatPromptTemplate.from_messages([ (system, 你需要判断基于当前的对话摘要和最新问题是否需要对完整对话历史进行压缩以优化后续回答。仅回答compress或respond。), (user, 对话摘要{summary}\n最新用户消息{new_message}\n是否需要压缩) ]) # 回答提示词模板 RESPOND_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。请基于以下背景信息和最近对话来回答问题。\n背景摘要{compressed_history}), (user, {recent_messages}) ])3.3 实现核心节点函数接下来我们实现三个核心节点函数。# 路由节点函数 def route_node(state: AgentState) - Literal[compress, respond]: 决定下一步是压缩还是直接响应 messages state[messages] if len(messages) 5: # 简单策略前5轮不压缩 return respond # 高级策略调用LLM判断更智能但更贵 # last_msg messages[-1].content if isinstance(messages[-1], HumanMessage) else # prompt ROUTE_PROMPT.invoke({ # summary: state[compressed_history], # new_message: last_msg # }) # response llm.invoke(prompt).content.strip().lower() # return response if response in [compress, respond] else respond # 中等策略基于轮次和摘要长度 if state[turn_count] % 7 0 or len(state[compressed_history]) 500: return compress return respond # 压缩节点函数 def compress_node(state: AgentState) - dict: 压缩对话历史更新摘要 all_messages state[messages] old_summary state[compressed_history] # 将消息列表转换为纯文本用于压缩 full_history_text \n.join([f{msg.type}: {msg.content} for msg in all_messages]) # 调用LLM进行压缩 prompt COMPRESS_PROMPT.invoke({ old_summary: old_summary, full_history: full_history_text }) response llm.invoke(prompt) new_summary response.content # 更新状态保留新的摘要并可以选择清空或保留最近几条原始消息 # 这里我们选择保留最近3条原始消息作为“短期记忆” recent_messages all_messages[-3:] if len(all_messages) 3 else all_messages return { compressed_history: new_summary, messages: recent_messages, # 替换为近期消息实现记忆窗口 turn_count: state[turn_count] # 轮次计数器保持不变 } # 回答节点函数 def respond_node(state: AgentState) - dict: 生成对用户最新消息的回复 user_message state[messages][-1] # 最新的一条应该是用户消息 compressed_background state[compressed_history] # 构建给LLM的上下文压缩摘要 最近2-3轮对话短期记忆 recent_convo state[messages][-4:] # 获取最近的几条消息包括最新问题 recent_convo_text \n.join([f{msg.type}: {msg.content} for msg in recent_convo]) prompt RESPOND_PROMPT.invoke({ compressed_history: compressed_background, recent_messages: recent_convo_text }) response llm.invoke(prompt) ai_message AIMessage(contentresponse.content) # 更新状态将AI的回复追加到消息列表中并增加轮次计数 updated_messages state[messages] [ai_message] return { messages: updated_messages, turn_count: state[turn_count] 1 }3.4 组装LangGraph工作流最后我们把节点和边组装起来形成完整的工作流。# 创建图构建器 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(route, route_node) workflow.add_node(compress, compress_node) workflow.add_node(respond, respond_node) # 设置入口点 workflow.set_entry_point(route) # 添加条件边从route出发 workflow.add_conditional_edges( route, # 下一个节点由route_node的返回值决定 route_node, { compress: compress, respond: respond } ) # 添加普通边 workflow.add_edge(compress, respond) # 压缩完后必然去响应 workflow.add_edge(respond, END) # 响应完成后本轮结束 # 编译图 app workflow.compile() # 可视化需要安装graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: print(Graph compiled successfully. Visualize in LangGraph Studio.)3.5 运行与测试Agent现在我们可以像调用一个函数一样运行这个有状态的Agent了。# 初始化状态 initial_state: AgentState { messages: [SystemMessage(content你是一个有用的助手。)], compressed_history: , turn_count: 0 } # 模拟多轮对话 test_conversation [ 你好我叫小明。, 我喜欢打篮球和编程。, 我的编程语言主要是Python。, 我最近在学LangGraph。, 对了我养了一只狗叫旺财。, 我通常晚上去健身房。, “那么根据我们刚才的对话请总结一下我的兴趣爱好和个人信息。” ] current_state initial_state for i, user_input in enumerate(test_conversation): print(f\n[用户第{i1}轮]: {user_input}) # 将用户输入添加到消息列表 current_state[messages] current_state[messages] [HumanMessage(contentuser_input)] # 调用图应用 result app.invoke(current_state) # 更新当前状态 current_state result # 打印AI回复和当前记忆状态 ai_response result[messages][-1].content print(f[AI回复]: {ai_response}) print(f[压缩历史]: {result[compressed_history][:100]}...) # 打印前100字符 print(f[消息数/轮次]: {len(result[messages])} / {result[turn_count]})运行这段代码你会观察到在前几轮对话中compressed_history是空的Agent直接基于原始消息回复。当对话轮次达到触发条件例如第5轮或第7轮时route节点会指向compress你会看到compressed_history被更新为一段摘要同时messages列表被截断只保留最近3条。之后的回答将基于这段摘要和最近的对话生成。4. 高级优化与实战技巧基础版本跑通后我们可以从性能、成本、效果上进行深度优化。4.1 压缩策略的权衡与选择压缩不是免费的它需要调用一次LLM产生额外的成本和延迟。因此策略的选择至关重要。固定窗口压缩最简单。每N轮对话压缩一次。优点是稳定、可预测成本可控。缺点是可能在不必要时压缩如对话很简短或在需要时未压缩如对话突然变得复杂。基于令牌数压缩计算messages中所有内容的令牌总数超过LLM上下文窗口的一定比例如70%即触发压缩。这更贴近技术限制但需要能准确计算令牌数的工具如tiktoken。基于语义的智能压缩如我们之前提到的用一个小型LLM或分类器来判断。可以设计更精细的判断标准例如检测主题是否发生显著切换。判断最新问题是否严重依赖于早期历史如“回到我们一开始说的那个问题”。评估当前历史的信息冗余度。 这是效果最好的方法但实现最复杂且引入了额外的模型调用开销。实操心得在生产环境中我推荐混合策略。例如先使用一个廉价的固定窗口如每10轮作为基线再叠加一个基于令牌数的紧急检查如令牌数超过8000立即压缩。智能路由可以作为可选的升级项在对话质量要求极高的场景下开启。4.2 记忆层级与向量数据库集成单一的压缩摘要可能不足以应对极其复杂和长期的对话。我们可以引入多级记忆系统短期记忆原始的、未压缩的最近N条消息如最近10轮。访问速度最快细节最完整。中期记忆通过压缩节点生成的对话摘要。承载了对话的核心脉络和关键事实。长期记忆引入向量数据库如Chroma, Pinecone。将每一轮对话或每一个压缩后的摘要生成向量嵌入并存储。当用户提问时可以先从向量库中检索最相关的历史片段而不仅仅是最近的将其作为补充上下文注入。这相当于给Agent加了一个“联想记忆”的能力。在LangGraph中集成向量检索可以新增一个retrieve节点。在route或respond节点之前先查询向量库将检索到的相关文本片段也加入到构造给LLM的上下文中。4.3 成本控制与性能监控对于需要长期运行的Agent成本是必须考虑的因素。压缩模型选型压缩不一定需要用最顶级的模型如GPT-4。对于总结性任务gpt-3.5-turbo甚至更小的开源模型如Llama-3-8B的API通常就能胜任可以节省大量成本。可以在compress_node中配置一个专用的、成本更低的LLM。异步与批处理如果压缩操作比较耗时可以考虑将其异步化不让用户等待压缩完成。例如在respond节点正常响应用户后后台异步触发压缩任务为下一次交互做准备。监控指标记录每次压缩触发的时机、压缩前后的令牌数对比、压缩所用的模型和时间。这些数据有助于你持续优化压缩策略的阈值和算法。4.4 错误处理与边界情况一个健壮的Agent必须能处理各种意外。压缩失败LLM调用可能超时或返回非结构化内容。需要在compress_node中加入重试机制和异常捕获压缩失败时可以回退到不压缩的状态或者使用一个更简单的规则如只保留最近20条消息来清理历史。状态污染确保每个节点都正确返回状态更新。一个常见的坑是在compress_node中修改了messages但忘记返回turn_count导致状态不一致。使用TypedDict和良好的单元测试可以避免这个问题。无限循环理论上如果route节点逻辑有误可能导致compress-respond-route-compress的死循环。为图设置一个最大执行步数recursion_limit是必要的安全措施。5. 常见问题排查与调试实录在实际搭建和运行过程中你肯定会遇到各种问题。这里记录几个我踩过的坑和解决方法。问题1LangGraph状态更新不符合预期messages列表混乱。排查首先检查状态定义中的Annotated[Sequence[BaseMessage], operator.add]。这个注解确保了messages字段在节点间传递时是追加append操作而不是替换。如果你在某个节点内对state[‘messages’]进行了复杂的切片或重新赋值返回时一定要确保整个列表的更新逻辑正确。技巧在开发阶段在每个节点的开始和结束打印state的关键内容。或者使用LangGraph自带的**检查点Checkpoint和追踪Trace**功能在LangGraph Studio中可视化每一步的状态变化这是最强大的调试工具。问题2压缩后的摘要质量很差丢失关键信息。排查提示词工程你的压缩提示词COMPRESS_PROMPT是关键。确保指令清晰例如强调“保留所有关于事实、决策、数字、承诺的信息”。可以给出一个示例Few-shot会显著提升效果。输入格式检查传给LLM的full_history文本是否清晰可读。确保每条消息都有明确的说话人标识如Human:AI:。模型能力如果使用gpt-3.5-turbo效果不佳可以尝试gpt-4系列模型或者在提示词中要求模型以“要点列表”的形式先提取关键信息再组织成段落。技巧建立一个测试集包含多种类型的对话简单问答、多轮决策、事实陈述混杂观点等自动化运行并评估压缩摘要的ROUGE分数或人工检查持续迭代提示词。问题3Agent响应变慢尤其是触发压缩的轮次。排查这通常是预期的因为压缩节点需要额外调用一次LLM。使用time模块记录每个节点的执行时间。优化异步压缩如4.3节所述将compress_node设计为异步不阻塞本次响应。但这需要改变图的结构使得respond节点不等待compress完成。条件压缩优化route_node的逻辑避免不必要的压缩。例如如果最近几轮对话非常简短即使总轮次到了也可以不压缩。模型降级为压缩任务使用更小更快的模型。问题4在长对话后期AI似乎还是“忘记”了很早之前提过的关键信息。排查这说明你的压缩策略可能过于激进或者压缩过程丢失了信息。检查你的压缩提示词是否足够强调“保留所有关键事实”。解决引入向量检索长期记忆见4.2节。这是解决“长期遗忘”问题的终极方案之一。将历史对话切片存入向量库每次回答时进行检索确保任何历史信息只要相关都有机会被召回。问题5如何将这个Agent部署为可交互的API服务方案使用FastAPI或Flask封装编译好的app。将每次用户请求视为一次app.invoke(current_state)的调用。你需要将会话ID与对应的AgentState持久化存储如Redis、数据库。基本的流程是用户通过API发送消息和会话ID。服务端根据会话ID加载对应的AgentState。将用户消息添加到state[“messages”]。调用app.invoke(state)。获取新的状态和AI回复。将新状态保存并将AI回复返回给用户。构建一个带记忆压缩的LangGraph对话Agent就像给LLM装上一个智能的“记忆管理器”。它不再是一问一答的失忆者而是一个能随着对话演进不断提炼重点、优化资源使用的对话伙伴。从简单的轮次触发到智能路由判断再到结合向量数据库的多级记忆这个框架的扩展性非常强。