ZerdaZerda

RetrievalCache

On this page

core.retrieval_cache.RetrievalCache

Overview

A thread-safe, TTL-aware LRU cache for retrieval results, keyed by a normalized (query, top_k) pair. It exists to avoid repeating expensive query expansion + vector search + reranking work for repeated or near-duplicate queries within a short time window.

Exact signature

Python · 2 lines
class RetrievalCache:
    def __init__(self, maxsize: int = 256, ttl: Optional[float] = 300.0) -> None: ...

Constructor

Parameters

NameTypeRequiredDefaultNotes
maxsizeintNo256Maximum number of distinct (query, top_k) entries. Oldest entry evicted (LRU) when exceeded.
ttlOptional[float]No300.0Seconds before an entry expires. None disables expiry entirely.

Initialization flow

Sets up an RLock, an empty OrderedDict store (key → (results, inserted_at)), and _hits/_misses counters at 0. Logs an INFO line with maxsize and ttl (formatted as "∞" if ttl is None).

Attributes

AttributeTypeDescription
maxsizeintAs passed to constructor.
ttlOptional[float]As passed to constructor.

Key construction

_normalize_query (module-level)

Python
def _normalize_query(query: str) -> str

re.sub(r"\s+", " ", query.strip().lower()) — collapses internal whitespace, strips leading/trailing whitespace, lowercases. Note this is not Unicode-normalization-aware (e.g. it won't treat Arabic diacritics or different Unicode representations of the same character as equal).

_make_key (module-level)

Python
def _make_key(query: str, top_k: int) -> str

hashlib.md5(f"{normalized_query}|topk={top_k}".encode()).hexdigest() — an MD5 hex digest used as the cache dict key.

get

Python
def get(self, query: str, top_k: int) -> Optional[List[Tuple[Any, float]]]

Parameters

NameTypeRequired
querystrYes
top_kintYes

Returns

Optional[List[Tuple[Any, float]]] — cached results, or None if absent or expired.

Algorithm

  1. Compute the key.
  2. Under the lock: if the key is absent, increment _misses and return None.
  3. If ttl is set and time.monotonic() - inserted_at > ttl, delete the entry, increment _misses, log a DEBUG "TTL expired" line, and return None.
  4. Otherwise: move the entry to the end of the OrderedDict (LRU touch), increment _hits, log a DEBUG "HIT" line (with a 40-char query preview), and return the cached results.

Thread safety

Fully thread-safe — every operation is performed under self._lock (an RLock).

Complexity

O(1) amortized (dict lookup + OrderedDict.move_to_end).

set

Python
def set(self, query: str, top_k: int, results: List[Tuple[Any, float]]) -> None

Algorithm

  1. Compute the key.
  2. Under the lock: if the key already exists, move it to the end first.
  3. Store (results, time.monotonic()) under the key.
  4. If the store now exceeds maxsize, evict the single oldest entry (next(iter(self._store))) and log a DEBUG eviction line.

Side effects

Mutates the internal store; may evict an unrelated entry if at capacity.

Complexity

O(1) amortized.

invalidate

Python
def invalidate(self, query: str, top_k: int) -> bool

Removes a specific (query, top_k) entry if present. Returns True if something was removed, False otherwise.

clear

Python
def clear(self) -> None

Empties the entire store and resets _hits/_misses to 0. Logs RetrievalCache cleared at INFO. Called automatically by RAGSystem.add_documents() whenever new documents are successfully indexed.

stats

Python
def stats(self) -> Dict[str, Any]

Returns, under the lock:

Python · 8 lines
{
  "size":     len(self._store),
  "maxsize":  self.maxsize,
  "ttl":      self.ttl,
  "hits":     self._hits,
  "misses":   self._misses,
  "hit_rate": round(self._hits / total, 3) if total else 0.0,   # total = hits + misses
}

hit_rate is 0.0 (not None or NaN) when no lookups have occurred yet.

__repr__

Text
RetrievalCache(size={size}/{maxsize}, hit_rate={hit_rate:.1%})

Examples

Basic:

Python · 7 lines
from zerda.rag.core import RetrievalCache

cache = RetrievalCache(maxsize=512, ttl=600)
cached = cache.get("رقم هاتف محمد", top_k=5)
if cached is None:
    result = vector_db.search("رقم هاتف محمد", top_k=5)
    cache.set("رقم هاتف محمد", top_k=5, results=result)

Disabling expiry:

Python
cache = RetrievalCache(maxsize=1000, ttl=None)  # entries never expire by time

Manual invalidation after an external update:

Python
cache.invalidate("outdated query", top_k=5)

Monitoring hit rate:

Python · 2 lines
print(cache.stats())
# {'size': 12, 'maxsize': 256, 'ttl': 300.0, 'hits': 40, 'misses': 15, 'hit_rate': 0.727}

Production-ready usage — periodic cache health logging:

Python · 7 lines
import time

while True:
    time.sleep(60)
    s = cache.stats()
    logger.info("Cache: %d/%d entries, %.1f%% hit rate",
                s["size"], s["maxsize"], s["hit_rate"] * 100)

Best practices

  • Set ttl=None only for genuinely static corpora — otherwise stale results can persist indefinitely between explicit clear()/invalidate() calls.
  • Remember RAGSystem.add_documents() clears the entire cache on every successful ingestion — for high-frequency ingestion pipelines, consider batching document adds to preserve cache warmth.

Common mistakes

  • Assuming cache keys are case- or whitespace-sensitive — they are normalized (_normalize_query), so "Refund Policy" and "refund policy" hit the same cache entry.
  • Forgetting that top_k is part of the cache key — get("query", top_k=5) and get("query", top_k=10) are different entries entirely.
  • RAGSystem.retrieve — the primary consumer.
  • RAGConfig — cache_maxsize/cache_ttl/enable_retrieval_cache fields configure this class.

Related pages