100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
"""
|
||
tests/test_wp06_decision.py — Flexibler Integrationstest für WP-06
|
||
Führt eine Entscheidungsfrage gegen die API aus.
|
||
Unterstützt Parameter für Frage, Port und erwarteten Intent.
|
||
"""
|
||
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 10-30s dauern) ...")
|
||
|
||
response = requests.post(f"{api_url}/chat/", json=payload, timeout=120)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
# --- CHECKS ---
|
||
|
||
# 1. Intent Check
|
||
intent = data.get("intent", "UNKNOWN")
|
||
match = intent == expected_intent
|
||
print(f"\n1. INTENT DETECTION: [{'✅' if match else '❌'}]")
|
||
print(f" Erkannt: {intent} (Erwartet: {expected_intent})")
|
||
|
||
# 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)")
|
||
|
||
for i, source in enumerate(sources):
|
||
# Safe access auf Source-Dict
|
||
src_meta = source.get("source", {})
|
||
node_type = src_meta.get("type", "unknown")
|
||
title = source.get("note_id", "Unknown")
|
||
score = source.get("total_score", 0.0)
|
||
|
||
# Marker für Ausgabe
|
||
marker = " "
|
||
# Wir prüfen hier generisch auf alle strategischen Typen
|
||
if node_type in ["value", "principle", "goal", "experience", "belief", "profile"]:
|
||
marker = "🎯" # Strategischer Treffer
|
||
strategic_hits.append(title)
|
||
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.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).")
|
||
|
||
args = parser.parse_args()
|
||
|
||
test_decision_engine(args.query, args.port, args.expect) |