资讯动态

LangGraph长短期记忆实战:基于线程与Store构建有记忆的AI助手

发布时间:2026/10/17 16:41:17 来源:尧图企业网站定制
开始搞 AI 应用的朋友多少都会撞上同一堵墙模型很强但聊完就忘。用户上午刚告诉你他养了一只猫、对花生过敏、偏好夜间推送下午再问“你觉得我该养什么宠物”它愣是接不上。这不是模型笨是对话系统默认无状态每次请求都被当成第一面。今天不聊那些要烧钱的向量库方案也不谈需要专门团队维护的分布式缓存就拿 LangGraph 这套长短期记忆实现走一遍完整落地路径顺便把我在实战里踩过的坑一并倒出来文末还附了可直接抄的案例代码。这套方案适合谁你如果不是只做“单轮问答 Demo”而是打算真正做长期用户陪伴、客服机器人、智能助手、甚至带记忆的 Agent 工作流那这篇值得收藏。理解它不需要很深的强化学习背景但需要你写过基本的 Python见过LangGraph这个名字并且愿意亲手跑一遍代码。1. 内容整体设计与思路拆解1.1 为什么通用对话系统普遍“失忆”先看一个问题本质绝大多数大模型的 API 是“纯函数”同样的输入永远返回同样的输出温度取零时它自身不带任何对话历史。你发的消息被包装成 prompt填进上下文窗口模型只看得到窗口里的东西窗口之外一切皆空。那有人说了把完整历史拼到 prompt 里不就完事了短期可以长期必崩。一个重度用户对话三个月聊天记录几十万字token 费用先不谈上下文窗口一撑爆最早的记忆直接被挤掉。这就像一个人脑子里只能装最近五分钟的事你能忍更深层的问题在“记忆的类型”被混淆了。用户今天想吃什么这是临时状态明天就失效用户的过敏史、生日、常用收货地址这是长期事实三个月后还得用。如果不做区分一股脑塞进同一套存储里要么浪费 token要么关键信息被噪声淹没。1.2 LangGraph 怎么定义长短期记忆LangGraph 在这件事上做了一个非常优雅的切分短期记忆存在线程thread里长期记忆存在独立的记忆 Store 里。短期记忆贴近最近几轮对话的上下文本质上还是把历史消息放进模型的上下文窗口但它用“线程”隔离不同用户、不同会话避免串味。长期记忆经过提炼的、跨会话保留的核心事实通常以“命名空间 Key-Value”的形式存储每次新对话开始时按需加载。这种设计的核心思路是把“记住什么”和“怎么记”分开。模型负责从对话中提炼值得长期保存的信息LangGraph 负责把这些信息写入持久化存储并在适当时机取回注入上下文。这就好比人脑里的海马体和前额叶皮层分工海马体临时缓存前额叶决定哪些转成长期记忆。我当时选择这套方案图的就是两件事一是它原生支持 Redis、Sqlite 等多种持久化后端不用我自研存储二是它的记忆抽象和图的 checkpoint 机制天然绑定保存状态、恢复状态、跨线程访问都不需要额外写胶水代码。1.3 对比其他记忆方案的取舍动手之前我其实比过几类方案写出来给你参考省得你走弯路方案类型解决场景痛点整段历史拼 prompt短期、小规模窗口有限、token 爆炸引入向量库做相似度检索长期、开放式信息召回质量不稳定、需要 embedding 模型和索引维护外部数据库人工维护用户画像精确结构信息工程量巨大需要人工标注和更新规则LangGraph 长短记忆结合短期上下文 长期事实需要理解线程和 Store 模型有主动遗忘/更新的设计成本不是彻底否定向量库它适合“语义召回”比如用户三个月前说过“我特别怕蜘蛛”你很难用结构化字段存这个但像“用户名、生日、过敏源”这种高确定性信息用 Key-Value 的 Store 既省 token 又精准。真正稳定可用的系统往往是“结构化长期记忆 语义召回”双轨并行。今天的案例先打稳第一轨。2. 核心细节解析与实操要点2.1 线程Thread的正确理解如果你之前只把 LangGraph 当普通图执行框架那对 Thread 的概念可能比较模糊。我换一种说法Thread 就是一次完整会话的 ID。同一个用户的所有消息走同一个 Thread图的状态会自动保存在这个线程作用域内下次再带同一个 Thread ID 进来图能恢复上次执行结束时的状态。这个机制解决了一件很实际的事你不需要自己维护一个“历史消息列表”再手动喂给模型。LangGraph 的 checkpointer 会在每次节点执行后保存状态快照下次从断点续跑或直接继续跑都能读到之前的 state。对你而言只需要在客户端请求里带上config {configurable: {thread_id: user_123}}剩下的交给框架。这里有个容易犯的低级错误把 thread_id 设成全局唯一的一个。结果所有用户共享同一份短期记忆A 用户的对话历史出现在 B 用户面前场面极其尴尬。thread_id 务必由用户 ID 与会话 ID 组合生成比如f{user_id}:{session_id}如果你压根没有会话体系直接用用户 ID 也行但要做好“单用户单线程”的预期——上一轮对话必须和下一轮在同一个线程里才能接上。2.2 Store 模型命名空间与 Key-Value长期记忆的 Store 说白了就是一个带命名空间的 KV 数据库。LangGraph 定义了三个层级namespace相当于文件夹、key相当于文件名、value文件内容。你可以把命名空间设计成(user_id, preferences)这样同一个用户的偏好数据就被隔离在专属目录下。实际操作中命名空间不要设计得太深两层到三层就够了。太深会让检索代码写得啰嗦太浅又不好隔离。我的习惯是(user_id, facts)存长期事实(user_id, settings)存用户配置万一后面要做多租户再加一层(org_id, user_id, facts)也不迟。Value 这边建议存 JSON 序列化后的字典。LangGraph 的 Store 支持直接存字典不用你手动转字符串取出来直接就是 Python dict这对后续构造 prompt 太友好了。2.3 记忆的写入时机与提炼策略这是整个方案里最考验设计感的地方——不是每句话都值得写进长期记忆。我在第一版实现里犯过一个错把用户每条消息全存进 Store。结果 Store 里充满了“今天天气不错”“好的谢谢”这种垃圾真正重要的过敏源、生日反而分散难找。正确姿势是在对话流程里加一个“记忆提炼”节点。这个节点拿着当前对话轮次的消息让模型扮演“记忆抽取器”只输出结构化的重要事实。我是用一个专门的 prompt 来限定输出格式要求模型只抽取以下几类信息用户的个人信息名字、称呼、年龄、职业、所在地偏好饮食、作息、内容风格、沟通习惯约束条件过敏、禁忌、不可触碰的话题长期目标或正在进行中的项目然后把这些抽取结果写到 Store重复 key 就更新不存在的 key 就新建。2.4 记忆的读取策略全量还是按需另一个被忽略的点长期记忆不是每次都要全量加载。用户存了几百条事实全塞进 prompt 既费 token 又让模型分心反而抓不住当前对话真正需要的信息。我试过按“内容类型”分区加载比如用户聊到饮食就只加载(user_id, preferences)下和饮食相关的记录。但更省事的方式是先全量加载最近更新的一小部分再结合当前用户消息做关键词匹配。具体实现上LangGraph 的 Store 支持按 namespace scan也支持按 key 精确 get。我会在对话开始时并行做两件小事get 用户画像综合摘要scan 最近 30 天更新过的记忆。这两份数据合在一起基本覆盖大部分对话场景。3. 实操过程与核心环节实现3.1 环境准备与依赖安装开始写代码前先确认你的 Python 版本不低于 3.10然后装依赖。我用的是langgraph0.2.x 系列和langchain-openai做模型接入持久化后端选的 Sqlite——原因很简单本地开发零配置单文件就能跑等上线再切 Redis 也就换一行配置的事。pip install langgraph langchain-openai如果你的网络环境对默认源不友好用国内镜像装也完全没有问题。装完验证一下版本号LangGraph 迭代很快API 变化不小建议锁定一个主版本比如都装 0.2.x避免不同版本之间示例代码跑不通。3.2 初始化记忆存储与 Checkpointer我用的是 Sqlite 后端让长短期记忆都落在同一个文件里。短期记忆通过 checkpointer 保存长期记忆通过 Store 保存。两者一起初始化的时候要注意一点Store 的实例要传给图的compile()方法而 checkpointer 则通过compile(checkpointer...)传入。from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.store.sqlite import SqliteStore conn sqlite3.connect(memory.db, check_same_threadFalse) checkpointer SqliteSaver(conn) store SqliteStore(conn) graph workflow.compile(checkpointercheckpointer, storestore)这里有个很容易踩的坑Sqlite 的默认连接不允许跨线程访问而 LangGraph 内部执行时可能涉及多线程。所以连接必须带上check_same_threadFalse否则跑起来会报“SQLite objects created in a thread can only be used in that same thread”排查半天才发现是连接参数的问题。3.3 定义短期记忆状态与图节点按照 LangGraph 的习惯先定义状态 Schema。我这里用一个简单的字典状态包含消息列表和当前对话轮次的信息。短期记忆的载体就是messages字段它随着每次图执行被追加保存。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: stradd_messages这个 reducer 很关键。它在状态更新时自动做消息合并而不是覆盖旧列表。LangGraph 拿着这个 state 在 checkpoint 里存快照下次同线程进来旧的 messages 会原样恢复新的消息追加在后面相当于自动帮你拼好了对话历史。手动管理消息列表这件事从此可以删掉了。接下来是节点。核心节点就是聊天节点本身模型收到“系统提示 历史消息 新消息”后生成回复。def chat_node(state: AgentState): messages state[messages] memory_context load_relevant_memory(state[user_id], messages[-1].content) system_prompt f你是用户 {state[user_id]} 的私人助手。以下是关于用户的长期记忆\n{memory_context} response llm.invoke([{role: system, content: system_prompt}] messages) return {messages: [response]}注意看我先从 Store 里把相关长期记忆取出来拼成系统提示。这一步是长短期记忆衔接的核心短期记忆由消息历史天然携带长期记忆则通过手动加载进上下文。缺了任何一块对话都会显得“失忆”。3.4 记忆提炼节点的完整实现最见功力的还是记忆提炼节点。我在主对话之外单独加了一个“记忆写入”节点每次对话结束后异步触发。它的输入是这一轮的全部消息输出是待写入 Store 的结构化记忆。def summarize_and_store(state: AgentState): user_id state[user_id] messages state[messages] recent messages[-6:] # 只看最近几轮避免长会话混淆重点 extract_prompt f从以下对话中抽取值得长期记住的用户信息。 只输出 JSON不要额外解释。格式如下 {{facts: [{{key: 唯一标识, value: 事实内容, category: 偏好|个人信息|约束|目标}}]}} 对话内容 {recent} extraction llm.invoke([{role: user, content: extract_prompt}]) # 这里简单处理直接假设模型输出是合法 JSON import json data json.loads(extraction.content) namespace (user_id, facts) for fact in data[facts]: store.put(namespace, fact[key], {value: fact[value], category: fact[category]}) return {messages: state[messages]}这个节点的 prompt 我反复调过。最开始没限定 JSON 格式模型输出五花八门的散文解析又费劲又不稳定。后来改成强 JSON 约束输出质量明显稳定。但要注意不是所有模型都严格听话建议代码里加一层 try-except解析失败就跳过本轮记忆提炼也不要影响主对话。3.5 图组装与调用入口把上面几块拼成一个图。流程是用户输入 → 聊天节点 → 记忆提炼节点并行或串行→ 返回回复。不管记忆提炼是否成功都不应该阻断主对话回复所以我在编译时把记忆节点标记为非关键路径。builder StateGraph(AgentState) builder.add_node(chat, chat_node) builder.add_node(memory, summarize_and_store) builder.add_edge(START, chat) builder.add_edge(chat, memory) builder.add_edge(memory, END) graph builder.compile(checkpointercheckpointer, storestore)调用的时候一次完整对话长这样config {configurable: {thread_id: user_42:default}} response graph.invoke( {messages: [{role: user, content: 我叫小林对花生过敏以后别推荐含花生的食物}], user_id: user_42}, configconfig ) print(response[messages][-1].content)接下来第二次对话换个 session_id 也没关系只要 user_id 相同长期记忆依然能通过 Store 加载回来config2 {configurable: {thread_id: user_42:new_session}} response2 graph.invoke( {messages: [{role: user, content: 你还记得我有什么过敏吗}], user_id: user_42}, configconfig2 )只要 memory 加载逻辑写对了第二个问题模型能直接答出“你对花生过敏”。4. 常见问题与排查技巧实录4.1 记忆写入成功但加载不到这是最典型的失败场景。写入没问题Store 里也能查到记录但对话时模型还是“失忆”。查下来十有八九是命名空间不匹配。一个地方写入用的是(user_42, facts)另一个地方查询用的是(user_42, preferences)自然查不到。我后来给自己定了一条规矩命名空间所有地方都从常量模块里取禁止硬编码字符串。比如NAMESPACE_FACTS (user, facts) NAMESPACE_PREFS (user, preferences)这样写入和读取永远指向同一个“文件夹”彻底消灭低级拼写错误。4.2 短期记忆串线程现象是用户 A 的对话历史出现在用户 B 的上下文里。排查方法很简单打日志看每次请求进来的thread_id是什么。大概率你会发现所有用户都复用了同一个固定字符串。这种问题的根源在于把thread_id当成全局配置而不是动态请求参数。记住thread_id 必须在每次调用的时候由外部注入它属于请求上下文的一部分不属于图定义的一部分。4.3 长期记忆被污染当记忆提炼节点不加过滤什么话都往 Store 里塞时间一长Store 里全是噪声加载上下文也加载不到重点。解决方法是两层策略第一层prompt 里明确“只抽取事实忽略寒暄、忽略一般性问题、忽略情绪宣泄”。第二层代码里加去重和更新机制对已有 key 做预检查如果原值和新值冲突优先保留新值但记录冲突次数超过阈值则标记人工复核。4.4 模型输出非法 JSON 导致记忆写入失败我试过好几个模型总有一些输出会在 JSON 前后带行内注释或者 markdown 代码块标记。面对这种情况别硬解析。第一个技巧是在 prompt 里写死“输出纯 JSON不要带 json 标记”第二个技巧是解析前先做清理把反引号去掉只截取第一个{到最后一个}之间的内容。import re def safe_json_loads(text): text text.strip() start text.find({) end text.rfind(}) if start -1 or end -1: raise ValueError(no json found) return json.loads(text[start:end1])这个清理函数帮我避免了很多次线上事故。4.5 长期事实过期与遗忘机制时间长了你会发现一个哲学问题用户告诉你“我下个月要去北京”这条记忆三个月后就是错的。如果一直存着反而误导对话。我给每条长期记忆加了一个expires_at字段写入时根据内容类型预估有效期。个人信息类设一年目标类设一个月行程类设一周。加载的时候直接过滤掉过期项。这个“主动遗忘”机制很关键。人的记忆本身就在衰退AI 的记忆如果永远不删最终会累积成沉重的包袱既影响性能也影响回答质量。定期清理让记忆保持新鲜和准确比一味存储更重要。5. 经验总结与进阶思考5.1 实际项目中的落地节奏建议如果你正打算在一个真实项目里引入长短期记忆别一上来就追求“把所有对话都记住”。我从实战里总结出来的路径是这样第一周只做短期记忆给每个用户分配独立线程让对话能接上文。这一步能解决 80% 的“失忆”体验问题。第二周再加长期记忆先只抽取高确定性信息比如用户名、地域、偏好类别不要超过五种。第三周再做记忆整理每隔一段时间把 Store 里的碎片事实合并成结构化摘要减少碎片化带来的信息熵。这个节奏能让你在每一步都有可验证的收益而不是憋大招然后一次翻车。5.2 记忆可视化的价值代码跑通之后我强烈建议你花半天时间做一个极简单的记忆查询页面把某个用户的所有长期记忆按时间倒序展示出来。这件事的价值在调试时体现得淋漓尽致——你一眼就能看出哪些记忆是有效的哪些是噪声哪些过期了。我遇到过好几个问题都是靠看这个页面才定位到的比盯日志高效太多。5.3 数据安全与合规底线最后必须提醒一句长期记忆存储的是用户隐私数据存储安全、访问权限、数据删除机制都是绝对不能偷懒的部分。用户在对话里说出的个人信息至少应该支持“一键清除”的能力。LangGraph Store 的deleteAPI 很简单但你要保证产品里有一个能让用户触发这个操作的入口。这不是可选项是底线。按我的经验在最初设计阶段就把“数据可清除”考虑进去比上线后再补要省太多事。6. 结尾顺着这套思路走下去你的 Agent 就不再是每次见面都“你好我是谁”的陌生人了。它能记住用户的偏好能避开会触雷的禁忌能接住三个月前埋下的伏笔。这种体验上的跃升比任何花哨的功能都更有说服力。我个人在踩了无数坑之后的体会是记忆系统没有什么玄学核心就是“分清楚短期和长期写对了命名空间想好了遗忘策略”。把这三点做扎实你的 AI 助手就已经胜过市面上绝大多数只会聊天的机器人了。希望这篇能帮你省下我当初几周的摸索时间。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑