""" semantic.py — Semantic Memory Preferenze, concetti, conoscenza persistente. Database: Supabase (pgvector cosine similarity). Fallback: ilike + Jaccard. Backend server: HuggingFace Spaces (FastAPI). S568: similarity reale via Jaccard word-overlap (ilike path). S569: pgvector via HF Inference API — cosine similarity reale su embeddings BAAI/bge-small-en-v1.5. Attivazione: eseguire migration_pgvector.sql nel SQL Editor Supabase. Fallback automatico a ilike+Jaccard se pgvector non ancora attivato. S570: LRU cache degli embedding (256 entry, TTL 10 min) — evita ricalcolo HF API per query ripetute. """ from __future__ import annotations import hashlib import logging import os import time as _time import uuid from collections import OrderedDict from pathlib import Path # GAP-NEW-1: usa /data/chroma_db su HF Spaces (volume persistente), locale in dev # Priorità: 1) env CHROMA_DATA_DIR 2) /data/ se esiste (HF Spaces) 3) cwd (dev) import os as _os_gap1 # noqa: E402 — import qui per non rompere l'ordine top-level _CHROMA_DATA_DIR = _os_gap1.getenv("CHROMA_DATA_DIR") or ( "/data" if Path("/data").exists() else "." ) CHROMA_PATH = Path(_CHROMA_DATA_DIR) / "chroma_db" COLLECTION_NAME = "agente_semantic" EMBED_MODEL = "BAAI/bge-small-en-v1.5" # 384 dims EMBED_DIMS = 384 EMBED_MAX_CHARS = 1500 # bge-small max ~512 token ≈ 1500 chars try: import chromadb from chromadb.utils import embedding_functions CHROMA_AVAILABLE = True except ImportError: CHROMA_AVAILABLE = False _logger = logging.getLogger("semantic") # ── S570: LRU cache per embedding ───────────────────────────────────────────── _EMBED_CACHE_MAX = 256 # entry massime in memoria _EMBED_CACHE_TTL = 600.0 # secondi di validità (10 minuti) class _EmbedCache: """LRU cache O(1) per embedding vettoriali con TTL. Struttura: OrderedDict[sha256_key → (embedding, monotonic_timestamp)]. - get(): ritorna embedding se presente e non scaduto, else None. - set(): inserisce/aggiorna; evicta l'entry più vecchia se piena. - Sicuro in single-thread (asyncio è single-threaded per design). """ def __init__(self, max_size: int = _EMBED_CACHE_MAX, ttl: float = _EMBED_CACHE_TTL) -> None: self._cache: OrderedDict[str, tuple[list[float], float]] = OrderedDict() self._max = max_size self._ttl = ttl self.hits = 0 self.misses = 0 @staticmethod def _key(text: str) -> str: return hashlib.sha256(text.encode()).hexdigest()[:16] def get(self, text: str) -> list[float] | None: k = self._key(text) if k not in self._cache: self.misses += 1 return None emb, ts = self._cache[k] if _time.monotonic() - ts > self._ttl: del self._cache[k] self.misses += 1 return None self._cache.move_to_end(k) # LRU touch self.hits += 1 return emb def set(self, text: str, emb: list[float]) -> None: k = self._key(text) if k in self._cache: self._cache.move_to_end(k) elif len(self._cache) >= self._max: self._cache.popitem(last=False) # evicta LRU (il più vecchio) self._cache[k] = (emb, _time.monotonic()) def __len__(self) -> int: return len(self._cache) def stats(self) -> dict: total = self.hits + self.misses return { "size": len(self._cache), "max_size": self._max, "ttl_s": self._ttl, "hits": self.hits, "misses": self.misses, "hit_rate": round(self.hits / total, 3) if total else 0.0, } class SemanticMemory: def __init__(self): self._client = None # chromadb fallback self._collection = None self._embed_fn = None self._sb = None # Supabase client self._hf_client = None # HuggingFace InferenceClient (lazy) self._pgvector = False # S569: True quando match_semantic_memory RPC disponibile self._embed_cache = _EmbedCache() # S570: LRU 256 entry, TTL 10 min self.available = False def _try_supabase(self): url = os.getenv("SUPABASE_URL", "") key = os.getenv("SUPABASE_ANON_KEY") or os.getenv("SUPABASE_KEY", "") if not url or not key: return None try: from supabase import create_client return create_client(url, key) except Exception: return None def init(self): self._sb = self._try_supabase() if self._sb: try: self._sb.table("semantic_memory").select("id").limit(1).execute() self.available = True # S569: proba se match_semantic_memory RPC esiste (pgvector attivato) _zero_emb = "[" + ",".join(["0.0"] * EMBED_DIMS) + "]" try: self._sb.rpc("match_semantic_memory", { "query_embedding": _zero_emb, "match_count": 1, "match_threshold": 0.99, }).execute() self._pgvector = True _logger.info("SemanticMemory: Supabase ✓ pgvector ✓ (cosine similarity attiva)") except Exception: self._pgvector = False _logger.warning( "SemanticMemory: Supabase pgvector non attivo — ilike+Jaccard. Esegui migration_pgvector.sql per cosine similarity reale." ) return except Exception as e: if "PGRST205" in str(e) or "not found" in str(e).lower(): _logger.warning( "SemanticMemory: tabella semantic_memory mancante su Supabase. Crea la tabella base e poi esegui migration_pgvector.sql." ) else: _logger.warning("SemanticMemory: Supabase error: %s", e) self._sb = None # Fallback ChromaDB locale (HF Spaces senza Supabase) if not CHROMA_AVAILABLE: _logger.warning("SemanticMemory: disabilitata (chromadb non installato, tabella Supabase mancante).") return try: self._embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction( model_name=EMBED_MODEL ) self._client = chromadb.PersistentClient(path=str(CHROMA_PATH)) self._collection = self._client.get_or_create_collection( name=COLLECTION_NAME, embedding_function=self._embed_fn, metadata={"hnsw:space": "cosine"}, ) self.available = True _logger.info("SemanticMemory: ChromaDB locale (fallback) ✓") except Exception as e: _logger.warning("SemanticMemory non disponibile: %s", e) # ── Embedding ───────────────────────────────────────────────────────────── def _embed(self, text: str) -> list[float] | None: """S569/S570: Calcola embedding via HF Inference API con LRU cache. - S569: usa BAAI/bge-small-en-v1.5 (384 dims) via HuggingFace InferenceClient. - S570: controlla _embed_cache prima della chiamata HTTP — cache hit ≈ 0ms vs ~200ms API. Cache: 256 entry, TTL 10 min, eviction LRU. Key: sha256(text)[:16]. Ritorna None se HF_TOKEN mancante o API non raggiungibile → fallback trasparente. """ _txt_key = text[:EMBED_MAX_CHARS] # S570: cache hit → ritorna senza chiamata HF (≈0ms) cached = self._embed_cache.get(_txt_key) if cached is not None: return cached # Cache miss → chiama HF Inference API try: from huggingface_hub import InferenceClient if self._hf_client is None: token = os.getenv("HF_TOKEN", "") if not token: if not getattr(self, '_hf_token_warned', False): _logger.warning("SemanticMemory: HF_TOKEN mancante — embedding HF disabilitato, fallback a ilike search") self._hf_token_warned = True return None self._hf_client = InferenceClient(token=token) # S750-GAP-C: timeout 10s su HF Inference API (sync bloccante). # ThreadPoolExecutor.submit().result(timeout) solleva concurrent.futures.TimeoutError # se la chiamata supera 10s — evita hang infinito del worker asyncio. import concurrent.futures as _cf with _cf.ThreadPoolExecutor(max_workers=1) as _pool: _fut = _pool.submit( self._hf_client.feature_extraction, text=_txt_key, model=EMBED_MODEL, ) try: result = _fut.result(timeout=10.0) except _cf.TimeoutError: return None # huggingface_hub ritorna np.ndarray shape (384,) o (1, 384) if hasattr(result, "tolist"): flat = result.tolist() else: flat = list(result) # Normalizza shape (1, 384) → (384,) if flat and isinstance(flat[0], list): flat = flat[0] if len(flat) != EMBED_DIMS: return None # S570: salva in cache prima di ritornare self._embed_cache.set(_txt_key, flat) return flat except Exception: return None @staticmethod def _emb_to_pg(emb: list[float]) -> str: """Serializza embedding a stringa vettoriale per PostgREST vector type: '[x,y,...]'.""" return "[" + ",".join(f"{x:.8f}" for x in emb) + "]" # ── Write ───────────────────────────────────────────────────────────────── def add(self, text: str, metadata: dict | None = None, doc_id: str | None = None): _id = doc_id or str(uuid.uuid4()) _txt = text[:4000] # S568: era 2000, aumentato per output codice if self._sb: try: row: dict = {"id": _id, "content": _txt, "metadata": metadata or {}} if self._pgvector: # S569: calcola embedding e salva nel campo vector — abilita cosine search emb = self._embed(_txt) if emb is not None: row["embedding"] = self._emb_to_pg(emb) self._sb.table("semantic_memory").upsert(row).execute() return except Exception as _e: _logger.warning("semantic.add: Supabase upsert failed: %s", _e) if self._collection: self._collection.upsert( documents=[text], metadatas=[metadata or {}], ids=[_id], ) # ── Read ────────────────────────────────────────────────────────────────── @staticmethod def _word_overlap_similarity(query: str, content: str) -> float: """S568: stima similarity via word-overlap (Jaccard) tra query e content. Usato come fallback quando pgvector non è attivo. Range reale: 0.05-0.95 (mai 0.0 — ilike garantisce match parziale). """ q_words = set(query.lower().split()) c_words = set(content.lower().split()) if not q_words or not c_words: return 0.5 intersection = len(q_words & c_words) union = len(q_words | c_words) return round(intersection / union, 4) if union else 0.5 def search(self, query: str, n_results: int = 5) -> list[dict]: if self._sb: # S569: path pgvector — cosine similarity reale via RPC if self._pgvector: emb = self._embed(query) if emb is not None: try: rows = ( self._sb.rpc("match_semantic_memory", { "query_embedding": self._emb_to_pg(emb), "match_count": n_results, "match_threshold": 0.2, # soglia coseno: 0.2 = rilevante }).execute().data or [] ) return [ { "content": r["content"], "metadata": r.get("metadata", {}), "similarity": round(float(r.get("similarity", 0)), 4), } for r in rows ] except Exception: pass # RPC fallita → fallback ilike sotto # Fallback ilike + Jaccard (pgvector non attivo o embedding non disponibile) try: rows = ( self._sb.table("semantic_memory") .select("*") .ilike("content", f"%{query}%") .limit(n_results) .execute().data or [] ) return [ { "content": r["content"], "metadata": r.get("metadata", {}), "similarity": self._word_overlap_similarity(query, r["content"]), } for r in rows ] except Exception as _exc: _logger.debug("[semantic] silenced %s", type(_exc).__name__) # noqa: BLE001 if not self._collection: return [] try: results = self._collection.query( query_texts=[query], n_results=min(n_results, self._collection.count() or 1), ) docs = results.get("documents", [[]])[0] metas = results.get("metadatas", [[]])[0] dists = results.get("distances", [[]])[0] return [ {"content": doc, "metadata": meta, "similarity": round(1 - dist, 4)} for doc, meta, dist in zip(docs, metas, dists) ] except Exception: return [] def count(self) -> int: if self._sb: try: return self._sb.table("semantic_memory").select("id", count="exact").execute().count or 0 except Exception as _exc: _logger.debug("[semantic] silenced %s", type(_exc).__name__) # noqa: BLE001 return self._collection.count() if self._collection else 0 def stats(self) -> dict: backend = "supabase+pgvector" if (self._sb and self._pgvector) else \ ("supabase" if self._sb else ("chroma" if self._collection else "none")) return { "available": self.available, "documents": self.count(), "model": EMBED_MODEL, "backend": backend, "pgvector": self._pgvector, "embed_cache": self._embed_cache.stats(), # S570: hit_rate, size, TTL } # ── Export / Import ─────────────────────────────────────────────────────── def export_all(self, limit: int = 2000) -> list[dict]: """Esporta tutti i record della semantic memory come lista di dict portabile. Funziona su entrambi i backend (Supabase e ChromaDB). Usato da GET /api/memory/sync/export per snapshot cross-sessione. """ if self._sb: try: rows = ( self._sb.table("semantic_memory") .select("id, content, metadata") .limit(limit) .execute().data or [] ) return [ {"id": r["id"], "content": r["content"], "metadata": r.get("metadata") or {}} for r in rows ] except Exception as exc: _logger.warning("SemanticMemory.export_all (supabase) error: %s", exc) return [] if self._collection: try: n = self._collection.count() res = self._collection.get(limit=min(n, limit)) if n else {"ids": [], "documents": [], "metadatas": []} ids = res.get("ids", []) docs = res.get("documents", []) metas = res.get("metadatas", []) return [ {"id": i, "content": d, "metadata": m or {}} for i, d, m in zip(ids, docs, metas) ] except Exception as exc: _logger.warning("SemanticMemory.export_all (chroma) error: %s", exc) return [] return [] def import_all(self, records: list[dict], overwrite: bool = False) -> dict: """Importa una lista di record nella semantic memory (upsert idempotente). Ogni record: {"id": str, "content": str, "metadata": dict}. Se overwrite=False (default) usa upsert — record esistenti vengono aggiornati. Usato da POST /api/memory/sync/import per restore cross-sessione. """ import json as _json imported = 0 skipped = 0 for r in records: content = str(r.get("content", "")).strip() if not content: skipped += 1 continue doc_id = r.get("id") or None metadata = r.get("metadata", {}) if isinstance(metadata, str): try: metadata = _json.loads(metadata) except Exception: metadata = {} try: self.add(content, metadata or {}, doc_id=doc_id) imported += 1 except Exception: skipped += 1 return {"imported": imported, "skipped": skipped, "total": len(records)}