LangGraph实战：构建带知识库和长期记忆的AI助手

说实话，我第一次看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里可能没问题，但文档一多，延迟和成本都很感人。正确做法是：文档处理、分块、向量化、建索引，全部离线做一次，在线流程只做检索。

分块参数我给一组自己常用的经验值，未必是最优，但适合大多数场景：

Embedding模型方面，本地优先选开源模型，商业化API也有对应向量模型，都可以。向量数据库的选择看阶段：本地学习和验证用轻量方案，生产环境则换成更专业、支持高并发的服务。这里有个原则：检索节点是通用的，底层向量库应该可以随时替换，这也正是把它设计成独立节点的价值所在。

2.2 retrieve和augment两个节点怎么写

我把知识库流程拆成多个节点，每个节点只做一件事。这样做的好处是：以后想换检索方式、加重排序、调整prompt模板，都不会影响其他环节。

第一个节点负责接收用户的当前问题，去向量库检索，把结果写入context字段：

def retrieve_node(state): query = state["query"] docs = vector_store.similarity_search(query, k=3) # 换成你的向量库实例 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(content=response.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(content=response.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(checkpointer=checkpointer, store=store)

调用时，config里同时带上thread_id和user_id。前者区分会话窗口，后者区分用户：

config_1 = {"configurable": {"thread_id": "session-1", "user_id": "user-A"}} result_1 = graph.invoke( {"messages": [HumanMessage("退货政策是什么？")], "query": "退货政策是什么？"}, config=config_1, ) print(result_1["messages"][-1].content)

这里有个细节值得注意：我传给State的query字段和messages里的用户消息内容是一样的。不要觉得冗余，这是刻意设计——messages用于给模型看完整历史，query则给路由和检索节点做逻辑判断，各司其职。

4.3 两轮对话验证：记忆到底有没有生效

跑完上面的调用，助手应该能根据模拟知识库回答退货政策。接着我们验证长期记忆是否跨窗口生效。

第一次对话，用户在session-1里告诉助手"我喜欢简洁的回答"：

result_2 = graph.invoke( {"messages": [HumanMessage("我喜欢简洁的回答，不想要太长的解释")], "query": "我喜欢简洁的回答"}, config=config_1, )

这时remember_node会提取出"偏好：简洁的回答"，写入Store里user-A的命名空间。

然后关键操作来了：我们完全新开一个会话窗口，也就是新的thread_id，但同一个user_id：

config_2 = {"configurable": {"thread_id": "session-2", "user_id": "user-A"}} result_3 = graph.invoke( {"messages": [HumanMessage("你还记得我喜欢什么样的回答吗？")], "query": "你还记得我喜欢什么样的回答吗？"}, config=config_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这套流程基本就能玩转了。