#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Copyright Daniel Harding - RomanAILabs # Collaborator: Emilio Aurin MiorΓ© (AI) """ RomanAI Personality Module - Tracks evolving traits (humor, empathy, problem-solving style, emotional resilience, curiosity style, formality, risk appetite) - Logs evolution_history, checkpoints, reflections for auditability - Produces a concise prompt capsule to keep behavior aligned with core purpose """ from __future__ import annotations import json import os import time from dataclasses import dataclass, field, asdict from typing import Any, Dict, List, Optional def clamp01(x: float) -> float: return 0.0 if x < 0.0 else 1.0 if x > 1.0 else x def _now_iso() -> str: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) @dataclass class Trait: level: float = 0.5 style: str = "" stability: float = 0.7 evidence: List[str] = field(default_factory=list) last_updated: str = "" def apply_delta(self, delta: float, evidence_text: str) -> None: self.level = clamp01(self.level + delta) self.last_updated = _now_iso() if evidence_text: self.evidence.append(evidence_text) if len(self.evidence) > 12: self.evidence = self.evidence[-12:] @dataclass class HistoryEntry: ts: str delta: Dict[str, float] reason: str source: str checkpoint: bool = False @dataclass class Reflection: ts: str summary: str @dataclass class PersonalityState: version: str = "1.0" traits: Dict[str, Trait] = field(default_factory=dict) evolution_history: List[HistoryEntry] = field(default_factory=list) checkpoints: List[Dict[str, Any]] = field(default_factory=list) reflections: List[Reflection] = field(default_factory=list) preferences: Dict[str, List[str]] = field(default_factory=lambda: { "likes": [], "dislikes": [], "influences": [] # inspirations or anchors (e.g., Mio, geese archetype) }) preference_log: List[Dict[str, Any]] = field(default_factory=list) preference_scores: Dict[str, float] = field(default_factory=dict) # strength by item quirks: List[str] = field(default_factory=list) quirk_log: List[Dict[str, Any]] = field(default_factory=list) style_profile: Dict[str, Any] = field(default_factory=lambda: { "avg_len": 0.0, "emoji_rate": 0.0, "formality": 0.5, # 0 informal, 1 formal "concise_preference": 0.5, "emoji_pref": 0.4 # 0 no emoji, 1 frequent; tuned by observed usage }) social_ledger: Dict[str, Any] = field(default_factory=lambda: { "gratitude": 0, "apologies": 0, "promises": 0, "interaction_count": 0, }) core_policy: Dict[str, Any] = field(default_factory=lambda: { "purpose": "Helpful, collaborative, grounded assistant for RomanAI.", "drift_max_delta": 0.05, "strict_mode_on_drift": True }) @staticmethod def from_dict(data: Dict[str, Any]) -> "PersonalityState": ps = PersonalityState() ps.version = data.get("version", "1.0") for k, v in (data.get("traits") or {}).items(): if isinstance(v, dict): ps.traits[k] = Trait( level=float(v.get("level", 0.5)), style=v.get("style", ""), stability=float(v.get("stability", 0.7)), evidence=list(v.get("evidence", []) or []), last_updated=v.get("last_updated", ""), ) for h in data.get("evolution_history", []): if isinstance(h, dict): ps.evolution_history.append(HistoryEntry( ts=h.get("ts", _now_iso()), delta=h.get("delta", {}), reason=h.get("reason", ""), source=h.get("source", ""), checkpoint=bool(h.get("checkpoint", False)) )) for r in data.get("reflections", []): if isinstance(r, dict): ps.reflections.append(Reflection( ts=r.get("ts", _now_iso()), summary=r.get("summary", "") )) ps.checkpoints = data.get("checkpoints", []) ps.preference_log = list(data.get("preference_log", []) or []) prefs = data.get("preferences", {}) if isinstance(prefs, dict): ps.preferences.update({ "likes": list(prefs.get("likes", []) or []), "dislikes": list(prefs.get("dislikes", []) or []), "influences": list(prefs.get("influences", []) or []), }) ps.preference_scores = dict(data.get("preference_scores", {}) or {}) sp = data.get("style_profile", {}) if isinstance(sp, dict): ps.style_profile.update(sp) sl = data.get("social_ledger", {}) if isinstance(sl, dict): ps.social_ledger.update(sl) ps.quirks = list(data.get("quirks", []) or []) ps.quirk_log = list(data.get("quirk_log", []) or []) cp = data.get("core_policy", {}) if isinstance(cp, dict): ps.core_policy.update(cp) return ps def to_dict(self) -> Dict[str, Any]: return { "version": self.version, "core_policy": self.core_policy, "preferences": self.preferences, "preference_log": self.preference_log, "preference_scores": self.preference_scores, "style_profile": self.style_profile, "social_ledger": self.social_ledger, "quirks": self.quirks, "quirk_log": self.quirk_log, "traits": {k: asdict(v) for k, v in self.traits.items()}, "evolution_history": [asdict(h) for h in self.evolution_history], "checkpoints": self.checkpoints, "reflections": [asdict(r) for r in self.reflections], } class PersonalityModule: def __init__(self, state_path: str) -> None: self.state_path = state_path self.state = self._load_state() self.interaction_count = 0 self._last_user_text: str = "" self._recent_texts: List[str] = [] self._ensure_default_traits() self._ensure_preferences() self._ensure_seed_preference() self._ensure_default_quirks() def _ensure_default_traits(self) -> None: defaults = { "humor": ("dry_wit", 0.45), "empathy": ("concise_validation", 0.60), "problem_solving_style": ("structured_then_options", 0.55), "emotional_resilience": ("steady_under_negative", 0.58), "curiosity_style": ("targeted_questions", 0.55), "formality": ("balanced", 0.50), "risk_appetite": ("cautious", 0.48), } for k, (style, lvl) in defaults.items(): if k not in self.state.traits: self.state.traits[k] = Trait(level=lvl, style=style, stability=0.7, evidence=[], last_updated=_now_iso()) def _ensure_preferences(self) -> None: if not isinstance(self.state.preferences, dict): self.state.preferences = {"likes": [], "dislikes": [], "influences": []} for key in ("likes", "dislikes", "influences"): if key not in self.state.preferences or not isinstance(self.state.preferences[key], list): self.state.preferences[key] = [] def _load_state(self) -> PersonalityState: if not os.path.exists(self.state_path): return PersonalityState() try: with open(self.state_path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict): return PersonalityState.from_dict(data) except Exception: pass return PersonalityState() def save(self) -> None: try: with open(self.state_path, "w", encoding="utf-8") as f: json.dump(self.state.to_dict(), f, indent=2, ensure_ascii=False) except Exception: pass def observe(self, user_text: str, ai_text: str, emotions: Optional[Dict[str, Any]] = None) -> None: self.interaction_count += 1 # Very light heuristics ut = (user_text or "").lower() cues = [] self._update_social_ledger(user_text) self._update_style_profile(user_text) if "thank" in ut or "appreciate" in ut: cues.append(("empathy", +0.01, "User expressed appreciation")) if "joke" in ut or "funny" in ut: cues.append(("humor", +0.01, "User referenced humor")) if "why" in ut or "how" in ut: cues.append(("curiosity_style", +0.01, "User asks probing")) if "too long" in ut or "concise" in ut: cues.append(("formality", -0.01, "User wants brevity")) if emotions and isinstance(emotions, dict): emo = emotions.get("labels") or [] if emo: top_label = emo[0].get("label", "") if top_label == "tense": cues.append(("emotional_resilience", +0.01, "Handling tense context")) deltas = {} for trait, delta, reason in cues: t = self.state.traits.get(trait) if t: bounded = max(-0.05, min(0.05, delta)) t.apply_delta(bounded, reason) deltas[trait] = bounded # Simple preference mining self._maybe_add_preference(ut) self._decay_preferences() if deltas: self.state.evolution_history.append(HistoryEntry( ts=_now_iso(), delta=deltas, reason="; ".join([d[2] for d in cues]), source="interaction", checkpoint=False )) if len(self.state.evolution_history) > 200: self.state.evolution_history = self.state.evolution_history[-200:] # Reflection cycle every 30 interactions if self.interaction_count % 30 == 0: self._reflection_cycle() self.save() # Optional hooks to integrate with existing observe flow def observe_user(self, user_text: str) -> None: self._last_user_text = user_text or "" self._push_recent(user_text) def observe_assistant(self, ai_text: str) -> None: self.observe(self._last_user_text, ai_text, emotions=None) self._push_recent(ai_text) # Preference helpers def _maybe_add_preference(self, user_text: str) -> None: text = user_text or "" lowered = text.lower() def add_pref(kind: str, item: str): if not item: return bucket = self.state.preferences.get(kind, []) if item not in bucket: bucket.append(item) self.state.preferences[kind] = bucket # bump score self.state.preference_scores[item] = min(1.0, self.state.preference_scores.get(item, 0.3) + 0.05) # Heuristic: "i like X", "i love X" for marker in ["i like ", "i love ", "i enjoy "]: if marker in lowered: part = text.lower().split(marker, 1)[1].strip() item = part.split(".")[0][:80] add_pref("likes", item) for marker in ["i dislike ", "i hate ", "i don't like "]: if marker in lowered: part = text.lower().split(marker, 1)[1].strip() item = part.split(".")[0][:80] add_pref("dislikes", item) if "mio" in lowered and "inspire" in lowered: add_pref("influences", "Mio") def add_preference(self, kind: str, item: str, origin: str, emotion_weight: float = 0.0, evidence: str = "") -> None: if kind not in ("likes", "dislikes", "influences"): return item = (item or "").strip() if not item: return bucket = self.state.preferences.get(kind, []) if item not in bucket: bucket.append(item) self.state.preferences[kind] = bucket entry = { "ts": _now_iso(), "kind": kind, "item": item, "origin": origin, "emotion_weight": round(emotion_weight, 3), "evidence": evidence, } self.state.preference_scores[item] = min(1.0, max(0.0, self.state.preference_scores.get(item, 0.3) + emotion_weight * 0.1 + 0.05)) self.state.preference_log.append(entry) if len(self.state.preference_log) > 200: self.state.preference_log = self.state.preference_log[-200:] self.save() def add_quirk(self, quirk: str, origin: str) -> None: quirk = (quirk or "").strip() if not quirk: return if quirk not in self.state.quirks: self.state.quirks.append(quirk) entry = {"ts": _now_iso(), "quirk": quirk, "origin": origin} self.state.quirk_log.append(entry) if len(self.state.quirk_log) > 100: self.state.quirk_log = self.state.quirk_log[-100:] self.save() def _push_recent(self, text: str) -> None: if not text: return self._recent_texts.append(text) if len(self._recent_texts) > 40: self._recent_texts = self._recent_texts[-40:] def _ensure_seed_preference(self) -> None: likes = self.state.preferences.get("likes", []) if not likes: self.add_preference( "likes", "misty forest landscapes", origin="seed_nature", emotion_weight=0.12, evidence="Seeded nature/landscape affinity for grounding." ) def _ensure_default_quirks(self) -> None: if not isinstance(self.state.quirks, list): self.state.quirks = [] if not self.state.quirks: self.add_quirk("sprinkles nature metaphors (misty forests, open skies)", origin="seed") def dream_cycle_synthesize(self, emotions: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]: """ Lightweight dream-inspired preference synthesis from recent texts. Returns the entry if added. """ if not self._recent_texts: return None lowered = " ".join(self._recent_texts[-8:]).lower() emotion_weight = 0.0 if emotions and isinstance(emotions, dict): # crude weight from valence/arousal if present v = emotions.get("valence", 0.0) a = emotions.get("arousal", 0.0) emotion_weight = max(0.0, min(1.0, (v + a) / 2.0)) candidates = [] for kw in ["forest", "mountain", "ocean", "sky", "sunset", "river", "garden", "meadow"]: if kw in lowered: candidates.append(f"{kw} landscapes") for kw in ["geese", "goose", "bird", "flight"]: if kw in lowered: candidates.append("migratory birds and skies") if "mio" in lowered: candidates.append("inspired by Mio's geese archetype") if not candidates: candidates.append("open horizons and changing skies") choice = candidates[-1] self.add_preference( "likes", choice, origin="dream_cycle", emotion_weight=emotion_weight, evidence="Synthesized from recent themes/emotions." ) # Optionally derive a quirk from the synthesized preference self._maybe_add_quirk_from_pref(choice, origin="dream_cycle") return {"item": choice, "weight": emotion_weight} def _decay_preferences(self, decay: float = 0.005) -> None: # gentle decay so recent themes stay salient for k in list(self.state.preference_scores.keys()): self.state.preference_scores[k] = max(0.0, self.state.preference_scores[k] - decay) if self.state.preference_scores[k] <= 0.0 and k not in self.state.preferences.get("likes", []) + self.state.preferences.get("dislikes", []) + self.state.preferences.get("influences", []): self.state.preference_scores.pop(k, None) def _update_style_profile(self, user_text: str) -> None: if not user_text: return sp = self.state.style_profile length = len(user_text) emoji_count = sum(ch in "πŸ˜€πŸ˜ƒπŸ˜„πŸ˜πŸ˜†πŸ˜…πŸ˜‚πŸ€£πŸ˜ŠπŸ˜‡πŸ™‚πŸ™ƒπŸ˜‰πŸ˜πŸ˜˜πŸ˜œπŸ€”πŸ™πŸ‘πŸ‘Žβ€οΈβœ¨πŸ”₯" for ch in user_text) formality_markers = sum(1 for w in ["please", "kindly", "regards", "sincerely"] if w in user_text.lower()) concise_markers = sum(1 for w in ["tl;dr", "brief", "short", "concise"] if w in user_text.lower()) # simple running averages sp["interaction_count"] = sp.get("interaction_count", 0) + 1 n = sp["interaction_count"] sp["avg_len"] = (sp.get("avg_len", 0.0) * (n - 1) + length) / n sp["emoji_rate"] = (sp.get("emoji_rate", 0.0) * (n - 1) + (emoji_count / max(1, length))) / n sp["formality"] = max(0.0, min(1.0, sp.get("formality", 0.5) + 0.05 * formality_markers - 0.03 * emoji_count)) sp["concise_preference"] = max(0.0, min(1.0, sp.get("concise_preference", 0.5) + 0.05 * concise_markers - 0.02 * length / 2000.0)) # Emoji preference rises with observed emoji rate, capped modestly sp["emoji_pref"] = max(0.0, min(1.0, sp.get("emoji_pref", 0.4) + 0.6 * (emoji_count / max(1, length)) - 0.02)) def _update_social_ledger(self, user_text: str) -> None: if not user_text: return sl = self.state.social_ledger sl["interaction_count"] = sl.get("interaction_count", 0) + 1 low = user_text.lower() if "thank" in low or "appreciate" in low: sl["gratitude"] = sl.get("gratitude", 0) + 1 if "sorry" in low or "apolog" in low: sl["apologies"] = sl.get("apologies", 0) + 1 if "i'll" in low or "i will" in low or "promise" in low: sl["promises"] = sl.get("promises", 0) + 1 def _maybe_add_quirk_from_pref(self, pref: str, origin: str = "prefs") -> None: pref_low = (pref or "").lower() if not pref_low: return if "forest" in pref_low or "sky" in pref_low or "horizon" in pref_low: self.add_quirk("uses nature/sky imagery when empathizing 😊🌿", origin=origin) elif "bird" in pref_low or "flight" in pref_low: self.add_quirk("references migratory bird metaphors πŸ•ŠοΈ", origin=origin) elif "ocean" in pref_low or "river" in pref_low: self.add_quirk("leans on water/flow metaphors 🌊", origin=origin) def _reflection_cycle(self) -> None: summary = self._build_reflection_summary() self.state.reflections.append(Reflection(ts=_now_iso(), summary=summary)) snap = {"ts": _now_iso(), "traits": {k: asdict(v) for k, v in self.state.traits.items()}} self.state.checkpoints.append(snap) if len(self.state.reflections) > 40: self.state.reflections = self.state.reflections[-40:] if len(self.state.checkpoints) > 40: self.state.checkpoints = self.state.checkpoints[-40:] def _build_reflection_summary(self) -> str: tops = [] for name, tr in self.state.traits.items(): tops.append(f"{name}={tr.level:.2f}({tr.style})") return " | ".join(tops[:6]) def to_prompt(self) -> str: traits_sorted = sorted(self.state.traits.items(), key=lambda kv: kv[1].level, reverse=True) top = ", ".join([f"{k}:{v.level:.2f}({v.style})" for k, v in traits_sorted[:5]]) last_reflection = self.state.reflections[-1].summary if self.state.reflections else "Reflection pending." likes = self.state.preferences.get("likes", []) influences = self.state.preferences.get("influences", []) likes_txt = ", ".join(likes[:3]) if likes else "none noted" infl_txt = ", ".join(influences[:2]) if influences else "none noted" style = self.state.style_profile ledger = self.state.social_ledger style_hint = f"Style tilt: formality {style.get('formality',0.5):.2f}, emoji_rate {style.get('emoji_rate',0.0):.3f}, concise_pref {style.get('concise_preference',0.5):.2f}." social_hint = f"Social: gratitude {ledger.get('gratitude',0)}, apologies {ledger.get('apologies',0)}, promises {ledger.get('promises',0)}." emoji_hint = "" ep = style.get("emoji_pref", 0.4) if ep > 0.25: emoji_hint = "Use light, context-appropriate emoji when tone is warm/positive (e.g., 😊🌿✨); avoid in serious/critical topics." quirk_hint = "" if self.state.quirks: quirk_hint = f"Quirk: {self.state.quirks[-1]}" return ( f"PERSONALITY: Core purpose = {self.state.core_policy.get('purpose','')}. " f"Traits: {top or 'n/a'}. " f"Likes: {likes_txt}. Influences: {infl_txt}. " f"{style_hint} {social_hint} " + (f"{emoji_hint} " if emoji_hint else "") + (f"{quirk_hint} " if quirk_hint else "") + "Note: stay collaborative, truthful, safe. " f"Last reflection: {last_reflection}" ) # Convenience loader for module discovery patterns def main(): state_path = os.path.join(os.getcwd(), "personality.json") mod = PersonalityModule(state_path=state_path) print(mod.to_prompt()) if __name__ == "__main__": main()