一、为什么 LangGraph 需要“记忆”在普通的 LLM 调用中模型本身是无状态的。也就是说用户第一轮说“我叫小明”第二轮再问“我是谁”如果我们没有把历史消息重新传给模型模型并不知道“小明”是谁。LangGraph 的记忆机制就是为了解决这个问题。参考https://docs.langchain.com/oss/python/langgraph/add-memory可以把 LangGraph 的记忆分成三类线程级记忆同一个 thread_id 下保存完整图状态适合连续对话。长期记忆使用 store 保存用户偏好、个人信息等跨会话数据。上下文压缩记忆当消息过多时通过裁剪或摘要保留关键信息避免 token 爆炸。这三类记忆组合起来就可以构建一个既能“记住上下文”又能“长期了解用户偏好”的智能体。本文根据LangChainDocs官方文档写了test几个demo进行实践。二、线程级记忆Checkpointer thread_id在 test.py 中代码使用了 PostgresSaver 作为 LangGraph 的 checkpoint 存储fromlanggraph.checkpoint.postgresimportPostgresSaverfrompsycopgimportConnection connConnection.connect(DB_URI,autocommitTrue)checkpointerPostgresSaver(conn)agent_searchagent_builder.compile(checkpointercheckpointer)这里的核心是 checkpointer。它会把每次图执行后的状态保存下来包括消息列表、节点执行位置、状态字段等。调用时通过 configurable.thread_id 指定会话线程config{configurable:{thread_id:2}}responseagent_search.invoke({messages:[{role:user,content:今天成都天气怎么样}],llm_calls:0},configconfig)只要后续继续使用同一个 thread_idLangGraph 就能从 checkpoint 中恢复这条线程之前的状态。因此第二次可以继续问responseagent_search.invoke({messages:[{role:user,content:刚才天气怎么样}],llm_calls:0},configconfig)模型之所以知道“刚才天气”指什么不是因为模型真的记住了而是 LangGraph 根据 thread_id 找回了历史状态。注意实际发布代码时不建议把数据库连接字符串硬编码在代码中推荐放到 .env 里例如 POSTGRES_URI。三、查看历史状态与时间旅行LangGraph 还有些很有特色的能力查看历史状态和从某个 checkpoint 重放。print(agent_search.get_state(config))forstateSnapshotinagent_search.get_state_history(config):print(id :,stateSnapshot.config[configurable][checkpoint_id])print(下一个节点,stateSnapshot.next)get_state(config)可以查看当前线程的最新状态。get_state_history(config)可以查看这个线程下所有历史快照。每个快照都有自己的checkpoint_id。如果想从某个历史状态继续执行可以使用result agent_search.invoke(None, configto_replay.config)这就类似“时间旅行”从某个历史节点重新运行图。这个能力在调试 Agent、复现线上问题、观察工具调用链路时非常有用。四、短期记忆和长期记忆的区别在 test2.py 中代码同时使用了graphbuilder.compile(checkpointerInMemorySaver(),storestore)这里要区分两个概念类型 对应组件 作用短期记忆 checkpointer 保存某个 thread_id 下的对话状态长期记忆 store 保存用户信息、偏好等跨线程数据checkpointer 更像“当前会话的上下文记录”它依赖 thread_id。store 更像“用户档案库”它可以跨不同 thread_id 读取只要使用相同的 user_id 或命名空间即可。五、长期记忆InMemoryStore 与命名空间test2.py 中使用 InMemoryStore 创建了长期记忆fromlanggraph.store.memoryimportInMemoryStorefromlangchain.embeddingsimportinit_embeddings storeInMemoryStore(index{embed:init_embeddings(openai:text-embedding-3-small),dims:1536,fields:[$]})这里的 index 表示给存储内容建立向量索引后续可以做语义搜索。写入记忆时代码使用了分层命名空间user_id1user_123namespace_1(user_id1,prefs,food)memory_id_1str(uuid.uuid4())memory_value_1{food:披萨}store.put(namespace_1,memory_id_1,memory_value_1)这个命名空间(user_id, prefs, food)很关键。它可以把不同用户、不同类型的数据隔离开。例如namespace_1(user_123,prefs,food)namespace_2(user_456,prefs,music)这样 user_123 的饮食偏好和 user_456 的音乐偏好就不会混在一起。读取单条记忆single_itemstore.get(namespace_1,memory_id_1)print(single_item.value)语义搜索记忆all_memstore.search((user_123,prefs,food),query用户喜欢的食物,limit1)如果配置了 embeddingstore.search 不只是关键词匹配而是可以按照语义查找相关记忆。六、PostgresStore让长期记忆真正持久化InMemoryStore 只适合本地调试程序重启后数据就会丢失。如果希望长期记忆落库可以使用 PostgresStore。test2.py 如下fromlanggraph.store.postgresimportPostgresStorefromlanggraph.checkpoint.postgresimportPostgresSaverwith(PostgresSaver.from_conn_string(DB_URI)ascheckpointer,PostgresStore.from_conn_string(DB_URI)asstore,):store.setup()store.put(namespacenamespace,keymemory_id,valuememory_value)生产环境中更推荐这样写DB_URIos.environ[POSTGRES_URI]with(PostgresSaver.from_conn_string(DB_URI)ascheckpointer,PostgresStore.from_conn_string(DB_URI)asstore,):checkpointer.setup()store.setup()PostgresSaver 负责保存图的 checkpoint。PostgresStore 负责保存长期记忆。两者职责不同但可以使用同一个 PostgreSQL 数据库。七、跨线程读取用户记忆test3.py 最接近真实业务示例。第一轮对话config{configurable:{thread_id:1111,user_id:user_123}}responseagent_search.invoke({messages:[HumanMessage(content我叫小明我喜欢吃川菜)]},config)第二轮对话换了新的 thread_idconfig2{configurable:{thread_id:2222,user_id:user_123}}responseagent_search.invoke({messages:[HumanMessage(content给我推荐下相关餐厅)]},config2)这里最重要的是thread_id 变了但 user_id 没变。这意味着第二轮对话虽然是一个新会话但仍然可以通过 store 读取 user_123 的长期记忆比如姓名和饮食偏好。八、用结构化输出抽取用户信息test3.py 中定义了一个 Person 模型classPerson(BaseModel):name:Optional[str]Field(defaultNone,description人的姓名)height:Optional[str]Field(defaultNone,description人的身高以米为单位)favourite_food:Optional[list[str]]Field(defaultNone,description最喜欢的食物的列表)然后通过model_with_structured model.with_structured_output(Person)让大模型按照固定结构抽取用户信息。抽取节点如下defget_person_by_llm(state,config,*,store):people_infomodel_with_structured.invoke([SystemMessage(content你是一个提取信息的专家只从文本中提取用户本人的相关信息不能提取别人的信息。不知道的值返回null。)]state[messages][-3:])user_idconfig[configurable][user_id]namespace1(user_id,info)namespace2(user_id,prefs)store.put(namespace1,user_base,{name:people_info.name,height:people_info.height,})store.put(namespace2,user_food,{favourite_food:people_info.favourite_food,})这里没有使用随机 UUID而是使用固定 key“user_base”“user_food”这样做的好处是同一类信息可以被覆盖更新不会无限生成重复记忆。例如用户第一次说“我喜欢吃川菜”后面又说“我现在更喜欢粤菜”就可以覆盖更新偏好而不是存成两条互相冲突的记忆。九、在 LLM 调用前注入长期记忆抽取到用户信息后需要在下一次调用模型前把记忆注入上下文。test3.py 中的 llm_call 做了这件事defllm_call(state,config,*,store):user_idconfig[configurable][user_id]namespace1(user_id,info)namespace2(user_id,prefs)info_resultstore.search(namespace1)prefs_resultstore.search(namespace2)user_infoinfo_result[0].valueifinfo_resultelse暂无用户基础信息user_prefsprefs_result[0].valueifprefs_resultelse暂无用户饮食偏好然后把这些记忆拼到模型输入里resultmodel_with_tools.invoke([SystemMessage(contentsys_prompt)][HumanMessage(contentf调用聊天模型前可以参考以下信息用户基本情况{user_info}用户偏好情况{user_prefs})]messages)这样模型在回答“给我推荐下相关餐厅”时就能参考之前保存的“用户喜欢吃川菜”。这就是典型的长期记忆工作流从用户消息中抽取信息。写入长期记忆 store。新会话开始时按 user_id 查找记忆。把记忆注入模型上下文。模型生成个性化回答。十、消息过多怎么办裁剪与删除仅靠 checkpoint 保存完整消息会遇到一个问题对话越长消息越多最终 prompt 可能超过模型上下文窗口。test4.py 演示了两种处理方式。第一种是调用模型前进行消息裁剪messages trim_messages(state[“messages”],strategy“last”,max_tokens128,token_counterdeepseek_token_counter,start_on“human”,end_on(“human”, “tool”))这表示只保留最后一部分消息并控制在指定 token 范围内。第二种是从状态中删除旧消息iflen(state[messages])6:return{messages:[RemoveMessage(idmsg.id)formsginstate[messages]][:6]}RemoveMessage 是 LangGraph 中管理消息状态的常见方式。它不是简单地在 Python list 里 pop而是通过状态更新告诉 LangGraph 删除指定消息。需要注意test4.py 中出现了两个同名 call_model 函数后一个会覆盖前一个。实际项目里建议把它们拆成两个不同函数例如call_model_with_trim和call_model_with_remove。十一、摘要压缩删除前先总结直接删除历史消息虽然能节省 token但可能丢失关键信息。因此 test5.py 使用了更合理的方式先生成摘要再删除旧消息。状态中增加一个 summary 字段classState(MessagesState):summary:str模型调用时如果已有摘要就把摘要加入上下文defcall_model(state:State):summarystate.get(summary,)all_messages[]ifsummary:all_messages.append(HumanMessage(contentsummary))all_messages.extend(state[messages])respmodel.invoke(all_messages)return{messages:[resp]}当消息数量超过 6 条时进入摘要节点defshould_summarize(state:State):iflen(state[messages])6:returnsummarize_conversationreturnEND摘要节点会基于历史消息生成新的 summarydefsummarize_conversation(state:State):summarystate.get(summary,)ifsummary:summary_message(f这是目前为止的对话摘要{summary}\n基于上面的消息来扩展摘要:)else:summary_message创建关于上面对话的完整摘要resultmodel.invoke(state[messages][HumanMessage(contentsummary_message)])return{summary:result.content,messages:[RemoveMessage(idm.id)forminstate[messages][:-1]]}这样既删除了大部分历史消息又把关键信息保存到了 summary 字段中。最终效果是短期内保留最近几轮原始对话。长期保留历史摘要。避免 prompt 越来越长。用户问“what’s my name?” 时仍然可以从摘要里找回 “bob”。十二、完整记忆架构总结结合这几个示例可以把 LangGraph 的记忆模块理解成下面这张图是否是用户输入LangGraph StateGraph提取用户信息写入 Store 长期记忆LLM 节点是否需要工具调用Tool Node返回回答Checkpointer 保存线程状态消息是否过多摘要压缩或删除旧消息其中StateGraph 负责编排智能体流程。checkpointer 保存线程级上下文。thread_id 区分不同会话。store 保存长期记忆。user_id 用来跨线程关联同一个用户。RemoveMessage 和 summary 用来控制上下文长度。PostgresSaver 和 PostgresStore 用来实现生产级持久化。十三、开发LangGraph 记忆模块中的建议本地调试用 InMemorySaver 和 InMemoryStore 即可简单方便。生产环境使用 PostgresSaver 和 PostgresStore否则程序重启后记忆会丢失。thread_id 用来保存单次会话上下文user_id 用来关联长期用户画像。用户偏好、姓名、身高等稳定信息适合放到 store。最近几轮聊天适合交给 checkpointer 管理。对话变长后不要无限追加历史消息要做裁剪或摘要。抽取长期记忆时推荐使用结构化输出减少脏数据。存储记忆时固定 key 适合覆盖更新UUID key 适合保存多条独立事实。不要把数据库账号、API Key 写死在代码中。十四、结语LangGraph 的记忆模块不是单一组件而是一套组合能力。如果只需要让同一轮会话连续起来用 checkpointer thread_id 就够了。如果希望智能体跨会话记住用户偏好就需要引入 store user_id。如果对话会持续很久还要加入 trim_messages、RemoveMessage 或摘要压缩。参考文档写的demo的 test.py 到 test5.py 正好展示了这个递进过程test.pyPostgres checkpoint、状态恢复、历史快照、重放。test2.py长期记忆 store、命名空间、语义搜索、Postgres 持久化。test3.py结构化抽取用户信息并跨线程读取长期记忆。test4.py消息裁剪和消息删除。test5.py摘要压缩删除旧消息前先保留关键信息。掌握这几种模式后就可以把 LangGraph Agent 从“单轮问答机器人”升级成真正具备上下文记忆和用户画像能力的智能体。参考文档官方文档LangChainDocs实践demo: gitte仓库