MemoryCompressor

Purpose

Reduces the size of a MemoryEntry collection via three independently-configurable strategies: dropping noisy/trivial entries, merging near-duplicates (by embedding similarity), and LLM-summarizing large collections into a single compact entry. Used by AIMemoryManager.run_maintenance() to keep LongTermMemory from growing unbounded.

LLMProtocol

class LLMProtocol(Protocol):
    def generate(self, prompt: str, max_tokens: int = 256) -> str: ...

A structural (duck-typed) Protocol — any object with a matching .generate() method works, no explicit inheritance required. Note: an identically-shaped LLMProtocol is separately defined in memory.intelligent.memory_manager — the two are not the same class object, though structurally interchangeable, since Protocol types are checked structurally rather than nominally.

Constructor

def __init__(
    self,
    llm: Optional[LLMProtocol] = None,
    merge_similarity_threshold: float = 0.92,
    tenant_id: str = "default",
) -> None

Public methods

remove_noise(entries) -> Tuple[List[MemoryEntry], int]

Filters out entries where _is_noise(entry) is True:

def _is_noise(entry):
    text = str(entry.content).strip()
    if len(text) < 10: return True
    if len(text.split()) < 3: return True
    alphanum_ratio = sum(c.isalnum() for c in text) / max(len(text), 1)
    return alphanum_ratio < 0.3

Note this operates on str(entry.content) — for dict content (the typical shape), this stringifies the entire dict repr (e.g. "{'input': 'hi', 'output': 'there'}'"), not just the meaningful text fields. A genuinely short/trivial conversational exchange can still produce a long str(dict) representation that passes the length checks despite the actual input/output text being trivial — the noise heuristic is calibrated against raw content stringification, not the conversational text itself.

merge_duplicates(entries) -> Tuple[List[MemoryEntry], int]

Returns (entries, 0) unmodified if fewer than 2 entries. Otherwise: separates entries with embeddings from those without (entries without an embedding are never merge candidates and pass through untouched). For entries with embeddings, does an O(n²) pairwise comparison — for each unmerged entry i, scans forward through remaining unmerged entries j > i and merges the first pair found above merge_similarity_threshold (greedy, first-match — not the globally most-similar pair). Each entry can be merged into at most one other in a single merge_duplicates() call (no multi-way merging within one pass).

summarise(entries, max_tokens=256, target_type=MemoryType.LONG_TERM) -> Optional[MemoryEntry]

Returns None for an empty entries list. If self.llm is None: builds a naive summary by concatenating the first 80 characters of each entry's str(content), joined by " | ", truncated to 500 characters overall — not an actual summary, just a truncated concatenation. If self.llm is set: formats _SUMMARISE_PROMPT and calls self.llm.generate(...); on any exception from the LLM call, falls back to a similar truncated concatenation (60 chars/entry, 400 chars overall) rather than propagating the error.

The resulting entry's importance is min(1.0, average_importance_of_sources + 0.1) — summaries always get a small importance boost relative to their sources' average.

compress_batch(entries, *, remove_noise=True, merge=True, summarise=False, summarise_threshold=20) -> Dict[str, Any]

Runs the pipeline conditionally: noise removal → merge (only if merge=True and at least 2 entries remain) → summarization (only if summarise=True and len(entries) > summarise_threshold after noise removal and merging have already reduced the count). Returns {"entries": [...], "noise_removed": int, "merges": int, "summarised": bool}.

[!NOTE] summarise_threshold is checked against the post-filter count If you pass 100 noisy/duplicate entries with summarise_threshold=20, and noise removal + merging reduce them to 15 entries, summarization will not trigger (15 > 20 is False) even though your original batch was well over the threshold. The threshold check happens after the other two stages, using their output count, not the original input count.

Internal helpers (documented for completeness)

_cosine(a, b) -> float

Manual cosine similarity implementation (not using numpy) — 0.0 if either vector's norm is zero (avoids division by zero).

_merge_pair(a, b) -> MemoryEntry

New entry with content={"merged": text, "sources": [a.id, b.id]}, memory_type=a.memory_type (the first entry's type is used, b's type is discarded if it differs), importance=max(a.importance, b.importance), tags=list(set(a.tags + b.tags)) (deduplicated union, arbitrary resulting order since it round-trips through a set).

_make_summary_entry(text, source_entries, target_type) -> MemoryEntry

New entry with content={"summary": text, "source_count": len(source_entries)}, tags=["compressed", "summary"], metadata["source_ids"] listing every contributing entry's ID.

_entry_to_text(entry) -> str (staticmethod)

Same logic as ContextBuilder._entry_to_text (duplicated independently in both modules) — recognizes {"summary": ...} and {"merged": ...} content shapes and renders those directly, else formats input/output dict content, else str(content).

Raises

No exceptions propagate from remove_noise, merge_duplicates, summarise, or compress_batch under normal operation — LLM failures are caught internally.

Thread safety

Not thread-safe by design intent, though each method only reads self.llm/self.merge_similarity_threshold/self.tenant_id (set once at construction) and operates on the entries list/new objects passed in — there's no shared mutable instance state being written to, so concurrent calls with different entries lists on the same MemoryCompressor instance are safe in practice, even though this isn't an explicitly documented guarantee.

Complexity

Examples

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

class FakeLLM:
    def generate(self, prompt, max_tokens=256):
        return "User discussed several topics across many turns."

compressor = MemoryCompressor(llm=FakeLLM(), merge_similarity_threshold=0.9)

entries = [MemoryEntry(content={"input": f"note {i}", "output": "ok"},
                        timestamp=time.time(), memory_type=MemoryType.LONG_TERM,
                        importance=0.5)
           for i in range(30)]

# 1. Basic — noise removal only
clean, removed = compressor.remove_noise(entries)

# 2. Intermediate — full pipeline without summarization
result = compressor.compress_batch(clean, remove_noise=False, merge=True, summarise=False)
print(result["merges"])

# 3. Advanced — with summarization above threshold
result2 = compressor.compress_batch(entries, summarise=True, summarise_threshold=10)
if result2["summarised"]:
    summary_entry = result2["entries"][0]
    print(summary_entry.content["summary"])

# 4. Error handling — LLM failure during summarise() falls back silently
class BrokenLLM:
    def generate(self, prompt, max_tokens=256):
        raise RuntimeError("timeout")

broken_compressor = MemoryCompressor(llm=BrokenLLM())
summary = broken_compressor.summarise(entries[:5])
print(summary.content["summary"])  # truncated concatenation, not a real summary, no exception raised

# 5. Production — periodic LTM compression (mirrors AIMemoryManager.run_maintenance)
if len(entries) > 100:
    result3 = compressor.compress_batch(entries, merge=True, summarise=True, summarise_threshold=50)

Best practices

• Set summarise_threshold relative to your expected post-noise-removal, post-merge count, not your raw input count, given the ordering caveat above.
• Don't rely on merge_duplicates finding the globally optimal pairing — it's greedy/first-match, adequate for basic dedup but not optimal clustering.
• Always supply a real llm if you need actual semantic summaries — the no-LLM fallback is a blunt truncation, not a summary.

Common mistakes

• Assuming _is_noise operates on the meaningful conversational text — it operates on str(dict), which includes Python dict-repr punctuation and key names in its length/alphanumeric-ratio calculation.
• Assuming a failed LLM call during summarise() raises so you can retry — it's caught and silently replaced with a truncated concatenation.

Related APIs

• ContextBuilder._entry_to_text — recognizes the same {"summary": ...}/{"merged": ...} shapes this class produces.
• AIMemoryManager.run_maintenance() — the primary caller.