"""Unified memory service for the knowledge plane. Tied together all knowledge plane components into a single callable API: file registration → chunking → embedding → vector store → navigation → semantic search → retrieval """ from __future__ import annotations import json import logging import time from pathlib import Path from typing import Any, cast from cyberbrain26.data_plane.sqlite_store import SQLiteDatabase from cyberbrain26.knowledge_plane.caching import ( DocumentFingerprint, EmbeddingCache, SearchResultCache, ) from cyberbrain26.knowledge_plane.classifier import ( ClassificationResult, LLMClassifier, detect_sensitivity_from_bytes, ) from cyberbrain26.knowledge_plane.contracts import ( DocumentRegistration, DocumentStatus, DocumentTrust, KG1Node, KnowledgeDocument, KnowledgeSource, Sensitivity, ) from cyberbrain26.knowledge_plane.deduplication import ( DeduplicationEngine, DeduplicationResult, ) from cyberbrain26.knowledge_plane.embedding_config import EmbeddingConfig from cyberbrain26.knowledge_plane.embedding_service import EmbeddingService from cyberbrain26.knowledge_plane.local_source import LocalDirectorySource from cyberbrain26.knowledge_plane.markdown_store import MarkdownStore from cyberbrain26.knowledge_plane.monitoring import KnowledgeMetrics from cyberbrain26.knowledge_plane.navigation import ( KnowledgeNavigator, ) from cyberbrain26.knowledge_plane.repository import SQLiteKnowledgeRepository from cyberbrain26.knowledge_plane.vector_store import ( KnowledgeVectorStore, ) from cyberbrain26.security_plane.audit import AuditEventType, AuditLog logger = logging.getLogger(__name__) class MemoryService: """Jednotný vstupní bod pro paměťovou vrstvu knowledge plane. Spolupracuje: - SQLiteKnowledgeRepository pro metadata dokumentů - LocalDirectorySource pro čtení lokálních souborů - TextChunker pro dělení textu na chunky - EmbeddingService pro generování vektorů přes Ollama - KnowledgeVectorStore (LanceDB) pro vektorové uložiště a sémantické vyhledávání - KnowledgeNavigator pro topic/tag navigaci - DeduplicationEngine pro detekci duplicit """ def __init__( self, *, database: SQLiteDatabase, audit_log: AuditLog | None = None, config: EmbeddingConfig | None = None, repository: SQLiteKnowledgeRepository | None = None, vector_store: KnowledgeVectorStore | None = None, embedding_service: EmbeddingService | None = None, navigator: KnowledgeNavigator | None = None, deduplication_engine: DeduplicationEngine | None = None, markdown_root: str | Path | None = None, auto_classify: bool = True, ) -> None: self.database = database self.audit_log = audit_log self.config = config or EmbeddingConfig() self.repository = repository or SQLiteKnowledgeRepository(database, audit_log=audit_log) self.vector_store = vector_store or KnowledgeVectorStore(config=self.config) self.embedding_service = embedding_service or EmbeddingService(config=self.config) self.navigator = navigator or KnowledgeNavigator() self.deduplication_engine = ( deduplication_engine or DeduplicationEngine( embedding_service=self.embedding_service, vector_store=self.vector_store, ) ) self.local_source = LocalDirectorySource( source_id="local", root=Path(".").resolve(), trust=DocumentTrust.REVIEWED, sensitivity=Sensitivity.PERSONAL, ) # Auto-categorization pipeline (per ByteRover / Memex research) self.auto_classify = auto_classify self.classifier = LLMClassifier(config=self.config) # Markdown storage for auto-categorized documents (per ByteRover / NeuRAG research) self.markdown_store = MarkdownStore( root=markdown_root or Path(self.config.lancedb_directory).parent / "knowledge_docs", auto_classify=auto_classify, ) # Embedding content-addressable cache (per MemStrata pattern) # Prevents re-computation of embeddings for same content + model cache_dir = Path(self.config.lancedb_directory).parent / "embedding_cache" self.embedding_cache = EmbeddingCache(db_path=str(cache_dir)) # Search result cache for frequently queried topics self.search_cache = SearchResultCache(max_size=100, ttl_seconds=300.0) # Monitoring metrics for knowledge plane self.metrics = KnowledgeMetrics() def _classify_document( self, content: bytes, source_uri: str, source_id: str ) -> ClassificationResult | None: """Run LLM classification on document content. Returns ClassificationResult or None if classification is disabled or fails. Fail-closed: never returns a partial result — caller must handle the None case by storing document as unclassified. """ if not self.auto_classify: return None try: text = content.decode("utf-8", errors="replace") return self.classifier.classify_document(text) except Exception as exc: # noqa: BLE001 - fail closed: log, return None logger.warning("Classification failed for %s: %s", source_uri, exc) return None def register_file( self, path: str | Path, *, source_id: str | None = None, root: str | Path | None = None, trust: DocumentTrust | None = None, sensitivity: Sensitivity | None = None, approve: bool = True, ) -> tuple[KnowledgeDocument, bool]: """Zaregistruje lokální soubor do paměťového systému. Postup: 1. Čte soubor z disku 2. (NEW) Klasifikuje obsah přes LLM (pokud auto_classify=True) 3. Ukládá klasifikaci do Markdown store 4. Rozdělí na chunky 5. Vygeneruje embeddingy 6. Uloží do vektorového úložiště 7. Proindexuje do navigátoru (topics, tags) 8. Zaznamená do auditu """ file_path = Path(path).expanduser().resolve(strict=True) content = file_path.read_bytes() # --- Step 1: PII/sensitivity detection (MemStrata pattern) --- # Deterministic, no LLM call — auto-upgrade sensitivity if PII detected detected_sensitivity = detect_sensitivity_from_bytes(content) sensitivity_rank = { "public": 0, "personal": 1, "sensitive": 2, "restricted": 3, } if sensitivity is None: # Use detected sensitivity as default try: sensitivity = Sensitivity(detected_sensitivity) except ValueError: sensitivity = Sensitivity.PERSONAL else: # Auto-upgrade if detected sensitivity is higher than explicit if sensitivity_rank.get(detected_sensitivity, 0) > sensitivity_rank.get(sensitivity.value, 0): logger.warning( "Auto-upgrading sensitivity for %s from %s to %s (PII detected)", file_path.name, sensitivity.value, detected_sensitivity, ) try: sensitivity = Sensitivity(detected_sensitivity) except ValueError: pass # --- Step 2: Auto-classification (per ByteRover / Memex research) --- classification = self._classify_document(content, str(file_path), source_id or "local") if classification is not None: try: self.markdown_store.store_classified( classification=classification, content=content.decode("utf-8", errors="replace"), source_uri=str(file_path), source_id=source_id or "local", filename=file_path.name, ) except Exception as exc: # noqa: BLE001 - fail closed: log, continue logger.warning("Markdown store failed for %s: %s", file_path, exc) # --- Steps 3-8: Original registration pipeline --- if root is not None: root_path = Path(root).expanduser().resolve(strict=True) source = LocalDirectorySource( source_id=source_id or f"local-{root_path.name}", root=root_path, trust=trust or DocumentTrust.REVIEWED, sensitivity=sensitivity or Sensitivity.PERSONAL, ) document, created = source.register_file( self.repository, path, vector_store=self.vector_store, embedding_service=self.embedding_service, ) else: # Použij rodičovský adresář souboru jako dočasný root # (pro testy, kde soubor není v pracovním adresáři) root_path = file_path.parent.expanduser().resolve(strict=True) source = LocalDirectorySource( source_id=source_id or f"local-{root_path.name}", root=root_path, trust=trust or DocumentTrust.REVIEWED, sensitivity=sensitivity or Sensitivity.PERSONAL, ) document, created = source.register_file( self.repository, path, vector_store=self.vector_store, embedding_service=self.embedding_service, ) # Update metadata with classification result if available if classification is not None: document.metadata["auto_classified"] = True document.metadata["classification_topic"] = classification.topic document.metadata["classification_subtopic"] = classification.subtopic document.metadata["classification_tags"] = classification.tags document.metadata["classification_summary"] = classification.summary document.metadata["classification_confidence"] = classification.confidence document.metadata["classification_intent"] = classification.intent document.metadata["classification_entities"] = classification.entities # Set temporal validity (MemStrata pattern) import time current_time = time.time() from dataclasses import replace document = replace(document, valid_from=current_time) if created and self.audit_log is not None: self.audit_log.record( AuditEventType.MEMORY_UPDATED, action="memory_file_registered", resource=document.document_id, memory_write=document.document_id, metadata={ "source_id": document.source_id, "source_uri": document.source_uri, "trust_level": document.trust.value, "sensitivity": document.sensitivity.value, "status": document.status.value, "auto_classified": classification is not None, }, ) # Proindexovat do navigátoru self._index_in_navigator(document) return document, created def register_text( self, content: bytes, *, source_uri: str, source_id: str = "local-text", topic: str | None = None, tags: list[str] | None = None, entities: list[str] | None = None, trust: DocumentTrust = DocumentTrust.REVIEWED, sensitivity: Sensitivity = Sensitivity.PERSONAL, ) -> tuple[KnowledgeDocument, bool]: """Zaregistruje čistý text (v paměti) do paměťového systému.""" source = KnowledgeSource( source_id=source_id, source_uri=source_uri, trust=trust, sensitivity=sensitivity, approved=True, ) registration = DocumentRegistration( source=source, content=content, media_type="text/plain", metadata={"topic": topic or "", "tags": ",".join(tags or []), "entities": ",".join(entities or [])}, ) document, created = self.repository.register( registration, vector_store=self.vector_store, embedding_service=self.embedding_service, ) if created and self.audit_log is not None: self.audit_log.record( AuditEventType.MEMORY_UPDATED, action="memory_text_registered", resource=document.document_id, memory_write=document.document_id, metadata={ "source_id": document.source_id, "source_uri": source_uri, "topic": topic, "tags": tags or [], "entities": entities or [], }, ) # Proindexovat do navigátoru self._index_in_navigator(document, topic=topic, tags=tags, entities=entities) return document, created def semantic_search( self, query_text: str, *, limit: int = 10, topic: str | None = None, tag: str | None = None, ) -> list[dict[str, Any]]: """Sémantické vyhledávání v paměťovém systému. Postup: 1. Cache lookup přes SearchResultCache (TTL-based) 2. Vytvoří embedding pro dotaz (z content-addressable cache pokud možné) 3. Hledá v LanceDB vector store 4. Volitelně filtruje podle topiku/štítku přes navigátor """ import time start_time = time.monotonic() # Cache lookup (SearchResultCache) cached = self.search_cache.get(query_text, limit, topic=topic, tag=tag) cache_hit = cached is not None self.metrics.record_search_cache_hit(cache_hit) if cached is not None: self.metrics.record_vector_search( latency_seconds=time.monotonic() - start_time, best_distance=cached[0].get("score") if cached else None, ) return cached # Content-addressable embedding cache for query query_hash = DocumentFingerprint.from_content(query_text).content_sha256 query_vector = self.embedding_cache.get(query_hash, self.config.model, self.config.vector_dim) if query_vector is None: embed_start = time.monotonic() query_vector = self.embedding_service.embed_text(query_text) self.metrics.record_embedding_latency( latency_seconds=time.monotonic() - embed_start, cached=False, ) self.embedding_cache.set(query_hash, self.config.model, query_vector) else: self.metrics.record_embedding_latency( latency_seconds=0.0, cached=True, ) search_start = time.monotonic() results = self.vector_store.search(query_vector, limit=limit) search_latency = time.monotonic() - search_start # Volitelné filtrování přes navigátor if topic or tag: candidate_ids = self._get_candidate_ids(topic=topic, tag=tag) if candidate_ids: results = [ r for r in results if r.get("id") in candidate_ids ] # Uložit do search cache self.search_cache.set(query_text, limit, results, topic=topic, tag=tag) # Record vector search metrics best_distance = results[0].get("score") if results else None self.metrics.record_vector_search( latency_seconds=search_latency, best_distance=best_distance, ) return results @staticmethod def _rrf_fuse( results_a: list[dict[str, Any]], results_b: list[dict[str, Any]], k: int = 60, ) -> list[dict[str, Any]]: """Reciprocal Rank Fusion (RRF) of two result lists. Combines two ranked result lists by summing normalized reciprocal ranks. Higher final score = more relevant. Works even if the two lists have different sets of items (items only in one list get rank from that list only). Adapted from RAGraph research: https://arxiv.org/abs/2408.03245 """ # Build rank maps: id -> position (1-indexed) rank_a: dict[str, int] = {} for i, r in enumerate(results_a, start=1): doc_id = r.get("id") or r.get("document_id") or r.get("path") if doc_id and doc_id not in rank_a: rank_a[doc_id] = i rank_b: dict[str, int] = {} for i, r in enumerate(results_b, start=1): doc_id = r.get("id") or r.get("document_id") or r.get("path") if doc_id and doc_id not in rank_b: rank_b[doc_id] = i # Collect all unique doc_ids all_ids: set[str] = set(rank_a.keys()) | set(rank_b.keys()) # Compute fused scores fused_scores: dict[str, float] = {} for doc_id in all_ids: rank = rank_a.get(doc_id, float("inf")) + rank_b.get(doc_id, float("inf")) fused_scores[doc_id] = k / rank if rank != float("inf") else 0.0 # Build fused result list, sorted by fused score descending fused_results: list[dict[str, Any]] = [] for doc_id in sorted(all_ids, key=lambda did: fused_scores.get(did, 0), reverse=True): # Find the best original result for this doc_id best_result: dict[str, Any] | None = None best_score = -1.0 for r in results_a + results_b: rid = r.get("id") or r.get("document_id") or r.get("path") if rid == doc_id and r.get("score", 0) > best_score: best_result = r best_score = r.get("score", 0) if best_result is not None: best_result["rrf_score"] = fused_scores[doc_id] fused_results.append(best_result) return fused_results def keyword_search( self, query: str, *, limit: int = 10, ) -> list[dict[str, Any]]: """Keyword search over document metadata (topic, tags, summary). Searches across topic, tags, and summary fields in the SQLite knowledge_documents table metadata_json using LIKE patterns. This provides keyword-based retrieval complementary to semantic (vector) search. Research reference: RAGraph — hybrid BM25+semantic retrieval with RRF fusion. """ query_lower = query.lower() with self.database.connection() as connection: rows = connection.execute( """ SELECT document_id, source_uri, media_type, metadata_json FROM knowledge_documents WHERE LOWER(source_uri) LIKE ? OR LOWER(metadata_json) LIKE ? """, (f"%{query_lower}%", f"%{query_lower}%"), ).fetchall() results: list[dict[str, Any]] = [] for row in rows: metadata = json.loads(row["metadata_json"]) if row["metadata_json"] else {} topic = (metadata.get("topic") or "").lower() tags = [t.lower() for t in (metadata.get("tags") or [])] summary = (metadata.get("summary") or "").lower() # Match if query appears in topic, any tag, or summary matched = False score = 1.0 # base score if query_lower in topic: matched = True score += 10 if any(query_lower in t for t in tags): matched = True score += 5 if query_lower in summary: matched = True score += 1 if matched: results.append( { "id": row["document_id"], "source_uri": row["source_uri"], "media_type": row["media_type"], "text": metadata.get("summary", ""), "score": score, } ) # Sort by relevance score results.sort(key=lambda r: r.get("score", 0), reverse=True) return results[:limit] def hybrid_search( self, query_text: str, *, limit: int = 10, topic: str | None = None, tag: str | None = None, ) -> list[dict[str, Any]]: """Hybrid search: keyword (metadata) + semantic (vector) + RRF fusion. Combines keyword search over document metadata (topic, tags, summary) with semantic (vector) search via Ollama embeddings, then fuses the two result lists using Reciprocal Rank Fusion (RRF) per RAGraph research. Postup: 1. Keyword search over topic/tags/summary → results_a 2. Semantic search via Ollama embeddings → results_b 3. RRF fusion → ranked fused results with rrf_score Research reference: RAGraph (arXiv:2408.03245) — hybrid BM25+vector retrieval with reciprocal rank fusion for improved precision. """ # Step 1: Keyword search keyword_results = self.keyword_search(query_text, limit=limit * 2) # Step 2: Semantic search semantic_results = self.semantic_search( query_text, limit=limit * 2, topic=topic, tag=tag ) # Step 3: RRF fusion fused = self._rrf_fuse(keyword_results, semantic_results, k=60) return fused[:limit] def graph_search( self, entity_query: str, *, limit: int = 10, depth: int = 2, relationship_type: str | None = None, ) -> list[dict[str, Any]]: """Graph-based retrieval via entity relationship traversal (MemGraphRAG). Performs breadth-first traversal over the entity relationship graph starting from the query entity, collecting all documents reachable within `depth` hops. Returns documents sorted by graph proximity. Research reference: MemGraphRAG — memory-aware hierarchical retrieval through entity co-occurrence graphs. Postup: 1. Najde entity v navigátoru podle entity_query 2. Provede BFS do hloubky `depth` přes KnowledgeNavigator.get_related_entities 3. Sbírá všechny dokumenty spojené s nalezenými entity 4. Řadí dokumenty podle vzdálenosti v grafu (blíž = relevantnější) """ # Normalize query to match entity names (case-insensitive substring) query_lower = entity_query.lower() all_entities = sorted(self.navigator._entity_graph.entities.keys()) # Find starting entities that match the query starting_entities: list[str] = [] for entity in all_entities: if query_lower in entity.lower(): starting_entities.append(entity) if not starting_entities: return [] # BFS traversal collecting (entity, distance) pairs visited_entities: set[str] = set() entity_distances: dict[str, int] = {} for start_entity in starting_entities: if start_entity in visited_entities: continue visited_entities.add(start_entity) entity_distances[start_entity] = 0 related = self.navigator.get_related_entities( start_entity, depth=depth, relationship_type=relationship_type ) for distance, entity_list in related.items(): for entity in entity_list: if entity not in visited_entities: visited_entities.add(entity) entity_distances[entity] = int(min( entity_distances.get(entity, 999), distance )) # Collect all documents linked to visited entities document_scores: dict[str, float] = {} for entity, distance in entity_distances.items(): doc_ids = self.navigator.get_entity_documents(entity) for doc_id in doc_ids: # Closer entities contribute higher score # distance 0 = direct match (score 1.0), distance 1 = 0.5, etc. contribution = 1.0 / (1.0 + distance) document_scores[doc_id] = document_scores.get(doc_id, 0.0) + contribution # Build result list sorted by graph score descending results: list[dict[str, Any]] = [] for doc_id, score in sorted( document_scores.items(), key=lambda item: item[1], reverse=True ): document = self.get_document(doc_id) if document is not None: results.append( { "id": doc_id, "source_uri": document.source_uri, "media_type": document.media_type, "text": document.metadata.get("summary", ""), "score": score, "graph_distance": min( entity_distances.get(e, 999) for e in self.navigator._entity_graph.entity_documents if doc_id in self.navigator._entity_graph.entity_documents.get(e, set()) ), "matched_entities": [ e for e in visited_entities if doc_id in self.navigator.get_entity_documents(e) ], } ) return results[:limit] def get_document(self, document_id: str) -> KnowledgeDocument | None: """Vrátí dokument podle ID.""" return self.repository.get(document_id) def search_by_topic(self, topic: str) -> list[str]: """Vrátí dokumentová ID podle topiku.""" return self.navigator.get_documents_by_topic(topic) def search_by_tag(self, tag: str) -> list[str]: """Vrátí dokumentová ID podle štítku.""" return self.navigator.get_documents_by_tag(tag) def get_topics(self) -> list[str]: """Vrátí všechny známé topiky.""" return self.navigator.get_topics() def get_tags(self) -> list[str]: """Vrátí všechny známé štítky.""" return self.navigator.get_all_tags() def check_duplicate( self, content: bytes, source_uri: str, ) -> DeduplicationResult: """Zkontroluje, zda je obsah duplicitní.""" existing_documents = [ self.get_document(doc_id) for doc_id in self._all_document_ids() ] existing_documents = [d for d in existing_documents if d is not None] cast_documents: list[KnowledgeDocument] = cast( list[KnowledgeDocument], existing_documents ) return self.deduplication_engine.check_duplicate( content, source_uri, cast_documents ) def health_check(self) -> dict[str, Any]: """Rychlá kontrola stavu všech komponent paměťového systému.""" embedding_health = self.embedding_service.health_check() return { "embedding": embedding_health, "vector_store_count": self.vector_store.count(), "navigator_topics": len(self.navigator.get_topics()), "navigator_tags": len(self.navigator.get_all_tags()), "healthy": embedding_health.get("healthy", False), } def register_kg1_node( self, node: KG1Node, *, audit: bool = True, ) -> KG1Node: """Jurassic-pattern: register an Instance Resource Graph node. Stores the node in the knowledge graph layer (navigation + audit). The node must have business_semantics (IPOMSQ) and domain_constraints. """ if not node.business_semantics: raise ValueError("KG1Node requires business_semantics (IPOMSQ framework)") # Register in navigator as special entity self.navigator.add_document( KnowledgeDocument( document_id=node.id, source_id=node.source.source_id, source_uri=node.source.source_uri, content_sha256="", media_type="knowledge/graph", trust=node.source.trust, sensitivity=node.source.sensitivity, status=DocumentStatus.INDEXED, metadata={ "kg1_type": node.type, "business_semantics": node.business_semantics, "domain_constraints": node.domain_constraints, "evidence": node.evidence, **node.metadata, }, created_at=time.time(), updated_at=time.time(), ) ) if audit and self.audit_log is not None: self.audit_log.record( AuditEventType.MEMORY_UPDATED, action="memory_kg1_registered", resource=node.id, memory_write=node.id, metadata={ "kg1_type": node.type, "source_id": node.source.source_id, "business_semantics_keys": list(node.business_semantics.keys()), "evidence_count": len(node.evidence), }, ) return node def close(self) -> None: """Uzavře všechna připojení.""" self.vector_store.close() self.embedding_service.close() # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _index_in_navigator( self, document: KnowledgeDocument, *, topic: str | None = None, tags: list[str] | None = None, entities: list[str] | None = None, ) -> None: """Přidá dokument do KnowledgeNavigator.""" # Pokud jsou zadány topic/tags/entities, použijeme je přímo if topic or tags or entities: metadata = document.metadata metadata["topic"] = topic or metadata.get("topic", "uncategorized") if tags: metadata["tags"] = tags if entities: metadata["entities"] = entities # Vytvořit kopii dokumentu s aktualizovanými metadaty from dataclasses import replace updated_document = replace(document, metadata=metadata) self.navigator.add_document(updated_document) else: self.navigator.add_document(document) def _get_candidate_ids( self, *, topic: str | None = None, tag: str | None = None, ) -> set[str]: """Získá sadu dokumentových ID pro daný topik/štítek.""" ids: set[str] = set() if topic: ids.update(self.navigator.get_documents_by_topic(topic)) if tag: ids.update(self.navigator.get_documents_by_tag(tag)) return ids def _all_document_ids(self) -> list[str]: """Vrátí všechny dokumentová ID z navigátoru.""" all_ids: list[str] = [] for topic in self.navigator.get_topics(): all_ids.extend(self.navigator.get_documents_by_topic(topic)) # Deduplikovat (dokument může být v více topikách) return list(set(all_ids)) __all__ = ["MemoryService"]