概览
Memory 是一个记住先前交互信息的系统。对于 AI 代理,memory 至关重要,因为它让代理能够记住先前交互、从反馈中学习,并适应用户偏好。随着代理处理包含大量用户交互的更复杂任务,这种能力对效率和用户满意度都变得必不可少。 短期记忆让你的应用能够记住单个线程或对话中的先前交互。线程会组织会话中的多次交互,类似电子邮件将消息分组到单个对话中的方式。
需要跨对话记住信息?使用 long-term memory 在不同线程和会话之间存储并召回用户特定或应用级数据。
用法
要向代理添加短期记忆(线程级持久化),需要在创建代理时指定checkpointer。
LangChain 的代理会将短期记忆作为代理状态的一部分进行管理。通过将这些内容存储在图状态中,代理可以访问给定对话的完整上下文,同时保持不同线程之间的隔离。状态会使用 checkpointer 持久化到数据库(或内存)中,因此线程可以随时恢复。当代理被调用或某个步骤(例如工具调用)完成时,短期记忆会更新,并且每个步骤开始时都会读取状态。
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
def get_user_info() -> str:
"""Look up information about the current user."""
return "No user profile on file."
agent = create_agent(
model="google_genai:gemini-3.5-flash",
tools=[get_user_info],
checkpointer=InMemorySaver(),
)
thread_config = {"configurable": {"thread_id": "1"}}
response = agent.invoke(
{"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
thread_config,
)["messages"][-1].content
print(response) # "Hi Bob! Nice to see you here. How are you doing?"
response = agent.invoke(
{"messages": [{"role": "user", "content": "What's my name?"}]},
thread_config,
)["messages"][-1].content
print(response) # "You are Bob!"
在生产环境中
在生产环境中,使用由数据库支持的 checkpointer:pip install langgraph-checkpoint-postgres
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
def get_user_info() -> str:
"""Look up information about the current user."""
return "No user profile on file."
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup() # auto create tables in PostgreSQL
agent = create_agent(
"gpt-5.5",
tools=[get_user_info],
checkpointer=checkpointer,
)
如需更多 checkpointer 选项,包括 SQLite、Postgres 和 Azure Cosmos DB,请参阅 Persistence 文档中的 checkpointer libraries 列表。
自定义代理记忆
默认情况下,代理使用AgentState 管理短期记忆,具体来说是通过 messages 键管理对话历史。
可以扩展 AgentState 以添加其他字段。自定义状态 schemas 使用 state_schema 参数传给 create_agent。
from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
class CustomAgentState(AgentState):
user_id: str
preferences: dict
agent = create_agent(
"gpt-5.5",
tools=[get_user_info],
state_schema=CustomAgentState,
checkpointer=InMemorySaver(),
)
# Custom state can be passed in invoke
result = agent.invoke(
{
"messages": [{"role": "user", "content": "Hello"}],
"user_id": "user_123",
"preferences": {"theme": "dark"}
},
{"configurable": {"thread_id": "1"}})
常见模式
启用短期记忆后,长对话可能超过 LLM 的上下文窗口。常见解决方案包括:裁剪消息
移除最前或最后 N 条消息(在调用 LLM 前)
删除消息
从 LangGraph 状态中永久删除消息
总结消息
总结历史中的较早消息,并用摘要替换它们
自定义策略
自定义策略(例如消息过滤等)
裁剪消息
大多数 LLM 都有最大支持上下文窗口(以 tokens 计)。 决定何时截断消息的一种方法是统计消息历史中的 tokens,并在接近限制时截断。如果你使用 LangChain,可以使用 trim messages 工具,并指定要从列表中保留的 token 数,以及用于处理边界的strategy(例如保留最后 max_tokens)。
要在代理中裁剪消息历史,请使用 @before_model middleware decorator:
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from typing import Any
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
"""Keep only the last few messages to fit context window."""
messages = state["messages"]
if len(messages) <= 3:
return None # No changes needed
first_msg = messages[0]
recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
new_messages = [first_msg] + recent_messages
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
*new_messages
]
}
agent = create_agent(
"gpt-5.5",
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.
"""
删除消息
可以从图状态中删除消息来管理消息历史。 当你想移除特定消息或清空整个消息历史时,这很有用。 要从图状态中删除消息,可以使用RemoveMessage。
要让 RemoveMessage 工作,需要使用带 add_messages reducer 的状态键。
默认 AgentState 已提供这一点。
要移除特定消息:
from langchain.messages import RemoveMessage
def delete_messages(state):
messages = state["messages"]
if len(messages) > 2:
# remove the earliest two messages
return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
from langgraph.graph.message import REMOVE_ALL_MESSAGES
def delete_messages(state):
return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}
删除消息时,务必确保生成的消息历史有效。检查所用 LLM provider 的限制。例如:
- 有些 providers 要求消息历史以
user消息开始 - 大多数 providers 要求带工具调用的
assistant消息后面跟着对应的tool结果消息。
from langchain.messages import RemoveMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
"""Remove old messages to keep conversation manageable."""
messages = state["messages"]
if len(messages) > 2:
# remove the earliest two messages
return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
return None
agent = create_agent(
"gpt-5-nano",
tools=[...],
system_prompt="Please be concise and to the point.",
middleware=[delete_old_messages],
checkpointer=InMemorySaver(),
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
for event in agent.stream(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
stream_mode="values",
):
print([(message.type, message.content) for message in event["messages"]])
for event in agent.stream(
{"messages": [{"role": "user", "content": "write a short poem about cats"}]},
config,
stream_mode="values",
):
print([(message.type, message.content) for message in event["messages"]])
for event in agent.stream(
{"messages": [{"role": "user", "content": "what's my name?"}]},
config,
stream_mode="values",
):
print([(message.type, message.content) for message in event["messages"]])
[('human', "hi! I'm bob")]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.')]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.'), ('human', "write a short poem about cats")]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today? I can answer questions, brainstorm ideas, draft text, explain things, or help with code.'), ('human', "write a short poem about cats"), ('ai', 'There once was a cat on a wall, Who barely moved at all...')]
[('human', 'write a short poem about cats'), ('ai', 'There once was a cat on a wall, Who barely moved at all...')]
[('human', 'write a short poem about cats'), ('ai', 'There once was a cat on a wall, Who barely moved at all...'), ('human', "what's my name?")]
[('human', 'write a short poem about cats'), ('ai', 'There once was a cat on a wall, Who barely moved at all...'), ('human', "what's my name?"), ('ai', "I don't know your name - you haven't told me!")]
[('human', "what's my name?"), ('ai', "I don't know your name - you haven't told me!")]
总结消息
如上所示,裁剪或移除消息的问题在于,你可能会因为削减消息队列而丢失信息。 因此,一些应用可以受益于更复杂的方法:使用聊天模型总结消息历史。
要在代理中总结消息历史,请使用内置 SummarizationMiddleware:
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
checkpointer = InMemorySaver()
agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[
SummarizationMiddleware(
model="gpt-5.4-mini",
trigger=("tokens", 4000),
keep=("messages", 20)
)
],
checkpointer=checkpointer,
)
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!
"""
SummarizationMiddleware。
访问 memory
可以通过多种方式访问和修改代理的短期记忆(状态):Tools
在工具中读取短期记忆
在工具中使用runtime 参数(类型为 ToolRuntime)访问短期记忆(状态)。
runtime 参数对工具签名隐藏(因此模型看不到它),但工具可以通过它访问状态。
from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime
class CustomState(AgentState):
user_id: str
@tool
def get_user_info(
runtime: ToolRuntime
) -> str:
"""Look up user info."""
user_id = runtime.state["user_id"]
return "User is John Smith" if user_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.
从工具写入短期记忆
要在执行期间修改代理的短期记忆(状态),可以直接从工具返回状态更新。 这适合持久化中间结果,或让后续工具或 prompts 可以访问某些信息。from langchain.tools import tool, ToolRuntime
from langchain_core.runnables import RunnableConfig
from langchain.messages import ToolMessage
from langchain.agents import create_agent, AgentState
from langgraph.types import Command
from pydantic import BaseModel
class CustomState(AgentState):
user_name: str
class CustomContext(BaseModel):
user_id: str
@tool
def update_user_info(
runtime: ToolRuntime[CustomContext, CustomState],
) -> Command:
"""Look up and update user info."""
user_id = runtime.context.user_id
name = "John Smith" if user_id == "user_123" else "Unknown user"
return Command(update={
"user_name": name,
# update the message history
"messages": [
ToolMessage(
"Successfully looked up user information",
tool_call_id=runtime.tool_call_id
)
]
})
@tool
def greet(
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)
if user_name is None:
return Command(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
)
]
})
return f"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
在 middleware 中访问短期记忆(状态),以基于对话历史或自定义状态字段创建动态 prompts。from langchain.agents import create_agent
from typing import TypedDict
from langchain.agents.middleware import dynamic_prompt, ModelRequest
class CustomContext(TypedDict):
user_name: str
def get_weather(city: str) -> str:
"""Get the weather in a city."""
return f"The weather in {city} is always sunny!"
@dynamic_prompt
def dynamic_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}."
return system_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"),
)
for msg in result["messages"]:
msg.pretty_print()
Output
================================ Human Message =================================
What is the weather in SF?
================================== Ai Message ==================================
Tool Calls:
get_weather (call_WFQlOGn4b2yoJrv7cih342FG)
Call ID: call_WFQlOGn4b2yoJrv7cih342FG
Args:
city: San Francisco
================================= Tool Message =================================
Name: get_weather
The weather in San Francisco is always sunny!
================================== Ai Message ==================================
Hi John Smith, the weather in San Francisco is always sunny!
Before model
在@before_model middleware 中访问短期记忆(状态),以在模型调用前处理消息。
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime
from typing import Any
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
"""Keep only the last few messages to fit context window."""
messages = state["messages"]
if len(messages) <= 3:
return None # No changes needed
first_msg = messages[0]
recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
new_messages = [first_msg] + recent_messages
return {
"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
在@after_model middleware 中访问短期记忆(状态),以在模型调用后处理消息。
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.runtime import Runtime
@after_model
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
"""Remove messages containing sensitive words."""
STOP_WORDS = ["password", "secret"]
last_message = state["messages"][-1]
if any(word in last_message.content for word in STOP_WORDS):
return {"messages": [RemoveMessage(id=last_message.id)]}
return None
agent = create_agent(
model="gpt-5-nano",
tools=[],
middleware=[validate_response],
checkpointer=InMemorySaver(),
)
Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

