WorkingMemory

Purpose

Holds the information actively relevant to the current reasoning turn. Unlike ShortTermMemory (a fixed FIFO window), WorkingMemory is rebuilt each turn from whatever candidates MemorySelector chose, ranked and capacity/token-budget-limited by importance rather than recency.

Constructor

def __init__(self, capacity: int = 10, token_budget: int = 3000, *, tenant_id: str = "default") -> None

Backed internally by a plain Dict[str, MemoryEntry] (id → entry), not an ordered structure — ordering for get_all() is always recomputed by sorting on effective_importance.

Public methods

load(entries: List[MemoryEntry]) -> None

Replaces the entire contents (self._store.clear() first). Sorts the input by effective_importance descending, then walks the sorted list up to self.capacity entries, accumulating a running token estimate and stopping (break) the moment the next entry would exceed self.token_budget. This is the method AIMemoryManager.build_context() calls every turn with the MemorySelector-chosen candidates.

[!NOTE] load() uses break, MemorySelector uses continue — two different budgeting strategies WorkingMemory.load()'s token-budget loop stops entirely at the first entry that doesn't fit (break), so entries after that point in the sorted order are never considered even if a smaller one further down would fit. MemorySelector._apply_token_budget instead uses continue — skipping an oversized entry and checking subsequent (smaller, lower-ranked) ones. The two components of the same pipeline (MemorySelector.select_entries feeding into WorkingMemory.load) apply genuinely different greedy strategies for what is conceptually the same problem. See Known Issues.

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

Constructs a MemoryEntry with memory_type=MemoryType.WORKING, calls _evict_if_needed() before inserting the new entry, then stores it. Note the default importance here is 0.6, differing from ShortTermMemory.add's default of 0.5.

remove(entry_id: str) -> Optional[MemoryEntry]

dict.pop(entry_id, None) — returns the removed entry or None if not present.

clear() -> None

self._store.clear().

get_all() -> List[MemoryEntry]

Returns entries sorted by effective_importance descending — not insertion order, and not the same order load() originally processed them in (though since load() already inserts in importance-descending order up to the budget, the two orderings typically coincide in practice unless add() was called afterward to insert additional entries out of order).

get_as_text() -> str

Joins all entries' content (via get_all()'s ordering) as "user: {input}\nassistant: {output}" for dict content, or str(content) otherwise, separated by "\n\n".

_evict_if_needed() (internal)

while len(self._store) >= self.capacity:
    worst_id = min(self._store, key=lambda k: self._store[k].effective_importance)
    del self._store[worst_id]

[!DANGER] capacity <= 0 causes an infinite loop / immediate crash depending on state If capacity is 0 (or negative) and _store is already empty, len(self._store) >= self.capacity (0 >= 0) is True on the very first add() call, but min(self._store, ...) is called on an empty dict — min() on an empty iterable raises ValueError (exact message varies by Python version, e.g. "min() arg is an empty sequence" or "min() iterable argument is empty"). So WorkingMemory(capacity=0).add(...) raises ValueError immediately rather than silently accepting entries or looping forever. This is not documented anywhere in the source and differs from the graceful ValueError at construction time seen in ShortTermMemory(window_size=0) — here the error surfaces later, on first add(), with a less specific message. See Known Issues.

stats() -> Dict[str, Any]

Returns type, capacity, token_budget, entries (count), estimated_tokens_used (sum of per-entry token estimates), tenant_id.

Raises

Thread safety

Not thread-safe — _store dict mutation in load/add/remove/clear has no locking. AIMemoryManager.build_context() calls bundle.working.load(selected) synchronously on the calling thread.

Complexity

• add: O(capacity) worst case, due to _evict_if_needed's min() scan over the whole store.
• load: O(n log n) for the sort, plus O(min(n, capacity)) for the budget walk.
• get_all: O(n log n) (re-sorts every call).

Examples

from zerda.memory import WorkingMemory, MemoryEntry, MemoryType
import time

wm = WorkingMemory(capacity=5, token_budget=500)

# 1. Basic
entry = wm.add({"input": "Remember this", "output": "Got it"}, importance=0.7)

# 2. Intermediate — load() replaces contents each turn
candidates = [MemoryEntry(content={"input": f"fact {i}", "output": "ok"},
                           timestamp=time.time(), memory_type=MemoryType.WORKING,
                           importance=0.5 + i * 0.05)
              for i in range(8)]
wm.load(candidates)
print(len(wm.get_all()))  # up to 5, budget/capacity permitting

# 3. Advanced — inspect as text for prompt injection
print(wm.get_as_text())

# 4. Error handling — degenerate capacity
try:
    WorkingMemory(capacity=0).add({"input": "x", "output": "y"})
except ValueError as e:
    print(e)  # "min() arg is an empty sequence" (message varies by Python version)

# 5. Production — check token pressure
stats = wm.stats()
if stats["estimated_tokens_used"] >= stats["token_budget"] * 0.9:
    print("Working memory near its token budget")

Best practices

• Always construct with capacity >= 1 — the class does not validate this at construction, so validate it yourself if capacity comes from user/config input.
• Call load() (not add()) when replacing the full working set each turn, as AIMemoryManager does — add() is for incrementally inserting a single new item mid-turn.

Common mistakes

• Constructing WorkingMemory(capacity=0) expecting it to simply hold nothing gracefully — the first add() call raises ValueError.
• Assuming get_all() preserves the order entries were add()-ed in — it always re-sorts by effective_importance.

Related APIs

• MemorySelector — typically produces the entries passed to load().
• ShortTermMemory — the fixed-window counterpart.