"""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