Lars
05dbb21c58
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 3s
121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
#
|
||
# test_retriever_smoke.py – End-to-End Smoke-Test für den mindnet Retriever
|
||
#
|
||
# Dieser Smoke-Test ist für manuelle End-to-End-Tests gedacht.
|
||
# Er wird NICHT automatisch von pytest ausgeführt.
|
||
#
|
||
# Empfohlener Speicherort: tests/test_retriever_smoke.py
|
||
#
|
||
# Aufrufbeispiele:
|
||
#
|
||
# python tests/test_retriever_smoke.py --query "embeddings" --mode semantic
|
||
#
|
||
# python tests/test_retriever_smoke.py \
|
||
# --query "embeddings" \
|
||
# --mode hybrid \
|
||
# --expand-depth 1 \
|
||
# --expand-edge-types references depends_on
|
||
#
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
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="URL des Query-Endpunkts")
|
||
parser.add_argument("--query", required=True, help="Text-Query")
|
||
parser.add_argument("--mode", choices=["semantic", "hybrid"], default="hybrid")
|
||
parser.add_argument("--top-k", type=int, default=5)
|
||
parser.add_argument("--expand-depth", type=int, default=0)
|
||
parser.add_argument(
|
||
"--expand-edge-types",
|
||
nargs="*",
|
||
default=None,
|
||
help="Edge-Typen (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:
|
||
print(f"[ERROR] HTTP-Request fehlgeschlagen: {exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
if resp.status_code != 200:
|
||
print(f"[ERROR] Statuscode: {resp.status_code}", file=sys.stderr)
|
||
print(resp.text, file=sys.stderr)
|
||
return 1
|
||
|
||
try:
|
||
data = resp.json()
|
||
except Exception:
|
||
print("[ERROR] Ungültiges JSON:", 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 []
|
||
print(f"used_mode: {data.get('used_mode')}")
|
||
print(f"latency: {data.get('latency_ms')} ms")
|
||
print(f"Treffer: {len(results)}")
|
||
print()
|
||
|
||
for i, hit in enumerate(results, start=1):
|
||
nid = hit.get("note_id")
|
||
cid = hit.get("node_id")
|
||
tot = hit.get("total_score")
|
||
sem = hit.get("semantic_score")
|
||
edge = hit.get("edge_bonus")
|
||
cen = hit.get("centrality_bonus")
|
||
src = hit.get("source") or {}
|
||
path = src.get("path")
|
||
sec = src.get("section")
|
||
|
||
print(f"[{i}] note_id={nid} node_id={cid}")
|
||
try:
|
||
print(
|
||
f" total={float(tot):.4f} "
|
||
f"(semantic={float(sem):.4f} edge={float(edge):.4f} centrality={float(cen):.4f})"
|
||
)
|
||
except Exception:
|
||
print(f" total={tot}, semantic={sem}, edge={edge}, centrality={cen}")
|
||
|
||
if path or sec:
|
||
print(f" source: {path or '-'} :: {sec or '-'}")
|
||
print()
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|