83dea7c8ca
Automated salvage commit — agent session ended (exit 1). Work in progress, may need continuation.
111 lines
4.2 KiB
Python
111 lines
4.2 KiB
Python
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/<user_id>/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/<profile_name>/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
|