"""Caching vrstva pro knowledge plane (Fáze C2). Poskytuje: 1. `EmbeddingCache` — content-addressable cache embedingů klíčená SHA-256(text + model), persistentní přes SQLite. Zabraňuje re-computaci embedingů při opakovaném ingestu stejného obsahu (ByteRover 5-tier retrieval Tier 0). 2. `SearchResultCache` — in-memory LRU cache výsledků sémantického vyhledávání klíčená hashem dotazu + parametrů. 3. `DocumentFingerprint` — SHA-256 otisk dokumentu pro deduplikaci a detekci změn. Vzory z výzkumu: - Content-addressable embedding cache (ragit LRU cache, Cognee embedding cache) - SQLite-based persistent cache (LLM cache vzory z LangChain) - Document fingerprinting pro dedup (Cognee minimalhash, MinHash dedup) Použití: cache = EmbeddingCache(db_path="/path/to/embedding_cache.sqlite") vector = cache.get(text_hash, model) if vector is None: vector = embedding_service.embed_text(text) cache.set(text_hash, model, vector) """ from __future__ import annotations import hashlib import json import logging import sqlite3 import threading import time from collections import OrderedDict from dataclasses import dataclass, field from pathlib import Path from typing import Any logger = logging.getLogger(__name__) @dataclass(frozen=True) class DocumentFingerprint: """Otisk dokumentu pro deduplikaci a detekci změn.""" content_sha256: str size_bytes: int mtime: float @classmethod def from_path(cls, path: Path) -> DocumentFingerprint: content = path.read_bytes() return cls( content_sha256=hashlib.sha256(content).hexdigest(), size_bytes=len(content), mtime=path.stat().st_mtime, ) @classmethod def from_content(cls, content: bytes | str) -> DocumentFingerprint: if isinstance(content, str): content = content.encode("utf-8") return cls( content_sha256=hashlib.sha256(content).hexdigest(), size_bytes=len(content), mtime=time.time(), ) def cache_key(self, model: str) -> str: """Stabilní klíč pro embedding cache: sha256(content + model).""" raw = f"{self.content_sha256}:{model}" return hashlib.sha256(raw.encode()).hexdigest() class EmbeddingCache: """Persistentní content-addressable cache embedingů přes SQLite. Klíč: SHA-256(content_sha256 + model_name). Hodnota: serializovaný vektor jako JSON + metadata (model, dimenze, created_at, hit_count). """ def __init__(self, db_path: str | Path | None = None) -> None: self.db_path = Path(db_path or Path.home() / ".cyberbrain26" / "embedding_cache.sqlite") self.db_path.parent.mkdir(parents=True, exist_ok=True) self._lock = threading.RLock() self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False) self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA synchronous=NORMAL") self._init_schema() def _init_schema(self) -> None: self._conn.execute( """ CREATE TABLE IF NOT EXISTS embedding_cache ( cache_key TEXT PRIMARY KEY, model TEXT NOT NULL, vector_dim INTEGER NOT NULL, vector_json TEXT NOT NULL, created_at REAL NOT NULL, hit_count INTEGER NOT NULL DEFAULT 0, last_hit_at REAL ) """ ) self._conn.execute( "CREATE INDEX IF NOT EXISTS idx_embedding_cache_model ON embedding_cache(model)" ) self._conn.commit() def get(self, content_hash: str, model: str, vector_dim: int) -> list[float] | None: """Vrátí embeding z cache nebo None při cache miss.""" cache_key = hashlib.sha256(f"{content_hash}:{model}".encode()).hexdigest() with self._lock: row = self._conn.execute( "SELECT vector_json, vector_dim, hit_count, last_hit_at FROM embedding_cache WHERE cache_key = ?", (cache_key,), ).fetchone() if row is None: return None vector_json, stored_dim, hit_count, last_hit_at = row if stored_dim != vector_dim: # Rozdílná dimenze — cache miss (model se změnil) return None vector = json.loads(vector_json) now = time.time() self._conn.execute( "UPDATE embedding_cache SET hit_count = ?, last_hit_at = ? WHERE cache_key = ?", (hit_count + 1, now, cache_key), ) self._conn.commit() logger.debug("Embedding cache hit: key=%s model=%s hits=%d", cache_key[:12], model, hit_count + 1) return vector def set(self, content_hash: str, model: str, vector: list[float]) -> None: """Uloží embeding do cache.""" cache_key = hashlib.sha256(f"{content_hash}:{model}".encode()).hexdigest() vector_json = json.dumps(vector) now = time.time() with self._lock: self._conn.execute( """ INSERT OR REPLACE INTO embedding_cache (cache_key, model, vector_dim, vector_json, created_at, hit_count, last_hit_at) VALUES (?, ?, ?, ?, ?, 0, NULL) """, (cache_key, model, len(vector), vector_json, now), ) self._conn.commit() logger.debug("Embedding cache set: key=%s model=%s dim=%d", cache_key[:12], model, len(vector)) def invalidate(self, content_hash: str, model: str) -> int: """Invaliduje cache položku pro daný obsah a model. Vrátí počet smazaných řádků.""" cache_key = hashlib.sha256(f"{content_hash}:{model}".encode()).hexdigest() with self._lock: cursor = self._conn.execute( "DELETE FROM embedding_cache WHERE cache_key = ?", (cache_key,) ) self._conn.commit() return cursor.rowcount def clear(self) -> int: """Vymaže celou cache. Vrátí počet smazaných řádků.""" with self._lock: cursor = self._conn.execute("DELETE FROM embedding_cache") self._conn.commit() return cursor.rowcount def stats(self) -> dict[str, Any]: """Vrátí statistiky cache.""" with self._lock: total_rows = self._conn.execute( "SELECT COUNT(*) FROM embedding_cache" ).fetchone()[0] total_hits = self._conn.execute( "SELECT COALESCE(SUM(hit_count), 0) FROM embedding_cache" ).fetchone()[0] db_size = self.db_path.stat().st_size if self.db_path.exists() else 0 return { "total_entries": total_rows, "total_hits": total_hits, "db_size_bytes": db_size, "db_path": str(self.db_path), } def close(self) -> None: with self._lock: self._conn.close() class SearchResultCache: """In-memory LRU cache pro výsledky sémantického vyhledávání. Klíč: SHA-256(dotazu + limit + topic + tag). Hodnota: seznam výsledků + timestamp. LRU eviction při překročení max_size. """ def __init__(self, max_size: int = 1000, ttl_seconds: float = 300.0) -> None: self.max_size = max_size self.ttl_seconds = ttl_seconds self._cache: OrderedDict[str, tuple[list[dict[str, Any]], float]] = OrderedDict() self._lock = threading.RLock() self._hits = 0 self._misses = 0 def _make_key( self, query: str, limit: int, topic: str | None, tag: str | None ) -> str: raw = f"{query}:{limit}:{topic or ''}:{tag or ''}" return hashlib.sha256(raw.encode()).hexdigest() def get( self, query: str, limit: int, topic: str | None = None, tag: str | None = None ) -> list[dict[str, Any]] | None: key = self._make_key(query, limit, topic, tag) with self._lock: if key not in self._cache: self._misses += 1 return None results, cached_at = self._cache[key] if time.time() - cached_at > self.ttl_seconds: del self._cache[key] self._misses += 1 return None # LRU: přesunout na konec self._cache.move_to_end(key) self._hits += 1 return results def set( self, query: str, limit: int, results: list[dict[str, Any]], topic: str | None = None, tag: str | None = None, ) -> None: key = self._make_key(query, limit, topic, tag) with self._lock: if key in self._cache: self._cache.move_to_end(key) self._cache[key] = (results, time.time()) while len(self._cache) > self.max_size: self._cache.popitem(last=False) def invalidate(self) -> int: """Vymaže celou cache. Vrátí počet smazaných položek.""" with self._lock: count = len(self._cache) self._cache.clear() return count def stats(self) -> dict[str, Any]: with self._lock: total = self._hits + self._misses hit_rate = self._hits / total if total > 0 else 0.0 return { "hits": self._hits, "misses": self._misses, "hit_rate": hit_rate, "size": len(self._cache), "max_size": self.max_size, "ttl_seconds": self.ttl_seconds, } __all__ = [ "DocumentFingerprint", "EmbeddingCache", "SearchResultCache", ]