From b815f6235fe89dde5573c7437a63879500c4fc95 Mon Sep 17 00:00:00 2001 From: Lars Date: Thu, 11 Dec 2025 14:59:59 +0100 Subject: [PATCH] =?UTF-8?q?mehrdimensionale=20matrix=20f=C3=BCr=20Kanten?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/discovery.py | 128 +++++++++++++++++++++++++++----------- 1 file changed, 92 insertions(+), 36 deletions(-) diff --git a/app/services/discovery.py b/app/services/discovery.py index 40f731c..995abde 100644 --- a/app/services/discovery.py +++ b/app/services/discovery.py @@ -1,7 +1,12 @@ """ app/services/discovery.py Service für Link-Vorschläge und Knowledge-Discovery (WP-11). -Optimiert: Deduplizierung pro Notiz & Footer-Fokus für kurze Texte. + +Features: +- Sliding Window Analyse für lange Texte. +- Footer-Scan für Projekt-Referenzen. +- 'Matrix-Logic' für intelligente Kanten-Typen (Experience -> Value = based_on). +- Async & Nomic-Embeddings kompatibel. """ import logging import asyncio @@ -23,33 +28,42 @@ class DiscoveryService: self.registry = self._load_type_registry() async def analyze_draft(self, text: str, current_type: str) -> Dict[str, Any]: + """ + Analysiert den Text und liefert Vorschläge mit kontext-sensitiven Kanten-Typen. + """ suggestions = [] + + # Fallback, falls keine spezielle Regel greift default_edge_type = self._get_default_edge_type(current_type) - # Tracking-Sets für Deduplizierung (Wir merken uns NOTE-IDs, nicht Chunk-IDs) + # Tracking-Sets für Deduplizierung (Wir merken uns NOTE-IDs) seen_target_note_ids = set() # --------------------------------------------------------- # 1. Exact Match: Titel/Aliases # --------------------------------------------------------- + # Holt Titel, Aliases UND Typen aus dem Index known_entities = self._fetch_all_titles_and_aliases() found_entities = self._find_entities_in_text(text, known_entities) for entity in found_entities: - # Duplikate vermeiden if entity["id"] in seen_target_note_ids: continue seen_target_note_ids.add(entity["id"]) + # INTELLIGENTE KANTEN-LOGIK (MATRIX) + target_type = entity.get("type", "concept") + smart_edge = self._resolve_edge_type(current_type, target_type) + suggestions.append({ "type": "exact_match", "text_found": entity["match"], "target_title": entity["title"], "target_id": entity["id"], - "suggested_edge_type": default_edge_type, - "suggested_markdown": f"[[rel:{default_edge_type} {entity['title']}]]", + "suggested_edge_type": smart_edge, + "suggested_markdown": f"[[rel:{smart_edge} {entity['title']}]]", "confidence": 1.0, - "reason": f"Exakter Treffer: '{entity['match']}'" + "reason": f"Exakter Treffer: '{entity['match']}' ({target_type})" }) # --------------------------------------------------------- @@ -64,33 +78,33 @@ class DiscoveryService: # Ergebnisse verarbeiten for hits in results_list: for hit in hits: - # WICHTIG: Note ID aus Payload holen (Chunk ID ist hit.node_id) note_id = hit.payload.get("note_id") - - # Fallback, falls Payload leer (sollte nicht passieren) - if not note_id: - continue + if not note_id: continue - # 1. Check: Haben wir diese NOTIZ schon? (Egal welcher Chunk) + # Deduplizierung (Notiz-Ebene) if note_id in seen_target_note_ids: continue - # 2. Score Check (Threshold) + # Score Check (Threshold 0.50 für nomic-embed-text) if hit.total_score > 0.50: - seen_target_note_ids.add(note_id) # Blockiere weitere Chunks dieser Notiz + seen_target_note_ids.add(note_id) target_title = hit.payload.get("title") or "Unbekannt" - suggested_md = f"[[rel:{default_edge_type} {target_title}]]" + + # INTELLIGENTE KANTEN-LOGIK (MATRIX) + # Den Typ der gefundenen Notiz aus dem Payload lesen + target_type = hit.payload.get("type", "concept") + smart_edge = self._resolve_edge_type(current_type, target_type) suggestions.append({ "type": "semantic_match", "text_found": (hit.source.get("text") or "")[:60] + "...", "target_title": target_title, - "target_id": note_id, # Wir verlinken auf die Notiz, nicht den Chunk - "suggested_edge_type": default_edge_type, - "suggested_markdown": suggested_md, + "target_id": note_id, + "suggested_edge_type": smart_edge, + "suggested_markdown": f"[[rel:{smart_edge} {target_title}]]", "confidence": round(hit.total_score, 2), - "reason": f"Semantisch ähnlich ({hit.total_score:.2f})" + "reason": f"Semantisch ähnlich zu {target_type} ({hit.total_score:.2f})" }) # Sortieren nach Confidence @@ -103,34 +117,63 @@ class DiscoveryService: "suggestions": suggestions[:10] } - # --- Optimierte Sliding Windows --- + # --------------------------------------------------------- + # Core Logic: Die Matrix + # --------------------------------------------------------- + + def _resolve_edge_type(self, source_type: str, target_type: str) -> str: + """ + Entscheidungsmatrix für komplexe Verbindungen. + Definiert, wie Typ A auf Typ B verlinken sollte. + """ + st = source_type.lower() + tt = target_type.lower() + + # Regeln für 'experience' (Erfahrungen) + if st == "experience": + if tt == "value": return "based_on" + if tt == "principle": return "derived_from" + if tt == "trip": return "part_of" + if tt == "lesson": return "learned" + if tt == "project": return "related_to" # oder belongs_to + + # Regeln für 'project' + if st == "project": + if tt == "decision": return "depends_on" + if tt == "concept": return "uses" + if tt == "person": return "managed_by" + + # Regeln für 'decision' (ADR) + if st == "decision": + if tt == "principle": return "compliant_with" + if tt == "requirement": return "addresses" + + # Fallback: Standard aus der types.yaml für den Source-Typ + return self._get_default_edge_type(st) + + # --------------------------------------------------------- + # Sliding Windows + # --------------------------------------------------------- def _generate_search_queries(self, text: str) -> List[str]: """ - Erzeugt intelligente Fenster. - Besonderheit: Erzwingt 'Footer-Scan' auch bei kurzen Texten, - damit "Referenzen am Ende" nicht im Kontext untergehen. + Erzeugt intelligente Fenster + Footer Scan. """ text_len = len(text) if not text: return [] queries = [] - # A) Der gesamte Text (oder Anfang) für den groben Kontext - # Bei sehr kurzen Texten ist das alles. + # 1. Start / Gesamtkontext queries.append(text[:600]) - # B) Der "Footer-Scan" (Das Ende) - # Wenn der Text > 150 Zeichen ist, nehmen wir die letzten 200 Zeichen separat. - # Grund: Oft steht dort "Gehört zu Projekt X". - # Wenn wir das isolieren, ist der Vektor "Projekt X" sehr rein. + # 2. Footer-Scan (Wichtig für "Projekt"-Referenzen am Ende) if text_len > 150: footer = text[-250:] - # Nur hinzufügen, wenn es sich signifikant vom Start unterscheidet if footer not in queries: queries.append(footer) - # C) Sliding Window für lange Texte (> 800 Chars) + # 3. Sliding Window für lange Texte if text_len > 800: window_size = 500 step = 1500 @@ -142,7 +185,9 @@ class DiscoveryService: return queries - # --- Standard Helper (Unverändert) --- + # --------------------------------------------------------- + # Standard Helpers + # --------------------------------------------------------- async def _get_semantic_suggestions_async(self, text: str): req = QueryRequest(query=text, top_k=5, explain=False) @@ -174,12 +219,21 @@ class DiscoveryService: col = f"{self.prefix}_notes" try: while True: - res, next_page = self.client.scroll(collection_name=col, limit=1000, offset=next_page, with_payload=True, with_vectors=False) + res, next_page = self.client.scroll( + collection_name=col, limit=1000, offset=next_page, + with_payload=True, with_vectors=False + ) for point in res: pl = point.payload or {} aliases = pl.get("aliases") or [] if isinstance(aliases, str): aliases = [aliases] - notes.append({"id": pl.get("note_id"), "title": pl.get("title"), "aliases": aliases}) + + notes.append({ + "id": pl.get("note_id"), + "title": pl.get("title"), + "aliases": aliases, + "type": pl.get("type", "concept") # WICHTIG: Typ laden für Matrix + }) if next_page is None: break except Exception: pass return notes @@ -188,12 +242,14 @@ class DiscoveryService: found = [] text_lower = text.lower() for entity in entities: + # Title Check title = entity.get("title") if title and title.lower() in text_lower: - found.append({"match": title, "title": title, "id": entity["id"]}) + found.append({"match": title, "title": title, "id": entity["id"], "type": entity["type"]}) continue + # Alias Check for alias in entity.get("aliases", []): if str(alias).lower() in text_lower: - found.append({"match": alias, "title": title, "id": entity["id"]}) + found.append({"match": alias, "title": title, "id": entity["id"], "type": entity["type"]}) break return found \ No newline at end of file