AI Agent入門第2部：メモリとデータ管理 - 継続的対話エージェントの実装

AI Agent開発シリーズの第2部へようこそ！第1部で基本的なエージェントを作成しましたが、今度は**「会話を記憶し、文脈を理解する」高度なエージェント**を実装します。

従来のチャットボットとの決定的な違いは、継続的な対話能力です。「私の名前は田中です」と伝えれば、次の会話でも「田中さん」として認識してくれる——そんな自然な対話体験を実現します。

この記事で実装するもの：

✅ 会話履歴を記憶するメモリ機能
✅ SQLiteデータベースによる永続化
✅ セッション管理システム
✅ 会話サマリー機能

第2部で身につけるスキル：

LangChainメモリシステムの活用
データベース設計とSQLite操作
セッション管理とデータ永続化
実用的なメモリ最適化手法

前提条件：

第1部: 基本概念と環境構築の完了
Python基礎知識とSQLiteの基本的な理解

それでは、記憶する力を持つAI Agentの世界に踏み込みましょう！

🧠 メモリ機能の重要性と仕組み

🤔 なぜメモリが重要なのか？

人間の会話を考えてみてください：

メモリなしの会話：

あなた: 私の名前は田中です
AI: はじめまして！
あなた: 私の趣味は読書です  
AI: 素晴らしい趣味ですね！
あなた: 私の名前を覚えてる？
AI: 申し訳ありませんが、お名前を教えていただけますか？

メモリありの会話：

あなた: 私の名前は田中です
AI: はじめまして、田中さん！
あなた: 私の趣味は読書です
AI: 読書がお好きなんですね、田中さん！
あなた: 私の名前を覚えてる？
AI: はい、田中さんとおっしゃいましたね。読書がご趣味でしたよね。

この違いが、ユーザーエクスペリエンスを劇的に向上させます。

🔄 LangChainメモリシステムの種類

LangChainは多様なメモリタイプを提供しています：

# メモリタイプの比較
memory_types = {
    "ConversationBufferMemory": {
        "特徴": "全ての会話を保存",
        "メリット": "完全な履歴保持",
        "デメリット": "トークン数が増加",
        "適用場面": "短い対話セッション"
    },
    "ConversationBufferWindowMemory": {
        "特徴": "直近N件の会話のみ保存",
        "メリット": "メモリ使用量を制限",
        "デメリット": "古い情報は忘却",
        "適用場面": "長期的な対話セッション"
    },
    "ConversationSummaryMemory": {
        "特徴": "過去の会話を要約して保存",
        "メリット": "効率的な情報圧縮",
        "デメリット": "詳細情報の喪失",
        "適用場面": "長期記憶が必要な場合"
    },
    "ConversationSummaryBufferMemory": {
        "特徴": "要約と直近会話の組み合わせ",
        "メリット": "効率と詳細のバランス",
        "デメリット": "実装が複雑",
        "適用場面": "高度な記憶戦略"
    }
}

この記事では、ConversationBufferWindowMemoryを使用します。実用性と効率性のバランスが優れているためです。

💾 データ永続化の戦略

メモリ機能を実装する際の重要な考慮点：

揮発性メモリ: プログラム終了時に消失
永続化メモリ: データベースに保存、再起動後も保持
ハイブリッド方式: 両方を組み合わせた最適化

今回はハイブリッド方式を採用し、以下を実現します：

短期記憶: メモリ内で高速アクセス
長期記憶: SQLiteデータベースで永続化
セッション管理: ユーザー別の会話分離

🗄️ データベース設計と実装

📊 データベーススキーマ設計

まず、効率的な会話履歴管理のためのデータベース設計を行います：

# memory/database_manager.py
import sqlite3
import json
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Any, Optional
import os

class ConversationDatabase:
    """会話履歴管理データベース"""
    
    def __init__(self, db_path: str = "memory/conversations.db"):
        """データベース初期化
        
        Args:
            db_path: データベースファイルのパス
        """
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """データベースとテーブルの初期化"""
        # ディレクトリが存在しない場合は作成
        Path(os.path.dirname(self.db_path)).mkdir(parents=True, exist_ok=True)
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 会話テーブルの作成
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS conversations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT NOT NULL,
                user_input TEXT NOT NULL,
                ai_response TEXT NOT NULL,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                metadata TEXT  -- JSON形式で追加情報を保存
            )
        ''')
        
        # セッション情報テーブルの作成
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS sessions (
                session_id TEXT PRIMARY KEY,
                user_name TEXT,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                total_messages INTEGER DEFAULT 0,
                metadata TEXT
            )
        ''')
        
        # インデックス作成（検索性能向上）
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_session_timestamp 
            ON conversations(session_id, timestamp)
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_session_updated 
            ON sessions(updated_at)
        ''')
        
        conn.commit()
        conn.close()
    
    def save_conversation(self, session_id: str, user_input: str, 
                         ai_response: str, metadata: Dict[str, Any] = None):
        """会話をデータベースに保存
        
        Args:
            session_id: セッション識別子
            user_input: ユーザーからの入力
            ai_response: AIの応答
            metadata: 追加のメタデータ
        """
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            # 会話を保存
            cursor.execute('''
                INSERT INTO conversations (session_id, user_input, ai_response, metadata)
                VALUES (?, ?, ?, ?)
            ''', (session_id, user_input, ai_response, 
                  json.dumps(metadata) if metadata else None))
            
            # セッション情報を更新
            cursor.execute('''
                INSERT OR REPLACE INTO sessions 
                (session_id, updated_at, total_messages)
                VALUES (?, ?, 
                    COALESCE((SELECT total_messages FROM sessions WHERE session_id = ?), 0) + 1)
            ''', (session_id, datetime.now().isoformat(), session_id))
            
            conn.commit()
            conn.close()
            
        except Exception as e:
            print(f"データベース保存エラー: {e}")
    
    def load_conversation_history(self, session_id: str, 
                                 limit: int = 10) -> List[Dict[str, Any]]:
        """過去の会話履歴を読み込み
        
        Args:
            session_id: セッション識別子
            limit: 読み込む会話数の上限
            
        Returns:
            会話履歴のリスト
        """
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                SELECT user_input, ai_response, timestamp, metadata
                FROM conversations 
                WHERE session_id = ? 
                ORDER BY timestamp DESC 
                LIMIT ?
            ''', (session_id, limit))
            
            results = cursor.fetchall()
            conn.close()
            
            # 結果を辞書形式で返す（時系列順）
            conversations = []
            for row in reversed(results):
                conversations.append({
                    "user_input": row[0],
                    "ai_response": row[1],
                    "timestamp": row[2],
                    "metadata": json.loads(row[3]) if row[3] else {}
                })
            
            return conversations
            
        except Exception as e:
            print(f"履歴読み込みエラー: {e}")
            return []
    
    def get_session_summary(self, session_id: str) -> Dict[str, Any]:
        """セッションのサマリー情報を取得"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            # 会話統計の取得
            cursor.execute('''
                SELECT COUNT(*) as total_messages,
                       MIN(timestamp) as first_message,
                       MAX(timestamp) as last_message
                FROM conversations 
                WHERE session_id = ?
            ''', (session_id,))
            
            conv_stats = cursor.fetchone()
            
            # セッション情報の取得
            cursor.execute('''
                SELECT user_name, created_at, total_messages, metadata
                FROM sessions 
                WHERE session_id = ?
            ''', (session_id,))
            
            session_info = cursor.fetchone()
            conn.close()
            
            return {
                "session_id": session_id,
                "total_messages": conv_stats[0] if conv_stats else 0,
                "first_message": conv_stats[1] if conv_stats else None,
                "last_message": conv_stats[2] if conv_stats else None,
                "user_name": session_info[0] if session_info else None,
                "created_at": session_info[1] if session_info else None,
                "db_total_messages": session_info[2] if session_info else 0,
                "metadata": json.loads(session_info[3]) if session_info and session_info[3] else {}
            }
            
        except Exception as e:
            return {"error": f"サマリー取得エラー: {e}"}
    
    def get_all_sessions(self, limit: int = 50) -> List[Dict[str, Any]]:
        """全セッションの一覧を取得"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                SELECT s.session_id, s.user_name, s.created_at, s.updated_at, 
                       s.total_messages, COUNT(c.id) as actual_messages
                FROM sessions s
                LEFT JOIN conversations c ON s.session_id = c.session_id
                GROUP BY s.session_id
                ORDER BY s.updated_at DESC 
                LIMIT ?
            ''', (limit,))
            
            results = cursor.fetchall()
            conn.close()
            
            return [
                {
                    "session_id": row[0],
                    "user_name": row[1],
                    "created_at": row[2],
                    "updated_at": row[3],
                    "total_messages": row[4],
                    "actual_messages": row[5]
                }
                for row in results
            ]
            
        except Exception as e:
            print(f"セッション一覧取得エラー: {e}")
            return []
    
    def clear_session_history(self, session_id: str):
        """特定セッションの履歴を削除"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('DELETE FROM conversations WHERE session_id = ?', (session_id,))
            cursor.execute('DELETE FROM sessions WHERE session_id = ?', (session_id,))
            
            conn.commit()
            conn.close()
            print(f"セッション '{session_id}' の履歴を削除しました")
            
        except Exception as e:
            print(f"履歴削除エラー: {e}")
    
    def update_session_metadata(self, session_id: str, user_name: str = None, 
                               metadata: Dict[str, Any] = None):
        """セッションのメタデータを更新"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            updates = []
            params = []
            
            if user_name:
                updates.append("user_name = ?")
                params.append(user_name)
            
            if metadata:
                updates.append("metadata = ?")
                params.append(json.dumps(metadata))
            
            if updates:
                updates.append("updated_at = ?")
                params.append(datetime.now().isoformat())
                params.append(session_id)
                
                query = f"UPDATE sessions SET {', '.join(updates)} WHERE session_id = ?"
                cursor.execute(query, params)
                
                conn.commit()
            
            conn.close()
            
        except Exception as e:
            print(f"メタデータ更新エラー: {e}")

🧠 メモリ機能付きエージェントの実装

次に、データベースと連携するメモリ機能付きエージェントを実装します：

# agents/memory_agent.py
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import SystemMessage
from typing import List, Dict, Any, Optional
import sys
from pathlib import Path

# プロジェクトルートをパスに追加
project_root = Path(__file__).parent.parent
sys.path.append(str(project_root))

from config.settings import settings, validate_environment
from memory.database_manager import ConversationDatabase

class MemoryAgent:
    """会話履歴を記憶するエージェント
    
    このクラスは会話の文脈を理解し、継続的な対話を可能にします。
    会話履歴はメモリとデータベースの両方に保存されます。
    """
    
    def __init__(self, model_name: str = None, memory_window: int = 10):
        """メモリ付きエージェントを初期化
        
        Args:
            model_name: 使用するOpenAIモデル名
            memory_window: メモリに保持する会話数
        """
        # 設定を取得
        model_config = settings.get_model_config(model_name)
        
        # OpenAI LLMの初期化
        self.llm = ChatOpenAI(
            model=model_config["model_name"], 
            temperature=model_config["temperature"],
            api_key=model_config["api_key"]
        )
        
        # メモリ設定（直近N件の会話を記憶）
        self.memory = ConversationBufferWindowMemory(
            k=memory_window,  # 保持する会話数
            return_messages=True,  # メッセージ形式で返す
            memory_key="chat_history"  # メモリのキー名
        )
        
        # データベース設定
        self.db = ConversationDatabase()
        
        # システムプロンプトの定義
        self.system_prompt = """あなたは会話の文脈を理解し、継続的な対話ができるAIアシスタントです。
過去の会話内容を参考にして、一貫性のある回答を提供してください。

重要な指針：
1. 過去の会話内容を適切に参照する
2. ユーザーが以前に話した内容について言及する場合は、
   「先ほどお話しした〜について」のように文脈を明確にする
3. 一貫性のある人格を保つ
4. 会話の流れを自然に継続する
5. 必要に応じて過去の内容を要約・整理する
6. ユーザーの名前や個人情報を適切に記憶・活用する

会話の継続性を重視し、まるで古い友人と話しているような自然な対話を心がけてください。"""
        
        # プロンプトテンプレートの作成（メモリ対応）
        self.prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content=self.system_prompt),
            MessagesPlaceholder(variable_name="chat_history"),  # 会話履歴の挿入点
            ("human", "{input}")
        ])
        
        # チェーンの構築
        self.chain = self.prompt | self.llm
    
    def chat_with_memory(self, user_input: str, session_id: str = "default") -> str:
        """メモリ機能付きチャット
        
        Args:
            user_input: ユーザーからの入力
            session_id: セッション識別子（ユーザー別の会話分離用）
            
        Returns:
            AI生成の回答
        """
        try:
            # 現在のメモリから会話履歴を取得
            chat_history = self.memory.chat_memory.messages
            
            # LLMに送信（プロンプト + 履歴 + 新しい入力）
            response = self.chain.invoke({
                "input": user_input,
                "chat_history": chat_history
            })
            
            ai_response = response.content
            
            # メモリに新しい会話を追加
            self.memory.chat_memory.add_user_message(user_input)
            self.memory.chat_memory.add_ai_message(ai_response)
            
            # データベースに永続化
            self.db.save_conversation(session_id, user_input, ai_response)
            
            # ユーザー名の抽出と保存（簡単な例）
            self._extract_and_save_user_info(user_input, session_id)
            
            return ai_response
            
        except Exception as e:
            error_msg = f"申し訳ありません。エラーが発生しました: {str(e)}"
            # エラーもメモリに記録
            self.memory.chat_memory.add_user_message(user_input)
            self.memory.chat_memory.add_ai_message(error_msg)
            return error_msg
    
    def _extract_and_save_user_info(self, user_input: str, session_id: str):
        """ユーザー情報の抽出と保存（簡単な実装）"""
        # 名前の抽出（簡単なパターンマッチング）
        import re
        
        name_patterns = [
            r"私の名前は(.+?)です",
            r"私は(.+?)です",
            r"(.+?)と申します",
            r"(.+?)と呼んでください"
        ]
        
        for pattern in name_patterns:
            match = re.search(pattern, user_input)
            if match:
                user_name = match.group(1).strip()
                self.db.update_session_metadata(session_id, user_name=user_name)
                break
    
    def load_conversation_history(self, session_id: str, limit: int = 10) -> List[Dict]:
        """過去の会話履歴をデータベースから読み込み
        
        Args:
            session_id: 読み込むセッションID
            limit: 読み込む会話数の上限
            
        Returns:
            会話履歴のリスト
        """
        try:
            # データベースから履歴を取得
            history = self.db.load_conversation_history(session_id, limit)
            
            # メモリに履歴を復元（既存のメモリをクリア）
            self.memory.clear()
            
            for conv in history:
                self.memory.chat_memory.add_user_message(conv["user_input"])
                self.memory.chat_memory.add_ai_message(conv["ai_response"])
            
            return history
            
        except Exception as e:
            print(f"履歴読み込みエラー: {e}")
            return []
    
    def clear_memory(self):
        """メモリをクリア（一時的な忘却）"""
        self.memory.clear()
    
    def clear_session_history(self, session_id: str):
        """特定セッションの履歴をデータベースから削除"""
        self.db.clear_session_history(session_id)
    
    def get_conversation_summary(self, session_id: str) -> Dict[str, Any]:
        """会話サマリーを生成"""
        summary = self.db.get_session_summary(session_id)
        summary["memory_size"] = len(self.memory.chat_memory.messages)
        return summary
    
    def get_all_sessions(self) -> List[Dict[str, Any]]:
        """すべてのセッション一覧を取得"""
        return self.db.get_all_sessions()

def demonstrate_memory_agent():
    """メモリエージェントのデモンストレーション"""
    print("🧠 メモリ付きエージェントのデモンストレーション")
    print("=" * 60)
    
    agent = MemoryAgent()
    session_id = "demo_session"
    
    # デモ会話
    demo_conversations = [
        "私の名前は田中太郎です",
        "私の趣味は読書と映画鑑賞です", 
        "特にSF小説が好きです",
        "私の名前を覚えていますか？",
        "私の趣味について教えてください",
        "おすすめのSF小説はありますか？"
    ]
    
    for i, user_input in enumerate(demo_conversations, 1):
        print(f"\n🙋 会話 {i}: {user_input}")
        response = agent.chat_with_memory(user_input, session_id)
        print(f"🧠 AI: {response}")
        print("-" * 40)
    
    # セッションサマリーの表示
    summary = agent.get_conversation_summary(session_id)
    print(f"\n📊 セッションサマリー:")
    print(f"   ユーザー名: {summary.get('user_name', '不明')}")
    print(f"   総メッセージ数: {summary['total_messages']}")
    print(f"   メモリ内メッセージ: {summary['memory_size']}")

def interactive_memory_chat():
    """対話型メモリチャット"""
    agent = MemoryAgent()
    
    print("🧠 メモリ付きチャットエージェントが起動しました！")
    print("このエージェントは会話の内容を記憶します。")
    print("'quit'、'exit'、'終了'で終了します。")
    print("'summary'でセッション情報を表示します。")
    print("'clear'でメモリをクリアします。\n")
    
    # セッションIDの設定
    session_id = input("セッションID（Enter で default）: ").strip() or "default"
    
    # 過去の履歴があれば読み込み
    history = agent.load_conversation_history(session_id, limit=5)
    if history:
        print(f"\n📚 過去の会話履歴を読み込みました（最新5件）：")
        for i, conv in enumerate(history[-3:], 1):  # 最新3件のみ表示
            print(f"  {i}. あなた: {conv['user_input'][:50]}...")
            print(f"     AI: {conv['ai_response'][:50]}...")
        print()
    
    while True:
        user_input = input("あなた: ")
        
        if user_input.lower() in ['quit', 'exit', '終了', 'q']:
            # 終了前にサマリーを表示
            summary = agent.get_conversation_summary(session_id)
            print(f"\n📊 今回の会話サマリー:")
            print(f"   ユーザー名: {summary.get('user_name', '不明')}")
            print(f"   総メッセージ数: {summary['total_messages']}")
            print(f"   メモリ内メッセージ: {summary['memory_size']}")
            print("👋 ありがとうございました！")
            break
        elif user_input.lower() == 'summary':
            summary = agent.get_conversation_summary(session_id)
            print(f"\n📊 セッションサマリー:")
            for key, value in summary.items():
                if key != 'error':
                    print(f"   {key}: {value}")
            continue
        elif user_input.lower() == 'clear':
            agent.clear_memory()
            print("🧹 メモリをクリアしました")
            continue
        elif not user_input.strip():
            print("❓ 何か話しかけてください。")
            continue
        
        # メモリ付きチャット実行
        print("🧠 考え中...")
        response = agent.chat_with_memory(user_input, session_id)
        print(f"🧠 AI: {response}\n")

# 使用例とテスト
if __name__ == "__main__":
    # 環境設定の検証
    if not validate_environment():
        print("環境設定を確認してから再実行してください。")
        exit(1)
    
    print("AI Agent メモリシステムが起動しました！\n")
    
    # 実行モードの選択
    mode = input("実行モード選択 [1: デモ, 2: 対話型]: ").strip()
    
    if mode == "1":
        demonstrate_memory_agent()
    elif mode == "2":
        interactive_memory_chat()
    else:
        print("デフォルトで対話型モードを開始します。")
        interactive_memory_chat()

🧪 実際の使用例とテスト

💾 メモリエージェントの実行

作成したメモリエージェントを実際に動かしてみましょう：

# メモリエージェントのテスト
python agents/memory_agent.py

実行例（継続的な会話）：

✅ 環境設定の検証が完了しました
AI Agent メモリシステムが起動しました！

実行モード選択 [1: デモ, 2: 対話型]: 2

🧠 メモリ付きチャットエージェントが起動しました！
セッションID（Enter で default）: user_tanaka

あなた: 私の名前は田中太郎です
🧠 考え中...
🧠 AI: はじめまして、田中太郎さん！よろしくお願いします。何かお手伝いできることはありますか？

あなた: 私の趣味は読書です
🧠 考え中...
🧠 AI: 読書が趣味でいらっしゃるのですね、田中さん！素晴らしい趣味ですね。どのようなジャンルの本がお好みでしょうか？

あなた: 特にSF小説が好きです
🧠 考え中...
🧠 AI: SF小説がお好きなんですね！田中さんのお好みを覚えておきます。最近読まれた印象的なSF作品はありますか？

あなた: 私の名前と趣味を覚えてる？
🧠 考え中...
🧠 AI: はい、もちろん覚えています！田中太郎さんで、ご趣味は読書、特にSF小説がお好きでしたね。先ほどお聞かせいただいた内容をしっかりと記憶しています。

あなた: summary
📊 セッションサマリー:
   session_id: user_tanaka
   total_messages: 8
   user_name: 田中太郎
   memory_size: 8

あなた: quit
📊 今回の会話サマリー:
   ユーザー名: 田中太郎
   総メッセージ数: 8
   メモリ内メッセージ: 8
👋 ありがとうございました！

🔄 セッション復元テスト

次回の実行時に、同じセッションIDで会話を再開すると：

python agents/memory_agent.py

セッションID（Enter で default）: user_tanaka

📚 過去の会話履歴を読み込みました（最新5件）：
  1. あなた: 私の名前は田中太郎です...
     AI: はじめまして、田中太郎さん！よろしくお願いします...
  2. あなた: 私の趣味は読書です...
     AI: 読書が趣味でいらっしゃるのですね、田中さん！...
  3. あなた: 特にSF小説が好きです...
     AI: SF小説がお好きなんですね！田中さんのお好みを覚えて...

あなた: こんにちは！
🧠 AI: こんにちは、田中太郎さん！お久しぶりです。読書の調子はいかがですか？新しいSF小説は読まれましたか？

このように、過去の会話内容を完全に記憶しており、自然な継続対話が実現されています。

📈 メモリ最適化とパフォーマンス

⚡ メモリ使用量の最適化

大量の会話データを効率的に管理するための最適化手法：

# memory/optimization.py
from typing import List, Dict, Any
import sqlite3
from datetime import datetime, timedelta

class MemoryOptimizer:
    """メモリ使用量の最適化クラス"""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
    
    def cleanup_old_conversations(self, days_to_keep: int = 30):
        """古い会話の削除（データベース容量削減）"""
        cutoff_date = datetime.now() - timedelta(days=days_to_keep)
        
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                DELETE FROM conversations 
                WHERE timestamp < ?
            ''', (cutoff_date.isoformat(),))
            
            deleted_count = cursor.rowcount
            conn.commit()
            conn.close()
            
            print(f"🧹 {deleted_count}件の古い会話を削除しました")
            return deleted_count
            
        except Exception as e:
            print(f"クリーンアップエラー: {e}")
            return 0
    
    def get_database_stats(self) -> Dict[str, Any]:
        """データベース統計情報の取得"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            # 会話数統計
            cursor.execute('SELECT COUNT(*) FROM conversations')
            total_conversations = cursor.fetchone()[0]
            
            # セッション数統計
            cursor.execute('SELECT COUNT(*) FROM sessions')
            total_sessions = cursor.fetchone()[0]
            
            # データベースサイズ
            cursor.execute("SELECT page_count * page_size as size FROM pragma_page_count(), pragma_page_size()")
            db_size = cursor.fetchone()[0]
            
            # 最新会話日時
            cursor.execute('SELECT MAX(timestamp) FROM conversations')
            latest_conversation = cursor.fetchone()[0]
            
            conn.close()
            
            return {
                "total_conversations": total_conversations,
                "total_sessions": total_sessions,
                "database_size_bytes": db_size,
                "database_size_mb": round(db_size / (1024 * 1024), 2),
                "latest_conversation": latest_conversation
            }
            
        except Exception as e:
            return {"error": f"統計取得エラー: {e}"}
    
    def compress_session_memory(self, session_id: str, target_messages: int = 20):
        """セッションメモリの圧縮（古いメッセージの要約）"""
        # 実装例: LLMを使用した会話要約
        # 実際の実装では、要約機能をここに追加
        pass

# 使用例
def optimize_memory_usage():
    """メモリ使用量の最適化実行"""
    optimizer = MemoryOptimizer("memory/conversations.db")
    
    # データベース統計の表示
    stats = optimizer.get_database_stats()
    print("📊 データベース統計:")
    for key, value in stats.items():
        if key != 'error':
            print(f"   {key}: {value}")
    
    # 30日以上古い会話の削除
    deleted = optimizer.cleanup_old_conversations(days_to_keep=30)
    print(f"\n🧹 最適化完了: {deleted}件の古いデータを削除")

if __name__ == "__main__":
    optimize_memory_usage()

🎯 効率的なクエリ設計

大量データでも高速に動作するクエリ設計のポイント：

適切なインデックス: session_id + timestampの複合インデックス
LIMIT句の活用: 必要な分だけデータを取得
準備済みステートメント: SQLインジェクション防止と性能向上
バッチ処理: 複数の操作をトランザクション内で実行

🎯 第2部のまとめ

📚 この記事で学んだこと

メモリ管理面：

✅ LangChainメモリシステムの活用方法
✅ ConversationBufferWindowMemoryの実装
✅ 揮発性メモリと永続化メモリの組み合わせ

データベース面：

✅ SQLiteによる会話履歴管理
✅ 効率的なデータベーススキーマ設計
✅ インデックスによる検索性能最適化

実装面：

✅ セッション管理システムの構築
✅ ユーザー情報の自動抽出・保存
✅ 会話サマリー機能の実装

最適化面：

✅ メモリ使用量の最適化戦略
✅ データベース容量管理
✅ パフォーマンス監視機能

🚀 第2部の成果物

この記事を通じて、以下が完成しました：

高度なメモリエージェント: 会話文脈を理解する実用的なAI
データベース管理システム: 効率的な会話履歴管理
セッション管理機能: ユーザー別の会話分離システム
最適化機能: 大規模運用に対応したパフォーマンス管理

💡 重要なポイント

実用性重視: ビジネスでも使える本格的な機能実装
スケーラビリティ: 大量データにも対応する設計
ユーザーエクスペリエンス: 自然で継続的な対話体験
メンテナンス性: 運用・保守を考慮した実装

🔗 次のステップ

第2部では、会話を記憶する高度なエージェントシステムを構築しました。第3部では、さらに強力なツール統合機能とWebUIを実装し、本格的な実用システムを完成させます。

📖 AI Agent開発シリーズ

第1部: 基本概念と環境構築
第2部: メモリとデータ管理 ← 現在の記事
第3部: ツール統合とWebUI - 高度なAI Agentシステム構築

🎓 関連する学習リソース

データベース設計基礎 - データベース設計をより深く理解
Pythonデータ分析基礎 - データ処理スキルの向上
FastAPIデータベース統合 - API連携での活用

💡 第2部をマスターしたら、次は計算・検索ツールを統合した最強のAI Agentシステム構築に挑戦しましょう！

AI Agent入門第2部：メモリとデータ管理 - 継続的対話エージェントの実装

AI Agent入門第2部：メモリとデータ管理 - 継続的対話エージェントの実装

🧠 メモリ機能の重要性と仕組み

🤔 なぜメモリが重要なのか？

🔄 LangChainメモリシステムの種類

💾 データ永続化の戦略

🗄️ データベース設計と実装

📊 データベーススキーマ設計

🧠 メモリ機能付きエージェントの実装

🧪 実際の使用例とテスト

💾 メモリエージェントの実行

🔄 セッション復元テスト

📈 メモリ最適化とパフォーマンス

⚡ メモリ使用量の最適化

🎯 効率的なクエリ設計

🎯 第2部のまとめ

📚 この記事で学んだこと

🚀 第2部の成果物

💡 重要なポイント

🔗 次のステップ

📖 AI Agent開発シリーズ

🎓 関連する学習リソース

関連記事

MCP（Model Context Protocol）完全ガイド：AIツール連携の新標準で構築する拡張性の高いシステム

CrewAI実践入門：複数AIエージェントでチーム開発を自動化【LangChainの次世代フレームワーク】

LangGraph入門：ステートフルなAI Agentの構築ガイド

LightGBM実践ガイド：不動産価格予測で学ぶ高精度な勾配ブースティング

コメント

AI Agent入門 第2部：メモリとデータ管理 - 継続的対話エージェントの実装

🧠 メモリ機能の重要性と仕組み

🤔 なぜメモリが重要なのか？

🔄 LangChainメモリシステムの種類

💾 データ永続化の戦略

🗄️ データベース設計と実装

📊 データベーススキーマ設計

🧠 メモリ機能付きエージェントの実装

🧪 実際の使用例とテスト

💾 メモリエージェントの実行

🔄 セッション復元テスト

📈 メモリ最適化とパフォーマンス

⚡ メモリ使用量の最適化

🎯 効率的なクエリ設計

🎯 第2部のまとめ

📚 この記事で学んだこと

🚀 第2部の成果物

💡 重要なポイント

🔗 次のステップ

📖 AI Agent開発シリーズ

🎓 関連する学習リソース

関連記事

MCP（Model Context Protocol）完全ガイド：AIツール連携の新標準で構築する拡張性の高いシステム

CrewAI実践入門：複数AIエージェントでチーム開発を自動化【LangChainの次世代フレームワーク】

LangGraph入門：ステートフルなAI Agentの構築ガイド

LightGBM実践ガイド：不動産価格予測で学ぶ高精度な勾配ブースティング

コメント

AI Agent入門第2部：メモリとデータ管理 - 継続的対話エージェントの実装