Short-Term Memory with Message Buffers

Duration: 15 min

Short-Term Memory with Message Buffers

Duration: 15 min

Overview

Short-term memory bridges the gap between the immediate context window and persistent long-term storage. It holds recent conversation history, intermediate reasoning steps, and temporary state that the agent may need within a session. This module covers buffer strategies - sliding windows, token-limited buffers, and summary buffers - and shows how to implement each with practical Python code.

Choosing the right buffer strategy determines whether your agent remembers what happened 5 messages ago or loses track of multi-step tasks.

Key Concepts & Foundations

  • What: Short-term memory stores recent interactions and intermediate state, typically lasting one session
  • Why: Agents need to remember what was said earlier in a conversation without loading everything into the context window
  • How: Buffer implementations control what stays, what goes, and how evicted content is preserved

Buffer Strategies

| Strategy | How It Works | Best For | |----------|-------------|----------| | Sliding Window | Keep last N messages | Simple chatbots | | Token Buffer | Keep messages until token limit hit | API cost control | | Summary Buffer | Summarize old messages, keep recent | Long conversations | | Entity Buffer | Track entities mentioned, discard details | Customer service |

When Short-Term Memory Fails

  • Lost instructions - User gave a preference 20 messages ago, buffer dropped it
  • Broken multi-step tasks - Agent forgets step 2 results when executing step 5
  • Repetitive questions - Agent asks for clarification it already received

Hands-On Implementation

from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class BufferMessage:
    role: str
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    tokens: int = 0
    def __post_init__(self):
        self.tokens = len(self.content) // 4

class SlidingWindowBuffer:
    """Keep the last N messages."""
    def __init__(self, max_messages: int = 20):
        self.max_messages = max_messages
        self.buffer: deque[BufferMessage] = deque(maxlen=max_messages)
    def add(self, role: str, content: str):
        self.buffer.append(BufferMessage(role=role, content=content))
    def get_messages(self) -> list[dict]:
        return [{"role": m.role, "content": m.content} for m in self.buffer]
    def __len__(self):
        return len(self.buffer)

class TokenLimitedBuffer:
    """Keep messages until a token budget is exhausted."""
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self.messages: list[BufferMessage] = []
    @property
    def total_tokens(self) -> int:
        return sum(m.tokens for m in self.messages)
    def add(self, role: str, content: str):
        msg = BufferMessage(role=role, content=content)
        self.messages.append(msg)
        # Evict oldest messages until under budget
        while self.total_tokens > self.max_tokens and len(self.messages) > 1:
            self.messages.pop(0)
    def get_messages(self) -> list[dict]:
        return [{"role": m.role, "content": m.content} for m in self.messages]

class SummaryBuffer:
    """Summarize old messages, keep recent ones verbatim."""
    def __init__(self, max_recent: int = 6, summarize_fn=None):
        self.max_recent = max_recent
        self.recent: list[BufferMessage] = []
        self.summary: str = ""
        self.summarize_fn = summarize_fn or self._default_summarize
    def _default_summarize(self, messages: list[BufferMessage]) -> str:
        """Simple extractive summary (replace with LLM call in production)."""
        points = []
        for m in messages:
            # Take first sentence of each message
            first_sentence = m.content.split(".")[0]
            points.append(f"- [{m.role}] {first_sentence}")
        return "Previous conversation summary:\n" + "\n".join(points)
    def add(self, role: str, content: str):
        self.recent.append(BufferMessage(role=role, content=content))
        if len(self.recent) > self.max_recent:
            # Move oldest messages into summary
            to_summarize = self.recent[: len(self.recent) - self.max_recent]
            self.recent = self.recent[len(self.recent) - self.max_recent:]
            self.summary = self.summarize_fn(
                [BufferMessage("system", self.summary)] + to_summarize
                if self.summary else to_summarize
            )
    def get_messages(self) -> list[dict]:
        messages = []
        if self.summary:
            messages.append({"role": "system", "content": self.summary})
        messages.extend({"role": m.role, "content": m.content} for m in self.recent)
        return messages


Demo: Compare buffer strategies

print("=== Sliding Window Buffer ===")
sw = SlidingWindowBuffer(max_messages=3)
for i in range(5):
    sw.add("user", f"Message {i+1}: Tell me about topic {i+1}")
print(f"Buffer holds {len(sw)} messages (dropped oldest 2)")
for m in sw.get_messages():
    print(f"  {m['content'][:50]}")
print("\n=== Token-Limited Buffer ===")
tb = TokenLimitedBuffer(max_tokens=100)
tb.add("user", "Short message")
tb.add("assistant", "This is a much longer response " * 10)
tb.add("user", "Follow-up question about the topic")
print(f"Buffer: {tb.total_tokens} tokens, {len(tb.messages)} messages")
print("\n=== Summary Buffer ===")
sb = SummaryBuffer(max_recent=3)
sb.add("user", "My name is Alex. I am learning Python.")
sb.add("assistant", "Welcome Alex! I will help you learn Python.")
sb.add("user", "I prefer short code examples with comments.")
sb.add("assistant", "Got it - concise code with inline comments.")
sb.add("user", "Now teach me about decorators.")
sb.add("assistant", "A decorator wraps a function to extend its behavior.")
print(f"Summary: {sb.summary}")
print(f"Recent messages: {len(sb.recent)}")

Advanced Techniques

Production short-term memory systems often combine multiple strategies:

1. Hybrid Buffers - Use sliding window for recent turns + summary for older context 2. Priority Tagging - Mark certain messages as "sticky" so they survive eviction 3. Session Segmentation - Detect topic changes and summarize completed topics 4. Time-Based Decay - Weight messages by recency; old messages evict before recent ones even at same priority 5. Buffer Checkpointing - Save buffer state so conversations can resume after disconnection

Quiz

Q1: What is the main advantage of a summary buffer over a sliding window?

  • A) It uses less code
  • B) It preserves the gist of old messages instead of losing them entirely ✓
  • C) It is faster to query
  • D) It never drops any information

Q2: When would a token-limited buffer be preferred over a message-count buffer?

  • A) When all messages are the same length
  • B) When you need to control API costs based on actual token usage ✓
  • C) When the conversation is very short
  • D) When you do not care about context quality

Q3: What is the primary risk of a pure sliding window buffer?

  • A) It uses too much memory
  • B) It may drop important early instructions or user preferences ✓
  • C) It is too slow
  • D) It cannot handle multiple users

← Back ▶ Run in Colab