mindnet/app/core/graph_adapter.py

249 lines
7.4 KiB
Python
Raw Normal View History

2025-10-07 11:30:08 +02:00
"""
2025-12-15 15:40:39 +01:00
FILE: app/core/graph_adapter.py
DESCRIPTION: Lädt Kanten aus Qdrant und baut einen In-Memory Subgraphen für Scoring (Centrality) und Explanation.
VERSION: 0.4.0
STATUS: Active
DEPENDENCIES: qdrant_client, app.core.qdrant
LAST_ANALYSIS: 2025-12-15
2025-10-07 11:30:08 +02:00
"""
from __future__ import annotations
2025-12-03 18:40:06 +01:00
2025-12-07 16:32:02 +01:00
from typing import Dict, List, Optional, DefaultDict, Any
2025-10-07 11:30:08 +02:00
from collections import defaultdict
2025-12-03 18:40:06 +01:00
2025-10-07 11:30:08 +02:00
from qdrant_client import QdrantClient
2025-12-03 18:40:06 +01:00
from qdrant_client.http import models as rest
from app.core.qdrant import collection_names
2025-12-07 16:32:02 +01:00
# Legacy-Import Fallback
2025-12-03 18:40:06 +01:00
try: # pragma: no cover
from app.core.qdrant_points import get_edges_for_sources # type: ignore
except Exception: # pragma: no cover
get_edges_for_sources = None # type: ignore
2025-12-07 16:32:02 +01:00
# Basisgewichte je Edge-Typ (WP-04a Config)
2025-12-03 18:40:06 +01:00
EDGE_BASE_WEIGHTS: Dict[str, float] = {
2025-12-07 16:32:02 +01:00
# Struktur
2025-12-04 11:16:03 +01:00
"belongs_to": 0.10,
"next": 0.06,
"prev": 0.06,
"backlink": 0.04,
"references_at": 0.08,
2025-12-07 16:32:02 +01:00
# Wissen
2025-12-04 11:16:03 +01:00
"references": 0.20,
"depends_on": 0.18,
"related_to": 0.15,
"similar_to": 0.12,
2025-10-07 11:30:08 +02:00
}
2025-12-03 18:40:06 +01:00
def _edge_weight(pl: Dict) -> float:
"""Berechnet das effektive Edge-Gewicht aus kind + confidence."""
kind = pl.get("kind", "edge")
base = EDGE_BASE_WEIGHTS.get(kind, 0.0)
conf_raw = pl.get("confidence", None)
try:
conf = float(conf_raw) if conf_raw is not None else None
except Exception:
conf = None
if conf is None:
return base
2025-12-07 16:32:02 +01:00
if conf < 0.0: conf = 0.0
if conf > 1.0: conf = 1.0
2025-12-03 18:40:06 +01:00
return base * conf
def _fetch_edges(
client: QdrantClient,
prefix: str,
seeds: List[str],
edge_types: Optional[List[str]] = None,
limit: int = 2048,
) -> List[Dict]:
"""
Holt Edges direkt aus der *_edges Collection.
2025-12-07 16:32:02 +01:00
Filter: source_id IN seeds OR target_id IN seeds OR note_id IN seeds
2025-12-03 18:40:06 +01:00
"""
if not seeds or limit <= 0:
return []
_, _, edges_col = collection_names(prefix)
seed_conditions = []
for field in ("source_id", "target_id", "note_id"):
for s in seeds:
seed_conditions.append(
2025-12-07 16:32:02 +01:00
rest.FieldCondition(key=field, match=rest.MatchValue(value=str(s)))
2025-12-03 18:40:06 +01:00
)
seeds_filter = rest.Filter(should=seed_conditions) if seed_conditions else None
type_filter = None
if edge_types:
type_conds = [
2025-12-07 16:32:02 +01:00
rest.FieldCondition(key="kind", match=rest.MatchValue(value=str(k)))
2025-12-03 18:40:06 +01:00
for k in edge_types
]
type_filter = rest.Filter(should=type_conds)
must = []
2025-12-07 16:32:02 +01:00
if seeds_filter: must.append(seeds_filter)
if type_filter: must.append(type_filter)
flt = rest.Filter(must=must) if must else None
2025-12-03 18:40:06 +01:00
pts, _ = client.scroll(
collection_name=edges_col,
scroll_filter=flt,
limit=limit,
with_payload=True,
with_vectors=False,
)
out: List[Dict] = []
for p in pts or []:
pl = dict(p.payload or {})
if pl:
out.append(pl)
return out
2025-10-07 11:30:08 +02:00
class Subgraph:
2025-12-07 16:32:02 +01:00
"""Leichtgewichtiger Subgraph mit Adjazenzlisten & Kennzahlen."""
2025-10-07 11:30:08 +02:00
2025-12-03 18:40:06 +01:00
def __init__(self) -> None:
2025-12-07 16:32:02 +01:00
# Forward: source -> [targets]
2025-10-07 11:30:08 +02:00
self.adj: DefaultDict[str, List[Dict]] = defaultdict(list)
2025-12-07 16:32:02 +01:00
# Reverse: target -> [sources] (Neu für WP-04b Explanation)
self.reverse_adj: DefaultDict[str, List[Dict]] = defaultdict(list)
2025-10-07 11:30:08 +02:00
self.in_degree: DefaultDict[str, int] = defaultdict(int)
self.out_degree: DefaultDict[str, int] = defaultdict(int)
2025-12-03 18:40:06 +01:00
def add_edge(self, e: Dict) -> None:
2025-12-04 11:52:18 +01:00
"""
2025-12-07 16:32:02 +01:00
Fügt eine Kante hinzu und aktualisiert Forward/Reverse Indizes.
e muss enthalten: source, target, kind, weight.
2025-12-04 11:52:18 +01:00
"""
src = e.get("source")
tgt = e.get("target")
kind = e.get("kind")
2025-10-07 11:30:08 +02:00
weight = e.get("weight", EDGE_BASE_WEIGHTS.get(kind, 0.0))
2025-12-04 11:52:18 +01:00
owner = e.get("note_id")
2025-12-03 18:40:06 +01:00
if not src or not tgt:
return
2025-12-07 16:32:02 +01:00
# 1. Primäre Adjazenz (Forward)
edge_data = {"target": tgt, "kind": kind, "weight": weight}
self.adj[src].append(edge_data)
2025-10-07 11:30:08 +02:00
self.out_degree[src] += 1
self.in_degree[tgt] += 1
2025-12-07 16:32:02 +01:00
# 2. Reverse Adjazenz (Neu für Explanation)
# Wir speichern, woher die Kante kam.
rev_data = {"source": src, "kind": kind, "weight": weight}
self.reverse_adj[tgt].append(rev_data)
# 3. Kontext-Note Handling (Forward & Reverse)
# Wenn eine Kante "im Kontext einer Note" (owner) definiert ist,
# schreiben wir sie der Note gut, damit der Retriever Scores auf Note-Ebene findet.
if owner and owner != src:
# Forward: Owner -> Target
self.adj[owner].append(edge_data)
self.out_degree[owner] += 1
# Reverse: Target wird vom Owner referenziert (indirekt)
2025-12-04 11:52:18 +01:00
if owner != tgt:
2025-12-07 16:32:02 +01:00
rev_owner_data = {"source": owner, "kind": kind, "weight": weight, "via_context": True}
self.reverse_adj[tgt].append(rev_owner_data)
self.in_degree[owner] += 1 # Leichter Centrality Boost für den Owner
2025-12-04 11:52:18 +01:00
2025-10-07 11:30:08 +02:00
def aggregate_edge_bonus(self, node_id: str) -> float:
2025-12-07 16:32:02 +01:00
"""Summe der ausgehenden Kantengewichte (Hub-Score)."""
2025-10-07 11:30:08 +02:00
return sum(edge["weight"] for edge in self.adj.get(node_id, []))
2025-12-04 11:16:03 +01:00
def edge_bonus(self, node_id: str) -> float:
2025-12-07 16:32:02 +01:00
"""API für Retriever (WP-04a Kompatibilität)."""
2025-12-04 11:16:03 +01:00
return self.aggregate_edge_bonus(node_id)
2025-10-07 11:30:08 +02:00
def centrality_bonus(self, node_id: str) -> float:
2025-12-07 16:32:02 +01:00
"""Log-gedämpfte Zentralität (In-Degree)."""
2025-10-07 11:30:08 +02:00
import math
2025-12-03 18:40:06 +01:00
indeg = self.in_degree.get(node_id, 0)
if indeg <= 0:
return 0.0
return min(math.log1p(indeg) / 10.0, 0.15)
2025-10-07 11:30:08 +02:00
2025-12-07 16:32:02 +01:00
# --- WP-04b Explanation Helpers ---
def get_outgoing_edges(self, node_id: str) -> List[Dict[str, Any]]:
"""Liefert Liste aller Ziele, auf die dieser Knoten zeigt."""
return self.adj.get(node_id, [])
def get_incoming_edges(self, node_id: str) -> List[Dict[str, Any]]:
"""Liefert Liste aller Quellen, die auf diesen Knoten zeigen."""
return self.reverse_adj.get(node_id, [])
2025-12-03 18:40:06 +01:00
def expand(
client: QdrantClient,
prefix: str,
seeds: List[str],
depth: int = 1,
edge_types: Optional[List[str]] = None,
) -> Subgraph:
2025-10-07 11:30:08 +02:00
"""
2025-12-07 16:32:02 +01:00
Expandiert ab Seeds entlang von Edges (bis `depth`).
2025-10-07 11:30:08 +02:00
"""
sg = Subgraph()
frontier = set(seeds)
visited = set()
2025-12-03 18:40:06 +01:00
max_depth = max(depth, 0)
for _ in range(max_depth):
2025-10-07 11:30:08 +02:00
if not frontier:
break
2025-12-03 18:40:06 +01:00
edges_payloads = _fetch_edges(
client=client,
prefix=prefix,
seeds=list(frontier),
edge_types=edge_types,
limit=2048,
)
2025-10-07 11:30:08 +02:00
next_frontier = set()
2025-12-03 18:40:06 +01:00
for pl in edges_payloads:
src = pl.get("source_id")
tgt = pl.get("target_id")
2025-12-07 16:32:02 +01:00
# Skip invalid edges
if not src or not tgt:
continue
2025-12-03 18:40:06 +01:00
2025-10-07 11:30:08 +02:00
e = {
2025-12-03 18:40:06 +01:00
"source": src,
"target": tgt,
2025-12-07 16:32:02 +01:00
"kind": pl.get("kind", "edge"),
2025-12-03 18:40:06 +01:00
"weight": _edge_weight(pl),
2025-12-07 16:32:02 +01:00
"note_id": pl.get("note_id"),
2025-10-07 11:30:08 +02:00
}
sg.add_edge(e)
2025-12-03 18:40:06 +01:00
2025-12-07 16:32:02 +01:00
# Nur weitersuchen, wenn Target noch nicht besucht
if tgt and tgt not in visited:
2025-12-03 18:40:06 +01:00
next_frontier.add(tgt)
2025-10-07 11:30:08 +02:00
visited |= frontier
frontier = next_frontier - visited
2025-12-07 16:32:02 +01:00
return sg