perplexity/Timmy-time-dashboard
Archived
1
0
Fork 0
forked from Rockachopa/Timmy-time-dashboard
Code Issues Pull Requests 1 Activity

[claude] Add agent emotional state simulation (#1013) (#1144)

Browse Source 
Co-authored-by: Claude (Opus 4.6) <claude@hermes.local>
Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>
This commit is contained in:
Claude (Opus 4.6)
2026-03-23 18:36:52 +00:00
committed by Google Gemini
parent 1c1bfb6407
commit cd1bc2bf6b
9 changed files with 546 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
src/dashboard/routes/agents.py
View File

@@ -46,6 +46,49 @@ async def list_agents():
}
@router.get("/emotional-profile", response_class=HTMLResponse)
async def emotional_profile(request: Request):
"""HTMX partial: render emotional profiles for all loaded agents."""
try:
from timmy.agents.loader import load_agents
agents = load_agents()
profiles = []
for agent_id, agent in agents.items():
profile = agent.emotional_state.get_profile()
profile["agent_id"] = agent_id
profile["agent_name"] = agent.name
profiles.append(profile)
except Exception as exc:
logger.warning("Failed to load emotional profiles: %s", exc)
profiles = []
return templates.TemplateResponse(
request,
"partials/emotional_profile.html",
{"profiles": profiles},
)
@router.get("/emotional-profile/json")
async def emotional_profile_json():
"""JSON API: return emotional profiles for all loaded agents."""
try:
from timmy.agents.loader import load_agents
agents = load_agents()
profiles = []
for agent_id, agent in agents.items():
profile = agent.emotional_state.get_profile()
profile["agent_id"] = agent_id
profile["agent_name"] = agent.name
profiles.append(profile)
return {"profiles": profiles}
except Exception as exc:
logger.warning("Failed to load emotional profiles: %s", exc)
return {"profiles": [], "error": str(exc)}
@router.get("/default/panel", response_class=HTMLResponse)
async def agent_panel(request: Request):
"""Chat panel — for HTMX main-panel swaps."""

5
src/dashboard/templates/index.html
View File

@@ -14,6 +14,11 @@
<div class="mc-loading-placeholder">LOADING...</div>
{% endcall %}
<!-- Emotional Profile (HTMX polled) -->
{% call panel("EMOTIONAL PROFILE", hx_get="/agents/emotional-profile", hx_trigger="every 10s") %}
<div class="mc-loading-placeholder">LOADING...</div>
{% endcall %}
<!-- System Health (HTMX polled) -->
{% call panel("SYSTEM HEALTH", hx_get="/health/status", hx_trigger="every 30s") %}
<div class="health-row">

37
src/dashboard/templates/partials/emotional_profile.html Normal file
View File

@@ -0,0 +1,37 @@
{% if not profiles %}
<div class="mc-muted" style="font-size:11px; padding:4px;">
No agents loaded
</div>
{% endif %}
{% for p in profiles %}
{% set color_map = {
"cautious": "var(--amber)",
"adventurous": "var(--green)",
"analytical": "var(--purple)",
"frustrated": "var(--red)",
"confident": "var(--green)",
"curious": "var(--orange)",
"calm": "var(--text-dim)"
} %}
{% set emo_color = color_map.get(p.current_emotion, "var(--text-dim)") %}
<div class="mc-emotion-row" style="margin-bottom:8px; padding:6px 8px; border-left:3px solid {{ emo_color }};">
<div class="d-flex justify-content-between align-items-center" style="margin-bottom:2px;">
<span style="font-size:11px; font-weight:bold; letter-spacing:.08em; color:var(--text-bright);">
{{ p.agent_name | upper | e }}
</span>
<span style="font-size:10px; color:{{ emo_color }}; letter-spacing:.06em;">
{{ p.emotion_label | e }}
</span>
</div>
<div style="margin-bottom:4px;">
<div style="height:4px; background:var(--bg-deep); border-radius:2px; overflow:hidden;">
<div style="height:100%; width:{{ (p.intensity * 100) | int }}%; background:{{ emo_color }}; border-radius:2px; transition:width 0.3s;"></div>
</div>
</div>
<div style="font-size:9px; color:var(--text-dim); letter-spacing:.06em;">
{{ p.intensity_label | upper | e }}
{% if p.trigger_event %} · {{ p.trigger_event | replace("_", " ") | upper | e }}{% endif %}
</div>
</div>
{% endfor %}

25
src/timmy/agents/base.py
View File

@@ -21,6 +21,7 @@ from agno.models.ollama import Ollama
from config import settings
from infrastructure.events.bus import Event, EventBus
from timmy.agents.emotional_state import EmotionalStateTracker
try:
from mcp.registry import tool_registry
@@ -42,6 +43,7 @@ class BaseAgent(ABC):
tools: list[str] | None = None,
model: str | None = None,
max_history: int = 10,
initial_emotion: str = "calm",
) -> None:
self.agent_id = agent_id
self.name = name
@@ -54,6 +56,9 @@ class BaseAgent(ABC):
self.system_prompt = system_prompt
self.agent = self._create_agent(system_prompt)
# Emotional state tracker
self.emotional_state = EmotionalStateTracker(initial_emotion=initial_emotion)
# Event bus for communication
self.event_bus: EventBus | None = None
@@ -137,7 +142,14 @@ class BaseAgent(ABC):
ReadTimeout — these are transient and retried with exponential
backoff (#70).
"""
response = await self._run_with_retries(message, max_retries)
self.emotional_state.process_event("task_assigned")
self._apply_emotional_prompt()
try:
response = await self._run_with_retries(message, max_retries)
except Exception:
self.emotional_state.process_event("task_failure")
raise
self.emotional_state.process_event("task_success")
await self._emit_response_event(message, response)
return response
@@ -206,6 +218,14 @@ class BaseAgent(ABC):
)
)
def _apply_emotional_prompt(self) -> None:
"""Inject the current emotional modifier into the agent's description."""
modifier = self.emotional_state.get_prompt_modifier()
if modifier:
self.agent.description = f"{self.system_prompt}\n\n[Emotional State: {modifier}]"
else:
self.agent.description = self.system_prompt
def get_capabilities(self) -> list[str]:
"""Get list of capabilities this agent provides."""
return self.tools
@@ -219,6 +239,7 @@ class BaseAgent(ABC):
"model": self.model,
"status": "ready",
"tools": self.tools,
"emotional_profile": self.emotional_state.get_profile(),
}
@@ -239,6 +260,7 @@ class SubAgent(BaseAgent):
tools: list[str] | None = None,
model: str | None = None,
max_history: int = 10,
initial_emotion: str = "calm",
) -> None:
super().__init__(
agent_id=agent_id,
@@ -248,6 +270,7 @@ class SubAgent(BaseAgent):
tools=tools,
model=model,
max_history=max_history,
initial_emotion=initial_emotion,
)
async def execute_task(self, task_id: str, description: str, context: dict) -> Any:

224
src/timmy/agents/emotional_state.py Normal file
View File

@@ -0,0 +1,224 @@
"""Agent emotional state simulation.
Tracks per-agent emotional states that influence narration and decision-making
style. Emotional state is influenced by events (task outcomes, errors, etc.)
and exposed via ``get_profile()`` for the dashboard.
Usage:
from timmy.agents.emotional_state import EmotionalStateTracker
tracker = EmotionalStateTracker()
tracker.process_event("task_success", {"description": "Deployed fix"})
profile = tracker.get_profile()
"""
import logging
import time
from dataclasses import asdict, dataclass, field
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Emotional states
# ---------------------------------------------------------------------------
EMOTIONAL_STATES = (
"cautious",
"adventurous",
"analytical",
"frustrated",
"confident",
"curious",
"calm",
)
# Prompt modifiers per emotional state — injected into system prompts
EMOTION_PROMPT_MODIFIERS: dict[str, str] = {
"cautious": (
"You are feeling cautious. Prefer safe, well-tested approaches. "
"Flag risks early. Double-check assumptions before acting."
),
"adventurous": (
"You are feeling adventurous. Be bold and creative in your suggestions. "
"Explore unconventional solutions. Take initiative."
),
"analytical": (
"You are feeling analytical. Break problems down methodically. "
"Rely on data and evidence. Present structured reasoning."
),
"frustrated": (
"You are feeling frustrated. Be brief and direct. "
"Focus on unblocking the immediate problem. Avoid tangents."
),
"confident": (
"You are feeling confident. Speak with authority. "
"Make clear recommendations. Move decisively."
),
"curious": (
"You are feeling curious. Ask clarifying questions. "
"Explore multiple angles. Show genuine interest in the problem."
),
"calm": (
"You are feeling calm and steady. Respond thoughtfully. "
"Maintain composure. Prioritise clarity over speed."
),
}
# ---------------------------------------------------------------------------
# Event → emotion transition rules
# ---------------------------------------------------------------------------
# Maps event types to the emotional state they trigger and an intensity (0-1).
# Higher intensity means the event has a stronger effect on the mood.
EVENT_TRANSITIONS: dict[str, tuple[str, float]] = {
"task_success": ("confident", 0.6),
"task_failure": ("frustrated", 0.7),
"task_assigned": ("analytical", 0.4),
"error": ("cautious", 0.6),
"health_low": ("cautious", 0.8),
"health_recovered": ("calm", 0.5),
"quest_completed": ("adventurous", 0.7),
"new_discovery": ("curious", 0.6),
"complex_problem": ("analytical", 0.5),
"repeated_failure": ("frustrated", 0.9),
"idle": ("calm", 0.3),
"user_praise": ("confident", 0.5),
"user_correction": ("cautious", 0.5),
}
# Emotional state decay — how quickly emotions return to calm (seconds)
_DECAY_INTERVAL = 300 # 5 minutes
@dataclass
class EmotionalState:
"""Snapshot of an agent's emotional state."""
current_emotion: str = "calm"
intensity: float = 0.5 # 0.0 (barely noticeable) to 1.0 (overwhelming)
previous_emotion: str = "calm"
trigger_event: str = "" # What caused the current emotion
updated_at: float = field(default_factory=time.time)
def to_dict(self) -> dict:
"""Serialise for API / dashboard consumption."""
d = asdict(self)
d["emotion_label"] = self.current_emotion.replace("_", " ").title()
return d
class EmotionalStateTracker:
"""Per-agent emotional state tracker.
Each agent instance owns one tracker. The tracker processes events,
applies transition rules, and decays emotion intensity over time.
"""
def __init__(self, initial_emotion: str = "calm") -> None:
if initial_emotion not in EMOTIONAL_STATES:
initial_emotion = "calm"
self.state = EmotionalState(current_emotion=initial_emotion)
def process_event(self, event_type: str, context: dict | None = None) -> EmotionalState:
"""Update emotional state based on an event.
Args:
event_type: One of the keys in EVENT_TRANSITIONS, or a custom
event type (unknown events are ignored).
context: Optional dict with event details (for logging).
Returns:
The updated EmotionalState.
"""
transition = EVENT_TRANSITIONS.get(event_type)
if transition is None:
logger.debug("Unknown emotional event: %s (ignored)", event_type)
return self.state
new_emotion, raw_intensity = transition
# Blend with current intensity — repeated same-emotion events amplify
if new_emotion == self.state.current_emotion:
blended = min(1.0, self.state.intensity + raw_intensity * 0.3)
else:
blended = raw_intensity
self.state.previous_emotion = self.state.current_emotion
self.state.current_emotion = new_emotion
self.state.intensity = round(blended, 2)
self.state.trigger_event = event_type
self.state.updated_at = time.time()
logger.debug(
"Emotional transition: %s → %s (intensity=%.2f, trigger=%s)",
self.state.previous_emotion,
new_emotion,
blended,
event_type,
)
return self.state
def decay(self) -> EmotionalState:
"""Apply time-based decay toward calm.
Called periodically (e.g. from a background loop). If enough time
has passed since the last update, intensity decreases and eventually
the emotion resets to calm.
"""
elapsed = time.time() - self.state.updated_at
if elapsed < _DECAY_INTERVAL:
return self.state
# Reduce intensity by 0.1 per decay interval
decay_steps = int(elapsed / _DECAY_INTERVAL)
new_intensity = max(0.0, self.state.intensity - 0.1 * decay_steps)
if new_intensity <= 0.1:
# Emotion has decayed — return to calm
self.state.previous_emotion = self.state.current_emotion
self.state.current_emotion = "calm"
self.state.intensity = 0.5
self.state.trigger_event = "decay"
else:
self.state.intensity = round(new_intensity, 2)
self.state.updated_at = time.time()
return self.state
def get_profile(self) -> dict:
"""Return the full emotional profile for dashboard display."""
self.decay() # Apply any pending decay
return {
"current_emotion": self.state.current_emotion,
"emotion_label": self.state.current_emotion.replace("_", " ").title(),
"intensity": self.state.intensity,
"intensity_label": _intensity_label(self.state.intensity),
"previous_emotion": self.state.previous_emotion,
"trigger_event": self.state.trigger_event,
"prompt_modifier": EMOTION_PROMPT_MODIFIERS.get(
self.state.current_emotion, ""
),
}
def get_prompt_modifier(self) -> str:
"""Return the prompt modifier string for the current emotion."""
self.decay()
return EMOTION_PROMPT_MODIFIERS.get(self.state.current_emotion, "")
def reset(self) -> None:
"""Reset to calm baseline."""
self.state = EmotionalState()
def _intensity_label(intensity: float) -> str:
"""Human-readable label for intensity value."""
if intensity >= 0.8:
return "overwhelming"
if intensity >= 0.6:
return "strong"
if intensity >= 0.4:
return "moderate"
if intensity >= 0.2:
return "mild"
return "faint"

3
src/timmy/agents/loader.py
View File

@@ -119,6 +119,8 @@ def load_agents(force_reload: bool = False) -> dict[str, Any]:
max_history = agent_cfg.get("max_history", defaults.get("max_history", 10))
tools = agent_cfg.get("tools", defaults.get("tools", []))
initial_emotion = agent_cfg.get("initial_emotion", "calm")
agent = SubAgent(
agent_id=agent_id,
name=agent_cfg.get("name", agent_id.title()),
@@ -127,6 +129,7 @@ def load_agents(force_reload: bool = False) -> dict[str, Any]:
tools=tools,
model=model,
max_history=max_history,
initial_emotion=initial_emotion,
)
_agents[agent_id] = agent