From 3c5c01998f0db7ee84ffa6469bf123ce1bfd6252 Mon Sep 17 00:00:00 2001 From: Lars Date: Mon, 8 Dec 2025 19:41:32 +0100 Subject: [PATCH] Parametrisierung und testscript --- config/decision_engine.yaml | 2 +- tests/test_wp06_decision.py | 83 ++++++++++++++++++++----------------- 2 files changed, 46 insertions(+), 39 deletions(-) diff --git a/config/decision_engine.yaml b/config/decision_engine.yaml index 58c090d..6145ad1 100644 --- a/config/decision_engine.yaml +++ b/config/decision_engine.yaml @@ -18,7 +18,7 @@ strategies: # HIER definierst du, was das 'Gewissen' ausmacht: # Aktuell: Werte & Prinzipien. # Später einfach ergänzen um: "goal", "experience", "belief" - inject_types: ["value", "principle"] + inject_types: ["value", "principle", "goal"] prompt_template: "decision_template" prepend_instruction: | !!! ENTSCHEIDUNGS-MODUS !!! diff --git a/tests/test_wp06_decision.py b/tests/test_wp06_decision.py index 6eb9e52..d0adf0d 100644 --- a/tests/test_wp06_decision.py +++ b/tests/test_wp06_decision.py @@ -1,33 +1,29 @@ """ -tests/test_wp06_decision.py — Integrationstest für WP-06 +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 -# KORREKTUR: Port auf 8002 (Dev-Environment) gesetzt -API_URL = "http://localhost:8002" +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") -def test_decision_engine(): - print(f"🔵 Starte WP-06 Decision Engine Test gegen {API_URL}...\n") - - # 1. Die Entscheidungsfrage - # Das Keyword "Soll ich" triggert die Heuristik in chat.py - question = "Soll ich SuperCloud Sync für meine privaten Tagebücher nutzen?" - payload = { - "message": question, + "message": query, "top_k": 3, "explain": True } try: # Request senden - print(f"FRAGE: '{question}'") + 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 = requests.post(f"{api_url}/chat/", json=payload, timeout=120) response.raise_for_status() data = response.json() @@ -36,34 +32,42 @@ def test_decision_engine(): # 1. Intent Check intent = data.get("intent", "UNKNOWN") print(f"\n1. INTENT DETECTION: [{'✅' if intent == 'DECISION' else '❌'}]") - print(f" Erkannt: {intent} (Erwartet: DECISION)") + print(f" Erkannt: {intent}") # 2. Source Check (Strategic Retrieval) - # Wir suchen nach einer Quelle vom Typ 'value' oder 'principle' sources = data.get("sources", []) - found_value = False - found_fact = False + strategic_hits = [] + fact_hits = [] - print(f"\n2. STRATEGIC RETRIEVAL:") + 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) - print(f" [{i+1}] {title} (Typ: {node_type})") - - if node_type in ["value", "principle"]: - found_value = True - if "SuperCloud" in title or node_type == "concept": - found_fact = True + # 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) - if found_value and found_fact: - print(" ✅ ERFOLG: Fakten UND Werte im Kontext gefunden.") - elif found_fact and not found_value: - print(" ⚠️ WARNUNG: Nur Fakten gefunden. Values-Injection fehlgeschlagen?") + 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(" ❌ FEHLER: Relevante Quellen fehlen.") + print(f"\n ⚠️ WARNUNG: Keine strategischen Quellen (Value/Principle/Goal) gefunden.") # 3. Reasoning Check (LLM Antwort) answer = data.get("answer", "") @@ -72,19 +76,22 @@ def test_decision_engine(): print(answer) print("-" * 60) - # Einfache Keyword-Prüfung in der Antwort - keywords_negative = ["nein", "nicht nutzen", "abraten", "konflikt", "verboten", "gegen"] - if any(k in answer.lower() for k in keywords_negative): - print("\n✅ FAZIT: Das System rät korrekt ab (basierend auf 'Privacy_First').") - else: - print("\n⚠️ FAZIT: Antwort scheint nicht strikt abzulehnen. Prüfe Prompt.") - except requests.exceptions.ConnectionError: - print(f"\n❌ FEHLER: Keine Verbindung zu {API_URL}. Läuft der Server?") + 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__": - test_decision_engine() \ No newline at end of file + 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) \ No newline at end of file