Module 6: Memory Systems for Agents
2. Short-term Memory: Conversation History
Overview
In the previous capsule you saw the types of memory an agent needs. Now you're going to solve the first and most immediate one: short-term memory — the active conversation. Every time the user sends a message and the agent replies, that exchange piles up in a message list that grows with every turn. The agent needs that list to stay coherent: if the user said "look up information about RAG" and then says "now compare it with fine-tuning," the agent needs to remember what "it" is.
The problem is that the list grows without limit. A 30-turn research run can generate 60+ messages with thousands of tokens. LLMs have finite context windows (128K tokens in GPT-4.1, but every token costs money and latency), and sending the entire history on every request is unviable in production. It's not just a technical problem — it's an economic one. An agent carrying 50K tokens of history on every call is burning budget for no reason.
Connection with the module: This capsule gives you the tools to manage the active conversation: window trimming (simple, loses context), token-based trimming (precise, budget-driven), and summarization (the best of both worlds, but more complex). In capsule 03, the history gets persisted with MemorySaver so it survives between invocations. In capsule 07, these patterns get folded into a complete conversation management strategy.
The infinite context problem
How a conversation grows
Every turn adds at least 2 messages (user + assistant). With tools, each tool call and response adds more. A 20-turn research run with the Research Agent accumulates ~15,000 tokens in history alone — and every turn sends all of it to the LLM.
The three costs of unmanaged history
| Cost | Impact | Example |
|---|---|---|
| Economic | More tokens = more money per request | 15K tokens of context × $2.50/1M tokens × 100 requests/day = $3.75/day in history alone |
| Latency | More tokens = more processing time | Time-to-first-token grows roughly linearly with prompt size |
| Quality | Long context windows dilute the model's attention | Relevant information from turn 2 gets "lost" when there are 50 messages after it |
The third point is the least obvious but the most critical. Studies like "Lost in the Middle" (Liu et al., 2023) show that LLMs pay less attention to information in the middle of the context. If the user's key instruction is at turn 3 of 50, the model may ignore it — not because it doesn't "fit" in the context window, but because its attention gets diluted.
MessagesState: how LangGraph handles messages
Before solving the problem, understand how LangGraph accumulates messages:
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
The add_messages reducer means that every time a node returns {"messages": [...]}, the messages get appended to the existing list instead of replacing it. Without intervention, the list only grows. The question is: how do you shrink that list without losing the coherence of the conversation?
Window-based trimming
The simplest idea
Keep only the last N messages, discard everything before:
Full conversation: [m1, m2, m3, m4, m5, m6, m7, m8, m9, m10]
Window of 6: [m5, m6, m7, m8, m9, m10]
Manual implementation
def trim_by_window(messages: list, window_size: int = 10) -> list:
"""Keep the last window_size messages."""
if len(messages) <= window_size:
return messages
return messages[-window_size:]
Simple, but it has a subtle problem: if m5 is a ToolMessage responding to an AIMessage with tool_calls in m4, and m4 fell outside the window, the LLM will see a tool response with no matching tool call. Some providers (OpenAI) reject this with an error.
Implementation with LangGraph trim_messages
LangGraph provides trim_messages, which handles these edge cases:
from langchain_core.messages import (
trim_messages,
SystemMessage,
HumanMessage,
AIMessage,
)
messages = [
SystemMessage(content="You are a research assistant."),
HumanMessage(content="What is RAG?"),
AIMessage(content="RAG is Retrieval-Augmented Generation..."),
HumanMessage(content="How is it implemented?"),
AIMessage(content="It's implemented with a retriever and an LLM..."),
HumanMessage(content="Give me a code example"),
AIMessage(content="Here's an example with LangChain..."),
HumanMessage(content="Now compare it with fine-tuning"),
]
trimmed = trim_messages(
messages,
max_tokens=500,
strategy="last",
token_counter=len, # simplified; in production use tiktoken
include_system=True,
allow_partial=False,
)
Key parameters: strategy="last" keeps the most recent messages; include_system=True always preserves the system message (without it, the agent loses its identity); allow_partial=False doesn't cut messages in half; token_counter defines how tokens are counted (in production, use tiktoken).
Trade-offs of window trimming
| Advantage | Disadvantage |
|---|---|
| Trivial to implement (1-2 lines) | Loses all context outside the window |
| Predictable: you know exactly how many messages you're sending | If the user gave a key instruction at turn 2, it's gone by turn 12 |
| Fixed cost per request | The window size is a compromise: large = more cost, small = more loss |
When to use it: Casual conversations, chatbots, interactions where old context doesn't matter. Don't use it for multi-step research where every turn depends on the previous ones.
Token-based trimming
More precise than counting messages
The problem with window-based trimming is that "10 messages" can mean 500 tokens or 15,000 tokens — a tool response can carry 2,000 tokens in a single message. Token-based trimming defines a token budget and keeps recent messages until that budget runs out, iterating from the end.
Implementation with real token counting
import tiktoken
def count_tokens(messages: list, model_name: str = "gpt-4.1-mini") -> int:
"""Count real tokens using tiktoken."""
encoding = tiktoken.encoding_for_model(model_name)
total = 0
for msg in messages:
total += len(encoding.encode(msg.content))
total += 4 # per-message overhead (role, separators)
return total
def trim_by_token_budget(
messages: list,
max_tokens: int = 4000,
model_name: str = "gpt-4.1-mini",
) -> list:
"""Keep recent messages within the token budget."""
encoding = tiktoken.encoding_for_model(model_name)
system_messages = [m for m in messages if isinstance(m, SystemMessage)]
non_system = [m for m in messages if not isinstance(m, SystemMessage)]
system_tokens = sum(
len(encoding.encode(m.content)) + 4 for m in system_messages
)
available_tokens = max_tokens - system_tokens
kept = []
current_tokens = 0
for msg in reversed(non_system):
msg_tokens = len(encoding.encode(msg.content)) + 4
if current_tokens + msg_tokens > available_tokens:
break
kept.insert(0, msg)
current_tokens += msg_tokens
return system_messages + kept
Using trim_messages with a real token_counter
The cleanest approach is to use LangChain's trim_messages with an exact token counter:
from langchain_core.messages import trim_messages
import tiktoken
def tiktoken_counter(messages: list) -> int:
"""Count tokens with tiktoken for trim_messages."""
encoding = tiktoken.encoding_for_model("gpt-4.1-mini")
total = 0
for msg in messages:
total += len(encoding.encode(msg.content))
total += 4
return total
def call_model_with_token_trim(state: AgentState) -> dict:
"""Trim by token budget before invoking the model."""
trimmed = trim_messages(
state["messages"],
max_tokens=4000,
strategy="last",
token_counter=tiktoken_counter,
include_system=True,
allow_partial=False,
)
response = model.invoke(trimmed)
return {"messages": [response]}
Choosing the token budget
Why 4,000 and not 50,000? Because more context isn't always better. The "Lost in the Middle" effect means the model pays more attention to the beginning and the end of the context. A history of 4,000 well-chosen tokens beats one of 50,000 with noise, and costs 12x less.
Practical rule: For conversational agents, 3,000-6,000 tokens of history. For deep research, 8,000-12,000. More than 15,000 rarely improves quality.
Trade-offs of token-based trimming
| Advantage | Disadvantage |
|---|---|
| Precise control over cost (you know exactly how many tokens you send) | More complex than window trimming (you need a token counter) |
| Adapts to messages of varying size | Still loses old context (it only keeps what's recent) |
| Predictable in cost and latency | Token counting adds overhead (minimal, but it exists) |
Summarization
The best of both worlds
Window and token trimming discard old messages. If at turn 3 the user said "Python implementation only, no JavaScript," that instruction is lost once the window moves forward.
Summarization solves this: instead of discarding old messages, it summarizes them. The agent keeps a compressed summary of the old conversation + the recent messages in full. Result: context from the whole conversation in ~200 tokens (the summary) + full detail of the most recent turns.
Implementing the summarizer
from langchain_core.messages import (
SystemMessage,
HumanMessage,
AIMessage,
RemoveMessage,
)
from langchain.chat_models import init_chat_model
model = init_chat_model("openai:gpt-4.1-mini")
SUMMARY_PROMPT = """Summarize this conversation in one concise paragraph.
Preserve:
- What the user asked for (main task and sub-tasks)
- What preferences they expressed (format, language, style, sources)
- What has been found/completed so far
- What is still pending
Do NOT include implementation details or full code — only the context
an agent needs to continue the conversation coherently.
Conversation to summarize:
{conversation}"""
def summarize_conversation(messages: list) -> str:
"""Summarize a list of messages into one paragraph."""
conversation_text = "\n".join(
f"{msg.__class__.__name__}: {msg.content[:300]}"
for msg in messages
)
response = model.invoke([
HumanMessage(
content=SUMMARY_PROMPT.format(conversation=conversation_text)
)
])
return response.content
Integrating it as a node in LangGraph
To integrate summarization into a StateGraph, you need a state that supports the summary:
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class MemoryState(TypedDict):
messages: Annotated[list, add_messages]
summary: str
The summarization node uses RemoveMessage to delete old messages from the state:
def summarize_node(state: MemoryState) -> dict:
"""Summarize old messages and clean up the history."""
messages = state["messages"]
if len(messages) <= 12:
return {}
old_messages = messages[:-6]
summary_text = summarize_conversation(old_messages)
existing = state.get("summary", "")
if existing:
summary_text = model.invoke([HumanMessage(
content=f"Merge into a single summary:\n\nPrevious:\n{existing}\n\nNew:\n{summary_text}"
)]).content
return {
"summary": summary_text,
"messages": [RemoveMessage(id=m.id) for m in old_messages],
}
RemoveMessage deletes messages from the active history. Combined with checkpointing (capsule 03), the originals remain accessible in the checkpoint history.
Trade-offs of summarization
| Advantage | Disadvantage |
|---|---|
| Preserves context from the whole conversation | Requires an extra LLM call to summarize (additional cost) |
| Cuts tokens drastically (50 messages → 1 summary of ~200 tokens) | The summary can lose specific details that mattered |
| The model stays coherent with old instructions | Additional latency from the summarization LLM call |
| Combines with trimming for maximum efficiency | More complexity in the implementation |
When to use each strategy
There's no universally best strategy. The choice depends on the type of interaction, the budget, and your tolerance for context loss.
| Criterion | Window Trimming | Token-based Trimming | Summarization |
|---|---|---|---|
| Implementation complexity | Trivial (1 line) | Low (token counter) | Medium (extra LLM call) |
| Cost per request | Low and fixed | Low and predictable | Medium (+ 1 LLM call to summarize) |
| Context preservation | None (outside the window = lost) | None (outside the budget = lost) | High (the summary preserves the essentials) |
| Additional latency | None | Minimal (token counting) | Noticeable (~1-2s for the summary) |
| Best for | Casual chat, FAQ bots | Agents with tools (messages of varying size) | Long research runs, agents with persistent instructions |
| Worst for | Multi-step research | Conversations where old context is critical | Casual chat (overhead not justified) |
Combining strategies
In production, the best solution combines both: first summarize if there are too many messages, then trim by tokens if it still exceeds the budget. Double protection. That's exactly what you'll implement in the complete StateGraph below.
Implementation in an agent
Complete StateGraph with memory management
Now put it all together in a working StateGraph. This agent has three nodes: manage_memory (trims/summarizes), agent (invokes the model), and tools (runs tools if the model asks for them).
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
from langchain.chat_models import init_chat_model
from langchain_core.messages import (
SystemMessage,
HumanMessage,
AIMessage,
RemoveMessage,
trim_messages,
)
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
class ResearchState(TypedDict):
messages: Annotated[list, add_messages]
summary: str
@tool
def web_search(query: str) -> str:
"""Search the web for information."""
from tavily import TavilyClient
client = TavilyClient()
results = client.search(query, max_results=3)
return "\n\n".join(
f"**{r['title']}**\n{r['content'][:300]}"
for r in results["results"]
)
model = init_chat_model("openai:gpt-4.1-mini")
tools = [web_search]
model_with_tools = model.bind_tools(tools)
SYSTEM_PROMPT = "You are a research agent. Use tools when you need current information."
SUMMARIZE_THRESHOLD = 12
KEEP_RECENT = 6
TOKEN_BUDGET = 4000
def manage_memory(state: ResearchState) -> dict:
"""Summarize if there are too many messages."""
messages = state["messages"]
if len(messages) <= SUMMARIZE_THRESHOLD:
return {}
old_messages = messages[:-KEEP_RECENT]
conv_text = "\n".join(
f"{m.__class__.__name__}: {m.content[:200]}"
for m in old_messages if hasattr(m, "content") and m.content
)
existing = state.get("summary", "")
prompt = (
f"Merge this summary with the new conversation:\n\n"
f"Previous summary:\n{existing}\n\nNew:\n{conv_text}"
) if existing else (
f"Summarize this conversation (preserve the user's preferences, "
f"what was found, what is still pending):\n\n{conv_text}"
)
summary = model.invoke([HumanMessage(content=prompt)]).content
return {"summary": summary, "messages": [RemoveMessage(id=m.id) for m in old_messages]}
def agent_node(state: ResearchState) -> dict:
"""Invoke the model with managed context."""
summary = state.get("summary", "")
sys = SYSTEM_PROMPT + (f"\n\nPrevious context:\n{summary}" if summary else "")
trimmed = trim_messages(
[SystemMessage(content=sys)] + state["messages"],
max_tokens=TOKEN_BUDGET, strategy="last",
token_counter=len, include_system=True, allow_partial=False,
)
return {"messages": [model_with_tools.invoke(trimmed)]}
def should_continue(state: ResearchState) -> str:
last = state["messages"][-1]
return "tools" if hasattr(last, "tool_calls") and last.tool_calls else "end"
graph = StateGraph(ResearchState)
graph.add_node("manage_memory", manage_memory)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "manage_memory")
graph.add_edge("manage_memory", "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "agent")
research_agent = graph.compile()
The flow: START → manage_memory → agent → (tools → agent loop | END). manage_memory runs before every invocation, guaranteeing the history never exceeds the threshold.
Running it
result = research_agent.invoke({
"messages": [HumanMessage(content="Research RAG best practices in 2025")],
"summary": "",
})
print(result["messages"][-1].content[:500])
print(f"\nMessages in state: {len(result['messages'])}")
print(f"Summary: {result['summary'][:200] if result['summary'] else 'N/A'}")
Connection with the project
The M5 Research Agent (planning + reflection) gets memory management: a new summary: str field in the state, a new manage_memory node that runs before planning, and the agent node injects the summary into the system prompt.
In capsule 03 (Checkpointing), the complete state — including summary — gets persisted with MemorySaver. In capsule 07 (Conversation Management), these strategies get combined with garbage collection for research runs of 100+ turns.
Troubleshooting
Problem 1: Orphaned tool messages after trimming
Symptom: API error: "A ToolMessage must be preceded by an AIMessage with tool_calls." It happens when trimming cuts the AIMessage with tool calls but keeps the matching ToolMessage.
Fix: Use trim_messages with allow_partial=False (already included). If you implement trimming manually, always check that no ToolMessage is left without its matching AIMessage:
def safe_trim(messages: list, window_size: int) -> list:
trimmed = messages[-window_size:]
while trimmed and hasattr(trimmed[0], "tool_call_id"):
trimmed = trimmed[1:]
return trimmed
Problem 2: The summary loses the user's instructions
Symptom: The user said "I only want academic sources" at turn 3. After summarization, the agent starts using blogs as sources.
Fix: The summary prompt must emphasize preserving user preferences. Add to the prompt: "Preserve ALL preferences the user expressed (format, sources, language, constraints)." Alternatively, extract preferences into a separate state field that never gets trimmed.
Problem 3: Recursive summarization degrades quality
Symptom: After 3 successive summarizations, the summary is so abstract it loses useful information. A "summary of a summary of a summary" is like photocopying a photocopy — every generation loses fidelity.
Fix: Limit yourself to 2-3 levels of summarization. After the third level, discard the oldest summary and start a new one. Or use an adaptive threshold: if the summary exceeds 500 tokens, compact it before adding new information.
Problem 4: The agent forgets its main task
Symptom: In long conversations, the agent "loses the thread" and stops pursuing the original task, replying only to the last message with no context.
Fix: Keep the main task outside the message history — in a separate state field (task: str) that never gets trimmed. Always inject it into the system prompt:
system_content = (
f"{SYSTEM_PROMPT}\n\n"
f"Main task: {state.get('task', 'N/A')}\n\n"
f"Previous context:\n{state.get('summary', 'N/A')}"
)
Problem 5: The token counter isn't accurate
Symptom: You exceed the context window even though you have trimming in place. The model rejects the request with a token error.
Fix: Use tiktoken with the right model, not len() (which counts characters, not tokens). Add a 10-15% safety margin and subtract the tool schema overhead (~500 tokens).
Exercises
Exercise 1: Window trimming that preserves the system message
Implement trim_with_system(messages, window_size) that always preserves the system message, even if it falls outside the window. If there's no system message, return only the window.
View solution
from langchain_core.messages import SystemMessage
def trim_with_system(messages: list, window_size: int = 10) -> list:
"""Trim while preserving the system message."""
system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
non_system = [m for m in messages if not isinstance(m, SystemMessage)]
trimmed = non_system[-window_size:] if len(non_system) > window_size else non_system
return system_msgs + trimmed
# Test
msgs = [
SystemMessage(content="You are a researcher"),
HumanMessage(content="Turn 1"),
AIMessage(content="Answer 1"),
HumanMessage(content="Turn 2"),
AIMessage(content="Answer 2"),
HumanMessage(content="Turn 3"),
AIMessage(content="Answer 3"),
HumanMessage(content="Turn 4"),
]
result = trim_with_system(msgs, window_size=4)
print(f"Messages: {len(result)}") # 5 (1 system + 4 recent)
print(f"First message: {result[0].__class__.__name__}") # SystemMessage
print(f"Second message: {result[1].content}") # Turn 2 (not Turn 1)
Exercise 2: Token budget calculator
Build a function that, given a model and its context window, computes the optimal token budget for history. Subtract: system prompt, tool schemas (estimated), room for the response, and a safety margin.
View solution
def calculate_history_budget(
context_window: int = 128_000,
system_prompt_tokens: int = 500,
tool_schemas_tokens: int = 600,
max_response_tokens: int = 2000,
safety_margin_pct: float = 0.10,
current_user_msg_tokens: int = 200,
) -> dict:
"""Compute the optimal token budget for history."""
fixed_overhead = (
system_prompt_tokens
+ tool_schemas_tokens
+ max_response_tokens
+ current_user_msg_tokens
)
available = context_window - fixed_overhead
safety_margin = int(available * safety_margin_pct)
history_budget = available - safety_margin
return {
"context_window": context_window,
"fixed_overhead": fixed_overhead,
"safety_margin": safety_margin,
"history_budget": history_budget,
"history_budget_pct": round(history_budget / context_window * 100, 1),
"recommendation": (
f"Use max_tokens={history_budget} for history. "
f"That's {history_budget / context_window * 100:.0f}% "
f"of the context window."
),
}
budget = calculate_history_budget(
context_window=128_000,
system_prompt_tokens=500,
tool_schemas_tokens=600,
max_response_tokens=2000,
)
print(f"Budget: {budget['history_budget']:,} tokens")
print(f"Percentage of the context window: {budget['history_budget_pct']}%")
print(budget["recommendation"])
budget_small = calculate_history_budget(context_window=8_192)
print(f"\nSmall model (8K): {budget_small['history_budget']:,} tokens")
Exercise 3: Summarizer that preserves preferences
Implement a summarizer that uses structured output to extract: (1) a general summary, (2) the user's preferences (format, sources, language), (3) pending tasks. The preferences go into a separate field that never gets trimmed.
View solution
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage
model = init_chat_model("openai:gpt-4.1-mini")
class ConversationSummary(BaseModel):
summary: str = Field(description="Concise summary of what was discussed")
user_preferences: list[str] = Field(description="The user's preferences")
pending_tasks: list[str] = Field(description="Tasks not yet completed")
summarizer = model.with_structured_output(ConversationSummary)
def summarize_with_preferences(messages: list) -> ConversationSummary:
conversation = "\n".join(
f"{m.__class__.__name__}: {m.content[:300]}"
for m in messages if hasattr(m, "content") and m.content
)
return summarizer.invoke(
f"Extract a summary, the user's preferences, and pending tasks:\n\n"
f"{conversation}"
)
test_msgs = [
HumanMessage(content="Research RAG for my project"),
AIMessage(content="RAG combines retrieval with generation..."),
HumanMessage(content="I only want academic sources, no blogs"),
AIMessage(content="Understood, I'll look for papers..."),
]
result = summarize_with_preferences(test_msgs)
print(f"Summary: {result.summary}")
print(f"Preferences: {result.user_preferences}")
print(f"Pending: {result.pending_tasks}")
Exercise 4: Adaptive trimming
Implement adaptive_trim(messages, summary) that picks the strategy automatically: if the conversation is short (< 10 messages), don't trim; if it's medium (10-20), use token trimming; if it's long (20+), use summarization + token trimming. Return (messages, summary, strategy_used).
View solution
from langchain_core.messages import SystemMessage, trim_messages
def adaptive_trim(
messages: list,
summary: str = "",
window_size: int = 10,
token_budget: int = 4000,
summarize_threshold: int = 20,
) -> tuple[list, str, str]:
system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
non_system = [m for m in messages if not isinstance(m, SystemMessage)]
msg_count = len(non_system)
if msg_count <= window_size:
return messages, summary, "none"
if msg_count <= summarize_threshold:
trimmed = trim_messages(
messages, max_tokens=token_budget, strategy="last",
token_counter=len, include_system=True, allow_partial=False,
)
return trimmed, summary, "window"
old = non_system[:-window_size]
recent = non_system[-window_size:]
conv_text = "\n".join(
f"{m.__class__.__name__}: {m.content[:200]}"
for m in old if hasattr(m, "content") and m.content
)
new_summary = f"{summary}\n\n{conv_text[:500]}" if summary else conv_text[:800]
result = system_msgs + recent
trimmed = trim_messages(
result, max_tokens=token_budget, strategy="last",
token_counter=len, include_system=True, allow_partial=False,
)
return trimmed, new_summary, "summarize+trim"
short = [HumanMessage(content=f"msg {i}") for i in range(5)]
medium = [HumanMessage(content=f"msg {i}") for i in range(15)]
long = [HumanMessage(content=f"msg {i}") for i in range(30)]
for name, msgs in [("short", short), ("medium", medium), ("long", long)]:
result, summary, strategy = adaptive_trim(msgs)
print(f"{name} ({len(msgs)} msgs): strategy={strategy}, "
f"output={len(result)} msgs")
Exercise 5: Memory monitor with metrics
Build a MemoryMonitor that tracks: tokens used per turn, when trimming kicked in, how many summarizations were done, and the compression ratio (messages_after / messages_before).
View solution
from dataclasses import dataclass, field
from collections import Counter
@dataclass
class MemoryMonitor:
trim_events: int = 0
summarize_events: int = 0
tokens_per_turn: list[int] = field(default_factory=list)
msg_counts: list[tuple[int, int]] = field(default_factory=list)
strategies: list[str] = field(default_factory=list)
def record(self, msgs_before: int, msgs_after: int, tokens: int, strategy: str):
self.msg_counts.append((msgs_before, msgs_after))
self.tokens_per_turn.append(tokens)
self.strategies.append(strategy)
if strategy in ("window", "token"):
self.trim_events += 1
elif "summarize" in strategy:
self.summarize_events += 1
@property
def compression_ratio(self) -> float:
total_before = sum(b for b, _ in self.msg_counts)
total_after = sum(a for _, a in self.msg_counts)
return total_after / total_before if total_before else 1.0
def report(self) -> str:
return (
f"Turns: {len(self.tokens_per_turn)} | "
f"Trims: {self.trim_events} | "
f"Summaries: {self.summarize_events} | "
f"Avg tokens: {sum(self.tokens_per_turn)/len(self.tokens_per_turn):.0f} | "
f"Compression: {self.compression_ratio:.2f} | "
f"Strategies: {dict(Counter(self.strategies))}"
)
monitor = MemoryMonitor()
for turn in range(25):
n = turn + 1
if n <= 10:
monitor.record(n, n, n * 100, "none")
elif n <= 20:
monitor.record(n, 10, 1000, "window")
else:
monitor.record(n, 8, 900, "summarize+trim")
print(monitor.report())
Summary
- Short-term memory is the history of the active conversation. It grows without limit with every turn, and every turn sends the entire history to the LLM — driving up cost, latency, and diluting the model's attention.
- Window trimming keeps the last N messages. Trivial to implement, but it loses all context outside the window. Useful for casual chat.
- Token-based trimming defines a token budget and keeps recent messages within that budget. More precise than window trimming (it adapts to messages of varying size), but it still discards old context.
- Summarization condenses old messages into a compact paragraph and keeps the recent ones intact. It preserves context from the whole conversation at a cost of ~200 tokens. It requires an additional LLM call.
- In production, combine strategies: summarization for long conversations + token trimming as a safety net. Keep user preferences in separate fields that never get trimmed.
- LangChain's
trim_messageshandles edge cases (orphaned tool messages, preserved system messages) that a manual implementation can miss. - The token budget for history must subtract: system prompt, tool schemas, room for the response, and a safety margin. A range of 3,000-6,000 tokens covers most cases.
Additional resources
- LangGraph Memory Concepts — Official documentation on memory in LangGraph
- trim_messages API Reference — Reference for the trim_messages function
- Lost in the Middle (Liu et al., 2023) — Paper on how LLMs lose attention in long contexts
- How to add summary of the conversation history — Official summarization tutorial in LangGraph
- tiktoken — OpenAI's tokenization library for precise token counting