Coverage for astrocyte/pipeline/recent_buffer.py: 93%
85 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""Recent memory buffer — ring buffer of recently stored chunks for Tier 1 fuzzy matching.
3Provides sub-5ms fuzzy text search on recent memories without embedding cost.
4Handles typos and morphological variations via character-level similarity.
6Sync, self-contained — Rust migration candidate.
7"""
9from __future__ import annotations
11import logging
12import threading
13from collections import deque
14from dataclasses import dataclass
15from difflib import SequenceMatcher
16from string import punctuation
18from astrocyte.pipeline.provenance import stored_source
19from astrocyte.types import MemoryHit
21logger = logging.getLogger(__name__)
23#: Default max recent items per bank.
24DEFAULT_MAX_PER_BANK = 100
26#: Minimum character-level similarity for a token to count as a fuzzy match.
27_TOKEN_SIMILARITY_THRESHOLD = 0.75
29# Query words too common to be meaningful for fuzzy matching.
30_STOP_WORDS = frozenset(
31 "a an the is are was were be been being have has had do does did "
32 "will would shall should may might can could of in on at to for "
33 "with by from and or but not no nor so yet both either neither "
34 "this that these those it its he she they them his her their "
35 "who what which when where how i me my we our you your "
36 "about also just like very much more than then some any".split()
37)
40@dataclass(slots=True)
41class _RecentEntry:
42 memory_id: str
43 text: str
44 tokens: list[str] # pre-tokenized for fast matching
45 bank_id: str
46 metadata: dict | None = None
49def _tokenize(text: str) -> list[str]:
50 """Lowercase, strip punctuation, remove stop words."""
51 return [w for raw in text.lower().split() if len(w := raw.strip(punctuation)) > 1 and w not in _STOP_WORDS]
54def _fuzzy_token_score(query_tokens: list[str], memory_tokens: list[str]) -> float:
55 """Score a memory against query tokens using character-level fuzzy matching.
57 For each query token, find the best fuzzy match among memory tokens.
58 A match counts if SequenceMatcher ratio >= threshold (handles typos).
59 Returns fraction of query tokens that matched (0.0–1.0).
60 """
61 if not query_tokens:
62 return 0.0
63 if not memory_tokens:
64 return 0.0
66 matched = 0
67 # Build set for exact-match fast path
68 memory_set = set(memory_tokens)
70 for qt in query_tokens:
71 # Fast path: exact match
72 if qt in memory_set:
73 matched += 1
74 continue
75 # Slow path: fuzzy match (character-level similarity)
76 best = 0.0
77 for mt in memory_tokens:
78 # Quick length filter — very different lengths can't match well
79 if abs(len(qt) - len(mt)) > max(len(qt), len(mt)) * 0.5:
80 continue
81 ratio = SequenceMatcher(None, qt, mt).ratio()
82 if ratio > best:
83 best = ratio
84 if best >= _TOKEN_SIMILARITY_THRESHOLD:
85 break # Good enough, stop searching
86 if best >= _TOKEN_SIMILARITY_THRESHOLD:
87 matched += 1
89 return matched / len(query_tokens)
92class RecentMemoryBuffer:
93 """Ring buffer of recently stored text chunks per bank for Tier 1 fuzzy matching.
95 Designed for sub-5ms search on the last N stored memories per bank.
96 Uses character-level fuzzy matching (SequenceMatcher) to handle typos
97 and morphological variations that BM25 would miss.
99 Thread-safe: all mutations are protected by a lock.
100 """
102 _MAX_BANKS = 500
104 def __init__(self, max_per_bank: int = DEFAULT_MAX_PER_BANK) -> None:
105 self.max_per_bank = max_per_bank
106 self._buffers: dict[str, deque[_RecentEntry]] = {}
107 self._lock = threading.Lock()
109 def add(self, bank_id: str, memory_id: str, text: str, metadata: dict | None = None) -> None:
110 """Add a recently stored chunk to the buffer."""
111 tokens = _tokenize(text)
112 if not tokens:
113 return # Skip empty/stop-word-only text
115 entry = _RecentEntry(
116 memory_id=memory_id,
117 text=text,
118 tokens=tokens,
119 bank_id=bank_id,
120 metadata=metadata,
121 )
123 with self._lock:
124 if bank_id not in self._buffers:
125 # Evict LRU bank if at capacity
126 if len(self._buffers) >= self._MAX_BANKS:
127 lru_bank = next(iter(self._buffers))
128 logger.warning(
129 "RecentMemoryBuffer at capacity (%d banks); evicting LRU bank '%s'",
130 self._MAX_BANKS,
131 lru_bank,
132 )
133 del self._buffers[lru_bank]
134 self._buffers[bank_id] = deque(maxlen=self.max_per_bank)
135 self._buffers[bank_id].append(entry)
137 def search(
138 self,
139 query: str,
140 bank_id: str,
141 limit: int = 10,
142 min_score: float = 0.3,
143 ) -> list[MemoryHit]:
144 """Fuzzy search recent memories for a bank.
146 Returns scored MemoryHits sorted by relevance, up to ``limit``.
147 Typically completes in <5ms for buffers of 100 entries.
148 """
149 query_tokens = _tokenize(query)
150 if not query_tokens:
151 return []
153 with self._lock:
154 entries = list(self._buffers.get(bank_id, []))
156 scored: list[tuple[float, _RecentEntry]] = []
157 for entry in entries:
158 score = _fuzzy_token_score(query_tokens, entry.tokens)
159 if score >= min_score:
160 scored.append((score, entry))
162 scored.sort(key=lambda x: x[0], reverse=True)
164 return [
165 MemoryHit(
166 text=entry.text,
167 score=score,
168 metadata=entry.metadata,
169 memory_id=entry.memory_id,
170 bank_id=bank_id,
171 source=stored_source(entry.metadata),
172 )
173 for score, entry in scored[:limit]
174 ]
176 def clear_bank(self, bank_id: str) -> None:
177 """Clear buffer for a bank."""
178 with self._lock:
179 self._buffers.pop(bank_id, None)
181 def size(self, bank_id: str | None = None) -> int:
182 """Number of buffered entries (total or per bank)."""
183 with self._lock:
184 if bank_id:
185 return len(self._buffers.get(bank_id, []))
186 return sum(len(b) for b in self._buffers.values())