ZerdaZerda

ConversationBufferWindowMemory

On this page

Purpose

A ConversationBufferMemory subclass that keeps only the last k turns, using a collections.deque(maxlen=k) instead of an unbounded list. Use this for long-running conversations where only recent context matters and you want automatic, allocation-free eviction of old turns.

Constructor

Python
def __init__(self, k: int = 5, **kwargs)
ParameterTypeRequiredDefaultNotes
kintNo5Window size. Must be > 0 — raises ValueError("k must be a positive integer") otherwise.
**kwargs—No—Forwarded to ConversationBufferMemory.__init__ (return_messages, input_key, output_key, memory_key).

After calling super().__init__(**kwargs) (which sets self.chat_memory = []), the constructor replaces self.chat_memory with deque(maxlen=k), discarding the empty list from the parent constructor.

Inherited behavior

save_context, load_memory_variables, clear, and the async wrappers are all inherited unmodified from ConversationBufferMemory. Because self.chat_memory is now a deque(maxlen=k), .append() in save_context automatically evicts the oldest entry once the deque is full — no explicit eviction logic exists in this subclass.

[!DANGER] get_messages(limit=N) crashes on this class get_messages is inherited from ConversationBufferMemory and does self.chat_memory[-limit:] when limit is truthy. deque does not support slice indexing in Python, so:

Python · 5 lines
from zerda.memory import ConversationBufferWindowMemory
mem = ConversationBufferWindowMemory(k=5)
mem.save_context({"input": "hi"}, {"output": "hello"})
mem.get_messages(limit=3)
# TypeError: sequence index must be integer, not 'slice'

get_messages() (no limit) works fine, since it falls through to .copy(), which deque supports. Only the limit-provided branch is broken. See Known Issues — this is the most impactful bug found in the memories subpackage since it's triggered by ordinary, documented usage (get_messages(limit=...) is a public method with no warning against calling it this way on a window memory).

Additional public methods

get_memory_stats() -> Dict[str, Any]

Extends the parent's stats dict with type: "ConversationBufferWindowMemory", window_size: self.k, and is_full: len(self.chat_memory) >= self.k.

get_window_info() -> Dict[str, Any]

Python · 6 lines
{
    'current_size': len(self.chat_memory),
    'max_size': self.k,
    'available_slots': self.k - len(self.chat_memory),
    'is_full': len(self.chat_memory) >= self.k,
}

Thread safety

Same caveats as ConversationBufferMemory — no explicit locking. deque.append() is itself thread-safe at the C level for single appends/pops from either end (CPython implementation detail), but the class provides no compound-operation safety guarantees beyond that.

Complexity

  • save_context: O(1) — deque append/evict is O(1) regardless of k.
  • get_window_info: O(1) (uses len(), which is O(1) for deque).

Examples

Python · 26 lines
from zerda.memory import ConversationBufferWindowMemory

# 1. Basic
mem = ConversationBufferWindowMemory(k=3)
for i in range(5):
    mem.save_context({"input": f"msg {i}"}, {"output": f"reply {i}"})
print(len(mem.get_messages()))   # 3 — only the last 3 survive

# 2. Intermediate — inspect window state
info = mem.get_window_info()
print(info)   # {'current_size': 3, 'max_size': 3, 'available_slots': 0, 'is_full': True}

# 3. Advanced — custom I/O keys forwarded via **kwargs
mem2 = ConversationBufferWindowMemory(k=10, input_key="q", output_key="a")
mem2.save_context({"q": "2+2?"}, {"a": "4"})

# 4. Error handling — invalid k
try:
    ConversationBufferWindowMemory(k=0)
except ValueError as e:
    print(e)  # "k must be a positive integer"

# 5. Production — DO NOT do this (will crash):
# mem.get_messages(limit=2)   # TypeError
# Instead, slice the full list yourself:
recent_2 = mem.get_messages()[-2:]   # works — get_messages() returns a plain list via deque.copy()...

[!WARNING] Example 5 caveat deque.copy() returns another deque, not a list — so mem.get_messages()[-2:] also raises TypeError for the same reason. To safely get the last N messages from a window memory, convert explicitly: list(mem.get_messages())[-2:].

Best practices

  • Never call get_messages(limit=...) on this class — use get_window_info() for size introspection, and list(mem.get_messages())[-n:] if you need a sub-slice.
  • Pick k based on your model's context window and the typical length of a single turn, not just a round number.

Common mistakes

  • Calling get_messages(limit=N) on a ConversationBufferWindowMemory instance, expecting the same behavior as on ConversationBufferMemory — it crashes.
  • Assuming get_messages()'s return value supports slicing directly — it's a deque, so wrap in list(...) first.

Related pages