111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
"""Local operational signals for a conversation. No extra LLM call, not a Writing Profile."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
EMOTION = re.compile(
|
|
r"aufgeregt|traurig|berührt|ängst|wütend|freude|ruhig|unsicher|gefühl|tiefer",
|
|
re.I,
|
|
)
|
|
PLAN = re.compile(r"geplant|wollen|werde|sollten|vorhaben", re.I)
|
|
CHRONICLE = re.compile(
|
|
r"\b(heute|gestern|morgens|mittags|abends|danach|vorher|später|zuerst|einkauf|"
|
|
r"spaziergang|markt|chronik)\b",
|
|
re.I,
|
|
)
|
|
CLOCK = re.compile(r"\b\d{1,2}:\d{2}\b|\b\d{1,2}\s*Uhr\b", re.I)
|
|
STOP = {
|
|
"dann", "noch", "schon", "wieder", "gegen", "waren", "wurde", "haben", "hatte",
|
|
"unter", "über", "nach", "beim", "eine", "einem", "einer", "dieser", "dieses",
|
|
"auch", "aber", "dass", "wenn", "sich", "uns", "euch", "mein", "dein", "sein",
|
|
"heute", "dann", "nach", "noch",
|
|
}
|
|
|
|
DEEP_OPS = {"erleben_vertiefen", "bedeutung"}
|
|
PLAN_OPS = {"plan_aufgreifen"}
|
|
|
|
|
|
def _words(text: str) -> list[str]:
|
|
return [
|
|
word
|
|
for word in re.findall(r"[a-zäöüß]{4,}", (text or "").lower())
|
|
if word not in STOP
|
|
]
|
|
|
|
|
|
def infer_signals(user_bodies: list[str], last_operation: str | None = None) -> dict:
|
|
"""Derive operational dialogue state. Not a Writing Profile or Interaction Preference."""
|
|
bodies = [item.strip() for item in user_bodies if (item or "").strip()]
|
|
blob = " ".join(bodies)
|
|
last = bodies[-1] if bodies else ""
|
|
op = (last_operation or "").strip().lower()
|
|
emotion = bool(EMOTION.search(blob))
|
|
plan = bool(PLAN.search(blob))
|
|
chronicle = bool(CHRONICLE.search(blob) or CLOCK.search(blob))
|
|
if emotion:
|
|
mode, depth = "reflective", "deep"
|
|
elif plan:
|
|
mode, depth = "plan", "surface"
|
|
elif chronicle:
|
|
mode, depth = "chronicle", "surface"
|
|
elif op in DEEP_OPS:
|
|
mode, depth = "reflective", "deep"
|
|
elif op in PLAN_OPS:
|
|
mode, depth = "plan", "surface"
|
|
else:
|
|
mode, depth = "open", "surface"
|
|
counts: dict[str, int] = {}
|
|
for word in _words(blob):
|
|
counts[word] = counts.get(word, 0) + 1
|
|
focus = " ".join(
|
|
item[0] for item in sorted(counts.items(), key=lambda pair: (-pair[1], pair[0]))[:8]
|
|
)
|
|
last_words = last.split()
|
|
long_story = len(last_words) >= 40 or last.count(".") + last.count("!") >= 3
|
|
if emotion and (long_story or depth == "deep"):
|
|
intensity = "high"
|
|
elif emotion:
|
|
intensity = "medium"
|
|
else:
|
|
intensity = "low"
|
|
return {
|
|
"narrative_mode": mode,
|
|
"reflection_depth": depth,
|
|
"current_focus": focus,
|
|
"emotional_intensity": intensity,
|
|
"long_story": 1 if long_story else 0,
|
|
}
|
|
|
|
|
|
def _focus_words(row: dict) -> set[str]:
|
|
return {item for item in str(row.get("current_focus") or "").split() if item}
|
|
|
|
|
|
def compatible(left: dict, right: dict) -> bool:
|
|
left_mode = (left.get("narrative_mode") or "open").strip() or "open"
|
|
right_mode = (right.get("narrative_mode") or "open").strip() or "open"
|
|
if left_mode == "open" or right_mode == "open":
|
|
return False
|
|
if left_mode != right_mode:
|
|
return False
|
|
left_depth = (left.get("reflection_depth") or "surface").strip() or "surface"
|
|
right_depth = (right.get("reflection_depth") or "surface").strip() or "surface"
|
|
if left_depth != right_depth:
|
|
return False
|
|
left_focus = _focus_words(left)
|
|
right_focus = _focus_words(right)
|
|
if len(left_focus) >= 2 and len(right_focus) >= 2 and not (left_focus & right_focus):
|
|
return False
|
|
return True
|
|
|
|
|
|
def similar_enough(conversations: list[dict]) -> bool:
|
|
usable = [item for item in conversations if item]
|
|
if len(usable) < 2:
|
|
return False
|
|
for index, left in enumerate(usable):
|
|
for right in usable[index + 1 :]:
|
|
if not compatible(left, right):
|
|
return False
|
|
return True
|