LangChain 短期记忆 --(Short-term Memory)

一、短期记忆是什么？

▎ 短期记忆让您的应用程序能够记住单个线程或对话中的先前交互。

打个比方：短期记忆就是"这一通电话里的记性"——你跟智能体在这一个对话中聊过的所有内容，它都记得。但换一个对话（挂断重拨）
，就从头开始。

关键概念：线程（thread）

▎ 线程将一次会话中的多次交互组织起来，类似于电子邮件将消息归入单个对话中的方式。

一句话：一个线程 = 一次连续对话。短期记忆就是"按线程保存的记忆"。

二、为什么需要专门管理短期记忆？

几个痛点：

1. 上下文窗口有限：完整对话历史可能塞不进 LLM 的上下文窗口，导致"上下文丢失或错误"。
2. 长上下文表现变差：即使模型支持长上下文，也会因陈旧/无关内容"分心"，且响应变慢、成本增加。
3. 消息会不断增长：聊天中人类输入和模型回复交替，消息列表越来越长。

所以需要"移除或遗忘陈旧信息"的技术来管理记忆。

▎ 区分：要记住跨对话的信息，请用长期记忆（store），本文讲的是短期记忆（单线程内）。

三、怎么启用短期记忆？—— checkpointer

核心一句话：

▎ 要为智能体添加短期记忆，需要在创建智能体时指定一个 checkpointer（检查点器）。

checkpointer 就是负责"把状态存下来、下次能恢复"的组件。最简单的内存版：

fromlangchain.agentsimportcreate_agentfromlanggraph.checkpoint.memoryimportInMemorySaver agent=create_agent("gpt-5.4",tools=[get_user_info],checkpointer=InMemorySaver(),# 关键：加 checkpointer)# 调用时用 thread_id 标识"这是哪个对话"agent.invoke({"messages":[{"role":"user","content":"Hi! My name is Bob."}]},{"configurable":{"thread_id":"1"}},# 线程 ID)

两个关键点：

• checkpointer：决定记忆存哪（内存/数据库）
• thread_id：标识哪个对话线程，相同 thread_id = 同一个对话

生产环境：用数据库版 checkpointer

InMemorySaver 重启就丢，生产要用持久化的，比如 Postgres：

pip install langgraph-checkpoint-postgresfromlanggraph.checkpoint.postgresimportPostgresSaver DB_URI="postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"withPostgresSaver.from_conn_string(DB_URI)ascheckpointer:checkpointer.setup()# 自动建表agent=create_agent("gpt-5.4",tools=[get_user_info],checkpointer=checkpointer,)

短期记忆的工作机制

• 短期记忆作为智能体状态的一部分管理，存在图状态里
• 通过 checkpointer 持久化到数据库（或内存），线程可随时恢复
• 更新时机：调用智能体或完成一步（如工具调用）时更新
• 读取时机：每一步开始时读取状态

四、自定义智能体记忆

默认智能体用 AgentState 管理短期记忆（通过 messages 键存对话历史）。你可以扩展 AgentState 添加额外字段：

fromlangchain.agentsimportcreate_agent,AgentStatefromlanggraph.checkpoint.memoryimportInMemorySaverclassCustomAgentState(AgentState):user_id:strpreferences:dictagent=create_agent("gpt-5.4",tools=[get_user_info],state_schema=CustomAgentState,# 传入自定义状态checkpointer=InMemorySaver(),)# 调用时可以传入自定义状态字段result=agent.invoke({"messages":[{"role":"user","content":"Hello"}],"user_id":"user_123","preferences":{"theme":"dark"}},{"configurable":{"thread_id":"1"}})

这样除了对话历史，还能在这个线程里记住 user_id、preferences 等额外信息。

五、长对话的管理策略（核心！）

启用短期记忆后，长对话可能超出上下文窗口。页面给出四种常见策略：

策略 1：截断消息（@before_model 中间件）

用 @before_model 中间件，在调用模型之前修剪消息历史。思路是：保留第一条消息 + 最近几条，中间的丢掉：

fromlangchain.messagesimportRemoveMessagefromlanggraph.graph.messageimportREMOVE_ALL_MESSAGESfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain.agents.middlewareimportbefore_modelfromlanggraph.runtimeimportRuntimefromtypingimportAny@before_modeldeftrim_messages(state:AgentState,runtime:Runtime)->dict[str,Any]|None:"""Keep only the last few messages to fit context window."""messages=state["messages"]iflen(messages)<=3:# 消息不多就不用动returnNonefirst_msg=messages[0]# 保留第一条recent_messages=messages[-3:]# 保留最近几条new_messages=[first_msg]+recent_messagesreturn{"messages":[RemoveMessage(id=REMOVE_ALL_MESSAGES),# 先清空*new_messages# 再放回保留的]}agent=create_agent("gpt-5-nano",tools=[],middleware=[trim_messages],# 挂上中间件checkpointer=InMemorySaver(),)

效果：即使中间聊了很多轮，模型也只看到"开头 + 最近几条"

策略 2：删除消息（RemoveMessage）

用 RemoveMessage 删除特定消息或全部消息。需要状态键带 add_messages reducer（默认 AgentState 已提供）。

删除特定消息（最早的 2 条）：

fromlangchain.messagesimportRemoveMessagedefdelete_messages(state):messages=state["messages"]iflen(messages)>2:return{"messages":[RemoveMessage(id=m.id)forminmessages[:2]]}

删除所有消息：

fromlanggraph.graph.messageimportREMOVE_ALL_MESSAGESdefdelete_messages(state):return{"messages":[RemoveMessage(id=REMOVE_ALL_MESSAGES)]}

⚠️ 重要提醒：删除消息后要确保消息历史仍然有效，否则模型会报错：

• 某些厂商要求消息历史以 user 消息开头
• 大多数厂商要求带工具调用的 assistant 消息后必须跟对应的 tool 结果消息（不能拆散）

实战中常用 @after_model 中间件在模型回复后删除旧消息：

fromlangchain.agents.middlewareimportafter_model@after_modeldefdelete_old_messages(state:AgentState,runtime:Runtime)->dict|None:"""Remove old messages to keep conversation manageable."""messages=state["messages"]iflen(messages)>2:return{"messages":[RemoveMessage(id=m.id)forminmessages[:2]]}returnNoneagent=create_agent("gpt-5-nano",tools=[],system_prompt="Please be concise and to the point.",middleware=[delete_old_messages],checkpointer=InMemorySaver(),)

策略 3：总结消息（SummarizationMiddleware）

截断/删除的问题：会丢失信息。更聪明的方法是用模型把早期消息总结成摘要，再用摘要替换它们。

LangChain 提供内置的 SummarizationMiddleware：

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportSummarizationMiddleware agent=create_agent(model="gpt-5.4",tools=[],middleware=[SummarizationMiddleware(model="gpt-5.4-mini",# 用小模型做总结，省钱trigger=("tokens",4000),# 触发条件：token 达到 4000keep=("messages",20)# 保留最近 20 条消息不总结)],checkpointer=InMemorySaver(),)

三个参数：

• model：用哪个模型做总结（推荐用便宜的小模型）
• trigger：什么时候触发总结（这里设为 token 数到 4000）
• keep：保留多少条最近消息不被总结

效果：长对话中，早期内容被压缩成摘要，模型既能看到历史梗概，又不会超出窗口

六、访问短期记忆的三种途径

短期记忆（状态）可以通过三个地方访问和修改：

途径 1：工具里访问（runtime.state）

用 ToolRuntime 参数在工具中读写状态（对模型隐藏）：

fromlangchain.toolsimporttool,ToolRuntimefromlangchain.agentsimportcreate_agent,AgentStateclassCustomState(AgentState):user_id:str@tooldefget_user_info(runtime:ToolRuntime)->str:"""Look up user info."""user_id=runtime.state["user_id"]# 读短期记忆return"User is John Smith"ifuser_id=="user_123"else"Unknown user"

工具还能写入短期记忆——直接返回 Command(update={…})：

fromlanggraph.typesimportCommandfromlangchain.messagesimportToolMessage@tooldefupdate_user_info(runtime:ToolRuntime[CustomContext,CustomState])->Command:"""Look up and update user info."""user_id=runtime.context.user_id name="John Smith"ifuser_id=="user_123"else"Unknown user"returnCommand(update={"user_name":name,# 写入自定义状态字段"messages":[ToolMessage("Successfully looked up user information",tool_call_id=runtime.tool_call_id)]})

途径 2：提示里访问（@dynamic_prompt 中间件）

根据状态/上下文生成动态系统提示：

fromlangchain.agents.middlewareimportdynamic_prompt,ModelRequestclassCustomContext(TypedDict):user_name:str@dynamic_promptdefdynamic_system_prompt(request:ModelRequest)->str:user_name=request.runtime.context["user_name"]returnf"You are a helpful assistant. Address the user as{user_name}."agent=create_agent(model="gpt-5-nano",tools=[get_weather],middleware=[dynamic_system_prompt],context_schema=CustomContext,)

效果：系统提示会变成 “Address the user as John Smith”，模型回复时就会称呼用户名字。

途径 3：模型执行前/后访问（@before_model / @after_model）

• @before_model：调用模型前处理消息（如截断历史）
• @after_model：调用模型后处理消息（如删除含敏感词的回复）

fromlangchain.agents.middlewareimportafter_model@after_modeldefvalidate_response(state:AgentState,runtime:Runtime)->dict|None:"""Remove messages containing sensitive words."""STOP_WORDS=["password","secret"]last_message=state["messages"][-1]ifany(wordinlast_message.contentforwordinSTOP_WORDS):return{"messages":[RemoveMessage(id=last_message.id)]}returnNone

七、访问记忆途径总览

八、一句话总结

▎ 短期记忆 = 单个线程（对话）内的记忆，靠 checkpointer + thread_id 实现"同一对话记得住、能恢复"。
▎ 长对话靠四种策略管理：截断 / 删除 / 总结 /自定义。
▎ 访问记忆有四个切入点：工具（runtime.state）、动态提示（@dynamic_prompt）、模型执行前（@before_model）、模型执行后（@after_model）。

九、和前几讲的关系

把这几讲串起来：

• 短期记忆：线程内对话历史，存 state[“messages”]，靠 checkpointer 持久化
• 长期记忆（store）：跨线程永久保存，靠 runtime.store
• 工具访问上下文：工具用 runtime.state / runtime.context / runtime.store 读写这些记忆
• 消息（messages）：短期记忆的载体，四种类型组成对话历史

一句话：短期记忆是"这次通话的草稿纸"，checkpointer 是"草稿纸的存档"，thread_id 是"通话编号"。