Lars
2b4c1b83bd
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 2s
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
scripts/wp04_smoketest.py — E2E-Schnelltest der WP-04 Endpunkte
|
||
|
||
Zweck:
|
||
- Holt exemplarisch einen Vektor aus {prefix}_chunks (falls verfügbar),
|
||
- ruft POST /query mit diesem Vektor auf,
|
||
- ruft GET /graph/<note_id> für die Top-Note auf,
|
||
- druckt Kurzresultate auf STDOUT.
|
||
|
||
Kompatibilität:
|
||
Python 3.12+, requests, qdrant-client
|
||
Version:
|
||
0.1.0 (Erstanlage)
|
||
Stand:
|
||
2025-10-07
|
||
Bezug:
|
||
WP-04 /query & /graph
|
||
Nutzung:
|
||
export QDRANT_URL="http://localhost:6333"
|
||
export MINDNET_PREFIX="mindnet"
|
||
python3 scripts/wp04_smoketest.py --api http://localhost:8000
|
||
Änderungsverlauf:
|
||
0.1.0 (2025-10-07) – Erstanlage.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
import os
|
||
import sys
|
||
import json
|
||
import argparse
|
||
import requests
|
||
from qdrant_client import QdrantClient
|
||
|
||
|
||
def get_any_chunk_vector(client: QdrantClient, prefix: str):
|
||
"""Holt eine beliebige Chunk-Vector/Note-ID-Kombi aus Qdrant (falls vorhanden)."""
|
||
col = f"{prefix}_chunks"
|
||
pts, _ = client.scroll(collection_name=col, limit=1, with_vectors=True, with_payload=True)
|
||
if not pts:
|
||
return None, None, None
|
||
p = pts[0]
|
||
vec = p.vector
|
||
payload = p.payload or {}
|
||
node_id = str(p.id)
|
||
note_id = payload.get("note_id")
|
||
return vec, node_id, note_id
|
||
|
||
|
||
def run(api_base: str, qdrant_url: str, prefix: str) -> int:
|
||
print(f"[i] API: {api_base}")
|
||
print(f"[i] Qdrant: {qdrant_url}")
|
||
client = QdrantClient(url=qdrant_url)
|
||
|
||
vec, node_id, note_id = get_any_chunk_vector(client, prefix)
|
||
if vec is None:
|
||
print("[!] Keine Chunks mit Vektoren gefunden. Bitte Import (WP-03) prüfen.")
|
||
return 2
|
||
|
||
print(f"[i] Beispiel-Chunk-ID: {node_id}, Note-ID: {note_id}")
|
||
print("[i] Rufe /query (hybrid) mit Beispielvektor auf ...")
|
||
|
||
q_payload = {
|
||
"mode": "hybrid",
|
||
"query_vector": vec, # 384d
|
||
"top_k": 5,
|
||
"expand": {"depth": 1, "edge_types": ["references", "belongs_to", "prev", "next"]},
|
||
}
|
||
r = requests.post(f"{api_base.rstrip('/')}/query", json=q_payload, timeout=60)
|
||
print(f"[i] /query Status: {r.status_code}")
|
||
if r.status_code != 200:
|
||
print(r.text)
|
||
return 3
|
||
|
||
data = r.json()
|
||
hits = data.get("results", [])
|
||
print(f"[i] Treffer: {len(hits)}")
|
||
for i, h in enumerate(hits, 1):
|
||
print(f" {i:02d}. node_id={h['node_id']} note_id={h.get('note_id')} total={h['total_score']:.3f}")
|
||
|
||
# Graph-Test (wenn Note-ID vorhanden)
|
||
gid = note_id or (hits[0].get("note_id") if hits else None)
|
||
if gid:
|
||
print(f"[i] Rufe /graph/{gid} ...")
|
||
r2 = requests.get(f"{api_base.rstrip('/')}/graph/{gid}?depth=1", timeout=60)
|
||
print(f"[i] /graph Status: {r2.status_code}")
|
||
if r2.status_code == 200:
|
||
g = r2.json()
|
||
print(f" Nodes: {len(g.get('nodes', []))}, Edges: {len(g.get('edges', []))}")
|
||
else:
|
||
print(r2.text)
|
||
|
||
print("[✔] Smoke-Test abgeschlossen.")
|
||
return 0
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--api", default="http://localhost:8000", help="Basis-URL der FastAPI.")
|
||
args = ap.parse_args()
|
||
|
||
qdrant_url = os.getenv("QDRANT_URL", "http://127.0.0.1:6333")
|
||
prefix = os.getenv("MINDNET_PREFIX", "mindnet")
|
||
sys.exit(run(args.api, qdrant_url, prefix))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|