mindnet/tests/diag_subgraph_for_note.py
Lars e93bab6ea7
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 4s
Fassadenauflösung unter app/core
2025-12-28 11:04:40 +01:00

121 lines
3.3 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Diagnose für den Graph-Adapter (Subgraph) in mindnet.
Nutzung (aus Projektwurzel):
(.venv) python tests/diag_subgraph_for_note.py \
--note-id 20251130-relshow-embeddings-101 \
--depth 1
Optional:
--prefix mindnet
--edge-type references --edge-type related_to ...
Ziel:
- Ruft app.core.graph_adapter.expand(...) für eine gegebene note_id auf.
- Gibt edge_bonus und centrality_bonus für diese note_id aus.
- Zeigt zusätzlich einige Nachbarn und deren Boni.
"""
from __future__ import annotations
import argparse
import os
import sys
from typing import List, Optional, Tuple
# sys.path anpassen, damit "import app...." aus tests/ funktioniert
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_ROOT = os.path.dirname(THIS_DIR)
if PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
from app.core.database.qdrant import QdrantConfig, get_client # type: ignore
import app.core.graph.graph_subgraph as ga # type: ignore
def build_subgraph_for_note(
note_id: str,
prefix: Optional[str],
depth: int,
edge_types: Optional[List[str]] = None,
) -> Tuple[ga.Subgraph, str]:
cfg = QdrantConfig.from_env()
if prefix:
cfg.prefix = prefix
client = get_client(cfg)
sg = ga.expand(client, cfg.prefix, [note_id], depth=depth, edge_types=edge_types)
return sg, cfg.prefix
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--note-id",
required=True,
help="Note-ID, z. B. 20251130-relshow-embeddings-101",
)
parser.add_argument(
"--depth",
type=int,
default=1,
help="Expansionstiefe (Default: 1)",
)
parser.add_argument(
"--prefix",
default=None,
help="Qdrant-Prefix (Default: aus ENV/QdrantConfig)",
)
parser.add_argument(
"--edge-type",
action="append",
default=None,
help="Edge-Typ-Filter (kann mehrfach angegeben werden), z. B. --edge-type references",
)
args = parser.parse_args()
note_id = args.note_id
depth = max(0, int(args.depth))
prefix = args.prefix
edge_types = args.edge_type
sg, eff_prefix = build_subgraph_for_note(note_id, prefix, depth, edge_types=edge_types)
print(f"Prefix: {eff_prefix}")
print(f"Note-ID (Seed): {note_id}")
print(f"Depth: {depth}")
print(f"Edge-Types-Filter: {edge_types or '(keiner)'}")
print("-" * 60)
# Boni für die Seed-Note
eb = sg.edge_bonus(note_id)
cb = sg.centrality_bonus(note_id)
print(f"edge_bonus({note_id}) = {eb:.4f}")
print(f"centrality_bonus({note_id}) = {cb:.4f}")
print("-" * 60)
# Ein paar weitere Knoten anzeigen (falls vorhanden)
print("Beispiele weiterer Knoten mit Boni (max. 10):")
printed = 0
for node, edges in sg.adj.items():
if node == note_id:
continue
if printed >= 10:
break
eb2 = sg.edge_bonus(node)
cb2 = sg.centrality_bonus(node)
print(f"- node={node}")
print(f" edge_bonus={eb2:.4f} centrality={cb2:.4f}")
printed += 1
if printed == 0:
print("(keine weiteren Knoten im Subgraph registriert)")
if __name__ == "__main__":
main()