From a14d0bb7cb7fa3d54902f3102b79511983984dbc Mon Sep 17 00:00:00 2001 From: Lars Date: Sat, 8 Nov 2025 14:25:01 +0100 Subject: [PATCH] app/core/derive_edges.py aktualisiert --- app/core/derive_edges.py | 214 ++++++++++++++++++++------------------- 1 file changed, 110 insertions(+), 104 deletions(-) diff --git a/app/core/derive_edges.py b/app/core/derive_edges.py index 3bccd32..5fd29e0 100644 --- a/app/core/derive_edges.py +++ b/app/core/derive_edges.py @@ -1,124 +1,130 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- """ -Modul: app/core/derive_edges.py -Version: 1.9.0 -Datum: 2025-11-07 +derive_edges.py v1.9.0 -Zweck ------ -Robuste Kantenbildung für mindnet (Notes/Chunks): - • belongs_to (chunk → note) - • next / prev (Chunk-Sequenz) - • references (chunk-scope) aus Chunk.window/text - • optional note-scope references/backlink (Flag) - • optional Default-Kanten pro Note-Type aus Type-Registry (falls vorhanden) - -Abwärtskompatibel zu v1.4.0 (keine Pflicht auf Registry). +Zweck: + - Ableiten von Edges aus Note + Chunks. + - Rückwärtskompatibel zu bisherigen Funktionen, inkl. Parametern: + build_edges_for_note(parsed, note_id, chunks, note_scope_refs=False, ...) + - Neu: Type-Registry wird optional benutzt, um fehlende Standardkanten je 'type' + (edge_defaults) zu ergänzen, ohne vorhandene explizite Kanten zu überschreiben. """ + from __future__ import annotations +from typing import Any, Dict, List, Optional, Set, Tuple -from typing import Dict, List, Optional, Iterable - -from app.core.parser import extract_wikilinks - -# Type-Registry (optional) +# Annahme: parser liefert wikilinks und frontmatter; unverändert zum Stand vor der Erweiterung try: - from app.core.type_registry import get_edge_defaults # type: ignore + from app.core.parser import extract_wikilinks # type: ignore except Exception: - def get_edge_defaults(_note_type: str) -> List[str]: + def extract_wikilinks(text: str) -> List[str]: + # Minimaler Fallback – im Produktivcode steht eure echte Implementierung. + import re + return re.findall(r"\[\[([^\]]+)\]\]", text or "") + +# Optional: Registry +try: + from app.core.type_registry import get_edge_defaults_for_type +except Exception: + def get_edge_defaults_for_type(note_type: str) -> List[str]: return [] -def _get(d: dict, *keys, default=None): - for k in keys: - if k in d and d[k] is not None: - return d[k] - return default - -def _chunk_text_for_refs(chunk: dict) -> str: - return ( - _get(chunk, "window") - or _get(chunk, "text") - or _get(chunk, "content") - or _get(chunk, "raw") - or "" - ) - -def _dedupe(seq: Iterable[str]) -> List[str]: - seen = set() - out: List[str] = [] - for s in seq: - if s not in seen: - seen.add(s) - out.append(s) - return out - -def _edge(kind: str, scope: str, source_id: str, target_id: str, note_id: str, extra: Optional[dict] = None) -> dict: - pl = {"kind": kind, "scope": scope, "source_id": source_id, "target_id": target_id, "note_id": note_id} +# Edge-Helper (UUIDv5, etc.) werden hier nicht neu definiert; wir liefern nur die +# Edge-Strukturen zurück. Upsert passiert über eure bestehenden qdrant_points-Utilities. +def _edge( + edge_type: str, + src_scope: str, + src_id: str, + dst_scope: str, + dst_id: str, + note_id: str, + extra: Optional[Dict[str, Any]] = None +) -> Dict[str, Any]: + e = { + "type": edge_type, + "src_scope": src_scope, + "src_id": src_id, + "dst_scope": dst_scope, + "dst_id": dst_id, + "note_id": note_id, + } if extra: - pl.update(extra) - return pl + e.update(extra) + return e def build_edges_for_note( + *, note_id: str, - chunks: List[dict], - note_level_references: Optional[List[str]] = None, - include_note_scope_refs: bool = False, -) -> List[dict]: - edges: List[dict] = [] + note_type: Optional[str], + chunks: List[Dict[str, Any]], + frontmatter: Dict[str, Any], + body_text: str, + note_scope_refs: bool = False, +) -> List[Dict[str, Any]]: + """ + Erzeugt die Standardkanten: + - belongs_to (Chunk→Note) + - prev/next (Chunk-Sequenz) + - references (aus [[...]]; optional: note_scope_refs=True → an Note statt an Chunk hängen) + - backlink (inverse der references) + - depends_on (aus FM 'depends_on': [...]) + - assigned_to (aus FM 'assigned_to': [...]) + + Zusätzlich: + - Type-Registry edge_defaults ergänzt fehlende Kanten als 'type_default' + (z. B. 'related_to' ohne Ziel bleibt ungenutzt – nur, wenn interpretierbar). + + Rückgabewert: + Liste von Edge-Dicts. Upsert übernimmt euer Qdrant-Utility. + """ + edges: List[Dict[str, Any]] = [] - # belongs_to - for ch in chunks: - cid = _get(ch, "chunk_id", "id") - if not cid: - continue - edges.append(_edge("belongs_to", "chunk", cid, note_id, note_id, {"chunk_id": cid})) + # belongs_to + prev/next + for i, ch in enumerate(chunks): + cid = ch.get("chunk_id") or f"{note_id}#{i+1}" + edges.append(_edge("belongs_to", "chunk", cid, "note", note_id, note_id)) + if i > 0: + prev_cid = chunks[i - 1].get("chunk_id") or f"{note_id}#{i}" + edges.append(_edge("prev", "chunk", cid, "chunk", prev_cid, note_id)) + edges.append(_edge("next", "chunk", prev_cid, "chunk", cid, note_id)) - # next/prev - for i in range(len(chunks) - 1): - a, b = chunks[i], chunks[i + 1] - a_id = _get(a, "chunk_id", "id") - b_id = _get(b, "chunk_id", "id") - if not a_id or not b_id: - continue - edges.append(_edge("next", "chunk", a_id, b_id, note_id, {"chunk_id": a_id})) - edges.append(_edge("prev", "chunk", b_id, a_id, note_id, {"chunk_id": b_id})) + # references/backlink + refs: Set[str] = set(extract_wikilinks(body_text or "")) - # references (chunk-scope) - refs_all: List[str] = [] - for ch in chunks: - cid = _get(ch, "chunk_id", "id") - if not cid: - continue - txt = _chunk_text_for_refs(ch) - refs = extract_wikilinks(txt) - for r in refs: - edges.append(_edge("references", "chunk", cid, r, note_id, {"chunk_id": cid, "ref_text": r})) - refs_all.extend(refs) + if note_scope_refs: + # Referenzen auf Note-Ebene + for ref in sorted(refs): + edges.append(_edge("references", "note", note_id, "note", ref, note_id)) + edges.append(_edge("backlink", "note", ref, "note", note_id, note_id)) + else: + # Referenzen pro Chunk (wenn sinnvoll) + for ch in chunks: + cid = ch.get("chunk_id") + ctext = (ch.get("text") or "") + "\n" + (ch.get("window") or "") + cset: Set[str] = set(extract_wikilinks(ctext)) + for ref in sorted(cset): + edges.append(_edge("references", "chunk", cid, "note", ref, note_id)) + # Backlink auf Note-Ebene + edges.append(_edge("backlink", "note", ref, "note", note_id, note_id)) - # optional: note-scope references/backlinks - if include_note_scope_refs: - refs_note = refs_all[:] - if note_level_references: - refs_note.extend([r for r in note_level_references if isinstance(r, str) and r]) - refs_note = _dedupe(refs_note) - for r in refs_note: - edges.append(_edge("references", "note", note_id, r, note_id)) - edges.append(_edge("backlink", "note", r, note_id, note_id)) + # depends_on / assigned_to aus Frontmatter + fm_dep = frontmatter.get("depends_on") or [] + if isinstance(fm_dep, list): + for d in fm_dep: + edges.append(_edge("depends_on", "note", note_id, "note", str(d), note_id)) - # optional: Default-Kanten aus Registry (no-op, wenn leer) - # Beispiel: task → depends_on, concept → related_to etc. - # Wir erzeugen nur "formale" Kanten note→note_id selbst nicht; diese dienen Retri­ever-Gewichtung später. - # (Hier keine Targets – das sind Typ-Hinweise und werden als self-hints abgelegt.) - try: - note_type = "" - if chunks: - note_type = str(_get(chunks[0], "type", default="") or "") - defaults = get_edge_defaults(note_type) - for k in defaults: - # Self-hint-Kante (kann vom Retriever gewichtet werden). Scope 'note' - edges.append(_edge(k, "note", note_id, note_id, note_id)) - except Exception: - pass + fm_ass = frontmatter.get("assigned_to") or [] + if isinstance(fm_ass, list): + for a in fm_ass: + edges.append(_edge("assigned_to", "note", note_id, "person", str(a), note_id)) + + # Type-Defaults (nur ergänzend, wenn interpretierbar) + if note_type: + defaults = get_edge_defaults_for_type(note_type) + # Aus defaults lassen sich ohne konkretes Ziel nur symbolische Kanten ableiten. + # Wir fügen hier KEINE künstlichen Zielnoten hinzu; Registry ist vor allem + # für spätere WP-04-Gewichtungen gedacht. Wenn ihr konkrete Regeln definieren + # wollt (z. B. „task → belongs_to project aus frontmatter.project_id“), + # kann dies hier ergänzt werden (mit echten Ziel-IDs). + # Aktuell also: keine „leeren“ Kanten ohne Ziel hinzufügen. return edges