说实话我第一次看LangGraph文档的时候脑子是懵的。又是节点又是边又是状态机看起来比普通链式调用复杂得多。但等我真正把它用在一个带知识库和长期记忆的AI助手上之后我只有一个感想这才是做有记忆的助手该有的姿势。很多人一开始做AI助手都是用最简单粗暴的方式把聊天历史拼到prompt里每次请求全量传给模型。前几轮没问题等对话超过二三十轮或者知识库文档一多模型就开始顾头不顾尾要么胡说要么忘记关键信息。我折腾了很久换大模型、调参数、加幻觉抑制最后发现问题的根子不在模型本身而在架构设计——你处理对话上下文、知识库检索、用户长期记忆的方式决定了这个助手能走多远。这篇内容就是把我从踩坑到跑通全程的经验整理出来核心是两条主线一是用LangGraph的图状态机制承载知识库检索增强生成流程二是把短期记忆和长期记忆彻底拆开分别用Checkpointer和Store来管理。面向的读者是那种已经会用LangChain调模型但一遇到要保存用户偏好跨会话记忆知识库问答就头疼的开发者。小白也能跟下来我会把关键原理讲透再给一份可以直接运行的最小示例。1. 别急着写代码LangGraph的图驱动到底解决了什么1.1 从一次崩溃的还记得吗开始我先讲一个自己项目里发生的真实场景。某个客服问答助手用户先问你们的退货政策是什么我回了完整说明几分钟后他又问我不想要那么长的回答能不能三句话说清楚再后来又问那你刚才说的第三条是什么意思。结果模型完全懵了。原因很简单我把历史记录全塞进prompt但消息长了以后前面的内容被截断模型根本看不到刚才说的第三条。最气人的是同一段回答有时用了Markdown表格有时用了编号列表模型自己都分不清第三条指向什么。后来我才意识到这类问题的深层原因有两个第一对话流程不是一条直线它中途有分支、有循环、有需要先决定要不要查知识库的逻辑判断第二对话上下文和长期用户档案是两码事混在一起必然互相干扰。LangGraph把这两个问题都摆到了明面上——它逼着你把流程和状态分开设计而不是靠prompt硬撑。1.2 链式调用 vs 图状态机差在哪如果你用过LangChain应该熟悉这样的代码prompt模板接模型、解析输出一步接一步像一条流水线。好处是简单直观坏处是它没有回路。而一个真正的对话助手需要这样的回路用户问了个需要查知识库的问题先进检索节点再生成回答用户只是闲聊直接回答跳过检索回答完之后还要判断这句话里有没有值得长期记住的信息有就写入记忆。这种带分支和回环的逻辑链式调用做起来非常痛苦要么用一堆if/else在外层套要么硬编码各种流程入口。LangGraph的模型是图节点是一段有明确职责的逻辑边决定下一步走到哪条件边可以根据当前状态动态选择方向。打个比方。链式调用像火车站的自助取票机流程是固定的刷身份证、选车次、出票走完就结束。而LangGraph更像一座地铁换乘站每次到达一个节点你可以根据当前情况选择下一站甚至可以走环线绕回原点对同一个节点多次访问。这正是对话系统的常见形态。1.3 状态State是记忆的骨架add_messages是它的魔法LangGraph里所有节点共享同一个State对象。你可以把它理解成一张放在工作台上的共享表格每个节点拿到这张表、看自己需要的信息、修改自己负责的字段再传回给下一个节点。如果中途某个步骤崩溃了表格的状态还在可以从断点继续这也是它能承载记忆的基础。这里必须说一个关键操作符add_messages。定义State时如果消息字段用它标注那么每次节点返回新的消息LangGraph不会覆盖旧消息而是自动追加。这对多轮对话至关重要你不需要手动维护一个全局消息数组。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] # 对话消息自动追加 query: str # 当前用户问题 context: str # 从知识库检索到的内容 facts: str # 从长期记忆中读取的用户档案这段代码看起来平平无奇但它决定了整个助手的信息流通方式。后面所有的节点本质上都是在读写这几个字段。先把这张表格设计好比写任何业务逻辑都重要。2. 知识库接入把RAG变成一个图形里的节点而不是一条链2.1 先搭数据骨架分块和向量化知识库问答也就是常说的RAG检索增强生成核心思路很简单用户提问之后先从准备好的文档库里检索出最相关的片段把片段塞进prompt再让模型基于这些片段回答。听起来简单但工程上有个高频误区把建立索引的过程放进了每轮对话流程里。有些人写代码用户每次提问就重新把整篇PDF分割、向量化、存库再检索。这个操作在小Demo里可能没问题但文档一多延迟和成本都很感人。正确做法是文档处理、分块、向量化、建索引全部离线做一次在线流程只做检索。分块参数我给一组自己常用的经验值未必是最优但适合大多数场景参数推荐值说明chunk_size800~1200字符太小检索碎片化太大超出模型注意力信息冗余chunk_overlap150~200字符防止关键句子被切断top_k3~5检索返回片段数太多了模型会分心相似度阈值0.5~0.6视Embedding模型而定低于阈值直接判定知识库没有答案Embedding模型方面本地优先选开源模型商业化API也有对应向量模型都可以。向量数据库的选择看阶段本地学习和验证用轻量方案生产环境则换成更专业、支持高并发的服务。这里有个原则检索节点是通用的底层向量库应该可以随时替换这也正是把它设计成独立节点的价值所在。2.2 retrieve和augment两个节点怎么写我把知识库流程拆成多个节点每个节点只做一件事。这样做的好处是以后想换检索方式、加重排序、调整prompt模板都不会影响其他环节。第一个节点负责接收用户的当前问题去向量库检索把结果写入context字段def retrieve_node(state): query state[query] docs vector_store.similarity_search(query, k3) # 换成你的向量库实例 context \n\n.join([doc.page_content for doc in docs]) return {context: context}第二个节点负责把检索结果和用户偏好拼进提示词让模型生成回答。注意这里的messages字段因为加了add_messages所以我只需要返回新的AIMessage它会被自动追加进历史from langchain_core.messages import AIMessage, HumanMessage, SystemMessage def answer_node(state): query state[query] context state.get(context, ) facts state.get(facts, ) system ( f你是客服助手。\n f用户长期偏好{facts}\n f知识库片段{context}\n f回答要求优先使用知识库片段不要编造如果知识库没有相关信息明确说不知道。 ) response chat_model.invoke([SystemMessage(system)] state[messages]) return {messages: AIMessage(contentresponse.content)}这里的chat_model是你项目里已经初始化好的大模型实例支持invoke方法就行无论是API调用还是本地模型。我故意把这两个节点独立开来就是要说明检索是检索生成是生成别揉在一起。RAG调优时90%的问题出在检索端把它独立出来排查就容易多了。2.3 不知道就说不知道阈值与意图路由知识库问答最怕的不是没有答案而是模型用训练数据里的旧知识硬答。比如你的产品退货政策明明改了模型不知道新政策反而给出过时信息这比不说话更糟。我习惯在检索节点里加一个相似度判断。如果最高的相似度分低于阈值就认为知识库没有相关答案让模型走不知道分支。不同Embedding模型的分数范围差异很大所以阈值要实际测我一般先从0.55试起再根据用户反馈微调。另外我非常推荐加一个意图路由节点。用户可能只是随口说你好谢谢这些根本不需要查知识库。如果每次都去检索既浪费资源又可能因为检索到弱相关内容而干扰模型。路由节点的做法很简单def route_query(state): query state[query] if any(kw in query for kw in [退货, 运费, 发票, 政策, 价格]): return kb return chat这个函数返回一个字符串LangGraph的add_conditional_edges会根据返回值走到不同节点。这看起来像一层浅薄的if判断但在真实系统里非常有用。等你的助手需要更复杂的路由时完全可以把这里的规则换成一个小模型来决策节点结构不用变。3. 长期记忆Checkpointer是短期记忆Store才是档案室3.1 线程与会话为什么历史记录不是长期记忆很多教程会告诉你加上MemorySaver这个Checkpointer就能让AI记住对话。但它记住的只是当前这条线程内的状态也就是短期记忆。LangGraph里有个重要概念叫thread_id你可以把它理解成浏览器里的一个聊天窗口标签。同一个thread_id下的所有消息都被保存下来换一个新thread_id就像新开了一个窗口之前窗口里的内容不会自动出现在新窗口里。你在购物网站上和客服聊了一整天的历史属于窗口短期记忆但这个用户是金牌会员喜欢简洁回答最近在关注显卡这种跨窗口依然存在的信息属于长期记忆它必须另外存储。我最初就是在这一步栽了跟头以为Checkpointer能解决所有记忆问题结果用户关了窗口再进来AI完全不记得他了。所以记住这个公式短期记忆 当前窗口的对话历史存在Checkpointer里用thread_id隔离。长期记忆 跨窗口的用户档案与事实存在Store里用命名空间区分。3.2 用Store写入记忆提取要克制存储要分区LangGraph的Store是一个带命名空间的键值存储。命名空间的意思是你可以用(user, user_id)这种结构来划分数据的归属。拿用户偏好举例我对每个用户保存一份档案写入时用它自己的命名空间互相之间完全隔离。写入记忆最忌讳的是什么都记。如果你让模型把每一句话都当成记忆写进去几天后Store里就会堆满用户今天下午三点说想吃西瓜这种毫无价值的噪音。正确做法是克制提取只从用户消息里找稳定的事实、偏好、状态变化。这是我的记忆写入节点的核心逻辑from langgraph.store.memory import InMemoryStore store InMemoryStore() def simple_extract(query): 从用户消息里提取简单事实生产环境建议换成LLM少样本抽取 facts [] if 我喜欢 in query or 偏好 in query: tail query.split()[-1].strip() facts.append(f偏好{tail}) if 住在 in query: tail query.split(住在)[-1].strip() facts.append(f常居地{tail}) if 我是 in query and 客户 in query: tail query.split(我是)[-1].strip() facts.append(f身份{tail}) return facts def remember_node(state, config): user_id config[configurable].get(user_id, default) new_facts simple_extract(state[query]) if not new_facts: return {} # 没有值得记的直接返回 item store.get((user, user_id), preferences) old_facts item.value[facts] if item else [] # 去重合并 merged old_facts[:] for fact in new_facts: if fact not in merged: merged.append(fact) store.put((user, user_id), preferences, {facts: merged}) return {}你可能注意到我的提取规则非常简陋更像一个示例模板。因为在真实项目里这里应该用一个小模型来做信息抽取并配上类似这样的少样本定义从用户消息中提取值得长期记住的事实忽略临时情绪和琐碎内容输出JSON数组。规则写法的好处是零成本、结果稳定适合Demo和学习模型抽取则更灵活能理解复杂的表达。两个方案我都用过建议生产环境直接用模型抽取但一定要加输出格式约束否则模型偶尔会发挥过头。读取的时候同样简单在流程的最前面加载档案写进facts字段def load_memory_node(state, config): user_id config[configurable].get(user_id, default) item store.get((user, user_id), preferences) facts item.value[facts] if item else [] return {facts: .join(facts) if facts else 暂无}3.3 记忆更新与删除被用户纠正后怎么办相比写记忆更新和删除长期记忆更容易被忽视也就更容易出事。用户说我不住杭州了搬到深圳如果你只在旧档案里追加一条常居地深圳那这个人的档案就会同时出现杭州和深圳两个地点以后所有依赖地点的判断都会出错。我的做法是让记忆写入节点带上冲突消解逻辑。最简单的方案是每条记忆都给定类型比如location、preference、identity三类。写入前先取出同类型旧记忆如果发现类型重复就用新值覆盖旧值。这种实现很直白但比较机械。更聪明的方案是让模型直接输出要新增的记忆和要删除的旧记忆ID。这意味着你的档案里不再是一个存事实的数组而是一组带ID和时间戳的记录。用户纠正时模型对比新旧内容把被推翻的旧记录标成删除。这种方式维护成本高一点但用户体验好很多毕竟长期记忆的精髓不是越攒越多而是关键时刻能记住对的信息。4. 跑通一个最小示例带知识库和记忆的助手长什么样4.1 状态定义与四个核心节点这一节把前面的概念串起来给一份几乎可以直接运行的最小代码。我先声明为了让示例聚焦在LangGraph的记忆与编排知识库检索我用了简单的关键词匹配代替向量检索实际项目中替换为向量库的similarity_search即可。状态设计比之前多一个user_id思路——它不放在State里而是通过调用时的config传入让节点函数读取from typing import Annotated, TypedDict from langchain_core.messages import AIMessage, HumanMessage, SystemMessage from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver from langgraph.store.memory import InMemoryStore class AgentState(TypedDict): messages: Annotated[list, add_messages] query: str context: str facts: str # 模拟知识库实际项目换成向量检索 KB_DOCS [ 运费标准满99元包邮未满99元收8元运费。, 退货政策签收后7天内可无理由退货需保持商品完好。, 发票说明本店支持电子发票下单后可在订单页申请。, ] def search_kb(query): results [doc for doc in KB_DOCS if any(kw in query for kw in [运费, 退货, 发票, 包邮])] return \n\n.join(results[:3]) if results else 四个节点分别是加载长期记忆、知识库检索、生成回答、写入新记忆。完整定义如下def load_memory_node(state, config): user_id config[configurable].get(user_id, default) item store.get((user, user_id), preferences) facts item.value[facts] if item else [] return {facts: .join(facts) if facts else 暂无} def retrieve_node(state): return {context: search_kb(state[query])} def answer_node(state): context state.get(context, ) facts state.get(facts, 暂无) system ( f你是客服助手。\n用户长期偏好{facts}\n知识库片段{context}\n f回答要求优先使用知识库片段不要编造没有相关信息就直说不知道。 ) # 替换成你自己的大模型实例只要支持 invoke 即可 response chat_model.invoke([SystemMessage(system)] state[messages]) return {messages: AIMessage(contentresponse.content)}remember_node和之前一致负责提取新事实并合并写入Store这里不再重复贴代码。为了演示简洁模型实例chat_model由你按自己的环境初始化支持invoke方法即可。4.2 图编译与调用流程现在把这些节点串成一个图。观察整个流程加载记忆 - 路由判断 - 检索 - 回答 或 直接回答- 写回记忆 - 结束。def route_query(state): query state[query].lower() if any(kw in query for kw in [运费, 退货, 发票, 政策, 价格]): return kb return chat builder StateGraph(AgentState) builder.add_node(load_memory, load_memory_node) builder.add_node(retrieve, retrieve_node) builder.add_node(answer, answer_node) builder.add_node(remember, remember_node) builder.add_edge(START, load_memory) builder.add_conditional_edges( load_memory, route_query, {kb: retrieve, chat: answer} ) builder.add_edge(retrieve, answer) builder.add_edge(answer, remember) builder.add_edge(remember, END) checkpointer MemorySaver() store InMemoryStore() graph builder.compile(checkpointercheckpointer, storestore)调用时config里同时带上thread_id和user_id。前者区分会话窗口后者区分用户config_1 {configurable: {thread_id: session-1, user_id: user-A}} result_1 graph.invoke( {messages: [HumanMessage(退货政策是什么)], query: 退货政策是什么}, configconfig_1, ) print(result_1[messages][-1].content)这里有个细节值得注意我传给State的query字段和messages里的用户消息内容是一样的。不要觉得冗余这是刻意设计——messages用于给模型看完整历史query则给路由和检索节点做逻辑判断各司其职。4.3 两轮对话验证记忆到底有没有生效跑完上面的调用助手应该能根据模拟知识库回答退货政策。接着我们验证长期记忆是否跨窗口生效。第一次对话用户在session-1里告诉助手我喜欢简洁的回答result_2 graph.invoke( {messages: [HumanMessage(我喜欢简洁的回答不想要太长的解释)], query: 我喜欢简洁的回答}, configconfig_1, )这时remember_node会提取出偏好简洁的回答写入Store里user-A的命名空间。然后关键操作来了我们完全新开一个会话窗口也就是新的thread_id但同一个user_idconfig_2 {configurable: {thread_id: session-2, user_id: user-A}} result_3 graph.invoke( {messages: [HumanMessage(你还记得我喜欢什么样的回答吗)], query: 你还记得我喜欢什么样的回答吗}, configconfig_2, )因为load_memory_node在流程最开始就会按user_id去Store读取档案session-2里哪怕一条历史消息都没有模型依然能从facts字段看到偏好简洁的回答。这就是跨会话长期记忆生效的证据。如果你把上面的config_2里的user_id改成user-B再跑一次模型就完全不记得这个偏好了。用这个对照实验能很直观地理解thread_id负责窗口上下文user_id负责人的身份。两套东西缺一不可。5. 小白最容易踩的坑和我的优化建议5.1 历史消息越堆越长模型越聊越笨用上Checkpointer之后很多人的第一个直觉是太好了终于可以无限聊下去了。但实际跑几轮就会发现窗口里的消息越来越多每次请求都要把所有历史发给模型。对话到了三四十轮光是历史就有好几万字不仅慢模型还容易在中段丢失注意力质量直线下降。我的建议是给历史加一套压缩机制。当消息数量超过某个阈值比如20条就把旧消息交给一个快速模型生成摘要用摘要替换最老的那一批消息最近的几条保留原样。LangChain有现成的消息裁剪工具但更推荐你主动控制因为不同业务对历史保留粒度的要求差别很大。一个省钱又有效的技巧给每轮消息打时间戳或序号历史截断时优先保留有工具调用、有知识库引用记录的消息这些上下文往往比普通寒暄更有价值。5.2 检索内容不对宁可少检也不能乱检RAG流程跑通之后质量瓶颈几乎都出在检索端。最典型的毛病是知识库里没有相关内容但向量检索强行找出几条最像的片段模型看着这些毫不相干的内容反而开始胡说八道。应对思路有三层。第一层是阈值过滤这个我前面提过了没有高置信度匹配就老实说不知道。第二层是调整分块策略如果你的文档是长表格、长代码段或者小标题很多固定字符切块会严重破坏语义最好按照文档结构切块。第三层是重排序先多召回一些候选片段再用重排模型精排前几名对答案质量的提升非常明显。我自己的经验是先跑通再回头仔细看检索结果。如果你发现完全相同的用户问题每次检索出来的片段都不一样多半是Embedding模型对短文本不够友好可以考虑在用户问题上做扩展把退货扩展成退货政策 退款 七天无理由再检索。5.3 长期记忆不要写成流水账我最后悔的一个设计就是早期让记忆节点把所有能提取的事实全存下来。结果运行一段时间后某个用户的档案里堆了上百条记录加载到prompt里本身就好几千字而且新旧矛盾经常出现。后来我总结出三条经验现在一直在用给记忆分类型只保留identity、location、preference、goal这四类稳定信息临时状态一律不存每条记忆带一个最后更新时间戳定期清理长期未更新的旧条目提供用户查看和删除记忆的入口哪怕只是一个简单命令如忘记我刚才说的。长期记忆是给人服务的不是给存储空间冲业绩的。少而准远胜多而乱。5.4 什么时候必须换掉内存版存储前面示例用的MemorySaver和InMemoryStore都是内存实现进程一重启记忆全丢。这在学习和单机Demo阶段完全够用但只要你的应用打算上线就一定要换成持久化方案。Checkpointer可以换乘数据库级的实现比如SQLite、Postgres把每次状态快照持久化到磁盘Store也一样换成数据库后端或者任意分布式键值存储。这一步改动本身不大但它决定了你的助手是重启后什么都不知道还是服务重启但用户记忆健在。并发方面也要注意。内存版实现适合单线程测试线上多用户同时访问时务必要确保存储层支持并发读写并且不同用户的user_id命名空间完全隔离别让A用户读到B用户的档案。我个人在设计这套架构时最大的体会是**AI助手的能力边界很大程度取决于你如何组织状态。**图状态机带来的不是代码量的减少而是把杂乱无章的上下文、知识库、用户档案这些信息安放到了它们各自该待的位置。现在我可以随时给某个节点加缓存、换模型、改路由规则而不用推倒重来。再分享一个调试小技巧。如果你在跑流程时发现状态不对可以打印某一个时刻的完整状态# 某个会话当前状态的快照 snapshot graph.get_state(config_1) print(snapshot.values) # 也可以在流程中间强制覆盖某个字段用来做测试 graph.update_state(config_1, {facts: 测试用偏好})这样你就能清清楚楚地看到messages、context、facts到底在哪一步被写成了什么样。调试记忆类问题靠肉眼读日志远不如直接看状态快照直观。把这个工具用好LangGraph这套流程基本就能玩转了。
