Lars/mindnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

mehrdimensionale matrix für Kanten

Browse Source
This commit is contained in:
Lars 2025-12-11 14:59:59 +01:00
parent a1a58727fd
commit b815f6235f
1 changed files with 92 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

128
app/services/discovery.py
View File

@ -1,7 +1,12 @@
""" """
app/services/discovery.py app/services/discovery.py
Service für Link-Vorschläge und Knowledge-Discovery (WP-11). 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 logging
import asyncio import asyncio
@ -23,33 +28,42 @@ class DiscoveryService:
self.registry = self._load_type_registry() self.registry = self._load_type_registry()
async def analyze_draft(self, text: str, current_type: str) -> Dict[str, Any]: 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 = [] suggestions = []
# Fallback, falls keine spezielle Regel greift
default_edge_type = self._get_default_edge_type(current_type) 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() seen_target_note_ids = set()
# --------------------------------------------------------- # ---------------------------------------------------------
# 1. Exact Match: Titel/Aliases # 1. Exact Match: Titel/Aliases
# --------------------------------------------------------- # ---------------------------------------------------------
# Holt Titel, Aliases UND Typen aus dem Index
known_entities = self._fetch_all_titles_and_aliases() known_entities = self._fetch_all_titles_and_aliases()
found_entities = self._find_entities_in_text(text, known_entities) found_entities = self._find_entities_in_text(text, known_entities)
for entity in found_entities: for entity in found_entities:
# Duplikate vermeiden
if entity["id"] in seen_target_note_ids: if entity["id"] in seen_target_note_ids:
continue continue
seen_target_note_ids.add(entity["id"]) 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({ suggestions.append({
"type": "exact_match", "type": "exact_match",
"text_found": entity["match"], "text_found": entity["match"],
"target_title": entity["title"], "target_title": entity["title"],
"target_id": entity["id"], "target_id": entity["id"],
"suggested_edge_type": default_edge_type, "suggested_edge_type": smart_edge,
"suggested_markdown": f"[[rel:{default_edge_type} {entity['title']}]]", "suggested_markdown": f"[[rel:{smart_edge} {entity['title']}]]",
"confidence": 1.0, "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 # Ergebnisse verarbeiten
for hits in results_list: for hits in results_list:
for hit in hits: for hit in hits:
# WICHTIG: Note ID aus Payload holen (Chunk ID ist hit.node_id)
note_id = hit.payload.get("note_id") note_id = hit.payload.get("note_id")
if not note_id: continue
# Fallback, falls Payload leer (sollte nicht passieren) # Deduplizierung (Notiz-Ebene)
if not note_id:
continue
# 1. Check: Haben wir diese NOTIZ schon? (Egal welcher Chunk)
if note_id in seen_target_note_ids: if note_id in seen_target_note_ids:
continue continue
# 2. Score Check (Threshold) # Score Check (Threshold 0.50 für nomic-embed-text)
if hit.total_score > 0.50: 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" 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({ suggestions.append({
"type": "semantic_match", "type": "semantic_match",
"text_found": (hit.source.get("text") or "")[:60] + "...", "text_found": (hit.source.get("text") or "")[:60] + "...",
"target_title": target_title, "target_title": target_title,
"target_id": note_id, # Wir verlinken auf die Notiz, nicht den Chunk "target_id": note_id,
"suggested_edge_type": default_edge_type, "suggested_edge_type": smart_edge,
"suggested_markdown": suggested_md, "suggested_markdown": f"[[rel:{smart_edge} {target_title}]]",
"confidence": round(hit.total_score, 2), "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 # Sortieren nach Confidence
@ -103,34 +117,63 @@ class DiscoveryService:
"suggestions": suggestions[:10] "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]: def _generate_search_queries(self, text: str) -> List[str]:
""" """
Erzeugt intelligente Fenster. Erzeugt intelligente Fenster + Footer Scan.
Besonderheit: Erzwingt 'Footer-Scan' auch bei kurzen Texten,
damit "Referenzen am Ende" nicht im Kontext untergehen.
""" """
text_len = len(text) text_len = len(text)
if not text: return [] if not text: return []
queries = [] queries = []
# A) Der gesamte Text (oder Anfang) für den groben Kontext # 1. Start / Gesamtkontext
# Bei sehr kurzen Texten ist das alles.
queries.append(text[:600]) queries.append(text[:600])
# B) Der "Footer-Scan" (Das Ende) # 2. Footer-Scan (Wichtig für "Projekt"-Referenzen am 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.
if text_len > 150: if text_len > 150:
footer = text[-250:] footer = text[-250:]
# Nur hinzufügen, wenn es sich signifikant vom Start unterscheidet
if footer not in queries: if footer not in queries:
queries.append(footer) queries.append(footer)
# C) Sliding Window für lange Texte (> 800 Chars) # 3. Sliding Window für lange Texte
if text_len > 800: if text_len > 800:
window_size = 500 window_size = 500
step = 1500 step = 1500
@ -142,7 +185,9 @@ class DiscoveryService:
return queries return queries
# --- Standard Helper (Unverändert) --- # ---------------------------------------------------------
# Standard Helpers
# ---------------------------------------------------------
async def _get_semantic_suggestions_async(self, text: str): async def _get_semantic_suggestions_async(self, text: str):
req = QueryRequest(query=text, top_k=5, explain=False) req = QueryRequest(query=text, top_k=5, explain=False)
@ -174,12 +219,21 @@ class DiscoveryService:
col = f"{self.prefix}_notes" col = f"{self.prefix}_notes"
try: try:
while True: 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: for point in res:
pl = point.payload or {} pl = point.payload or {}
aliases = pl.get("aliases") or [] aliases = pl.get("aliases") or []
if isinstance(aliases, str): aliases = [aliases] 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 if next_page is None: break
except Exception: pass except Exception: pass
return notes return notes
@ -188,12 +242,14 @@ class DiscoveryService:
found = [] found = []
text_lower = text.lower() text_lower = text.lower()
for entity in entities: for entity in entities:
# Title Check
title = entity.get("title") title = entity.get("title")
if title and title.lower() in text_lower: 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 continue
# Alias Check
for alias in entity.get("aliases", []): for alias in entity.get("aliases", []):
if str(alias).lower() in text_lower: 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 break
return found return found