Digital-Galatea / agents /emotional_agent.py
codejedi's picture
Replace quantum random number API with regular random numbers
273a32a
"""Emotional State Agent - responsible for managing and updating emotional state"""
import os
import sys
import json
import random
import logging
import requests
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import MODEL_CONFIG
class EmotionalStateAgent:
"""Agent responsible for managing and updating emotional state"""
def __init__(self, initial_state=None, config=None):
self.config = config or MODEL_CONFIG or {}
# Get JSON file path from config
emotions_config = self.config.get('emotions', {}) if self.config else {}
self.json_path = emotions_config.get('json_path', './emotions.json')
# Load emotional state from JSON file if it exists, otherwise use initial_state or defaults
self.emotional_state = self._load_from_json() or initial_state or {"joy": 0.2, "sadness": 0.2, "anger": 0.2, "fear": 0.2, "curiosity": 0.2}
# Slower learning rate for more gradual emotion changes
self.learning_rate = 0.03
# Save initial state to JSON
self._save_to_json()
def _load_from_json(self):
"""Load emotional state from JSON file"""
try:
if os.path.exists(self.json_path):
with open(self.json_path, 'r', encoding='utf-8') as f:
state = json.load(f)
# Validate state structure
required_emotions = ["joy", "sadness", "anger", "fear", "curiosity"]
if all(emotion in state for emotion in required_emotions):
logging.info(f"[EmotionalStateAgent] Loaded emotional state from {self.json_path}")
return state
else:
logging.warning(f"[EmotionalStateAgent] Invalid state structure in {self.json_path}, using defaults")
else:
logging.info(f"[EmotionalStateAgent] No existing emotional state file found, starting fresh")
except Exception as e:
logging.warning(f"[EmotionalStateAgent] Error loading emotional state from JSON: {e}")
return None
def _save_to_json(self):
"""Save emotional state to JSON file"""
try:
with open(self.json_path, 'w', encoding='utf-8') as f:
json.dump(self.emotional_state, f, indent=2, ensure_ascii=False)
logging.debug(f"[EmotionalStateAgent] Saved emotional state to {self.json_path}")
except Exception as e:
logging.error(f"[EmotionalStateAgent] Error saving emotional state to JSON: {e}")
def get_random_float(self, min_val=0.0, max_val=1.0):
"""Get a random float between min_val and max_val"""
return random.uniform(min_val, max_val)
def update_with_sentiment(self, sentiment_score):
"""Update emotional state based on sentiment"""
# Slower decay factor to prevent emotions from crashing to minimum too fast
# Changed from 0.9 to 0.97 (only 3% decay per interaction instead of 10%)
decay_factor = self.get_random_float(0.95, 0.99)
for emotion in self.emotional_state:
# Decay emotions more slowly (preserves emotional state longer)
self.emotional_state[emotion] *= decay_factor
# Clamp to valid range
self.emotional_state[emotion] = max(0.0, min(1.0, self.emotional_state[emotion]))
# Apply sentiment with slower learning rate for gradual changes
learning_rate = self.get_random_float(0.02, 0.04)
# Update emotions based on sentiment (slower, more gradual)
self.emotional_state["joy"] += sentiment_score * learning_rate
self.emotional_state["sadness"] -= sentiment_score * learning_rate
# Add randomness to curiosity (making responses more unpredictable)
curiosity_boost = self.get_random_float(-0.03, 0.03)
self.emotional_state["curiosity"] = max(0.0, min(1.0,
self.emotional_state["curiosity"] + curiosity_boost))
# Soft normalization - only normalize if emotions get too extreme
# This prevents emotions from being forced to equal values
total_emotion = sum(self.emotional_state.values())
if total_emotion > 1.5 or total_emotion < 0.5:
# Only normalize if emotions are way out of balance
for emotion in self.emotional_state:
self.emotional_state[emotion] = self.emotional_state[emotion] / total_emotion if total_emotion > 0 else 0.2
else:
# Just ensure no emotion goes below minimum threshold
for emotion in self.emotional_state:
if self.emotional_state[emotion] < 0.05:
self.emotional_state[emotion] = 0.05
# Save to JSON after update
self._save_to_json()
logging.info(f"[EmotionalStateAgent] Updated emotional state: {self.emotional_state}")
return self.emotional_state
def get_state(self):
"""Get current emotional state"""
return self.emotional_state.copy()
def is_ready(self):
"""Check if agent is ready"""
return True # Emotional state is always ready