From 981b0cba1f04a5373aab682919a8f29991dbe08f Mon Sep 17 00:00:00 2001 From: Lars Date: Fri, 9 Jan 2026 23:29:41 +0100 Subject: [PATCH] Update ingestion_db.py and ingestion_processor.py to version 3.3.1: Enhance documentation for clarity, refine edge validation logic, and improve logging mechanisms. Implement strict separation of explicit writes and symmetry validation in the two-phase ingestion workflow, ensuring data authority and integrity. Adjust comments for better understanding and maintainability. --- app/core/ingestion/ingestion_db.py | 3 +- app/core/ingestion/ingestion_processor.py | 119 ++++++++++++++-------- 2 files changed, 76 insertions(+), 46 deletions(-) diff --git a/app/core/ingestion/ingestion_db.py b/app/core/ingestion/ingestion_db.py index cad4c0c..c90fdfa 100644 --- a/app/core/ingestion/ingestion_db.py +++ b/app/core/ingestion/ingestion_db.py @@ -50,6 +50,7 @@ def is_explicit_edge_present(client: QdrantClient, prefix: str, edge_id: str) -> """ _, _, edges_col = collection_names(prefix) try: + # retrieve erwartet eine Liste von IDs res = client.retrieve( collection_name=edges_col, ids=[edge_id], @@ -83,7 +84,7 @@ def purge_artifacts(client: QdrantClient, prefix: str, note_id: str): # Dies umfasst: # - Alle ausgehenden Kanten (A -> B) # - Alle inversen Kanten, die diese Note in anderen Notizen "deponiert" hat (B -> A) - # Fremde inverse Kanten (C -> A) bleiben erhalten. + # Fremde inverse Kanten (C -> A), die von anderen Notizen stammen, bleiben erhalten. edges_filter = rest.Filter(must=[ rest.FieldCondition(key="origin_note_id", match=rest.MatchValue(value=note_id)) ]) diff --git a/app/core/ingestion/ingestion_processor.py b/app/core/ingestion/ingestion_processor.py index 67ade45..e3ae9e6 100644 --- a/app/core/ingestion/ingestion_processor.py +++ b/app/core/ingestion/ingestion_processor.py @@ -5,9 +5,9 @@ DESCRIPTION: Der zentrale IngestionService (Orchestrator). WP-25a: Integration der Mixture of Experts (MoE) Architektur. WP-15b: Two-Pass Workflow mit globalem Kontext-Cache. WP-20/22: Cloud-Resilienz und Content-Lifecycle integriert. - AUDIT v3.3.0: Einführung des 2-Phasen-Upserts. Garantiert, dass - explizite Kanten niemals durch Symmetrien überschrieben werden. -VERSION: 3.3.0 (WP-24c: Two-Phase Writing Strategy) + AUDIT v3.3.1: Strikte Trennung von Explicit-Write (Phase 1) und + Symmetry-Validation (Phase 2). 100% Datenhoheit für den Nutzer. +VERSION: 3.3.1 (WP-24c: Authority-First Ingestion) STATUS: Active """ import logging @@ -22,7 +22,7 @@ from app.core.parser import ( validate_required_frontmatter, NoteContext ) from app.core.chunking import assemble_chunks -# WP-24c: Import für die deterministische ID-Vorabberechnung +# WP-24c: Import für die deterministische UUID-Vorabberechnung from app.core.graph.graph_utils import _mk_edge_id # MODULARISIERUNG: Neue Import-Pfade für die Datenbank-Ebene @@ -79,7 +79,7 @@ class IngestionService: # Festlegen, welcher Hash für die Change-Detection maßgeblich ist self.active_hash_mode = self.settings.CHANGE_DETECTION_MODE - self.batch_cache: Dict[str, NoteContext] = {} # WP-15b LocalBatchCache + self.batch_cache: Dict[str, NoteContext] = {} # Globaler Kontext-Cache (Pass 1) try: # Aufruf der modularisierten Schema-Logik @@ -90,13 +90,13 @@ class IngestionService: def _is_valid_note_id(self, text: str) -> bool: """ - WP-24c: Prüft Ziel-Strings auf Validität. - Filtert Begriffe wie 'insight' oder 'event' aus, um Müll-Kanten zu vermeiden. + WP-24c: Prüft Ziel-Strings auf fachliche Validität. + Verhindert das Anlegen von Kanten zu reinen System-Platzhaltern. """ if not text or len(text.strip()) < 2: return False - - # Symmetrie-Filter gegen Typ-Strings + + # Blacklist für Begriffe, die keine echten Notizen sind blacklisted = {"insight", "event", "source", "task", "project", "person", "concept", "related_to", "referenced_by"} if text.lower().strip() in blacklisted: return False @@ -106,12 +106,12 @@ class IngestionService: async def run_batch(self, file_paths: List[str], vault_root: str) -> List[Dict[str, Any]]: """ - WP-15b: Implementiert den Two-Pass Ingestion Workflow. - Führt nun zusätzlich das 2-Phasen-Schreiben aus. + WP-15b: Two-Pass Ingestion Workflow mit 2-Phasen-Schreibstrategie. """ + self.batch_cache.clear() logger.info(f"--- 🔍 START BATCH IMPORT ({len(file_paths)} Dateien) ---") - # 1. Schritt: Context-Cache füllen + # SCHRITT 1: Pre-Scan (Context-Cache füllen) for path in file_paths: try: ctx = pre_scan_markdown(path, registry=self.registry) @@ -121,32 +121,36 @@ class IngestionService: fname = os.path.splitext(os.path.basename(path))[0] self.batch_cache[fname] = ctx except Exception as e: - logger.warning(f"⚠️ Pre-scan failed for {path}: {e}") + logger.warning(f" ⚠️ Pre-scan fehlgeschlagen für {path}: {e}") - # 2. Schritt: Verarbeitung & Schreiben (PHASE 1: AUTHORITY) - # Wir sammeln alle Symmetrie-Kandidaten, um sie in Phase 2 zu prüfen. + # SCHRITT 2: PHASE 1 (Authority-Schreiben) + # Wir verarbeiten alle Dateien und schreiben NUR explizite Kanten in die DB. results = [] all_virtual_candidates = [] for p in file_paths: - res, candidates = await self.process_file(p, vault_root, apply=True, purge_before=True, skip_virtuals=True) + # process_file liefert in dieser Version (res, virtual_candidates) zurück + res, candidates = await self.process_file(p, vault_root, apply=True, purge_before=True) results.append(res) all_virtual_candidates.extend(candidates) - # 3. Schritt: Symmetrie-Einspeisung (PHASE 2: SYMMETRY) + # SCHRITT 3: PHASE 2 (Symmetrie-Ergänzung) + # Nachdem alle expliziten Kanten fest in Qdrant liegen, prüfen wir die Inversen. if all_virtual_candidates: - logger.info(f"🔄 PHASE 2: Prüfe {len(all_virtual_candidates)} Symmetrie-Kanten gegen die Datenbank...") + logger.info(f"🔄 PHASE 2: Validiere {len(all_virtual_candidates)} Symmetrie-Kandidaten gegen Live-DB...") final_virtuals = [] for v_edge in all_virtual_candidates: - # Eindeutige ID für diese Symmetrie-Kante berechnen + # Eindeutige ID berechnen (muss exakt der ID in Phase 1 entsprechen) v_id = _mk_edge_id(v_edge["kind"], v_edge["note_id"], v_edge["target_id"], "note") - # Wenn in Phase 1 KEINE manuelle Kante mit dieser ID geschrieben wurde, darf die Symmetrie rein + # Check: Liegt dort bereits eine manuelle Kante? if not is_explicit_edge_present(self.client, self.prefix, v_id): final_virtuals.append(v_edge) - + else: + logger.debug(f" 🛡️ Symmetrie übersprungen (Manuelle Kante hat Vorrang): {v_id}") + if final_virtuals: - logger.info(f"📤 Schreibe {len(final_virtuals)} validierte Symmetrie-Kanten in den Graphen.") + logger.info(f"📤 Schreibe {len(final_virtuals)} geschützte Symmetrie-Kanten.") e_pts = points_for_edges(self.prefix, final_virtuals)[1] upsert_batch(self.client, f"{self.prefix}_edges", e_pts) @@ -156,12 +160,15 @@ class IngestionService: async def process_file(self, file_path: str, vault_root: str, **kwargs) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]: """ Transformiert eine Markdown-Datei. - Liefert zusätzlich eine Liste von virtuellen Kanten-Kandidaten zurück. + Schreibt Notes/Chunks/Explicit Edges sofort (Phase 1). + Gibt potenzielle Symmetrien für Phase 2 zurück. """ apply = kwargs.get("apply", False) force_replace = kwargs.get("force_replace", False) purge_before = kwargs.get("purge_before", False) - skip_virtuals = kwargs.get("skip_virtuals", False) + note_scope_refs = kwargs.get("note_scope_refs", False) + hash_source = kwargs.get("hash_source", "parsed") + hash_normalize = kwargs.get("hash_normalize", "canonical") result = {"path": file_path, "status": "skipped", "changed": False, "error": None} virtual_candidates = [] @@ -184,24 +191,34 @@ class IngestionService: # 2. Payload & Change Detection note_type = resolve_note_type(self.registry, fm.get("type")) - note_pl = make_note_payload(parsed, vault_root=vault_root, file_path=file_path, types_cfg=self.registry) + note_pl = make_note_payload( + parsed, vault_root=vault_root, file_path=file_path, + hash_source=hash_source, hash_normalize=hash_normalize, + types_cfg=self.registry + ) note_id = note_pl["note_id"] logger.info(f"📄 Bearbeite: '{note_id}' (Typ: {note_type})") old_payload = None if force_replace else fetch_note_payload(self.client, self.prefix, note_id) - check_key = f"{self.active_hash_mode}:{note_pl.get('hashes', {}).get('hash_source', 'parsed')}:{note_pl.get('hashes', {}).get('hash_normalize', 'canonical')}" - # (Hashing Logik hier vereinfacht zur Lesbarkeit, entspricht aber Ihrer Codebasis) - + check_key = f"{self.active_hash_mode}:{hash_source}:{hash_normalize}" + old_hash = (old_payload or {}).get("hashes", {}).get(check_key) + new_hash = note_pl.get("hashes", {}).get(check_key) + c_miss, e_miss = artifacts_missing(self.client, self.prefix, note_id) - if not (force_replace or not old_payload or c_miss or e_miss): + + if not (force_replace or not old_payload or old_hash != new_hash or c_miss or e_miss): return {**result, "status": "unchanged", "note_id": note_id}, [] + + if not apply: + return {**result, "status": "dry-run", "changed": True, "note_id": note_id}, [] # 3. Deep Processing (Chunking, Validation, Embedding) try: body_text = getattr(parsed, "body", "") or "" edge_registry.ensure_latest() profile = note_pl.get("chunk_profile", "sliding_standard") + chunk_cfg = get_chunk_config_by_profile(self.registry, profile, note_type) enable_smart = chunk_cfg.get("enable_smart_edge_allocation", False) @@ -215,8 +232,8 @@ class IngestionService: is_valid = await validate_edge_candidate( ch.text, cand, self.batch_cache, self.llm, profile_name="ingest_validator" ) - label = cand.get('target_id') or cand.get('note_id') or "Unknown" - logger.info(f" 🧠 [SMART EDGE] {label} -> {'✅ OK' if is_valid else '❌ SKIP'}") + t_id = cand.get('target_id') or cand.get('note_id') or "Unknown" + logger.info(f" 🧠 [SMART EDGE] {t_id} -> {'✅ OK' if is_valid else '❌ SKIP'}") if is_valid: new_pool.append(cand) else: new_pool.append(cand) @@ -225,49 +242,60 @@ class IngestionService: chunk_pls = make_chunk_payloads(fm, note_pl["path"], chunks, file_path=file_path, types_cfg=self.registry) vecs = await self.embedder.embed_documents([c.get("window") or "" for c in chunk_pls]) if chunk_pls else [] - # Kanten-Extraktion - raw_edges = build_edges_for_note(note_id, chunk_pls, note_level_references=note_pl.get("references", [])) + # Aggregation aller Kanten + raw_edges = build_edges_for_note( + note_id, chunk_pls, + note_level_references=note_pl.get("references", []), + include_note_scope_refs=note_scope_refs + ) - # PHASE 1: Authority Edges (Explizit) + # PHASE 1: Authority-Check & Kanonisierung explicit_edges = [] for e in raw_edges: target_raw = e.get("target_id") + # ID-Resolution über den Context-Cache (Titel -> Note_ID) target_ctx = self.batch_cache.get(target_raw) target_id = target_ctx.note_id if target_ctx else target_raw - # Junk-Filter if not self._is_valid_note_id(target_id): continue resolved_kind = edge_registry.resolve(e.get("kind", "related_to"), provenance=e.get("provenance", "explicit")) - # Echte physische Kante markieren + # Echte explizite Kante für Phase 1 e.update({ "kind": resolved_kind, "target_id": target_id, "origin_note_id": note_id, "virtual": False, "confidence": 1.0 }) explicit_edges.append(e) - # Kandidat für Symmetrie (Phase 2) + # Symmetrie-Kandidat für Phase 2 vorbereiten inv_kind = edge_registry.get_inverse(resolved_kind) if inv_kind and target_id != note_id: v_edge = e.copy() v_edge.update({ "note_id": target_id, "target_id": note_id, "kind": inv_kind, "virtual": True, "provenance": "structure", "confidence": 1.0, - "origin_note_id": note_id + "origin_note_id": note_id }) virtual_candidates.append(v_edge) # 4. DB Upsert (Phase 1) if apply: - if purge_before and old_payload: purge_artifacts(self.client, self.prefix, note_id) + if purge_before and old_payload: + purge_artifacts(self.client, self.prefix, note_id) + upsert_batch(self.client, f"{self.prefix}_notes", points_for_note(self.prefix, note_pl, None, self.dim)[1]) - if chunk_pls and vecs: upsert_batch(self.client, f"{self.prefix}_chunks", points_for_chunks(self.prefix, chunk_pls, vecs)[1]) - if explicit_edges: upsert_batch(self.client, f"{self.prefix}_edges", points_for_edges(self.prefix, explicit_edges)[1]) - - logger.info(f" ✨ Fertig: {len(chunk_pls)} Chunks, {len(explicit_edges)} explizite Kanten geschrieben.") - return {"status": "success", "note_id": note_id, "edges_count": len(explicit_edges)}, virtual_candidates + if chunk_pls and vecs: + upsert_batch(self.client, f"{self.prefix}_chunks", points_for_chunks(self.prefix, chunk_pls, vecs)[1]) + if explicit_edges: + upsert_batch(self.client, f"{self.prefix}_edges", points_for_edges(self.prefix, explicit_edges)[1]) + logger.info(f" ✨ Phase 1 fertig: {len(chunk_pls)} Chunks, {len(explicit_edges)} explizite Kanten.") + return { + "path": file_path, "status": "success", "changed": True, "note_id": note_id, + "chunks_count": len(chunk_pls), "edges_count": len(explicit_edges) + }, virtual_candidates + except Exception as e: logger.error(f"❌ Fehler bei {file_path}: {e}", exc_info=True) return {**result, "error": str(e)}, [] @@ -276,7 +304,8 @@ class IngestionService: """Erstellt eine Note aus einem Textstream.""" target_path = os.path.join(vault_root, folder, filename) os.makedirs(os.path.dirname(target_path), exist_ok=True) - with open(target_path, "w", encoding="utf-8") as f: f.write(markdown_content) + with open(target_path, "w", encoding="utf-8") as f: + f.write(markdown_content) await asyncio.sleep(0.1) res, _ = await self.process_file(file_path=target_path, vault_root=vault_root, apply=True, force_replace=True, purge_before=True) return res \ No newline at end of file