AIコーディング 2026.04.19

LangChainで会話メモリを実装する3ステップ|生成AIの履歴保存・永続化【コード付き】

タグ:生成AI / LangChain / 会話メモリ / 初心者向け / プログラミング

なぜ会話メモリが必要か

ChatGPTやClaudeのAPIをそのまま呼び出すと、毎回「初対面」の状態から会話が始まります。「さっき言った件はどうなった?」「前回教えてもらったコードに追加して」といった継続的な対話ができません。

LangChainの会話メモリ機能を使うと、過去のやり取りをコードで管理できるようになります。チャットボット・カスタマーサポートBot・個人アシスタントなど、継続的な文脈が重要なアプリケーションに必須の実装です。

準備

pip install langchain langchain-openai openai

環境変数にAPIキーを設定します:

export OPENAI_API_KEY="sk-..."

ステップ1:基本的な会話メモリを実装する

ConversationBufferMemory(全履歴保持)

会話をそのまま全て保持する最もシンプルな実装です。

from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain

# モデルとメモリの初期化
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
memory = ConversationBufferMemory()

# 会話チェーンの作成
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True  # Trueにすると送信されるプロンプトが表示される
)

# 会話実行
response1 = conversation.predict(input="私の名前は田中です。Pythonを勉強しています。")
print(response1)

response2 = conversation.predict(input="私の名前と勉強していることを教えてください")
print(response2)
# → 「田中さん、Pythonを勉強されているんですよね」のように前の会話を参照できる

会話履歴を直接確認する

# メモリの内容を確認
print(memory.chat_memory.messages)
# [HumanMessage(content='私の名前は...'), AIMessage(content='田中さん...')]

# 文字列として確認
print(memory.buffer)

ステップ2:会話履歴をファイルに保存・復元する

セッションをまたいで会話を継続するには、履歴をファイルやDBに保存する必要があります。

JSONファイルへの保存と復元

import json
from pathlib import Path
from langchain.memory import ConversationBufferMemory
from langchain_core.messages import HumanMessage, AIMessage

MEMORY_FILE = "conversation_history.json"

def save_memory(memory: ConversationBufferMemory, filepath: str) -> None:
    """会話履歴をJSONに保存"""
    messages = []
    for msg in memory.chat_memory.messages:
        messages.append({
            "type": msg.__class__.__name__,
            "content": msg.content
        })
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(messages, f, ensure_ascii=False, indent=2)

def load_memory(filepath: str) -> ConversationBufferMemory:
    """JSONから会話履歴を復元"""
    memory = ConversationBufferMemory()
    if not Path(filepath).exists():
        return memory
    
    with open(filepath, "r", encoding="utf-8") as f:
        messages = json.load(f)
    
    for msg in messages:
        if msg["type"] == "HumanMessage":
            memory.chat_memory.add_user_message(msg["content"])
        elif msg["type"] == "AIMessage":
            memory.chat_memory.add_ai_message(msg["content"])
    
    return memory

# 使用例:初回起動時
memory = load_memory(MEMORY_FILE)  # ファイルがなければ空のメモリで開始
conversation = ConversationChain(llm=llm, memory=memory)

response = conversation.predict(input="前回の続きを教えてください")
print(response)

# セッション終了時に保存
save_memory(memory, MEMORY_FILE)
print("会話履歴を保存しました")

SQLiteへの永続化(複数ユーザー対応)

from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain.memory import ConversationBufferMemory

def get_user_memory(user_id: str) -> ConversationBufferMemory:
    """ユーザーIDごとの会話メモリを取得"""
    chat_history = SQLChatMessageHistory(
        session_id=user_id,
        connection_string="sqlite:///conversations.db"
    )
    return ConversationBufferMemory(
        chat_memory=chat_history,
        return_messages=True
    )

# ユーザーA の会話
memory_a = get_user_memory("user_001")
conv_a = ConversationChain(llm=llm, memory=memory_a)
conv_a.predict(input="私は東京在住です")

# ユーザーB の会話(別の履歴)
memory_b = get_user_memory("user_002")
conv_b = ConversationChain(llm=llm, memory=memory_b)
conv_b.predict(input="私は大阪在住です")

ステップ3:長い会話を要約して圧縮する

会話が長くなるとトークン消費が増え、コストと応答速度に影響します。ConversationSummaryMemory は古い会話を自動的に要約して圧縮します。

from langchain.memory import ConversationSummaryMemory

summary_memory = ConversationSummaryMemory(
    llm=llm,
    return_messages=True
)

conversation = ConversationChain(
    llm=llm,
    memory=summary_memory
)

# 長い会話を続ける
for i in range(10):
    response = conversation.predict(
        input=f"質問{i+1}:Pythonのリスト操作について教えてください"
    )

# 要約された履歴を確認
print(summary_memory.moving_summary_buffer)
# → 会話の重要なポイントだけを要約したテキストが保存されている

トークン数で履歴を制限する方法

from langchain.memory import ConversationTokenBufferMemory

token_memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=2000  # 最大2000トークンに制限
)

conversation = ConversationChain(llm=llm, memory=token_memory)

上限を超えた古い会話は自動的に削除されます。

メモリの種類と使い分け

メモリタイプ特徴向いているケース
ConversationBufferMemory全履歴をそのまま保持短い会話・重要な細部が多い
ConversationSummaryMemory古い会話を自動要約長期的な会話・コスト削減優先
ConversationTokenBufferMemoryトークン数で上限設定コスト管理が重要な本番環境
ConversationBufferWindowMemory直近N回のみ保持最近の文脈だけ必要なケース
from langchain.memory import ConversationBufferWindowMemory

# 直近3回の会話だけ保持する例
window_memory = ConversationBufferWindowMemory(k=3)

プライバシーと注意点

会話履歴には個人情報が含まれる可能性があります。本番運用では以下を検討してください:

  • 暗号化: ファイル保存する場合はAES暗号化を適用
  • 有効期限: 30日後に自動削除するなどのポリシーを設定
  • ユーザー同意: プライバシーポリシーに会話履歴の保存と用途を明記
  • アクセス制限: DBへのアクセスはアプリサーバーのみに限定

あわせて読みたい

参考ソース