ShortTermMemory

Purpose

A sliding-window store of the most recent N interactions, where every slot is a full MemoryEntry (not a plain dict, unlike the memory.memories classes) so importance/decay bookkeeping works uniformly with the rest of the intelligent-tier machinery. This is the tier AIMemoryManager writes to on every save_interaction call, before anything is promoted to SemanticMemory or LongTermMemory.

Constructor

def __init__(self, window_size: int = 20, *, tenant_id: str = "default") -> None

Internally backed by collections.deque(maxlen=window_size).

Public methods

add(content, *, importance=0.5, tags=None, metadata=None) -> MemoryEntry

Constructs a MemoryEntry with memory_type=MemoryType.SHORT_TERM, timestamp=time.time(), and metadata={**(metadata or {}), "tenant_id": self.tenant_id}, then appends it. If the deque is already full, the oldest entry is silently evicted (no return value or hook for the evicted entry from add() itself — use evict_below() if you need to observe evictions).

get_all() -> List[MemoryEntry]

list(self._buffer) — a fresh list copy in insertion order (oldest first).

get_recent(n: int) -> List[MemoryEntry]

entries = list(self._buffer)
return entries[-n:] if n < len(entries) else entries

[!DANGER] get_recent(0) returns everything, not nothing Because the condition is n < len(entries) (true for n=0 whenever the buffer is non-empty) and Python's entries[-0:] is equivalent to entries[0:] (i.e. the entire list — -0 == 0 in Python), calling get_recent(0) returns all entries, not an empty list, as you might expect from a "give me the last 0 entries" call. get_recent(n) for any n ≥ 1 behaves as documented (last n entries, or all of them if n exceeds the buffer size). See Known Issues.

clear() -> None

self._buffer.clear().

apply_decay(decay_rate: float = 0.1) -> None

Calls entry.apply_decay(decay_rate) (the linear decay formula on MemoryEntry) on every entry currently in the buffer, in place.

evict_below(min_importance: float) -> List[MemoryEntry]

Partitions all entries by whether effective_importance < min_importance, replaces self._buffer with a new deque(keep, maxlen=self.window_size) containing only the survivors, and returns the evicted list. Called by AIMemoryManager.run_maintenance() after apply_decay().

Properties: size, is_full

size is len(self._buffer). is_full is size == window_size (exactly equal, not >= — though since the deque has maxlen=window_size, size can never exceed it, so this is equivalent to >= in practice).

stats() -> Dict[str, Any]

Returns type, window_size, current_size, is_full, tenant_id.

Raises

Thread safety

Not thread-safe — no locking around _buffer mutation. AIMemoryManager.save_interaction calls bundle.stm.add(...) synchronously on the calling thread (unlike the semantic/LTM writes, which are offloaded to a thread pool), so as long as your application serializes calls to save_interaction per tenant, ShortTermMemory itself sees no concurrent writers from that code path — but nothing in this class prevents a race if you call its methods directly from multiple threads.

Complexity

• add: O(1) amortized (deque append + auto-evict).
• get_all/get_recent: O(n) (materializes a list from the deque).
• apply_decay/evict_below: O(n).

Examples

from zerda.memory import ShortTermMemory

stm = ShortTermMemory(window_size=5, tenant_id="acme")

# 1. Basic
entry = stm.add({"input": "Hi", "output": "Hello!"}, importance=0.5)
print(stm.size)  # 1

# 2. Intermediate — window eviction
for i in range(10):
    stm.add({"input": f"msg {i}", "output": f"reply {i}"})
print(stm.size)  # 5 — capped at window_size

# 3. Advanced — decay + eviction maintenance pass
stm.apply_decay(decay_rate=0.1)
evicted = stm.evict_below(min_importance=0.15)
print(f"Evicted {len(evicted)} low-importance entries")

# 4. Error handling
try:
    ShortTermMemory(window_size=0)
except ValueError as e:
    print(e)  # "window_size must be ≥ 1"

# 5. Production — careful with get_recent(0)
recent_zero = stm.get_recent(0)
print(len(recent_zero))  # NOT 0 — returns everything currently in the buffer
recent_two = stm.get_recent(2)  # correctly returns the last 2

Best practices

• Never call get_recent(0) expecting an empty list — check for n <= 0 yourself before calling if you need that guarantee, or just use get_all()[:0]/[] directly for the "give me nothing" case.
• Call apply_decay() before evict_below() in your maintenance routine (as AIMemoryManager.run_maintenance() does) — otherwise effective_importance won't reflect elapsed time since last access.

Common mistakes

• Passing 0 to get_recent as a defensive "no entries" sentinel — it returns all entries instead.
• Assuming tenant_id filters reads across multiple tenants sharing one instance — it doesn't; each tenant needs its own ShortTermMemory instance (which AIMemoryManager does correctly via _TenantBundle).

Related APIs

• MemoryEntry
• WorkingMemory — the next tier, rebuilt each turn from candidates that may include this tier's entries.
• ConversationBufferWindowMemory — a conceptually similar but independent fixed-window implementation using plain dicts instead of MemoryEntry.

.