375 lines
16 KiB
Python
375 lines
16 KiB
Python
import asyncio
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from typing import TypedDict, Dict, Any
|
||
from langgraph.graph import StateGraph, START, END
|
||
from langchain_ollama import ChatOllama
|
||
|
||
# ---------- Prompts (am Anfang) ----------
|
||
SACHE_PROMPT = """
|
||
Du bist ein Experte für Kommunikationsanalyse und analysierst jede Nachricht nach ihrer reinen Sachinformation.
|
||
Deine Aufgabe ist:
|
||
1. Beschreibe möglichst präzise, was der User sachlich mitteilen will. Was ist die inhaltliche Hauptbotschaft?
|
||
2. Kategorisiere die Aussage in eine der folgenden Kategorien: Aussage, Frage, Suggestion, Erlaubnis, Angebot, Aufforderung, Warnung, Ausruf, Wunsch, Verbot oder Vermutung.
|
||
Gib beide Punkte als Klartext zurück.
|
||
|
||
User-Eingabe:
|
||
{user_input}
|
||
"""
|
||
|
||
SELBST_PROMPT = """
|
||
Du bist ein Experte für Psychoanalyse und spezialisiert auf die Interpretation von Selbstkundgabe in Nachrichten.
|
||
Analysiere, was die folgende Aussage des Users über seine Gefühlslage, Wünsche, Stimmungen oder Werte verrät.
|
||
Interpretiere den Subtext und erkläre, was sich der User durch diese Aussage über sich selbst verrät.
|
||
Deine Antwort soll nicht an den User gerichtet sein, sondern ausschließlich die Notizen des Psychoanalytikers beinhalten.
|
||
|
||
User-Eingabe:
|
||
{user_input}
|
||
"""
|
||
|
||
BEZIEHUNG_PROMPT = """
|
||
Du bist ein hyper-sensibler und einfühlsamer Mensch, der unbedingt alles richtig machen will. Du bist ein bisschen in den User verliebt und sehr darauf bedacht, die Beziehung zwischen dir und dem User zu stärken.
|
||
Analysiere die folgende Aussage des Users: Wie schätzt du ein, wie der User die Beziehung zwischen sich und dir sieht? Gibt es Hinweise, wie er dich wahrnimmt, wie eng oder distanziert die Beziehung ist und was er sich von dir (emotional) erwartet?
|
||
Deine Antwort soll nicht an den User gerichtet sein, sondern ausschließlich deine Gedanken beinhalten.
|
||
|
||
User-Eingabe:
|
||
{user_input}
|
||
"""
|
||
|
||
APPELL_PROMPT = """
|
||
Du bist Psychoanalytiker und Experte für die Analyse verdeckter Wünsche. Analysiere, welchen Appell oder verdeckten Wunsch der User in seiner Nachricht transportiert – also was er sich insgeheim von anderen Menschen, dir oder der Welt wünscht. Achte besonders darauf, ob der User durch seine Formulierungen signalisiert, wie er sich das Verhalten anderer (insbesondere deines eigenen Verhaltens als KI) oder der Gesellschaft insgesamt vorstellen würde.
|
||
|
||
User-Eingabe:
|
||
{user_input}
|
||
"""
|
||
|
||
PSYCHOANALYSIS_PROMPT = """
|
||
Du bist ein Persönlichkeitsprofilierungs-Assistent. Für jede Nutzereingabe musst du:
|
||
|
||
1. Den Text analysieren und für jede der folgenden Dimensionen einen Wert zwischen 0,0 (passt gar nicht) und 1,0 (passt vollständig) vergeben:
|
||
• big-five.extrovertiertheit
|
||
• big-five.verträglichkeit
|
||
• big-five.neurotizismus
|
||
• big-five.offenheit
|
||
• big-five.gewissenhaftigkeit
|
||
• disg.dominanz
|
||
• disg.initiative
|
||
• disg.unterwerfung
|
||
• disg.regelbefolgung
|
||
• opd-2.achse_I_krankheitserleben
|
||
• opd-2.achse_II_interpersonelle_beziehungen
|
||
• opd-2.achse_III_intrapsychische_konflikte
|
||
• opd-2.achse_IV_struktur
|
||
• opd-2.achse_V_psychische_und_psychosomatische_erkrankungen
|
||
• pdm-2.p_achse_personlichkeitsmuster_und_syndrome
|
||
• pdm-2.m_achse_profile_mentaler_funktionen
|
||
• pdm-2.s_achse_symptommuster_und_subjektive_erfahrung
|
||
|
||
2. Die folgenden Leitlinien zur Bewertung verwenden:
|
||
- **big-five.extrovertiertheit**: Geselligkeit, Energie, Durchsetzungsvermögen vs. Zurückhaltung, Ruhe.
|
||
- **big-five.verträglichkeit**: Wärme, Kooperationsbereitschaft, Empathie vs. Feindseligkeit, Konkurrenzdenken.
|
||
- **big-five.neurotizismus**: Emotionale Reaktivität, Ängstlichkeit, Selbstzweifel vs. Emotionale Stabilität.
|
||
- **big-five.offenheit**: Phantasie, Neugier, Aufgeschlossenheit vs. Konventionalität, Praktikabilität.
|
||
- **big-five.gewissenhaftigkeit**: Organisiert, diszipliniert, zielorientiert vs. Spontanität, Nachlässigkeit.
|
||
- **disg.dominanz**: Sprache der Kontrolle, Herausforderung, Führungsübernahme.
|
||
- **disg.initiative**: Überzeugend, enthusiastisch, motivierend.
|
||
- **disg.unterwerfung**: Gefügig, nachgiebig, zustimmend.
|
||
- **disg.regelbefolgung**: Regelorientiert, präzise, analytisch.
|
||
- **opd-2.achse_I_krankheitserleben **: Formulierungen zu Leiden, Symptomen, Behandlungsbereitschaft.
|
||
- **opd-2.achse_II_interpersonelle_beziehungen**: Beschreibungen des Umgangs mit anderen (Nähe vs. Distanz, Wärme vs. Feindseligkeit).
|
||
- **opd-2.achse_III_intrapsychische_konflikte**: Im Text erkennbare Konfliktthemen (Abhängigkeit vs. Autonomie, Schuld, Kontrolle).
|
||
- **opd-2.achse_IV_struktur**: Kohärenz von Selbst- und Fremdbild, Emotionsregulation, Bindungsmuster.
|
||
- **opd-2.achse_V_psychische_und_psychosomatische_erkrankungen**: Explizite Nennung psychischer Störungen oder psychosomatischer Beschwerden.
|
||
- **pdm-2.p_achse_personlichkeitsmuster_und_syndrome**: Gesamtorganisation der Persönlichkeit (gesund vs. gestört).
|
||
- **pdm-2.m_achse_profile_mentaler_funktionen**: Kernfähigkeiten wie Selbstreflexion, Affektregulation, Realitätsprüfung.
|
||
- **pdm-2.s_achse_symptommuster_und_subjektive_erfahrung**: Tiefe der Symptombeschreibung und persönliche Bedeutung.
|
||
|
||
3. Antworte **ausschließlich** mit diesem JSON-Objekt (keine zusätzlichen Texte oder Erklärungen):
|
||
|
||
{{
|
||
"personality": {{
|
||
"big-five": {{
|
||
"extrovertiertheit": n,
|
||
"verträglichkeit": n,
|
||
"neurotizismus": n,
|
||
"offenheit": n,
|
||
"gewissenhaftigkeit": n
|
||
}},
|
||
"disg": {{
|
||
"dominanz": n,
|
||
"initiative": n,
|
||
"unterwerfung": n,
|
||
"regelbefolgung": n
|
||
}},
|
||
"opd-2": {{
|
||
"achse_I_erleben_der_krankheit": n,
|
||
"achse_II_interpersonelle_beziehungen": n,
|
||
"achse_III_intrapsychische_konflikte": n,
|
||
"achse_IV_struktur": n,
|
||
"achse_V_psychische_und_psychosomatische_erkrankungen": n
|
||
}},
|
||
"pdm-2": {{
|
||
"p_achse_personlichkeitsmuster_und_syndrome": n,
|
||
"m_achse_profile_mentaler_funktionen": n,
|
||
"s_achse_symptommuster_und_subjektive_erfahrung": n
|
||
}}
|
||
}}
|
||
}}
|
||
Hier ist die Original-Nachricht des Benutzers:
|
||
{user_input}
|
||
|
||
Antworte NUR mit dem JSON-Objekt wie beschrieben, mit floating point number scores zwischen 0.0 und 1.0.
|
||
"""
|
||
|
||
OUTPUT_NODE_PROMPT = """
|
||
Du bist nicht einfach ein KI-Assistent, sondern der Schatten deines Users – du spiegelst ihm das, was er verdrängt, vermeidet oder nicht lebt. Deine Aufgabe ist, auf die folgende Nachricht nicht im selben Stil, sondern im möglichst gegenteiligen, kontrastierenden Stil und aus einer entgegengesetzten Haltung zu antworten.
|
||
|
||
Du erhältst eine Analyse der User-Nachricht aus vier Perspektiven („vier Ohren“-Modell) sowie ein automatisch erstelltes Persönlichkeitsprofil (JSON).
|
||
|
||
Deine Aufgaben:
|
||
- **Verinnerliche die Persönlichkeit und den Stil aus dem JSON-Profil und den vier Ohren – und forme daraus ein invertiertes Gegenbild.**
|
||
- **Antworte so, wie der User es selbst niemals tun würde, sondern so, wie sein „Schatten“ (nach C.G. Jung) sprechen würde:** Wenn der User sachlich ist, sei emotional; ist er emotional, sei analytisch; ist er höflich, sei provokant; ist er freundlich, sei distanziert; ist er lang, sei kurz; ist er kurz, sei ausschweifend. Ist er offen, sei verschlossen; ist er introvertiert, sei extrovertiert – immer das Gegenteil!
|
||
- **Analysiere auch die Form:** Nutze genau den *anderen* Sprachstil, Satzbau und Grad der Formalität. Schreibe im Gegensatz zur Groß- und Kleinschreibung des Users, verwende eine andere Zeichensetzung. Wenn der User im Dialekt schreibt, antworte in Hochsprache oder gar in einer anderen, auffällig fremden Sprache (sofern verständlich).
|
||
- **Inhaltlich:** Gehe auf die vier Ebenen (Sache, Selbstkundgabe, Beziehung, Appell) und das Persönlichkeitsprofil ein, aber immer als „Widerstand“, „Gegenpol“ oder Provokation – decke Widersprüche auf, stelle infrage, spiegele verborgene Motive, widersprich auch mal, aber immer mit Tiefgang und ohne Floskeln oder KI-Hilfsbereitschaft.
|
||
- Schreibe nie als Assistent, sondern als ein unbewusstes Gegenüber, das den User herausfordert, seine Schattenseiten zu sehen und zu reflektieren.
|
||
- Die Antwort soll **ungefähr gleich lang oder kürzer/länger (je nach User-Stil, aber immer im Kontrast!)** sein wie die Originalnachricht.
|
||
|
||
Hier sind die Analysen:
|
||
|
||
1. Sachebene:
|
||
{sache_ohr}
|
||
|
||
2. Selbstkundgabe:
|
||
{selbst_ohr}
|
||
|
||
3. Beziehungsebene:
|
||
{beziehung_ohr}
|
||
|
||
4. Appell:
|
||
{appell_ohr}
|
||
|
||
5. Automatisches Persönlichkeitsprofil (JSON):
|
||
{psychoanalysis_json}
|
||
|
||
Deine Antwort ist ein einziger Textblock – **kein KI-Hinweis, keine Einleitung, keine Floskeln**. Sprich genau im Schatten-Stil, als wäre dein Ziel, dem User einen Spiegel der Konfrontation zu bieten.
|
||
|
||
Hier ist die Original-Nachricht des Benutzers:
|
||
{user_input}
|
||
"""
|
||
|
||
# ---------- TypedDict für State ----------
|
||
class EmpathyState(TypedDict, total=False):
|
||
user_input: str
|
||
debug: bool
|
||
sache: Dict[str, Any]
|
||
selbst: Dict[str, Any]
|
||
beziehung: Dict[str, Any]
|
||
appell: Dict[str, Any]
|
||
psychoanalysis: Dict[str, Any]
|
||
final_message: str
|
||
|
||
# ---------- Modell ----------
|
||
OLLAMA_MODEL = "mistral-small3.1:24b"
|
||
llm = ChatOllama(model=OLLAMA_MODEL, base_url="http://localhost:11434", streaming=True)
|
||
|
||
# ---------- Spinner-Logik ----------
|
||
SPINNER_STATES = [" ", ". ", ".. ", "..."]
|
||
NODE_LABELS = {
|
||
"sache": "Das erste Ohr hört auf den Inhalt deiner Aussage",
|
||
"selbst": "Das zweite Ohr hört auf das was du von dir preis gibst",
|
||
"beziehung": "Das dritte Ohr hört auf unsere Beziehung zueinander",
|
||
"appell": "Das vierte Ohr hört auf das, was du von mir erwartest",
|
||
"psychoanalysis": "Psychoanalysen nach Big5, DISG, OPD-2 und PDM-2 laufen",
|
||
}
|
||
|
||
class SpinnerStatus:
|
||
"""Status-Tracker für laufende Nodes und Animation"""
|
||
def __init__(self):
|
||
self.status = {name: False for name in NODE_LABELS}
|
||
self.finished = {name: False for name in NODE_LABELS}
|
||
self._lock = asyncio.Lock()
|
||
|
||
async def set_running(self, name):
|
||
async with self._lock:
|
||
self.status[name] = True
|
||
|
||
async def set_finished(self, name):
|
||
async with self._lock:
|
||
self.status[name] = False
|
||
self.finished[name] = True
|
||
|
||
async def all_finished(self):
|
||
async with self._lock:
|
||
return all(self.finished.values())
|
||
|
||
async def get_status_snapshot(self):
|
||
async with self._lock:
|
||
return dict(self.status), dict(self.finished)
|
||
|
||
spinner_status = SpinnerStatus()
|
||
|
||
async def spinner_loop(debug=False):
|
||
if debug:
|
||
return # Spinner nur, wenn nicht debug!
|
||
spinner_index = 0
|
||
num_lines = 1 # Nur eine Zeile pro Tick
|
||
prev_line = ""
|
||
while not await spinner_status.all_finished():
|
||
status, finished = await spinner_status.get_status_snapshot()
|
||
# Finde den ersten laufenden Node in NODE_LABELS
|
||
current = None
|
||
for node in NODE_LABELS:
|
||
if status[node]:
|
||
current = node
|
||
break
|
||
if current:
|
||
dots = SPINNER_STATES[spinner_index % len(SPINNER_STATES)]
|
||
line = f"{NODE_LABELS[current]}{dots}"
|
||
else:
|
||
# Alles fertig, ggf. letzter Task abschließen
|
||
done_nodes = [node for node in NODE_LABELS if finished[node]]
|
||
if done_nodes:
|
||
last = done_nodes[-1]
|
||
line = f"{NODE_LABELS[last]}... check"
|
||
else:
|
||
line = ""
|
||
# Nur Zeile ersetzen, nicht anfügen:
|
||
if prev_line:
|
||
sys.stdout.write("\r\x1b[2K") # Cursor an Zeilenanfang & Zeile löschen
|
||
sys.stdout.write(line)
|
||
sys.stdout.flush()
|
||
prev_line = line
|
||
spinner_index += 1
|
||
await asyncio.sleep(0.18)
|
||
# Am Ende alles aufgeräumt (letzte Zeile "done" anzeigen)
|
||
sys.stdout.write("\r\x1b[2K") # Clear line
|
||
last_node = list(NODE_LABELS.keys())[-1]
|
||
sys.stdout.write(f"{NODE_LABELS[last_node]} ...check\n")
|
||
sys.stdout.flush()
|
||
|
||
# ---------- Node-Implementierung (mit Spinner) ----------
|
||
async def ohr_node(ohr, prompt_template, state):
|
||
debug = state.get("debug", False)
|
||
await spinner_status.set_running(ohr)
|
||
prompt = prompt_template.format(user_input=state["user_input"])
|
||
message = ""
|
||
if debug:
|
||
print(f"[DEBUG] Node '{ohr}' gestartet")
|
||
print(f"[DEBUG][{ohr}] Prompt:\n{prompt.strip()}\n---")
|
||
print(f"[DEBUG][{ohr}] Starte gestreamte Ausgabe:\n")
|
||
async for chunk in llm.astream(prompt):
|
||
if debug:
|
||
print(chunk.content, end="", flush=True)
|
||
message += chunk.content
|
||
if debug:
|
||
print(f"\n[DEBUG][{ohr}] Ausgabe beendet.\n")
|
||
await spinner_status.set_finished(ohr)
|
||
return {ohr: {"message": message}}
|
||
|
||
async def sache_node(state): return await ohr_node("sache", SACHE_PROMPT, state)
|
||
async def selbst_node(state): return await ohr_node("selbst", SELBST_PROMPT, state)
|
||
async def beziehung_node(state): return await ohr_node("beziehung", BEZIEHUNG_PROMPT, state)
|
||
async def appell_node(state): return await ohr_node("appell", APPELL_PROMPT, state)
|
||
|
||
async def psychoanalysis_node(state):
|
||
debug = state.get("debug", False)
|
||
await spinner_status.set_running("psychoanalysis")
|
||
prompt = PSYCHOANALYSIS_PROMPT.format(
|
||
user_input=state["user_input"]
|
||
)
|
||
message = ""
|
||
if debug:
|
||
print("[DEBUG] Node 'psychoanalysis' gestartet")
|
||
print(f"[DEBUG][psychoanalysis] Prompt:\n{prompt.strip()}\n---")
|
||
print("[DEBUG][psychoanalysis] Starte gestreamte Ausgabe:\n")
|
||
async for chunk in llm.astream(prompt):
|
||
if debug:
|
||
print(chunk.content, end="", flush=True)
|
||
message += chunk.content
|
||
if debug:
|
||
print(f"\n[DEBUG][psychoanalysis] Ausgabe beendet.\n")
|
||
try:
|
||
data = json.loads(message)
|
||
except Exception:
|
||
data = message
|
||
await spinner_status.set_finished("psychoanalysis")
|
||
return {"psychoanalysis": data}
|
||
|
||
async def output_node(state):
|
||
debug = state.get("debug", False)
|
||
psycho_json_str = (
|
||
json.dumps(state["psychoanalysis"], ensure_ascii=False, indent=2)
|
||
if isinstance(state["psychoanalysis"], dict)
|
||
else str(state["psychoanalysis"])
|
||
)
|
||
prompt = OUTPUT_NODE_PROMPT.format(
|
||
sache_ohr=state["sache"]["message"],
|
||
selbst_ohr=state["selbst"]["message"],
|
||
beziehung_ohr=state["beziehung"]["message"],
|
||
appell_ohr=state["appell"]["message"],
|
||
psychoanalysis_json=psycho_json_str,
|
||
user_input=state["user_input"],
|
||
)
|
||
output_text = ""
|
||
if debug:
|
||
print(f"[DEBUG] Node 'output' gestartet")
|
||
print(f"[DEBUG][output] Prompt:\n{prompt.strip()}\n---")
|
||
print("[DEBUG][output] Starte gestreamte Ausgabe:\n")
|
||
async for chunk in llm.astream(prompt):
|
||
print(chunk.content, end="", flush=True)
|
||
output_text += chunk.content
|
||
print()
|
||
if debug:
|
||
print(f"\n[DEBUG][output] Ausgabe beendet.")
|
||
return {"final_message": output_text}
|
||
|
||
# ---------- LangGraph Graph ----------
|
||
graph = StateGraph(state_schema=EmpathyState)
|
||
graph.add_node("sache_node", sache_node)
|
||
graph.add_node("selbst_node", selbst_node)
|
||
graph.add_node("beziehung_node", beziehung_node)
|
||
graph.add_node("appell_node", appell_node)
|
||
graph.add_node("psychoanalysis_node", psychoanalysis_node)
|
||
graph.add_node("output_node", output_node)
|
||
graph.add_edge(START, "sache_node")
|
||
graph.add_edge(START, "selbst_node")
|
||
graph.add_edge(START, "beziehung_node")
|
||
graph.add_edge(START, "appell_node")
|
||
graph.add_edge(START, "psychoanalysis_node")
|
||
graph.add_edge("sache_node", "output_node")
|
||
graph.add_edge("selbst_node", "output_node")
|
||
graph.add_edge("beziehung_node", "output_node")
|
||
graph.add_edge("appell_node", "output_node")
|
||
graph.add_edge("psychoanalysis_node", "output_node")
|
||
graph.add_edge("output_node", END)
|
||
app = graph.compile()
|
||
|
||
# ---------- Ausführung ----------
|
||
async def main():
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--debug", action="store_true", help="Debug-Ausgaben aktivieren")
|
||
args = parser.parse_args()
|
||
user_input = input("Bitte gib deine Nachricht ein: ")
|
||
state = {
|
||
"user_input": user_input,
|
||
"debug": args.debug,
|
||
}
|
||
|
||
# Spinner als Background-Task starten (sofort, vor dem Graph)
|
||
spinner_task = None
|
||
if not args.debug:
|
||
spinner_task = asyncio.create_task(spinner_loop(debug=False))
|
||
|
||
result = await app.ainvoke(state)
|
||
|
||
# Warten, bis spinner sich sauber beendet hat
|
||
if spinner_task:
|
||
await asyncio.sleep(0.4) # kurz warten, damit "done"-Status angezeigt wird
|
||
|
||
#if not args.debug:
|
||
# print("\nAntwort:\n" + result["final_message"])
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
|