122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
"""
|
||
tests/test_wp06_decision.py — Flexibler Integrationstest für WP-06
|
||
Update:
|
||
- Timeout 300s.
|
||
- Zeigt Intent Source an.
|
||
- Payload/Source Fallback für Metadaten.
|
||
- Debug-Dump bei unknown Type.
|
||
"""
|
||
import requests
|
||
import json
|
||
import sys
|
||
import argparse
|
||
|
||
def test_decision_engine(query: str, port: int, expected_intent: str):
|
||
api_url = f"http://localhost:{port}"
|
||
print(f"🔵 Starte WP-06 Decision Engine Test gegen {api_url}...\n")
|
||
|
||
payload = {
|
||
"message": query,
|
||
"top_k": 3,
|
||
"explain": True
|
||
}
|
||
|
||
try:
|
||
# Request senden
|
||
print(f"FRAGE: '{query}'")
|
||
print("... warte auf LLM (kann auf CPU >120s dauern) ...")
|
||
|
||
# FIX: Timeout auf 300 erhöht
|
||
response = requests.post(f"{api_url}/chat/", json=payload, timeout=300)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
# --- CHECKS ---
|
||
|
||
# 1. Intent Check
|
||
intent = data.get("intent", "UNKNOWN")
|
||
source_method = data.get("intent_source", "Unknown Source")
|
||
|
||
match = intent.upper() == expected_intent.upper()
|
||
|
||
print(f"\n1. INTENT DETECTION: [{'✅' if match else '❌'}]")
|
||
print(f" Erkannt: {intent}")
|
||
print(f" Erwartet: {expected_intent}")
|
||
print(f" Methode: {source_method}")
|
||
|
||
# 2. Source Check (Strategic Retrieval)
|
||
sources = data.get("sources", [])
|
||
strategic_hits = []
|
||
fact_hits = []
|
||
|
||
print(f"\n2. RETRIEVED SOURCES:")
|
||
if not sources:
|
||
print(" (Keine Quellen gefunden)")
|
||
|
||
debug_printed = False
|
||
|
||
for i, source in enumerate(sources):
|
||
# --- FIX: Robusterer Zugriff auf Metadaten ---
|
||
src_meta = source.get("payload") or source.get("source") or {}
|
||
|
||
node_type = src_meta.get("type", "unknown")
|
||
title = source.get("note_id", "Unknown")
|
||
score = source.get("total_score", 0.0)
|
||
|
||
# DEBUG: Wenn Typ unknown ist, dumpen wir das erste Objekt
|
||
if node_type == "unknown" and not debug_printed:
|
||
print(f"\n 🔴 DEBUG: Raw Data von Quelle {i+1} (da Typ unknown):")
|
||
print(json.dumps(source, indent=2, ensure_ascii=False))
|
||
print(" ------------------------------------------------")
|
||
debug_printed = True
|
||
|
||
# Marker für Ausgabe
|
||
marker = " "
|
||
if node_type in ["value", "principle", "goal", "experience", "belief", "profile", "decision"]:
|
||
marker = "🎯" # Strategischer Treffer
|
||
strategic_hits.append(f"{title} ({node_type})")
|
||
else:
|
||
marker = "📄"
|
||
fact_hits.append(title)
|
||
|
||
print(f" {marker} [{i+1}] {title} (Typ: {node_type}, Score: {score:.2f})")
|
||
|
||
# Analyse der Strategie
|
||
if strategic_hits:
|
||
print(f"\n ✅ ERFOLG: Strategische Quellen geladen: {strategic_hits}")
|
||
else:
|
||
print(f"\n ℹ️ INFO: Keine strategischen Quellen (Value/Experience/etc.) gefunden.")
|
||
|
||
# 3. Reasoning Check (LLM Antwort)
|
||
answer = data.get("answer", "")
|
||
print(f"\n3. REASONING (LLM Antwort):")
|
||
print("-" * 60)
|
||
print(answer)
|
||
print("-" * 60)
|
||
|
||
except requests.exceptions.ReadTimeout:
|
||
print(f"\n❌ TIMEOUT: Der Server hat nicht innerhalb von 300s geantwortet.")
|
||
print(" Tipp: Prüfe die Server-Logs (Terminal 1). Lädt er noch das Modell?")
|
||
sys.exit(1)
|
||
except requests.exceptions.ConnectionError:
|
||
print(f"\n❌ FEHLER: Keine Verbindung zu {api_url}. Läuft der Server?")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"\n❌ CRITICAL ERROR: {e}")
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description="Testet die WP-06 Decision Engine.")
|
||
parser.add_argument("--query", "-q", type=str,
|
||
default="Soll ich SuperCloud Sync für meine privaten Tagebücher nutzen?",
|
||
help="Die Frage, die an das System gestellt wird.")
|
||
parser.add_argument("--port", "-p", type=int,
|
||
default=8002,
|
||
help="Der Port der API (Default: 8002 für Dev).")
|
||
parser.add_argument("--expect", "-e", type=str,
|
||
default="DECISION",
|
||
help="Der erwartete Intent (z.B. DECISION, EMPATHY, FACT).")
|
||
|
||
args = parser.parse_args()
|
||
|
||
test_decision_engine(args.query, args.port, args.expect) |