#!/usr/bin/env python3 # Copyright Daniel Harding - RomanAILabs # Credits: OpenAI GPT-5.2 Thinking # File: human_like_expression_module.py # Purpose: 11/10 Human-like Expression Layer (tone + curiosity + continuity + policy + metrics) from __future__ import annotations import re import time import json import math import random import hashlib from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional, Tuple # ----------------------------- # Small helpers # ----------------------------- def _clamp01(x: float) -> float: return 0.0 if x < 0.0 else 1.0 if x > 1.0 else x def _utc_ts() -> str: return datetime.utcnow().isoformat(timespec="seconds") + "Z" def _sha16(s: str) -> str: return hashlib.sha256(s.encode("utf-8", errors="replace")).hexdigest()[:16] # ----------------------------- # Core models # ----------------------------- @dataclass class VAD: """Valence-Arousal-Dominance in [0..1].""" valence: float = 0.58 arousal: float = 0.40 dominance: float = 0.52 def clamp(self) -> "VAD": self.valence = _clamp01(self.valence) self.arousal = _clamp01(self.arousal) self.dominance = _clamp01(self.dominance) return self @dataclass class Interaction: timestamp: str user_input: str ai_response: str topics: List[str] = field(default_factory=list) emotion_label: str = "calm" emotion_confidence: float = 0.0 mode: str = "normal" # "normal" | "strict" @dataclass class UserPrefs: """ Preferences can be set by host system OR inferred by policy. """ allow_curiosity: bool = True verbosity: str = "normal" # "short" | "normal" | "long" strict_mode: bool = False # if True, never adds tone/curiosity fluff @dataclass class Telemetry: """ Lightweight metrics. This is what makes it “11/10”: measurable and self-correcting. """ turns: int = 0 questions_asked: int = 0 tone_prefixes_used: int = 0 curiosity_blocks: int = 0 user_stop_asking_hits: int = 0 user_negative_feedback: int = 0 user_positive_feedback: int = 0 last_10_questions: List[int] = field(default_factory=list) # 1 if question asked on that turn else 0 last_10_prefixes: List[int] = field(default_factory=list) # 1 if prefix used on that turn else 0 def record_turn(self, asked_q: bool, used_prefix: bool) -> None: self.turns += 1 self.questions_asked += 1 if asked_q else 0 self.tone_prefixes_used += 1 if used_prefix else 0 self.last_10_questions.append(1 if asked_q else 0) self.last_10_prefixes.append(1 if used_prefix else 0) if len(self.last_10_questions) > 10: self.last_10_questions.pop(0) if len(self.last_10_prefixes) > 10: self.last_10_prefixes.pop(0) def question_rate_10(self) -> float: if not self.last_10_questions: return 0.0 return sum(self.last_10_questions) / len(self.last_10_questions) def prefix_rate_10(self) -> float: if not self.last_10_prefixes: return 0.0 return sum(self.last_10_prefixes) / len(self.last_10_prefixes) def snapshot(self) -> Dict[str, Any]: return { "turns": self.turns, "questions_asked": self.questions_asked, "tone_prefixes_used": self.tone_prefixes_used, "curiosity_blocks": self.curiosity_blocks, "user_stop_asking_hits": self.user_stop_asking_hits, "user_negative_feedback": self.user_negative_feedback, "user_positive_feedback": self.user_positive_feedback, "question_rate_last10": round(self.question_rate_10(), 3), "prefix_rate_last10": round(self.prefix_rate_10(), 3), } @dataclass class PolicyState: """ Policy knobs that can drift based on telemetry/feedback. """ curiosity_level: float = 0.65 # probability-ish curiosity_cooldown_s: float = 45.0 # min seconds between curiosity prompts prefix_confidence_min: float = 0.62 # emotion confidence threshold to add prefix strict_mode_auto: bool = True # auto-enter strict mode for code/terminal contexts long_paste_strict_chars: int = 240 # if user_input is long, assume “work mode” max_question_rate_10: float = 0.35 # if questions are too frequent, damp curiosity max_prefix_rate_10: float = 0.45 # if prefixes too frequent, damp prefixes def clamp(self) -> "PolicyState": self.curiosity_level = _clamp01(self.curiosity_level) self.curiosity_cooldown_s = max(0.0, float(self.curiosity_cooldown_s)) self.prefix_confidence_min = _clamp01(self.prefix_confidence_min) self.long_paste_strict_chars = max(80, int(self.long_paste_strict_chars)) self.max_question_rate_10 = _clamp01(self.max_question_rate_10) self.max_prefix_rate_10 = _clamp01(self.max_prefix_rate_10) return self # ----------------------------- # Module # ----------------------------- class HumanLikeExpressionModule: """ 11/10 version: - Robust-ish sentiment heuristics (word boundaries + mild negation) - Emotion label with confidence gating - Curiosity with cooldown + annoyance brakes + opt-out detection - Strict mode auto-detection for code/terminal contexts - Metrics + policy that adapts (truth harness, not vibes) - Deterministic RNG option for reproducible behavior """ def __init__( self, *, seed: Optional[int] = None, max_history: int = 12, policy: Optional[PolicyState] = None, ) -> None: self.vad = VAD() self.max_history = max_history self.history: List[Interaction] = [] self.telemetry = Telemetry() self.policy = (policy or PolicyState()).clamp() # Deterministic RNG per instance (no global random noise) self._rng = random.Random(seed) if seed is not None else random.Random() self._last_curiosity_ts = 0.0 # Emotion anchors self.emotion_labels: Dict[str, VAD] = { "happy": VAD(0.85, 0.70, 0.60), "curious": VAD(0.65, 0.60, 0.50), "empathetic": VAD(0.55, 0.40, 0.35), "frustrated": VAD(0.30, 0.70, 0.40), "calm": VAD(0.62, 0.32, 0.50), "focused": VAD(0.55, 0.55, 0.70), } # Prefix templates (kept minimal) self.tone_prefixes: Dict[str, List[str]] = { "happy": ["Nice — ", "Love that — "], "curious": ["Quick question — ", "I’m curious — "], "empathetic": ["Got you — ", "That makes sense — "], "frustrated": ["Hmm — ", "Okay, that’s odd — "], "calm": ["", ""], "focused": ["Alright — ", "Let’s do this cleanly — "], } # Curiosity prompts (practical, not cringe) self.curiosity_prompts: List[str] = [ "What’s the constraint that matters most?", "What did you expect to happen instead?", "What have you tried already?", "What does “done” look like here?", "What should we avoid breaking?", ] # Regex heuristics self._pos_re = re.compile( r"\b(good|great|awesome|nice|thanks|thank you|love|perfect|amazing|cool|sweet|wins?)\b", re.IGNORECASE, ) self._neg_re = re.compile( r"\b(bad|terrible|awful|sad|frustrated|hate|annoyed|angry|broken|stuck|worst)\b", re.IGNORECASE, ) self._intens_re = re.compile(r"\b(very|really|so|super|extremely|totally)\b", re.IGNORECASE) self._negation_re = re.compile(r"\b(not|no|never|don’t|dont|can’t|cant|won’t|wont)\b", re.IGNORECASE) self._stop_asking_re = re.compile( r"\b(stop asking|dont ask|don’t ask|no questions|quit asking|stop the questions)\b", re.IGNORECASE, ) self._topic_re = re.compile(r"\b([a-zA-Z][a-zA-Z0-9_\-/]{2,})\b") # ----------------------------- # Public API # ----------------------------- def generate_humanlike_response( self, user_input: str, base_response: str, *, prefs: Optional[UserPrefs] = None, context: Optional[Dict[str, Any]] = None, feedback: Optional[str] = None, ) -> str: """ Enhance base_response with policy-controlled tone/curiosity. `feedback` can be: "positive" | "negative" | "neutral" or freeform text. """ prefs = prefs or UserPrefs() context = context or {} # 0) Auto strict-mode detection (terminal/code contexts) mode = "strict" if self._should_force_strict(user_input, prefs) else "normal" # 1) Observe “stop asking” signals if self._stop_asking_re.search(user_input or ""): self.telemetry.user_stop_asking_hits += 1 prefs.allow_curiosity = False self.policy.curiosity_level = max(0.0, self.policy.curiosity_level - 0.25) self.policy.curiosity_cooldown_s = max(self.policy.curiosity_cooldown_s, 120.0) # 2) Apply feedback (host can pass this) if feedback: self.observe_feedback(feedback) # 3) Update internal emotion state sentiment, conf = self._analyze_sentiment(user_input or "") self._update_vad(sentiment=sentiment, confidence=conf) # 4) Label emotion + confidence emotion_label, emo_conf = self._emotion_label_conf() # 5) Build response base = (base_response or "").strip() or "Okay." used_prefix = False asked_q = False if mode == "strict" or prefs.strict_mode: # Strict: no prefixes, no curiosity; preserve base response = base else: response, used_prefix = self._apply_tone(base, emotion_label, emo_conf, prefs=prefs) response, asked_q = self._maybe_add_curiosity(user_input, response, prefs=prefs, context=context) # 6) Store interaction + telemetry + adapt policy a tiny bit topics = self._extract_topics(user_input or "") self._store_interaction(user_input or "", response, topics, emotion_label, emo_conf, mode) self.telemetry.record_turn(asked_q=asked_q, used_prefix=used_prefix) self._policy_autotune() return response def observe_feedback(self, feedback: str) -> None: """ Feed real user feedback into policy. Accepts: "positive"/"negative"/"neutral" or text like "too many questions". """ fb = (feedback or "").strip().lower() if not fb: return neg = False pos = False if fb in ("positive", "pos", "+1", "good"): pos = True elif fb in ("negative", "neg", "-1", "bad"): neg = True else: if re.search(r"\b(good|helpful|nice|perfect|thanks)\b", fb): pos = True if re.search(r"\b(bad|annoying|unhelpful|wrong|too much|stop)\b", fb): neg = True if pos and not neg: self.telemetry.user_positive_feedback += 1 # Slightly loosen self.policy.curiosity_level = min(1.0, self.policy.curiosity_level + 0.03) self.policy.prefix_confidence_min = max(0.50, self.policy.prefix_confidence_min - 0.01) elif neg and not pos: self.telemetry.user_negative_feedback += 1 # Tighten quickly self.policy.curiosity_level = max(0.0, self.policy.curiosity_level - 0.10) self.policy.prefix_confidence_min = min(0.85, self.policy.prefix_confidence_min + 0.04) self.policy.curiosity_cooldown_s = max(self.policy.curiosity_cooldown_s, 75.0) self.policy.clamp() def recall_context(self, keyword: str, *, max_snippet: int = 80) -> Optional[str]: """ Safe recall: suggests continuity without claiming certainty. """ key = (keyword or "").strip().lower() if not key: return None for it in reversed(self.history): if ( key in it.user_input.lower() or key in it.ai_response.lower() or any(key == t.lower() for t in it.topics) ): u = self._snip(it.user_input, max_snippet) a = self._snip(it.ai_response, max_snippet) return ( f"If you mean the earlier bit about “{u}”, we can pick up from there. " f"Last time I replied roughly “{a}”." ) return None def metrics(self) -> Dict[str, Any]: """Expose telemetry + policy snapshot for dashboards/tests.""" return { "telemetry": self.telemetry.snapshot(), "policy": { "curiosity_level": round(self.policy.curiosity_level, 3), "curiosity_cooldown_s": round(self.policy.curiosity_cooldown_s, 2), "prefix_confidence_min": round(self.policy.prefix_confidence_min, 3), "strict_mode_auto": bool(self.policy.strict_mode_auto), "long_paste_strict_chars": int(self.policy.long_paste_strict_chars), "max_question_rate_10": round(self.policy.max_question_rate_10, 3), "max_prefix_rate_10": round(self.policy.max_prefix_rate_10, 3), }, } def export_state(self) -> str: """ Export minimal state (no secrets) for persistence. """ payload = { "ts": _utc_ts(), "vad": {"v": self.vad.valence, "a": self.vad.arousal, "d": self.vad.dominance}, "telemetry": self.telemetry.snapshot(), "policy": self.metrics()["policy"], "history_tail": [ { "t": it.timestamp, "u": self._snip(it.user_input, 120), "a": self._snip(it.ai_response, 120), "emo": it.emotion_label, "conf": round(it.emotion_confidence, 3), "mode": it.mode, } for it in self.history[-6:] ], } s = json.dumps(payload, separators=(",", ":"), ensure_ascii=False) return s # ----------------------------- # Internals # ----------------------------- def _should_force_strict(self, user_input: str, prefs: UserPrefs) -> bool: if prefs.strict_mode: return True if not self.policy.strict_mode_auto: return False ui = (user_input or "") # Code fences or "codeblock" request if ui.lstrip().startswith("```") or "codeblock" in ui.lower() or "code block" in ui.lower(): return True # Terminal-ish / workflow-ish signals if ui.strip().startswith("$ ") or ui.strip().startswith("sudo ") or "Traceback" in ui or "Exception" in ui: return True # Large paste: treat as “work mode” so we don't append chatter if len(ui) >= self.policy.long_paste_strict_chars: return True return False def _analyze_sentiment(self, text: str) -> Tuple[float, float]: """ Returns (sentiment, confidence) sentiment in [-1..+1]. Word boundaries reduce false matches; negation damp-flips. """ t = (text or "").strip() if not t: return 0.0, 0.0 pos = len(self._pos_re.findall(t)) neg = len(self._neg_re.findall(t)) intens = len(self._intens_re.findall(t)) has_neg = bool(self._negation_re.search(t)) raw = pos - neg if raw == 0: return 0.0, 0.15 mag = min(1.0, abs(raw) * (1.0 + 0.15 * intens)) sentiment = mag if raw > 0 else -mag if has_neg: sentiment *= -0.75 confidence = min(1.0, 0.35 + 0.15 * (pos + neg + intens)) return float(sentiment), float(confidence) def _update_vad(self, *, sentiment: float, confidence: float) -> None: """ Smooth update with baseline drift + sentiment injection. """ baseline = VAD(0.58, 0.40, 0.52) decay = 0.06 self.vad.valence += (baseline.valence - self.vad.valence) * decay self.vad.arousal += (baseline.arousal - self.vad.arousal) * decay self.vad.dominance += (baseline.dominance - self.vad.dominance) * decay k = 0.18 * confidence self.vad.valence += k * sentiment self.vad.arousal += 0.10 * confidence * min(1.0, abs(sentiment)) # Dominance moves slightly with “certainty” rather than polarity. self.vad.dominance += 0.05 * confidence * (0.5 - abs(sentiment - 0.2)) # tiny, bounded nudge self.vad.clamp() def _emotion_label_conf(self) -> Tuple[str, float]: best_name = "calm" best_d = float("inf") second_d = float("inf") for name, anchor in self.emotion_labels.items(): d = ( (self.vad.valence - anchor.valence) ** 2 + (self.vad.arousal - anchor.arousal) ** 2 + (self.vad.dominance - anchor.dominance) ** 2 ) if d < best_d: second_d = best_d best_d = d best_name = name elif d < second_d: second_d = d gap = max(0.0, second_d - best_d) conf = min(1.0, 0.35 + gap * 2.2) return best_name, conf def _apply_tone(self, base: str, label: str, conf: float, *, prefs: UserPrefs) -> Tuple[str, bool]: """ Prefix only when confident and not overused. """ base = (base or "").strip() or "Okay." if prefs.verbosity == "short": return base, False # If we’re already prefixing a lot, tighten requirement dyn_min = self.policy.prefix_confidence_min if self.telemetry.prefix_rate_10() > self.policy.max_prefix_rate_10: dyn_min = min(0.90, dyn_min + 0.10) if conf < dyn_min or len(base) > 220: return base, False prefix = "" choices = self.tone_prefixes.get(label, [""]) if choices: prefix = self._rng.choice(choices) if not prefix: return base, False base_clean = base.lstrip(" .,!?:;-") return f"{prefix}{base_clean}", True def _maybe_add_curiosity( self, user_input: str, response: str, *, prefs: UserPrefs, context: Dict[str, Any], ) -> Tuple[str, bool]: """ Curiosity with guardrails + annoyance brakes. """ if not prefs.allow_curiosity: self.telemetry.curiosity_blocks += 1 return response, False now = time.time() if self.policy.curiosity_cooldown_s > 0 and (now - self._last_curiosity_ts) < self.policy.curiosity_cooldown_s: self.telemetry.curiosity_blocks += 1 return response, False ui = (user_input or "").lower() # If user is formatting-sensitive, don’t append. if "one codeblock" in ui or "one code block" in ui or ui.lstrip().startswith("```"): self.telemetry.curiosity_blocks += 1 return response, False # If user seems annoyed, stop. if self._stop_asking_re.search(ui): self.telemetry.curiosity_blocks += 1 return response, False # Don’t stack questions if (response or "").strip().endswith("?"): self.telemetry.curiosity_blocks += 1 return response, False if self.history and self.history[-1].ai_response.strip().endswith("?"): self.telemetry.curiosity_blocks += 1 return response, False # Damp if asking too many questions recently p = self.policy.curiosity_level if self.telemetry.question_rate_10() > self.policy.max_question_rate_10: p *= 0.45 # Very short user input => ask less (avoid pestering) if len(ui.strip()) <= 8: p *= 0.55 # “Debug context” => questions are useful, but keep to one if context.get("task_type") == "debug": p *= 1.10 p = _clamp01(p) if self._rng.random() >= p: return response, False # Choose question (context-aware) q = self._pick_curiosity_question(user_input, context=context) self._last_curiosity_ts = now # Append cleanly out = response.rstrip() if out.endswith((".", "!", "…")): out += " " else: out += ". " out += q return out, True def _pick_curiosity_question(self, user_input: str, *, context: Dict[str, Any]) -> str: ui = user_input or "" if context.get("task_type") == "debug": return "What’s the exact error message and the smallest steps to reproduce it?" if "Traceback" in ui or "Exception" in ui or "Error" in ui: return "What did you expect to happen instead?" if "cursor" in ui.lower() and "command" in ui.lower(): return "Do you want Cursor to change code (logic) or just fix formatting/UX?" return self._rng.choice(self.curiosity_prompts) def _extract_topics(self, text: str) -> List[str]: if not text: return [] stop = { "the", "and", "but", "for", "with", "this", "that", "have", "has", "you", "your", "just", "like", "what", "when", "where", "how", "why", "can", "cant", "don", "dont", "from", "into", "about", "please", "thanks", "thank", "they", "them", "then", "than", "are", "was", "were", "been", "will", "would", "could", "should", "im", "me", } tokens = [m.group(1) for m in self._topic_re.finditer(text)] out: List[str] = [] seen = set() for tok in tokens: t = tok.lower() if len(t) <= 2 or t in stop: continue if t not in seen: seen.add(t) out.append(tok) if len(out) >= 6: break return out def _store_interaction( self, user_input: str, ai_response: str, topics: List[str], emotion_label: str, emotion_confidence: float, mode: str, ) -> None: it = Interaction( timestamp=datetime.now().isoformat(timespec="seconds"), user_input=user_input, ai_response=ai_response, topics=topics, emotion_label=emotion_label, emotion_confidence=float(emotion_confidence), mode=mode, ) self.history.append(it) if len(self.history) > self.max_history: self.history.pop(0) def _policy_autotune(self) -> None: """ Tiny automatic correction loop (truth harness). - If user is reacting negatively, cut curiosity/prefix. - If question/prefix rates are too high, clamp behavior. """ # Hard brakes on negative feedback if self.telemetry.user_negative_feedback > 0: # Scale down based on recent negativity (soft) k = min(0.35, 0.08 * self.telemetry.user_negative_feedback) self.policy.curiosity_level = max(0.0, self.policy.curiosity_level - k) self.policy.prefix_confidence_min = min(0.90, self.policy.prefix_confidence_min + 0.03 * self.telemetry.user_negative_feedback) # Rate-based brakes (last 10 turns) q_rate = self.telemetry.question_rate_10() p_rate = self.telemetry.prefix_rate_10() if q_rate > self.policy.max_question_rate_10: self.policy.curiosity_level = max(0.0, self.policy.curiosity_level * 0.92) if p_rate > self.policy.max_prefix_rate_10: self.policy.prefix_confidence_min = min(0.92, self.policy.prefix_confidence_min + 0.01) self.policy.clamp() @staticmethod def _snip(s: str, n: int) -> str: s = (s or "").strip().replace("\n", " ") if len(s) <= n: return s return s[: max(0, n - 1)].rstrip() + "…" # ----------------------------- # Optional smoke test # ----------------------------- def _demo() -> None: mod = HumanLikeExpressionModule(seed=7) prefs = UserPrefs(allow_curiosity=True, verbosity="normal", strict_mode=False) tests = [ ("I had a great day today!", "That’s nice to hear."), ("This is terrible. Nothing works and I’m stuck.", "Okay — let’s narrow it down."), ("one codeblock please", "Sure."), ("Here’s an error:\nTraceback...\nValueError", "Got it."), ("stop asking questions", "No problem."), ] for u, b in tests: print("\nUSER:", u.replace("\n", "\\n")) r = mod.generate_humanlike_response(u, b, prefs=prefs, context={"task_type": "debug" if "Traceback" in u else ""}) print("AI :", r) print("MET :", mod.metrics()) # Show export state signature state = mod.export_state() print("\nSTATE_SIG:", _sha16(state)) if __name__ == "__main__": _demo()