373 lines
14 KiB
Python
373 lines
14 KiB
Python
"""
|
|
FILE: app/core/retriever.py
|
|
DESCRIPTION: Implementiert die Hybrid-Suche (Vektor + Graph-Expansion) und das Scoring-Modell (Explainability).
|
|
WP-22 Update: Dynamic Edge Boosting, Lifecycle Scoring & Provenance Awareness.
|
|
Enthält detaillierte Debug-Informationen für die mathematische Verifizierung.
|
|
VERSION: 0.6.10 (WP-22 Full, Debug & Stable)
|
|
STATUS: Active
|
|
DEPENDENCIES: app.config, app.models.dto, app.core.qdrant*, app.services.embeddings_client, app.core.graph_adapter
|
|
LAST_ANALYSIS: 2025-12-18
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
import logging
|
|
from functools import lru_cache
|
|
from typing import Any, Dict, List, Tuple, Iterable, Optional
|
|
|
|
from app.config import get_settings
|
|
from app.models.dto import (
|
|
QueryRequest,
|
|
QueryResponse,
|
|
QueryHit,
|
|
Explanation,
|
|
ScoreBreakdown,
|
|
Reason,
|
|
EdgeDTO
|
|
)
|
|
import app.core.qdrant as qdr
|
|
import app.core.qdrant_points as qp
|
|
import app.services.embeddings_client as ec
|
|
import app.core.graph_adapter as ga
|
|
|
|
try:
|
|
import yaml # type: ignore[import]
|
|
except Exception: # pragma: no cover
|
|
yaml = None # type: ignore[assignment]
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ==============================================================================
|
|
# 1. CORE HELPERS & CONFIG LOADERS
|
|
# ==============================================================================
|
|
|
|
@lru_cache
|
|
def _get_scoring_weights() -> Tuple[float, float, float]:
|
|
"""
|
|
Liefert die Basis-Gewichtung (semantic_weight, edge_weight, centrality_weight).
|
|
Prio: 1. retriever.yaml -> 2. Environment -> 3. Hardcoded Defaults
|
|
"""
|
|
settings = get_settings()
|
|
sem = float(getattr(settings, "RETRIEVER_W_SEM", 1.0))
|
|
edge = float(getattr(settings, "RETRIEVER_W_EDGE", 0.0))
|
|
cent = float(getattr(settings, "RETRIEVER_W_CENT", 0.0))
|
|
|
|
config_path = os.getenv("MINDNET_RETRIEVER_CONFIG", "config/retriever.yaml")
|
|
if yaml is None:
|
|
return sem, edge, cent
|
|
try:
|
|
if os.path.exists(config_path):
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f) or {}
|
|
scoring = data.get("scoring", {}) or {}
|
|
sem = float(scoring.get("semantic_weight", sem))
|
|
edge = float(scoring.get("edge_weight", edge))
|
|
cent = float(scoring.get("centrality_weight", cent))
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load weights from {config_path}: {e}")
|
|
return sem, edge, cent
|
|
return sem, edge, cent
|
|
|
|
|
|
def _get_client_and_prefix() -> Tuple[Any, str]:
|
|
"""Liefert das initialisierte Qdrant-Client-Objekt und das Collection-Präfix."""
|
|
cfg = qdr.QdrantConfig.from_env()
|
|
client = qdr.get_client(cfg)
|
|
return client, cfg.prefix
|
|
|
|
|
|
def _get_query_vector(req: QueryRequest) -> List[float]:
|
|
"""Wandelt Text-Queries via EmbeddingsClient um oder nutzt vorhandenen Vektor."""
|
|
if req.query_vector:
|
|
return list(req.query_vector)
|
|
|
|
if not req.query:
|
|
raise ValueError("QueryRequest benötigt entweder 'query' oder 'query_vector'")
|
|
|
|
settings = get_settings()
|
|
model_name = settings.MODEL_NAME
|
|
|
|
try:
|
|
return ec.embed_text(req.query, model_name=model_name)
|
|
except TypeError:
|
|
return ec.embed_text(req.query)
|
|
|
|
|
|
def _semantic_hits(
|
|
client: Any,
|
|
prefix: str,
|
|
vector: List[float],
|
|
top_k: int,
|
|
filters: Dict[str, Any] | None = None,
|
|
) -> List[Tuple[str, float, Dict[str, Any]]]:
|
|
"""Führt eine Vektorsuche in Qdrant aus."""
|
|
flt = filters or None
|
|
raw_hits = qp.search_chunks_by_vector(client, prefix, vector, top=top_k, filters=flt)
|
|
results: List[Tuple[str, float, Dict[str, Any]]] = []
|
|
for pid, score, payload in raw_hits:
|
|
results.append((str(pid), float(score), dict(payload or {})))
|
|
return results
|
|
|
|
# ==============================================================================
|
|
# 2. WP-22 SCORING LOGIC (LIFECYCLE & FORMULA)
|
|
# ==============================================================================
|
|
|
|
def _get_status_multiplier(payload: Dict[str, Any]) -> float:
|
|
"""
|
|
WP-22 A: Lifecycle-Scoring.
|
|
- stable: 1.2 (Validiertes Wissen fördern)
|
|
- draft: 0.5 (Entwürfe de-priorisieren)
|
|
"""
|
|
status = str(payload.get("status", "active")).lower().strip()
|
|
if status == "stable":
|
|
return 1.2
|
|
if status == "draft":
|
|
return 0.5
|
|
return 1.0
|
|
|
|
|
|
def _compute_total_score(
|
|
semantic_score: float,
|
|
payload: Dict[str, Any],
|
|
edge_bonus_raw: float = 0.0,
|
|
cent_bonus_raw: float = 0.0,
|
|
dynamic_edge_boosts: Dict[str, float] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Die zentrale mathematische Scoring-Formel von WP-22.
|
|
Score = (Similarity * StatusMult) * (1 + (Weight-1) + DynamicBoost)
|
|
"""
|
|
_sem_w, edge_w_cfg, cent_w_cfg = _get_scoring_weights()
|
|
status_mult = _get_status_multiplier(payload)
|
|
node_weight = float(payload.get("retriever_weight", 1.0))
|
|
|
|
# 1. Base Score (Semantik * Lifecycle)
|
|
base_val = float(semantic_score) * status_mult
|
|
|
|
# 2. Graph Boost Factor (WP-22 C)
|
|
graph_boost_factor = 1.5 if dynamic_edge_boosts and (edge_bonus_raw > 0 or cent_bonus_raw > 0) else 1.0
|
|
|
|
# 3. Graph Contributions
|
|
edge_impact = (edge_w_cfg * edge_bonus_raw) * graph_boost_factor
|
|
cent_impact = (cent_w_cfg * cent_bonus_raw) * graph_boost_factor
|
|
dynamic_graph_impact = edge_impact + cent_impact
|
|
|
|
# 4. Final Merge
|
|
total = base_val * (1.0 + (node_weight - 1.0) + dynamic_graph_impact)
|
|
|
|
return {
|
|
"total": max(0.001, float(total)),
|
|
"edge_bonus": float(edge_bonus_raw),
|
|
"cent_bonus": float(cent_bonus_raw),
|
|
"status_multiplier": status_mult,
|
|
"graph_boost_factor": graph_boost_factor,
|
|
"type_impact": node_weight - 1.0,
|
|
"base_val": base_val,
|
|
"edge_impact_final": edge_impact,
|
|
"cent_impact_final": cent_impact
|
|
}
|
|
|
|
# ==============================================================================
|
|
# 3. EXPLANATION LAYER (DEBUG & VERIFIABILITY)
|
|
# ==============================================================================
|
|
|
|
def _build_explanation(
|
|
semantic_score: float,
|
|
payload: Dict[str, Any],
|
|
scoring_debug: Dict[str, Any],
|
|
subgraph: Optional[ga.Subgraph],
|
|
target_note_id: Optional[str],
|
|
applied_boosts: Optional[Dict[str, float]] = None
|
|
) -> Explanation:
|
|
"""Erstellt ein detailliertes Explanation-Objekt mit WP-22 Metriken."""
|
|
_, edge_w_cfg, cent_w_cfg = _get_scoring_weights()
|
|
|
|
type_weight = float(payload.get("retriever_weight", 1.0))
|
|
status_mult = scoring_debug["status_multiplier"]
|
|
graph_bf = scoring_debug["graph_boost_factor"]
|
|
note_type = payload.get("type", "unknown")
|
|
base_val = scoring_debug["base_val"]
|
|
|
|
# 1. Score Breakdown
|
|
breakdown = ScoreBreakdown(
|
|
semantic_contribution=base_val,
|
|
edge_contribution=base_val * scoring_debug["edge_impact_final"],
|
|
centrality_contribution=base_val * scoring_debug["cent_impact_final"],
|
|
raw_semantic=semantic_score,
|
|
raw_edge_bonus=scoring_debug["edge_bonus"],
|
|
raw_centrality=scoring_debug["cent_bonus"],
|
|
node_weight=type_weight,
|
|
status_multiplier=status_mult,
|
|
graph_boost_factor=graph_bf
|
|
)
|
|
|
|
reasons: List[Reason] = []
|
|
edges_dto: List[EdgeDTO] = []
|
|
|
|
# 2. Reasons generieren
|
|
if semantic_score > 0.70:
|
|
reasons.append(Reason(kind="semantic", message="Textuelle Übereinstimmung.", score_impact=base_val))
|
|
|
|
if type_weight != 1.0:
|
|
msg = "Bevorzugt" if type_weight > 1.0 else "Abgewertet"
|
|
reasons.append(Reason(kind="type", message=f"{msg} durch Typ '{note_type}'.", score_impact=base_val * (type_weight - 1.0)))
|
|
|
|
if status_mult != 1.0:
|
|
txt = "Bonus" if status_mult > 1.0 else "Malus"
|
|
reasons.append(Reason(kind="lifecycle", message=f"Status-{txt} ({payload.get('status')}).", score_impact=0.0))
|
|
|
|
# 3. Kanten-Details (WP-22 B)
|
|
if subgraph and target_note_id and scoring_debug["edge_bonus"] > 0:
|
|
raw_edges = []
|
|
if hasattr(subgraph, "get_incoming_edges"):
|
|
raw_edges.extend(subgraph.get_incoming_edges(target_note_id) or [])
|
|
if hasattr(subgraph, "get_outgoing_edges"):
|
|
raw_edges.extend(subgraph.get_outgoing_edges(target_note_id) or [])
|
|
|
|
for edge in raw_edges:
|
|
src, tgt = edge.get("source"), edge.get("target")
|
|
k = edge.get("kind", "edge")
|
|
prov = edge.get("provenance", "rule")
|
|
conf = float(edge.get("confidence", 1.0))
|
|
|
|
is_incoming = (tgt == target_note_id)
|
|
direction = "in" if is_incoming else "out"
|
|
|
|
# neighbor_id Scope-Fix
|
|
neighbor_id = src if is_incoming else tgt
|
|
|
|
edge_obj = EdgeDTO(
|
|
id=f"{src}->{tgt}:{k}", kind=k, source=src, target=tgt,
|
|
weight=conf, direction=direction,
|
|
provenance=prov, confidence=conf
|
|
)
|
|
edges_dto.append(edge_obj)
|
|
|
|
top_edges = sorted(edges_dto, key=lambda e: e.confidence, reverse=True)
|
|
for e in top_edges[:3]:
|
|
prov_txt = "Explizite" if e.provenance == "explicit" else "Heuristische"
|
|
boost_txt = f" [Boost x{applied_boosts.get(e.kind)}]" if applied_boosts and e.kind in applied_boosts else ""
|
|
|
|
# Nachbar-ID innerhalb der Loop sicherstellen
|
|
target_name = e.source if e.direction == "in" else e.target
|
|
msg = f"{prov_txt} Kante '{e.kind}'{boost_txt} von/zu '{target_name}'."
|
|
reasons.append(Reason(kind="edge", message=msg, score_impact=edge_w_cfg * e.confidence))
|
|
|
|
if scoring_debug["cent_bonus"] > 0.01:
|
|
reasons.append(Reason(kind="centrality", message="Knoten liegt zentral im Kontext.", score_impact=breakdown.centrality_contribution))
|
|
|
|
return Explanation(
|
|
breakdown=breakdown,
|
|
reasons=reasons,
|
|
related_edges=edges_dto if edges_dto else None,
|
|
applied_boosts=applied_boosts
|
|
)
|
|
|
|
|
|
def _extract_expand_options(req: QueryRequest) -> Tuple[int, List[str] | None]:
|
|
"""Extrahiert Expansion-Tiefe und Kanten-Filter."""
|
|
expand = getattr(req, "expand", None)
|
|
if not expand: return 0, None
|
|
if isinstance(expand, dict):
|
|
return int(expand.get("depth", 1)), expand.get("edge_types")
|
|
if hasattr(expand, "depth"):
|
|
return int(getattr(expand, "depth", 1)), getattr(expand, "edge_types", None)
|
|
return 1, None
|
|
|
|
|
|
def _build_hits_from_semantic(
|
|
hits: Iterable[Tuple[str, float, Dict[str, Any]]],
|
|
top_k: int,
|
|
used_mode: str,
|
|
subgraph: ga.Subgraph | None = None,
|
|
explain: bool = False,
|
|
dynamic_edge_boosts: Dict[str, float] = None
|
|
) -> QueryResponse:
|
|
"""Wandelt semantische Roh-Treffer in strukturierte QueryHits um."""
|
|
t0 = time.time()
|
|
enriched = []
|
|
|
|
for pid, semantic_score, payload in hits:
|
|
edge_bonus, cent_bonus = 0.0, 0.0
|
|
target_note_id = payload.get("note_id")
|
|
|
|
if subgraph is not None and target_note_id:
|
|
try:
|
|
edge_bonus = float(subgraph.edge_bonus(target_note_id))
|
|
cent_bonus = float(subgraph.centrality_bonus(target_note_id))
|
|
except Exception: pass
|
|
|
|
debug_data = _compute_total_score(
|
|
semantic_score, payload, edge_bonus_raw=edge_bonus,
|
|
cent_bonus_raw=cent_bonus, dynamic_edge_boosts=dynamic_edge_boosts
|
|
)
|
|
enriched.append((pid, float(semantic_score), payload, debug_data))
|
|
|
|
enriched_sorted = sorted(enriched, key=lambda h: h[3]["total"], reverse=True)
|
|
limited_hits = enriched_sorted[: max(1, top_k)]
|
|
|
|
results: List[QueryHit] = []
|
|
for pid, semantic_score, payload, debug in limited_hits:
|
|
explanation_obj = None
|
|
if explain:
|
|
explanation_obj = _build_explanation(
|
|
semantic_score=float(semantic_score),
|
|
payload=payload, scoring_debug=debug,
|
|
subgraph=subgraph, target_note_id=payload.get("note_id"),
|
|
applied_boosts=dynamic_edge_boosts
|
|
)
|
|
|
|
results.append(QueryHit(
|
|
node_id=str(pid),
|
|
note_id=payload.get("note_id", "unknown"),
|
|
semantic_score=float(semantic_score),
|
|
edge_bonus=debug["edge_bonus"],
|
|
centrality_bonus=debug["cent_bonus"],
|
|
total_score=debug["total"],
|
|
source={"path": payload.get("path"), "text": payload.get("page_content") or payload.get("text")},
|
|
payload=payload,
|
|
explanation=explanation_obj
|
|
))
|
|
|
|
return QueryResponse(results=results, used_mode=used_mode, latency_ms=int((time.time() - t0) * 1000))
|
|
|
|
# ==============================================================================
|
|
# 4. PUBLIC INTERFACE
|
|
# ==============================================================================
|
|
|
|
def hybrid_retrieve(req: QueryRequest) -> QueryResponse:
|
|
"""Hybrid-Suche: Semantik + WP-22 Graph Intelligence."""
|
|
client, prefix = _get_client_and_prefix()
|
|
vector = list(req.query_vector) if req.query_vector else _get_query_vector(req)
|
|
top_k = req.top_k or 10
|
|
hits = _semantic_hits(client, prefix, vector, top_k=top_k, filters=req.filters)
|
|
|
|
expand_depth, edge_types = _extract_expand_options(req)
|
|
boost_edges = getattr(req, "boost_edges", {}) or {}
|
|
|
|
subgraph: ga.Subgraph | None = None
|
|
if expand_depth > 0 and hits:
|
|
seed_ids = list({h[2].get("note_id") for h in hits if h[2].get("note_id")})
|
|
if seed_ids:
|
|
try:
|
|
subgraph = ga.expand(client, prefix, seed_ids, depth=expand_depth, edge_types=edge_types)
|
|
if subgraph and hasattr(subgraph, "graph"):
|
|
for u, v, data in subgraph.graph.edges(data=True):
|
|
# Provenance Weighting (Concept 2.6)
|
|
prov = data.get("provenance", "rule")
|
|
prov_w = 1.0 if prov == "explicit" else (0.9 if prov == "smart" else 0.7)
|
|
|
|
# Intent Boost mapping
|
|
k = data.get("kind")
|
|
intent_b = boost_edges.get(k, 1.0)
|
|
data["weight"] = data.get("weight", 1.0) * prov_w * intent_b
|
|
except Exception: subgraph = None
|
|
|
|
return _build_hits_from_semantic(hits, top_k, "hybrid", subgraph, req.explain, boost_edges)
|
|
|
|
|
|
class Retriever:
|
|
"""Asynchroner Wrapper für FastAPI-Integration."""
|
|
async def search(self, request: QueryRequest) -> QueryResponse:
|
|
return hybrid_retrieve(request) |