mindnet/tests/test_wp06_decision.py

97 lines
3.3 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 und Port.
"""
import requests
import json
import sys
import argparse
def test_decision_engine(query: str, port: int):
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")
print(f"\n1. INTENT DETECTION: [{'' if intent == 'DECISION' else ''}]")
print(f" Erkannt: {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 = " "
if node_type in ["value", "principle", "goal"]:
marker = "🎯" # Strategischer Treffer
strategic_hits.append(title)
elif node_type in ["decision", "experience"]:
marker = "🧠"
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 ⚠️ WARNUNG: Keine strategischen Quellen (Value/Principle/Goal) 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).")
args = parser.parse_args()
test_decision_engine(args.query, args.port)