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

CostImpactExample
EconomicMore tokens = more money per request15K tokens of context × $2.50/1M tokens × 100 requests/day = $3.75/day in history alone
LatencyMore tokens = more processing timeTime-to-first-token grows roughly linearly with prompt size
QualityLong context windows dilute the model's attentionRelevant 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

AdvantageDisadvantage
Trivial to implement (1-2 lines)Loses all context outside the window
Predictable: you know exactly how many messages you're sendingIf the user gave a key instruction at turn 2, it's gone by turn 12
Fixed cost per requestThe 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

AdvantageDisadvantage
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 sizeStill loses old context (it only keeps what's recent)
Predictable in cost and latencyToken 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

AdvantageDisadvantage
Preserves context from the whole conversationRequires 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 instructionsAdditional latency from the summarization LLM call
Combines with trimming for maximum efficiencyMore 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.

CriterionWindow TrimmingToken-based TrimmingSummarization
Implementation complexityTrivial (1 line)Low (token counter)Medium (extra LLM call)
Cost per requestLow and fixedLow and predictableMedium (+ 1 LLM call to summarize)
Context preservationNone (outside the window = lost)None (outside the budget = lost)High (the summary preserves the essentials)
Additional latencyNoneMinimal (token counting)Noticeable (~1-2s for the summary)
Best forCasual chat, FAQ botsAgents with tools (messages of varying size)Long research runs, agents with persistent instructions
Worst forMulti-step researchConversations where old context is criticalCasual 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_messages handles 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

  1. LangGraph Memory Concepts — Official documentation on memory in LangGraph
  2. trim_messages API Reference — Reference for the trim_messages function
  3. Lost in the Middle (Liu et al., 2023) — Paper on how LLMs lose attention in long contexts
  4. How to add summary of the conversation history — Official summarization tutorial in LangGraph
  5. tiktoken — OpenAI's tokenization library for precise token counting