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": {
"特徴": "過去の会話を要約して保存",
"メリット": "効率的な情報圧縮",
"デメリット": "詳細情報の喪失"
}
}
この記事では、ConversationBufferWindowMemoryを使用します。実用性と効率性のバランスが優れているためです。
💾 データ永続化の戦略
メモリ機能を実装する際の重要な考慮点：
- 揮発性メモリ: プログラム終了時に消失
- 永続化メモリ: データベースに保存、再起動後も保持
- ハイブリッド方式: 両方を組み合わせた最適化
今回はハイブリッド方式を採用し、以下を実現します：
- 短期記憶: メモリ内で高速アクセス
- 長期記憶: SQLiteデータベースで永続化
- セッション管理: ユーザー別の会話分離
🗄️ データベース設計と実装
📊 データベーススキーマ設計
まず、効率的な会話履歴管理のためのデータベース設計を行います：
-- 会話テーブル
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
);
-- セッション情報テーブル
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
);
-- インデックス（検索性能向上）
CREATE INDEX IF NOT EXISTS idx_session_timestamp
ON conversations(session_id, timestamp);
データベース管理クラス（主要メソッドのみ）
class ConversationDatabase:
def __init__(self, db_path: str = "memory/conversations.db"):
self.db_path = db_path
self._init_database()
def save_conversation(self, session_id: str, user_input: str,
ai_response: str, metadata: Dict = None):
"""会話をデータベースに保存"""
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))
conn.commit()
conn.close()
def load_conversation_history(self, session_id: str,
limit: int = 10) -> List[Dict]:
"""過去の会話履歴を読み込み"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT user_input, ai_response, timestamp
FROM conversations
WHERE session_id = ?
ORDER BY timestamp DESC
LIMIT ?
''', (session_id, limit))
results = cursor.fetchall()
conn.close()
return [{"user_input": r[0], "ai_response": r[1],
"timestamp": r[2]} for r in reversed(results)]
完全なコードはpontanuki.comで公開しています。
🧠 メモリ機能付きエージェントの実装
データベースと連携するメモリ機能付きエージェントの主要部分：
class MemoryAgent:
def __init__(self, model_name: str = None, memory_window: int = 10):
# OpenAI LLMの初期化
self.llm = ChatOpenAI(
model=model_config["model_name"],
temperature=model_config["temperature"]
)
# メモリ設定（直近N件の会話を記憶）
self.memory = ConversationBufferWindowMemory(
k=memory_window,
return_messages=True,
memory_key="chat_history"
)
# データベース設定
self.db = ConversationDatabase()
# プロンプトテンプレート（メモリ対応）
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:
"""メモリ機能付きチャット"""
# 現在のメモリから会話履歴を取得
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)
return ai_response
def load_conversation_history(self, session_id: str,
limit: int = 10) -> List[Dict]:
"""過去の会話履歴をデータベースから読み込み"""
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
🧪 実際の使用例とテスト
💾 メモリエージェントの実行
作成したメモリエージェントを実際に動かしてみましょう：
# メモリエージェントのテスト
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小説は読まれましたか？
このように、過去の会話内容を完全に記憶しており、自然な継続対話が実現されています。
📈 メモリ最適化とパフォーマンス
⚡ メモリ使用量の最適化
大量の会話データを効率的に管理するための最適化手法：
class MemoryOptimizer:
def cleanup_old_conversations(self, days_to_keep: int = 30):
"""古い会話の削除"""
cutoff_date = datetime.now() - timedelta(days=days_to_keep)
cursor.execute('''
DELETE FROM conversations
WHERE timestamp < ?
''', (cutoff_date.isoformat(),))
def get_database_stats(self) -> Dict[str, Any]:
"""データベース統計情報の取得"""
return {
"total_conversations": total_conversations,
"total_sessions": total_sessions,
"database_size_mb": round(db_size / (1024 * 1024), 2),
"latest_conversation": latest_conversation
}
🎯 効率的なクエリ設計
大量データでも高速に動作するクエリ設計のポイント：
- 適切なインデックス: session_id + timestampの複合インデックス
- LIMIT句の活用: 必要な分だけデータを取得
- 準備済みステートメント: SQLインジェクション防止と性能向上
- バッチ処理: 複数の操作をトランザクション内で実行
🎯 第2部のまとめ
📚 この記事で学んだこと
メモリ管理面：
- ✅ LangChainメモリシステムの活用方法
- ✅ ConversationBufferWindowMemoryの実装
- ✅ 揮発性メモリと永続化メモリの組み合わせ
データベース面：
- ✅ SQLiteによる会話履歴管理
- ✅ 効率的なデータベーススキーマ設計
- ✅ インデックスによる検索性能最適化
実装面：
- ✅ セッション管理システムの構築
- ✅ ユーザー情報の自動抽出・保存
- ✅ 会話サマリー機能の実装
最適化面：
- ✅ メモリ使用量の最適化戦略
- ✅ データベース容量管理
- ✅ パフォーマンス監視機能
🚀 第2部の成果物
この記事を通じて、以下が完成しました：
- 高度なメモリエージェント: 会話文脈を理解する実用的なAI
- データベース管理システム: 効率的な会話履歴管理
- セッション管理機能: ユーザー別の会話分離システム
- 最適化機能: 大規模運用に対応したパフォーマンス管理
💡 重要なポイント
- 実用性重視: ビジネスでも使える本格的な機能実装
- スケーラビリティ: 大量データにも対応する設計
- ユーザーエクスペリエンス: 自然で継続的な対話体験
- メンテナンス性: 運用・保守を考慮した実装
🔗 次のステップ
第2部では、会話を記憶する高度なエージェントシステムを構築しました。第3部では、さらに強力なツール統合機能とWebUIを実装し、本格的な実用システムを完成させます。
📖 AI Agent開発シリーズ
- 第1部: 基本概念と環境構築
- 第2部: メモリとデータ管理 ← 現在の記事
- 第3部: ツール統合とWebUI - 高度なAI Agentシステム構築
完全なソースコードと詳細な実装解説はpontanuki.comで公開しています。
🎓 関連する学習リソース
- データベース設計基礎 - データベース設計をより深く理解
- Pythonデータ分析基礎 - データ処理スキルの向上
- FastAPIデータベース統合 - API連携での活用
💡 第2部をマスターしたら、次は計算・検索ツールを統合した最強のAI Agentシステム構築に挑戦しましょう！