Conversation history is the most common form of short-term memory
Fact1: message includes instructions and inpus
Usage#
To add short-term memory to an agent ,you need to specify a checkpointer when creating an agent
note: short-term memory is part of your agent’s state. State is persisted in a database using a checkpointer so thread can resume it an any time.
Short-term memory updates when the agent is invoked or a step (like a tool call) is completed, and the state is read at the start of each step.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| 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="openai:gpt-5.5",
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!"
|
In production#
use a checkpointer backed by a database
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| 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="openai:gpt-5.5",
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!"
|
Customizing agent memory#
Agents use AgentState to manage short-term memory, specifically the
conversation history via a message key
You can extend AgentState to ad additional fields
Custom state schema are passed to create_agent using the state_scheam parameter
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| 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"}})
|
Common patterns#
Common solutions to exceed the LLM’s context window
Trim message(截断消息)#
A way: keep track of the token count and truncate when it approaches that limit
Using trim messages utility and specify the number if tokens to keep ang the strategy to use for handing the boundary
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
@before_mfrom 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.
"""
|
Delete message#
Useful when you want to remove specific message or clear the entire message history
Use RemoveMessage to delete message from the graph state
and you need to use a state key with add_messages reducer and default theAgentState provides it
To remove specific messages:#
1
2
3
4
5
6
7
| 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]]}
|
To remove all messages:#
1
2
3
4
| from langgraph.graph.message import REMOVE_ALL_MESSAGES
def delete_messages(state):
return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}
|
ps:
删除消息时,请确保生成的消息历史有效。检查你正在使用的 LLM 提供商的限制。例如:
- 某些提供程序期望消息历史记录以
user 消息开头 - 大多数提供程序要求带有工具调用的
assistant 消息后跟相应的 tool 结果消息。
Summarize messages#
above pattern will lose information
Summarizing message with a LLM and use the built-in SUmmarizationMiddleware
1
2
3
4
5
6
7
8
9
10
11
12
| agent = create_agent(
model="gpt-5.5",
tools=[...],
middleware=[
SummarizationMiddleware(
model="gpt-5.4-mini",
trigger=("tokens", 4000),
keep=("messages", 20)
)
],
checkpointer=checkpointer,
)
|
Access memory#
Using runtime parameter(typed as ToolRuntime)
runtime parameter can not be seen by model but the tool can access the state through it
1
2
3
4
5
6
7
| @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"
|
Using return Command with a ToolMessage
Prompt#
Using middleware to access to ~ to create dynamic prompts on conversation history or custom state fields
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
| 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()
|
Before model#
Using @before_model middleware to process message before model calls
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
| 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
Using @after_model
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| 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(),
)
|