LangGraph 记忆系统设计实战
从"失忆"痛点说起
想象一个场景:用户和 Agent 聊了 50 轮之后突然问一句"刚才说的那个方案怎么实现?",结果对方答:“抱歉,我不知道之前说了什么。”
这时候别急着怪 AI 笨。真正的问题是——长对话系统天生就带着三大矛盾:
| 矛盾 | 根源 | 实际代价 |
|---|---|---|
| 需要记住太多 vs 窗口有限 | LLM 通常 128K tokens 上限 | 超出后只能硬截断,直接丢掉 |
| 完整历史成本高 vs 预算有限 | 1000 轮对话就是数百万 tokens | 成本从几美元飙升到几十美元 |
| 信息密度低 vs 关键需求高 | 用户偏好埋在闲聊里 | 模型根本抓不住重点 |
这些矛盾光靠"把窗口调大"解决不了。真正可行的办法是用分层的方式设计记忆系统。
第一层:短期记忆——把一次会话聊完整
State 到底怎么设计
LangGraph 靠State(TypedDict)做数据的中枢。不过很多人拿到手就把 messages 往里扔,然后就没下文了。这里头其实有讲究。
推荐写法:
class AgentState(TypedDict): # 必选项 messages: list[BaseMessage] # 对话历史 # 控制项(用于压缩/召回决策) should_summarize: bool user_id: str # 扩展项(根据业务需求添加) current_task: str task_history: list[dict]这么设计的考虑:
messages是主链路,几乎所有节点都要用它
should_summarize是个开关信号,用来触发压缩逻辑
user_id作为长期记忆的钥匙,可以跨节点传递
Checkpoint:给对话做个快照
Checkpoint最实在的作用就一个:用户关掉页面几小时后再回来,Agent 还能接着之前的话题聊。
它的运行逻辑是这样的:
这三个字段最实用:
| 字段 | 干啥用的 | 长什么样 |
|---|---|---|
thread_id | 区分不同会话,避免串线 | "user-123" |
checkpoint_id | 标记这是第几轮保存 | "1ef2a..." |
configurable | 扩展配置用 | {"thread_id": "..."} |
这些情况都能用上:
- 用户刷新页面回来接着聊
- 同一个用户多条对话线并行跑
- 调试的时候想看某个时刻的状态
MessageGraph 为什么这么设计
为啥要用messages列表,而不是一整个字符串打包塞给 LLM?核心原因有两个:可控和灵活。
实际用的时候记住三点:
顺序怎么排
:system prompt 开头,中间插摘要,旧消息往后挪,最后留最近 10 条
数量怎么控
:最多保留 50 条,再多了容易爆窗口
要不要工具结果
:生产环境基本不用,调试的时候再看
第二层:长期记忆——隔几天还能认出来
短期记忆管"一次会话不卡壳",长期记忆管的是"过了几天再来,Agent 还能接着之前的话茬聊"。
到底啥值得存?
不是每条内容都配进长期记忆。看这张表就明白了:
| 适合长期存 | 放短期就行 |
|---|---|
| 名字、偏好、个人设置 | 这轮的具体答案 |
| 聊过的项目背景 | 随手查的临时数据 |
| 拍板的技术方案 | 中间的脑补过程 |
快速判断法:想象一下明天用户问"我之前提过啥",这事儿值得 Agent 记着吗?值得就值得存。
三种存储方案,怎么取舍
长期记忆一般就这三套打法:
Redis 怎么设计 Key
一个实用格式:user:{user_id}:prefs:preference_name,value 直接存内容就行。
比如:
user:user_123:prefs:name = "小明" user:user_123:prefs:tech_preference = "简洁方案优先"向量库怎么存、怎么查
什么时候该召回记忆
不用每条消息都去查长期记忆,那样 token 烧得飞快,响应也慢。常见有四招:
| 策略 | 啥时候用 | 好处 | 坑 |
|---|---|---|---|
| 抓关键词 | 出现"之前"“上次”"记得"这些词 | 直接命中需求 | 得靠用户会说话 |
| 会话开局 | 前 5 条消息内 | 主动调画像出来 | 有时候没必要 |
| 定时扫一眼 | 每隔 N 轮一次 | 性能和体验折中 | 有延迟 |
| 抓实体 | 检测到人名、项目名 | 可能有旧账可翻 | 容易误报 |
旧记忆该怎么处理
存久了不处理,存储迟早爆。有个经验做法是按时间加权:
| 存放多久 | 权重给多少 | 怎么召回 |
|---|---|---|
| 7 天内 | 1.0 | 随便用,频率拉满 |
| 7-30 天 | 0.5-1.0 | 每周打个九折 |
| 30 天以上 | 0.1-0.5 | 低调点,低频召回 |
| 半年+ | 0.1 | 归档或者直接删 |
第三层:上下文压缩——窗口不够用时的补救措施
短期 + 长期记忆都上了,为啥还要压缩?原因很简单:就算只存 messages,一个会话聊到 100 轮也是常事,到了那会儿就得靠压缩腾地方。
啥能删,啥得留着
打死不能动:
- System Prompt:这是 AI 的行为底线
- 用户真实需求:现在到底想干啥
- 硬约束条件:预算、时间、技术栈这些
- 已经拍板的决定:“就用 React”
可以放心压缩的:
- 中间的思考过程
- 反复确认过的细枝末节
- 被新信息覆盖掉的旧背景
- 啰嗦重复的表达
压缩和 RAG,怎么选?
这俩不是非此即彼的关系。看你的使用场景来定:
| 场景 | 选哪个 | 为啥 | 常见例子 |
|---|---|---|---|
| 客服聊天 | 上下文压缩 | 得记住前后问题的关联 | 查订单、售后咨询 |
| 知识问答 | RAG | 答案在外部文档里 | 产品手册、API 文档 |
| 超长大对话 | 压缩 + RAG | 历史要记,外部知识也要 | 技术顾问、编程助手 |
| 个性化助手 | 长期记忆 | 关键是记住用户偏好 | 私人助理、健身教练 |
压缩是怎么跑的
压缩的关键就是让 LLM 自己概括,prompt 大致是这样:
把下面这段对话压缩成 bullet points,只要三样东西: 1. 用户真正想要啥 2. 已经拍板的决定 3. 重要的限制条件 [旧消息放这儿]实战:搭一个完整的全能记忆 Agent
前面章节分别讲了短期、长期和压缩,现在把它们拼成一整个系统。
整体长啥样
每个组件干啥的
| 组件 | 负责啥 | 操作哪些 State 字段 | 依赖啥存储 |
|---|---|---|---|
| recall_node | 根据用户问题拉长期记忆 | user_id | Redis+ 向量库 |
| chat_node | 调用 LLM 生成回复 | messages | - |
| compress_node | 消息太多就压缩历史 | messages,should_summarize | - |
| save_node | 聊完把重要信息存下来 | user_id,messages | Redis+ 向量库 |
| checkpointer | 每一步都自动存档 | 所有字段 | Postgres/Memory |
关键节点怎么写
1. 召回节点怎么干
def recall_memory(state: AgentState) -> dict: """ 新消息进来先拉长期记忆,把相关背景塞进去 """ user_id = state["user_id"] last_message = state["messages"][-1].content # 第一步:Redis 拿用户的固定偏好(准没错) preferences = redis_client.hgetall(f"prefs:{user_id}") # 第二步:向量库找语义相关的历史(带点模糊匹配) relevant_docs = vectorstore.similarity_search(last_message, k=3) # 第三步:把两者合一块 memory_context = [] if preferences: memory_context.append("【用户偏好】\n" + "\n".join(preferences.keys())) if relevant_docs: memory_context.append("【历史相关】\n" + "\n".join(d.page_content for d in relevant_docs)) # 第四步:把拼好的东西塞进系统提示 if memory_context: system_msg = SystemMessage(content="\n\n".join(memory_context)) return {"messages": [system_msg] + state["messages"]} return {}这么写的考虑:
- Redis 先上:名字、设置这些固定信息,基本不会变
- 向量库跟进:找历史上类似的情况,带点模糊匹配
- 最后合流:把两块内容拼好,一起丢给 LLM
2. 压缩节点怎么干
def compress_node(state: AgentState) -> dict: """ 消息太多时,把旧的消息压成摘要 """ # 留最近 10 条原文,剩下的压缩 old_messages = state["messages"][:-10] recent_messages = state["messages"][-10:] summary = llm.invoke(f""" 把下面这段对话压成 bullet points,只要三样: 1. 用户真正想要啥 2. 已经拍板的决定 3. 重要的限制条件 {format_messages(old_messages)} """) return { "messages": [ SystemMessage(content=f"【历史摘要】\n{summary.content}") ] + recent_messages }这么写的考虑:
- 原文只留最近 10 条,其余全部压缩
- 摘要标成
SystemMessage,后面好识别 - 30 条这个阈值是经验值,按你的窗口大小调
3. 记忆怎么保存
def save_memory_node(state: AgentState) -> dict: """ 聊完了就把关键信息拎出来存长期记忆 """ user_id = state["user_id"] # 提偏好:发现新 info 就存 Redis extract_and_save_preferences(state) # 存摘要:丢进向量库,以后靠语义找 conversation_summary = generate_conversation_summary(state["messages"]) vectorstore.add_texts( [conversation_summary], metadatas=[{"user_id": user_id, "type": "conversation", "date": now()}] ) return {}这么写的考虑:
- 偏好提取:盯着对话里有没有新的人物信息,有就更新 Redis
- 摘要保存:每轮对话都存个短摘要,攒久了就是语义记忆库
数据咋流转的
实际跑起来啥样
第一回聊 (2026-01-01): 用户:我叫小明,喜欢简洁的技术方案 AI: 很高兴认识小明!简洁确实是高效的选择。我会按这个偏好推荐方案。 → Redis 存下:user:user_123:prefs:name = "小明" → Redis 存下:user:user_123:prefs:tech_pref = "简洁方案优先" → 向量库记一笔:对话摘要 v1 --- 过了三天 --- 第二回聊 (2026-01-04): 用户:帮我选个框架 AI: 根据你的偏好(你喜欢简洁的方案),我建议 Next.js... → 先从 Redis 捞出用户名字和偏好 → 再查向量库找到之前的讨论 → 合起来给个个性化回复 第三回聊 (2026-01-10): 用户:我之前说过啥偏好? AI: 你之前提过:"我喜欢简洁的技术方案" → 向量库翻出历史对话 → 引用原文回复实战经验总结
按需求挑方案
| 想干啥 | 第一层选啥 | 第二层补啥 | 复杂程度 |
|---|---|---|---|
| 基础对话恢复 | Checkpoint (MemorySaver) | - | ★☆☆ |
| 跨会话简单场景 | Checkpoint (PostgresSaver) | Redis(偏好) | ★★☆ |
| 完整用户画像 | 全套 | Redis+ 向量库 | ★★★ |
| 企业级应用 | 全套+RAG | 接知识库 | ★★★★ |
五条实操建议
- 先拿 Checkpoint 上手:最简单的方案往往最实用,能搞定 80% 的场景。真遇到跨进程/多实例共享时再切 PostgresSaver 也不迟。
- 长期记忆要有写入策略:别一股脑全塞进去。每轮对话结束拎出 3-5 条核心的就够,定期把三个月前的旧账清一清。
- 压缩阈值设个预警线:30-50 条是个参考值,别等到爆满了才动手。根据实际窗口大小提前定好红线。
- 盯着 token 成本:长期记忆 + 压缩都会多烧 token。最好做个成本看板,心里有数。
- 分层做记忆系统:工作内存 (当前对话) + 情景记忆 (会话恢复) + 语义记忆 (用户画像),各司其职。
还能往哪深化
记忆索引
:按时间/主题/重要性建多级索引
自我反思
:让 Agent 定期整理记忆,去重合并
选择性遗忘
:像人一样,自动丢掉低权重记忆
多模态记忆
:支持图、音频多种类型
传统产品经理,正在成为下个被淘汰的“传统岗位”。
过去画原型、写 PRD、跟进度的“传统技能包”,在AI时代正迅速贬值。63% 的企业转型做 AI 产品!当下的问题不再是“要不要学 AI ”,而是“如何构建 AI 产品”。
前段时间还跟字节、腾讯的资深 AI 产品经理沟通,他们反馈:在大量招人,只要有 AI 相关的项目经验,基本都能拿到面试机会,而且领导很舍得给钱,涨薪 40-60% 很正常!
01
接下来的产品人,得卷AI能力了!
如今AI大火,行业极速发展的背后,懂AI 产品人才却严重稀缺。这不是要你转技术岗,而是要掌握构建 AI 产品的核心方法:
- 如何将你的领域知识,转化为 AI 产品的核心竞争力?
- 如何用 AI 技术实现你的产品需求?
- 如何设计真正懂用户的 AI 交互体验?
- ……
懂AI,就是产品经理的“救命稻草”!
风口之下,与其焦虑被行业淘汰
不如先人一步享受AI技术带来的红利!
我把AI产品经理的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~
这份完整版的大模型 AI 学习资料已经上传CSDN,朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
(不限年龄!不限岗位!没有代码基础也能学!)
🎁现在扫码,完课还送:
《AI产品面试题库》《AI大模型应用案例集》
02
掌握技术+实战,快速转型!
想成为一名卓越的AI大模型产品经理,需要从技术、到项目实战的全方位转型指南!
**1)**AI产品应用原理解析,产品经理也能听懂!
对于产品经理来说,如果你不懂技术,做不了业务和AI大模型技术衔接、定义不了数据需求,是没法完整的落地一个产品的!
本次课程,专门面向产品经理人群,解析当下最热门的AI产品应用的必备的「大模型」、「多模态」的实际应用和算法原理!解析AI产品应用技术,积累大模型能力!简单易懂,不需要会代码,小白也能掌握!
- 大模型微调:掌握主流大模型(如DeepSeek、Qwen等)的微调技术,针对特定场景优化模型性能。学习如何利用领域数据(如制造、医药、金融等)进行模型定制
- AI Agent智能体搭建:学习如何设计和开发AI Agent,实现多任务协同、自主决策和复杂问题解决。构建垂类场景下的智能助手产品(如制造业中的设备故障诊断Agent、金融领域的投资分析Agent等)
2)超全行业案例解析!
课程详细讲解现阶段,大模型在各个行业和领域的应用现状!包括:零售与电商、教育、医疗、泛娱乐、法律等等10大行业!
详细讲解案例的思路、应用场景,以及背后的技术原理、核心技术!揭秘各个行业、场景的真实现状,和未来产品的发展与机遇!
可以说,讲解完一个案例,就能积累一个AI产品实践的经验!
课程中所涉及到的实战项目,都可以直接在自己的工作中使用,让自己的产品/项目有可借鉴的成功案例!
3)AI产品经理求职专项辅导
课程中会系统的帮助大家拆解字节、腾讯、百度等大厂AI PM岗位JD关键词,掌握AI PM高频面试题型与回答框架;展示 AI 相关能力的关键技巧:Prompt设计、模型评估、A/B测试、成本意识、与算法/工程协作经验;
- To B类AI产品经理:突出“行业理解 + 技术落地 + 商业闭环”能力的简历结构设计,展示项目成果;从客户需求洞察到技术方案设计,展现端到产品思维;如何评估To B AI产品的可行性、客户付费意愿与实施成本
- To C类AI产品经理:拆解头部公司岗位JD,将过往尽力转化为AI产品叙事逻辑;从行业趋势、产品设计题、案例分析&数据分析题、技术理解边界等全流程辅导面试;避免无效海投、锁定最适合的AI产品岗位;
03
本次课程,全程直播讲解,能直接对话大佬和专业助教,不懂就问,超详细的案例,小白也能轻松get!
完课后,还赠送《AI产品经理面试题库》、《AI大模型应用案例集》!不断更新中……
适合人群:
- 想转型AI产品经理、AI项目管理专家、AI产品解决方案等岗位
- 想进行AI产品创业的创业者
- 想成为制作AI产品的程序员
- 想利用AI解决企业问题的管理岗
- 想在AI方向寻找就业方向的毕业生
- AI方向前景广阔、待遇好!
目前,很多产品人已经通过完整学习拿到大厂高薪offer,收入嗷嗷涨!
我把AI产品经理的学习全流程已经整理📚好了!抓住AI时代风口,轻松解锁职业新可能,希望大家都能把握机遇,实现薪资/职业跃迁~