diff --git a/agent/pca.py b/agent/pca.py new file mode 100644 index 000000000..4937a62a9 --- /dev/null +++ b/agent/pca.py @@ -0,0 +1,110 @@ +import json +import logging +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + +@dataclass +class PersonalizedCognitiveProfile: + """ + Represents a personalized cognitive profile for a user. + """ + user_id: str + preferred_tone: Optional[str] = None + # Add more fields as the PCA evolves + + def to_dict(self) -> dict: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict) -> "PersonalizedCognitiveProfile": + return cls(**data) + +def _get_profile_path(user_id: str) -> Path: + """ + Returns the path to the personalized cognitive profile file for a given user. + """ + # Assuming profiles are stored under ~/.hermes/profiles//pca_profile.json + # This needs to be integrated with the existing profile system more robustly. + from hermes_constants import get_hermes_home + hermes_home = get_hermes_home() + # Profiles are stored under ~/.hermes/profiles//pca_profile.json + # where profile_name could be the user_id or a derived value. + # For now, we'll assume the user_id is the profile name for simplicity. + profile_dir = hermes_home / "profiles" / user_id + if not profile_dir.is_dir(): + # Fallback to default HERMES_HOME if no specific user profile dir exists + return hermes_home / "pca_profile.json" + return profile_dir / "pca_profile.json" + +def load_cognitive_profile(user_id: str) -> Optional[PersonalizedCognitiveProfile]: + """ + Loads the personalized cognitive profile for a user. + """ + profile_path = _get_profile_path(user_id) + if not profile_path.exists(): + return None + try: + with open(profile_path, "r", encoding="utf-8") as f: + data = json.load(f) + return PersonalizedCognitiveProfile.from_dict(data) + except Exception as e: + logger.warning(f"Failed to load cognitive profile for user {user_id}: {e}") + return None + +def save_cognitive_profile(profile: PersonalizedCognitiveProfile) -> None: + """ + Saves the personalized cognitive profile for a user. + """ + profile_path = _get_profile_path(profile.user_id) + profile_path.parent.mkdir(parents=True, exist_ok=True) + try: + with open(profile_path, "w", encoding="utf-8") as f: + json.dump(profile.to_dict(), f, indent=2, ensure_ascii=False) + except Exception as e: + logger.error(f"Failed to save cognitive profile for user {profile.user_id}: {e}") + +def _get_sessions_by_user_id(db, user_id: str) -> list[dict]: + """Helper to get sessions for a specific user_id from SessionDB.""" + def _do(conn): + cursor = conn.execute( + "SELECT id FROM sessions WHERE user_id = ? ORDER BY started_at DESC", + (user_id,) + ) + return [row["id"] for row in cursor.fetchall()] + return db._execute_read(_do) + +def analyze_interactions(user_id: str) -> Optional[PersonalizedCognitiveProfile]: + """ + Analyzes historical interactions for a user to infer their cognitive profile. + This is a placeholder and will be implemented with actual analysis logic. + """ + logger.info(f"Analyzing interactions for user {user_id}") + + from hermes_state import SessionDB + db = SessionDB() + + sessions = _get_sessions_by_user_id(db, user_id) + all_messages = [] + for session_id in sessions: + all_messages.extend(db.get_messages_as_conversation(session_id)) + + # Simple heuristic for preferred_tone (placeholder) + # In a real implementation, this would involve NLP techniques. + preferred_tone = "neutral" + if user_id == "Alexander Whitestone": # Example: Replace with actual detection + # This is a very simplistic example. Real analysis would be complex. + # For demonstration, let's assume Alexander prefers a 'formal' tone + # if he has had more than 5 interactions. + if len(all_messages) > 5: + preferred_tone = "formal" + else: + preferred_tone = "informal" # Default for less interaction + elif "technical" in " ".join([m.get("content", "").lower() for m in all_messages]): + preferred_tone = "technical" + + profile = PersonalizedCognitiveProfile(user_id=user_id, preferred_tone=preferred_tone) + save_cognitive_profile(profile) + return profile