Lars
95affe87e5
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 5s
123 lines
4.1 KiB
Python
123 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
scripts/test_retriever_smoke.py - einfacher Smoke-Test für den mindnet-Retriever
|
|
|
|
Zweck:
|
|
- Startet eine Query gegen die laufende FastAPI-Instanz
|
|
- Prüft, ob der /query-Endpunkt antwortet und sinnvolle Treffer liefert
|
|
- Gibt Top-K-Ergebnisse inkl. Scores auf stdout aus
|
|
|
|
Verwendung:
|
|
1) Stelle sicher, dass der mindnet-Service läuft, z. B.:
|
|
uvicorn app.main:create_app --factory --reload
|
|
|
|
2) Führe dieses Skript aus:
|
|
python scripts/test_retriever_smoke.py --query "karate" --mode hybrid
|
|
|
|
3) Optional:
|
|
python scripts/test_retriever_smoke.py --query "embeddings" --mode semantic --top-k 5
|
|
|
|
Hinweis:
|
|
- Standard-URL ist http://127.0.0.1:8000
|
|
- Endpoint ist /query (bestehender Router), der QueryRequest erwartet
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from typing import Any, Dict
|
|
|
|
import requests
|
|
|
|
|
|
def _build_payload(args: argparse.Namespace) -> Dict[str, Any]:
|
|
payload: Dict[str, Any] = {
|
|
"mode": args.mode,
|
|
"query": args.query,
|
|
"top_k": args.top_k,
|
|
}
|
|
if args.expand_depth is not None and args.expand_depth > 0:
|
|
expand: Dict[str, Any] = {"depth": args.expand_depth}
|
|
if args.expand_edge_types:
|
|
expand["edge_types"] = args.expand_edge_types
|
|
payload["expand"] = expand
|
|
return payload
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Smoke-Test für mindnet-Retriever (/query)")
|
|
parser.add_argument("--url", default="http://127.0.0.1:8000/query", help="Basis-URL des Query-Endpunkts")
|
|
parser.add_argument("--query", required=True, help="Text-Query für den Retriever")
|
|
parser.add_argument("--mode", choices=["semantic", "hybrid"], default="hybrid", help="Retrieval-Modus")
|
|
parser.add_argument("--top-k", type=int, default=5, help="Anzahl der gewünschten Treffer")
|
|
parser.add_argument("--expand-depth", type=int, default=0, help="Edge-Expansionstiefe (nur hybrid)")
|
|
parser.add_argument(
|
|
"--expand-edge-types",
|
|
nargs="*",
|
|
default=None,
|
|
help="Liste von Edge-Typen für die Expansion (z. B. references depends_on)",
|
|
)
|
|
|
|
args = parser.parse_args(argv)
|
|
|
|
payload = _build_payload(args)
|
|
|
|
try:
|
|
resp = requests.post(args.url, json=payload, timeout=10)
|
|
except Exception as exc: # pragma: no cover - Netzwerkfehler
|
|
print(f"[ERROR] HTTP-Request fehlgeschlagen: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
if resp.status_code != 200:
|
|
print(f"[ERROR] Unerwarteter Statuscode: {resp.status_code}", file=sys.stderr)
|
|
print(resp.text, file=sys.stderr)
|
|
return 1
|
|
|
|
try:
|
|
data = resp.json()
|
|
except Exception as exc: # pragma: no cover - JSON-Fehler
|
|
print(f"[ERROR] Antwort ist kein gültiges JSON: {exc}", file=sys.stderr)
|
|
print(resp.text, file=sys.stderr)
|
|
return 1
|
|
|
|
print("=== mindnet Retriever Smoke-Test ===")
|
|
print(f"URL: {args.url}")
|
|
print(f"Mode: {args.mode}")
|
|
print(f"Query: {args.query!r}")
|
|
print(f"Top-K: {args.top_k}")
|
|
print()
|
|
|
|
results = data.get("results") or []
|
|
used_mode = data.get("used_mode")
|
|
latency_ms = data.get("latency_ms")
|
|
|
|
print(f"used_mode = {used_mode}, latency = {latency_ms} ms")
|
|
print(f"Treffer insgesamt: {len(results)}")
|
|
print()
|
|
|
|
for i, hit in enumerate(results, start=1):
|
|
note_id = hit.get("note_id")
|
|
node_id = hit.get("node_id")
|
|
total = hit.get("total_score")
|
|
sem = hit.get("semantic_score")
|
|
edge_bonus = hit.get("edge_bonus")
|
|
cent_bonus = hit.get("centrality_bonus")
|
|
src = hit.get("source") or {}
|
|
path = src.get("path")
|
|
section = src.get("section")
|
|
|
|
print(f"[{i}] note_id={note_id} node_id={node_id}")
|
|
print(f" total={total:.4f} (semantic={sem:.4f} edge={edge_bonus:.4f} centrality={cent_bonus:.4f})")
|
|
if path or section:
|
|
print(f" source: {path or '-'} :: {section or '-'}")
|
|
print()
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
raise SystemExit(main())
|