LangChain核心组件之Short-term memory

概述

记忆是一种用于记录先前交互信息的系统。对于 AI Agent（智能体）而言，记忆至关重要——它使智能体能够记住过往对话、从用户反馈中学习，并适应用户的偏好。随着智能体处理的任务越来越复杂、交互轮次越来越多，这种能力对提升效率和用户体验变得不可或缺。

短期记忆允许你的应用在单个线程（thread）或一次会话（conversation）中记住之前的交互内容。

注：
一个线程（thread）用于组织一次会话中的多次交互，类似于电子邮件中将多条消息归为同一对话的方式。

最常见的短期记忆形式就是对话历史。然而，长对话对当前的大语言模型（LLM）构成了挑战：完整的对话历史可能超出 LLM 的上下文窗口（context window），导致上下文丢失或出错。

即使你的模型支持超长上下文，大多数 LLM 在处理过长上下文时表现依然不佳——它们容易被陈旧或无关的内容“干扰”，同时还会带来响应变慢和成本上升的问题。

聊天模型通过消息（messages）接收上下文，这些消息包括指令（system message）和用户输入（human message）。在聊天应用中，消息通常在用户输入与模型回复之间交替出现，随着时间推移形成一个不断增长的消息列表。由于上下文窗口有限，许多应用都需要采用一些策略来移除或“遗忘”过时的信息。

使用方法

要为智能体添加短期记忆（即线程级别的持久化能力），你需要在创建智能体时指定一个 checkpointer。

说明：
LangChain 的智能体将短期记忆作为其状态（state）的一部分进行管理。
通过将这些状态存储在图（graph）中，智能体可以在保持不同线程隔离的同时，访问特定会话的完整上下文。
状态通过 checkpointer 持久化到数据库（或内存）中，使得线程可以在任意时刻恢复。
每次调用智能体或完成一个步骤（如工具调用）后，短期记忆会被更新；而在每个步骤开始时，状态会被读取。

fromlangchain.agentsimportcreate_agentfromlanggraph.checkpoint.memoryimportInMemorySaver agent=create_agent("gpt-5",tools=[get_user_info],checkpointer=InMemorySaver(),)agent.invoke({"messages":[{"role":"user","content":"Hi! My name is Bob."}]},{"configurable":{"thread_id":"1"}},)

在生产环境中

在生产环境中，应使用基于数据库的 checkpointer：

pipinstalllanggraph-checkpoint-postgres

fromlangchain.agentsimportcreate_agentfromlanggraph.checkpoint.postgresimportPostgresSaver DB_URI="postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"withPostgresSaver.from_conn_string(DB_URI)ascheckpointer:checkpointer.setup()# 自动在 PostgreSQL 中创建所需表agent=create_agent("gpt-5",tools=[get_user_info],checkpointer=checkpointer,)

自定义智能体记忆

默认情况下，智能体使用 AgentState 来管理短期记忆，特别是通过 messages 键保存对话历史。

你可以通过继承 AgentState 来添加自定义字段。自定义的状态结构（state schema）可通过 create_agent 的 state_schema 参数传入。

fromlangchain.agentsimportcreate_agent,AgentStatefromlanggraph.checkpoint.memoryimportInMemorySaverclassCustomAgentState(AgentState):user_id:strpreferences:dictagent=create_agent("gpt-5",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"}})

常见模式

启用短期记忆后，长对话仍可能超出 LLM 的上下文窗口。常见解决方案包括：

移除最前或最后 N 条消息（在调用 LLM 前）从 LangGraph 状态中永久删除消息将早期消息摘要后替换原消息自定义策略（如消息过滤等）
这些方法可帮助智能体在不超出上下文限制的前提下，持续跟踪对话上下文。

裁剪消息（Trim messages）

大多数 LLM 都有最大上下文长度限制（以 token 计）。

一种判断何时裁剪消息的方法是：统计消息历史中的 token 数量，当接近上限时进行截断。如果你使用 LangChain，可以借助内置的 trim messages 工具，指定保留多少 token，并选择裁剪策略（例如保留最近的 max_tokens）。

要在智能体中裁剪消息历史，可使用 @before_model 中间件装饰器：

fromlangchain.messagesimportRemoveMessagefromlanggraph.graph.messageimportREMOVE_ALL_MESSAGESfromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain.agents.middlewareimportbefore_modelfromlanggraph.runtimeimportRuntimefromlangchain_core.runnablesimportRunnableConfigfromtypingimportAny@before_modeldeftrim_messages(state:AgentState,runtime:Runtime)->dict[str,Any]|None:"""仅保留最近几条消息，以适配上下文窗口。"""messages=state["messages"]iflen(messages)<=3:returnNone# 无需修改first_msg=messages[0]recent_messages=messages[-3:]iflen(messages)%2==0elsemessages[-4:]new_messages=[first_msg]+recent_messagesreturn{"messages":[RemoveMessage(id=REMOVE_ALL_MESSAGES),*new_messages]}agent=create_agent(your_model_here,tools=your_tools_here,middleware=[trim_messages],checkpointer=InMemorySaver(),)config:RunnableConfig={"configurable":{"thread_id":"1"}}agent.invoke({"messages":"hi, my name is bob"},config)agent.invoke({"messages":"write a short poem about cats"},config)agent.invoke({"messages":"now do the same but for dogs"},config)final_response=agent.invoke({"messages":"what's my name?"},config)final_response["messages"][-1].pretty_print()""" ================================== Ai Message ================================== Your name is Bob. You told me that earlier. If you'd like me to call you a nickname or use a different name, just say the word. """

删除消息（Delete messages）

你可以从图状态中删除消息，以管理消息历史。这在需要移除特定消息或清空整个历史时非常有用。

要删除消息，需使用 RemoveMessage。

注意：RemoveMessage 要求状态键使用 add_messages 这个 reducer（归约器）。

默认的 AgentState 已经提供了该功能。

删除特定消息：

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)]}

警告：
删除消息时，请确保剩余的消息历史仍然有效。请参考你所用 LLM 提供商的限制，例如：
某些提供商要求消息历史必须以 user 消息开头；
大多数提供商要求带有工具调用的 assistant 消息后必须紧跟对应的 tool 结果消息。

示例：

# 使用 @after_model 中间件在模型调用后删除旧消息@after_modeldefdelete_old_messages(state:AgentState,runtime:Runtime)->dict|None:messages=state["messages"]iflen(messages)>2:return{"messages":[RemoveMessage(id=m.id)forminmessages[:2]]}returnNone# ...（后续调用略）

输出示例显示：随着新消息加入，旧消息被逐步移除，但关键信息（如用户名）仍被保留。

摘要消息（Summarize messages）
上述裁剪或删除消息的方法可能导致重要信息丢失。因此，某些应用更适合采用更高级的策略：使用 LLM 对历史消息进行摘要。

LangChain 提供了内置的 SummarizationMiddleware 来实现这一功能：

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportSummarizationMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver agent=create_agent(model="gpt-4o",tools=[],middleware=[SummarizationMiddleware(model="gpt-4o-mini",trigger=("tokens",4000),# 当 token 数超过 4000 时触发摘要keep=("messages",20)# 保留最近 20 条消息)],checkpointer=InMemorySaver(),)config={"configurable":{"thread_id":"1"}}agent.invoke({"messages":"hi, my name is bob"},config)# ... 多轮交互后final_response=agent.invoke({"messages":"what's my name?"},config)final_response["messages"][-1].pretty_print()""" ================================== Ai Message ================================== Your name is Bob! """

更多配置选项请参阅 SummarizationMiddleware 文档。

访问记忆

你可以通过多种方式访问和修改智能体的短期记忆（即状态）：

在工具（Tools）中

读取短期记忆

在工具函数中，可通过 runtime 参数（类型为 ToolRuntime）访问当前状态。

该参数对模型不可见（不会出现在工具签名中），但工具内部可读取状态。

fromlangchain.agentsimportcreate_agent,AgentStatefromlangchain.toolsimporttool,ToolRuntimeclassCustomState(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"agent=create_agent(model="gpt-5-nano",tools=[get_user_info],state_schema=CustomState,)result=agent.invoke({"messages":"look up user information","user_id":"user_123"})print(result["messages"][-1].content)# > User is John Smith.

写入短期记忆

工具可通过返回 Command 对象直接更新智能体状态，适用于保存中间结果或供后续步骤使用。

fromlangchain.toolsimporttool,ToolRuntimefromlangchain_core.runnablesimportRunnableConfigfromlangchain.messagesimportToolMessagefromlangchain.agentsimportcreate_agent,AgentStatefromlanggraph.typesimportCommandfrompydanticimportBaseModelclassCustomState(AgentState):user_name:strclassCustomContext(BaseModel):user_id:str@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,# update the message history"messages":[ToolMessage("Successfully looked up user information",tool_call_id=runtime.tool_call_id)]})@tooldefgreet(runtime:ToolRuntime[CustomContext,CustomState])->str|Command:"""Use this to greet the user once you found their info."""user_name=runtime.state.get("user_name",None)ifuser_nameisNone:returnCommand(update={"messages":[ToolMessage("Please call the 'update_user_info' tool it will get and update the user's name.",tool_call_id=runtime.tool_call_id)]})returnf"Hello{user_name}!"agent=create_agent(model="gpt-5-nano",tools=[update_user_info,greet],state_schema=CustomState,context_schema=CustomContext,)agent.invoke({"messages":[{"role":"user","content":"greet the user"}]},context=CustomContext(user_id="user_123"),)

在提示词（Prompt）中

通过中间件动态生成系统提示词，可基于对话历史或自定义状态字段构建个性化 prompt。

fromlangchain.agentsimportcreate_agentfromtypingimportTypedDictfromlangchain.agents.middlewareimportdynamic_prompt,ModelRequestclassCustomContext(TypedDict):user_name:strdefget_weather(city:str)->str:"""Get the weather in a city."""returnf"The weather in{city}is always sunny!"@dynamic_promptdefdynamic_system_prompt(request:ModelRequest)->str:user_name=request.runtime.context["user_name"]system_prompt=f"You are a helpful assistant. Address the user as{user_name}."returnsystem_prompt agent=create_agent(model="gpt-5-nano",tools=[get_weather],middleware=[dynamic_system_prompt],context_schema=CustomContext,)result=agent.invoke({"messages":[{"role":"user","content":"What is the weather in SF?"}]},context=CustomContext(user_name="John Smith"),)formsginresult["messages"]:msg.pretty_print()

在模型调用前（@before_model）

可在模型调用前处理消息，例如裁剪、过滤或注入上下文。

fromlangchain.messagesimportRemoveMessagefromlanggraph.graph.messageimportREMOVE_ALL_MESSAGESfromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain.agents.middlewareimportbefore_modelfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.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:returnNone# No changes neededfirst_msg=messages[0]recent_messages=messages[-3:]iflen(messages)%2==0elsemessages[-4:]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())config:RunnableConfig={"configurable":{"thread_id":"1"}}agent.invoke({"messages":"hi, my name is bob"},config)agent.invoke({"messages":"write a short poem about cats"},config)agent.invoke({"messages":"now do the same but for dogs"},config)final_response=agent.invoke({"messages":"what's my name?"},config)final_response["messages"][-1].pretty_print()""" ================================== Ai Message ================================== Your name is Bob. You told me that earlier. If you'd like me to call you a nickname or use a different name, just say the word. """

在模型调用后（@after_model）

可在模型生成回复后进行后处理，例如过滤敏感内容：

fromlangchain.messagesimportRemoveMessagefromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.agentsimportcreate_agent,AgentStatefromlangchain.agents.middlewareimportafter_modelfromlanggraph.runtimeimportRuntime@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)]}returnNoneagent=create_agent(model="gpt-5-nano",tools=[],middleware=[validate_response],checkpointer=InMemorySaver(),)