core_memory_manager.py

"""
メモリ管理モジュール
会話履歴から重要単語を抽出し、トークン使用量を最適化する
"""
import logging
import re
from typing import List, Dict, Tuple, Any
from collections import Counter
import json

logger = logging.getLogger(__name__)

class MemoryManager:
    """会話履歴のメモリ管理を行うクラス"""
    
    def __init__(self, history_threshold: int = 10):
        """
        Args:
            history_threshold: 履歴圧縮を実行する会話数の閾値
        """
        self.history_threshold = history_threshold
        self.important_words_cache = []
        self.special_memories = {}  # 手紙などの特別な記憶を保存
        
    def extract_important_words(self, messages: List[Dict[str, str]], 
                              dialogue_generator=None) -> List[str]:
        """
        会話履歴から重要単語を抽出する（ルールベースのみ）
        
        Args:
            messages: チャットメッセージのリスト
            dialogue_generator: 対話生成器（使用しない）
            
        Returns:
            重要単語のリスト
        """
        try:
            # メッセージからテキストを結合
            text_content = []
            for msg in messages:
                if msg.get("content"):
                    text_content.append(msg["content"])
            
            combined_text = " ".join(text_content)
            
            # ルールベースの抽出のみ使用
            return self._extract_with_rules(combined_text)
            
        except Exception as e:
            logger.error(f"重要単語抽出エラー: {e}")
            return self._extract_with_rules(" ".join([msg.get("content", "") for msg in messages]))
    

    
    def _extract_with_rules(self, text: str) -> List[str]:
        """
        ルールベースで重要単語を抽出する（強化版）
        
        Args:
            text: 抽出対象のテキスト
            
        Returns:
            重要単語のリスト
        """
        try:
            # 基本的なクリーニング
            text = re.sub(r'[^\w\s]', ' ', text)
            words = text.split()
            
            # ストップワードを除外
            stop_words = {
                'の', 'に', 'は', 'を', 'が', 'で', 'と', 'から', 'まで', 'より',
                'だ', 'である', 'です', 'ます', 'した', 'する', 'される',
                'これ', 'それ', 'あれ', 'この', 'その', 'あの',
                'ここ', 'そこ', 'あそこ', 'どこ', 'いつ', 'なに', 'なぜ',
                'ちょっと', 'とても', 'すごく', 'かなり', 'もう', 'まだ',
                'でも', 'しかし', 'だから', 'そして', 'また', 'さらに',
                'あたし', 'お前', 'ユーザー', 'システム', 'アプリ'
            }
            
            # 重要カテゴリのキーワード
            important_categories = {
                'food': ['コーヒー', 'お茶', '紅茶', 'ケーキ', 'パン', '料理', '食べ物', '飲み物'],
                'hobby': ['読書', '映画', '音楽', 'ゲーム', 'スポーツ', '散歩', '旅行'],
                'emotion': ['嬉しい', '悲しい', '楽しい', '怒り', '不安', '安心', '幸せ'],
                'place': ['家', '学校', '会社', '公園', 'カフェ', '図書館', '駅', '街'],
                'time': ['朝', '昼', '夜', '今日', '明日', '昨日', '週末', '平日'],
                'color': ['赤', '青', '緑', '黄色', '白', '黒', 'ピンク', '紫'],
                'weather': ['晴れ', '雨', '曇り', '雪', '暑い', '寒い', '暖かい', '涼しい']
            }
            
            # 重要そうなパターンを優先
            important_patterns = [
                r'[A-Za-z]{3,}',  # 英単語（3文字以上）
                r'[ァ-ヶー]{2,}',  # カタカナ（2文字以上）
                r'[一-龯]{2,}',   # 漢字（2文字以上）
            ]
            
            important_words = []
            
            # パターンマッチング
            for pattern in important_patterns:
                matches = re.findall(pattern, text)
                important_words.extend(matches)
            
            # カテゴリ別重要語句の検出
            for category, keywords in important_categories.items():
                for keyword in keywords:
                    if keyword in text:
                        important_words.append(keyword)
            
            # 頻度でフィルタリング
            word_counts = Counter(important_words)
            filtered_words = []
            
            for word, count in word_counts.items():
                if (len(word) >= 2 and 
                    word not in stop_words and 
                    not word.isdigit() and  # 数字のみは除外
                    count >= 1):  # 最低1回は出現
                    filtered_words.append(word)
            
            # 重要度でソート（頻度 + カテゴリ重要度）
            def get_importance_score(word):
                base_score = word_counts[word]
                # カテゴリに含まれる語句は重要度アップ
                for keywords in important_categories.values():
                    if word in keywords:
                        base_score += 2
                # 長い語句は重要度アップ
                if len(word) >= 4:
                    base_score += 1
                return base_score
            
            # 重要度順でソートして上位15個を返す
            sorted_words = sorted(filtered_words, key=get_importance_score, reverse=True)
            return sorted_words[:15]
            
        except Exception as e:
            logger.error(f"ルールベース抽出エラー: {e}")
            return []
    
    def should_compress_history(self, messages: List[Dict[str, str]]) -> bool:
        """
        履歴を圧縮すべきかどうかを判定する
        
        Args:
            messages: チャットメッセージのリスト
            
        Returns:
            圧縮が必要かどうか
        """
        # ユーザーとアシスタントのペア数をカウント
        user_messages = [msg for msg in messages if msg.get("role") == "user"]
        return len(user_messages) >= self.history_threshold
    
    def compress_history(self, messages: List[Dict[str, str]], 
                        dialogue_generator=None) -> Tuple[List[Dict[str, str]], List[str]]:
        """
        履歴を圧縮し、重要単語を抽出する
        
        Args:
            messages: チャットメッセージのリスト
            dialogue_generator: 対話生成器
            
        Returns:
            (圧縮後のメッセージリスト, 抽出された重要単語のリスト)
        """
        try:
            if not self.should_compress_history(messages):
                return messages, self.important_words_cache
            
            # 最新の数ターンを保持
            keep_recent = 4  # 最新4ターン（ユーザー2回、アシスタント2回）を保持
            
            # 古い履歴から重要単語を抽出
            old_messages = messages[:-keep_recent] if len(messages) > keep_recent else []
            recent_messages = messages[-keep_recent:] if len(messages) > keep_recent else messages
            
            if old_messages:
                # 重要単語を抽出
                new_keywords = self.extract_important_words(old_messages, dialogue_generator)
                
                # 既存のキーワードと統合（重複除去）
                all_keywords = list(set(self.important_words_cache + new_keywords))
                self.important_words_cache = all_keywords[:20]  # 最大20個のキーワードを保持
                
                logger.info(f"履歴を圧縮しました。抽出されたキーワード: {new_keywords}")
            
            return recent_messages, self.important_words_cache
            
        except Exception as e:
            logger.error(f"履歴圧縮エラー: {e}")
            return messages, self.important_words_cache
    
    def get_memory_summary(self) -> str:
        """
        保存されている重要単語から記憶の要約を生成する
        
        Returns:
            記憶の要約文字列
        """
        summary_parts = []
        
        # 通常の重要単語
        if self.important_words_cache:
            keywords_text = "、".join(self.important_words_cache)
            summary_parts.append(f"過去の会話で言及された重要な要素: {keywords_text}")
        
        # 特別な記憶（手紙など）
        if self.special_memories:
            for memory_type, memories in self.special_memories.items():
                if memories:
                    latest_memory = memories[-1]["content"]
                    if memory_type == "letter_content":
                        summary_parts.append(f"最近の手紙の記憶: {latest_memory}")
                    else:
                        summary_parts.append(f"{memory_type}: {latest_memory}")
        
        return "\n".join(summary_parts) if summary_parts else ""
    
    def add_important_memory(self, memory_type: str, content: str) -> str:
        """
        重要な記憶を追加する（手紙の内容など）
        
        Args:
            memory_type: 記憶の種類（例: "letter_content"）
            content: 記憶する内容
            
        Returns:
            ユーザーに表示する通知メッセージ
        """
        if memory_type not in self.special_memories:
            self.special_memories[memory_type] = []
        
        self.special_memories[memory_type].append({
            "content": content,
            "timestamp": logging.Formatter().formatTime(logging.LogRecord("", 0, "", 0, "", (), None))
        })
        
        # 最大5件まで保持
        if len(self.special_memories[memory_type]) > 5:
            self.special_memories[memory_type] = self.special_memories[memory_type][-5:]
        
        logger.info(f"特別な記憶を追加しました: {memory_type}")
        
        # 記憶の種類に応じた通知メッセージを生成
        if memory_type == "letter_content":
            return "🧠✨ 麻理の記憶に新しい手紙の内容が刻まれました。今後の会話でこの記憶を参照することがあります。"
        else:
            return f"🧠✨ 麻理の記憶に新しい{memory_type}が追加されました。"
    
    def get_special_memories(self, memory_type: str = None) -> Dict[str, Any]:
        """
        特別な記憶を取得する
        
        Args:
            memory_type: 取得する記憶の種類（Noneの場合は全て）
            
        Returns:
            記憶の辞書
        """
        if memory_type:
            return self.special_memories.get(memory_type, [])
        return self.special_memories
    
    def clear_memory(self):
        """メモリをクリアする"""
        self.important_words_cache = []
        self.special_memories = {}
        logger.info("メモリをクリアしました")
    
    def get_memory_stats(self) -> Dict[str, Any]:
        """
        メモリの統計情報を取得する
        
        Returns:
            統計情報の辞書
        """
        return {
            "cached_keywords_count": len(self.important_words_cache),
            "cached_keywords": self.important_words_cache,
            "special_memories_count": sum(len(memories) for memories in self.special_memories.values()),
            "special_memories": self.special_memories,
            "history_threshold": self.history_threshold
        }