AIエージェントが会話を忘れる原因と解決方法【RAG・メモリ設計3つの実装パターン】

AIエージェントが「記憶喪失」になる原因

ChatGPT APIや生成AIを使ったエージェント（自動で判断して動作するプログラム）を運用していると、こんな経験はないでしょうか。

• 10分前の会話の内容をエージェントが忘れている
• ユーザーが「さっき言ったあの件」と言っても、過去のやり取りが反映されない
• チャットボットが同じ質問を何度もしてきた

これは「セッション間記憶喪失」と呼ばれる問題です。LLMは本質的にステートレス（状態を持たない）であり、エージェントが新しいセッションを開始すると、前のセッションの文脈が完全に失われてしまいます。

RAGを使っても解決しないケースが多い点が2026年の現場での共通課題です。

RAGの本質：「記憶」ではなく「検索」

RAGに対する誤解が根強くあります。RAGは記憶ではありません。

RAG（Retrieval-Augmented Generation）の実態:

質問 → テキストをベクトル化 → 類似ドキュメントを検索 → プロンプトに追加

これは Ctrl+F（検索）を賢くしただけです。

RAGが苦手なこと:
  ❌ 「3週間前の会議でAさんが決めた方針」を記憶する
  ❌ 「ユーザーが過去に拒否した選択肢」を避ける  
  ❌ セッション間のコンテキスト継続
  ❌ 意思決定の記録と参照

RAGが得意なこと:
  ✅ 大量ドキュメントから関連箇所を検索
  ✅ 社内ナレッジベースの活用
  ✅ 最新情報の取得（学習データのカットオフを超える）

本物のメモリが必要な場合、RAGとは別の設計が必要です。

ChatGPT APIでメモリ機能を実装する理由

ChatGPT APIは基本的に「ステートレス」です。つまり、APIを呼び出すたびに、以前の会話内容は保持されません。毎回、ユーザー設定や好みを何度も入力し直す必要があります。

これを解決するのが「メモリ機能」です。メモリ機能を実装すれば、ユーザーの過去の会話や設定を自動的に記憶し、次の会話に活かすことができます。結果として：

• ユーザーが同じ指示を繰り返す必要がなくなる
• AIアシスタントがユーザーの好みや背景を理解した上で回答する
• 会話がより自然で効率的になる

これは特にチャットボット、カスタマーサポートAI、個人用コーディングアシスタントで重要です。

メモリ実装の3つの方法

ChatGPT APIでメモリを実装する方法は、大きく3つに分かれます。

1. 会話履歴の直接送信（シンプル）

最もシンプルな方法は、ChatGPT APIのmessagesパラメータに過去の会話をそのまま追加することです。ただし、この方法はトークン数が増え続けるため、長期運用には向きません。

2. LangChainのメモリ永続化（実装的）

LangChainのメモリクラスを使い、会話履歴をJSONファイルやデータベースに保存する方法です。これが今回の記事の中心となります。

3. Retrieval Augmented Generation（RAG）統合（高度）

会話履歴をベクトル化して保存し、関連する過去の会話を検索して活用する方法です。大量の会話データを効率的に管理できます。

前提環境と必要なもの

以下の環境を整備してください。

• Python: 3.9以上
• OpenAI APIキー: https://platform.openai.com/api-keys で取得
• LangChain: 0.1.0以上
• 必要なライブラリ:
  • python-dotenv (APIキーの管理)
  • langchain (メモリと会話チェーン)
  • openai (ChatGPT APIの呼び出し)

インストール手順

# Pythonバージョン確認
python --version

# 必要なライブラリをインストール
pip install langchain openai python-dotenv

LangChainのバージョン確認：

pip show langchain

0.1.0以上であることを確認してください。

解決策1：JSONファイルでセッション間メモリを実装する

最も手軽で実用的な方法です。データベース不要で始められます。

ステップ1: 環境変数の設定

.env ファイルを作成し、OpenAI APIキーを保存します。

# .env ファイル
OPENAI_API_KEY=sk-...your-api-key-here...

ステップ2: シンプルなメモリ実装コード

まずは「メモリ機能なし」と「メモリ機能あり」の違いを実感するコードです。

# memory_basic.py
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.llms import OpenAI
from dotenv import load_dotenv
import os

# 環境変数から APIキー を読み込む
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

# ステップ1: メモリオブジェクトを作成
memory = ConversationBufferMemory()

# ステップ2: OpenAI LLM を初期化
llm = OpenAI(
    api_key=api_key,
    model="gpt-3.5-turbo",  # または gpt-4
    temperature=0.7
)

# ステップ3: ConversationChain を作成（メモリ付き）
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True  # デバッグ用：やり取りを画面に表示
)

# ステップ4: 会話を実行
response1 = conversation.predict(input="私の名前はTaroです。好きなプログラミング言語はPythonです。")
print("Response 1:")
print(response1)
print("\n" + "="*50 + "\n")

# 2番目の入力：前の会話を覚えているか確認
response2 = conversation.predict(input="私の好きな言語は何だったか覚えていますか？")
print("Response 2:")
print(response2)

このコードを実行すると、2番目の質問で「Python」と答えられます。なぜなら、ConversationChainが自動的に過去の会話をメモリに保存し、それを次の質問に含めるからです。

ステップ3: メモリを永続化（保存と読み込み）

# memory_persistent.py
import json
import os
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.llms import OpenAI
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

# メモリ保存用のファイルパス
MEMORY_FILE = "conversation_memory.json"

def load_memory():
    """ファイルからメモリを読み込む"""
    memory = ConversationBufferMemory()
    
    if os.path.exists(MEMORY_FILE):
        with open(MEMORY_FILE, 'r', encoding='utf-8') as f:
            data = json.load(f)
            # 保存されたメモリを復元
            memory.buffer = data.get('buffer', '')
    
    return memory

def save_memory(memory):
    """メモリをファイルに保存"""
    with open(MEMORY_FILE, 'w', encoding='utf-8') as f:
        json.dump({'buffer': memory.buffer}, f, ensure_ascii=False, indent=2)

# ステップ1: 前回のメモリを読み込む
memory = load_memory()

# ステップ2: LLM と ConversationChain を初期化
llm = OpenAI(api_key=api_key, model="gpt-3.5-turbo", temperature=0.7)
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# ステップ3: ユーザーと対話ループ
print("AIアシスタント（メモリ機能付き）")
print("終了: 'quit' または 'exit' を入力")
print("="*50)

while True:
    user_input = input("\nあなた: ").strip()
    
    if user_input.lower() in ['quit', 'exit']:
        save_memory(memory)  # 終了時にメモリを保存
        print("メモリを保存しました。さようなら！")
        break
    
    if not user_input:
        continue
    
    response = conversation.predict(input=user_input)
    print(f"\nAI: {response}")
    
    # 毎回メモリを保存（オプション：セッション終了時のみでもOK）
    save_memory(memory)

このコードの重要な点：

1. load_memory() は前回のセッションから会話履歴を復元
2. save_memory() は毎回の会話後にファイルに保存
3. プログラムを再起動しても、会話の履歴が残る

実行例：

python memory_persistent.py

初回セッション：

あなた: 私はデータサイエンティストです。好きなツールはPandasです。
AI: （Pandasについて返答）

あなた: quit

次のセッション：

python memory_persistent.py

あなた: 何度も教えなくていいから、私の職業とツール選びについて簡潔に説明して
AI: あなたはデータサイエンティストで、Pandasをお好みですね。...

解決策2：グラフ検索で関係性を追跡する

RAGのベクトル検索が苦手な「関係性」の追跡をグラフDBで解決します。

# graph_memory.py
# グラフ構造でエンティティ間の関係を記録して検索する

import json
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, field, asdict

GRAPH_FILE = Path("knowledge_graph.json")

@dataclass
class Node:
    id: str
    type: str          # "person", "decision", "project", "concept"
    label: str
    properties: dict = field(default_factory=dict)
    created_at: str = field(default_factory=lambda: datetime.now().isoformat()[:10])

@dataclass
class Edge:
    source_id: str
    target_id: str
    relation: str      # "decided_by", "depends_on", "rejected", "approved"
    metadata: dict = field(default_factory=dict)

class KnowledgeGraph:
    def __init__(self):
        data = json.loads(GRAPH_FILE.read_text()) if GRAPH_FILE.exists() else {"nodes": {}, "edges": []}
        self.nodes: dict[str, Node] = {k: Node(**v) for k, v in data["nodes"].items()}
        self.edges: list[Edge] = [Edge(**e) for e in data["edges"]]
    
    def add_node(self, node: Node) -> None:
        self.nodes[node.id] = node
        self._save()
    
    def add_edge(self, edge: Edge) -> None:
        self.edges.append(edge)
        self._save()
    
    def get_related(self, node_id: str, max_depth: int = 2) -> list[Node]:
        """指定ノードから深さmax_depthまで関連ノードを取得する"""
        visited = set()
        result = []
        queue = [(node_id, 0)]
        
        while queue:
            current_id, depth = queue.pop(0)
            if current_id in visited or depth > max_depth:
                continue
            visited.add(current_id)
            
            if current_id in self.nodes:
                result.append(self.nodes[current_id])
            
            for edge in self.edges:
                if edge.source_id == current_id and edge.target_id not in visited:
                    queue.append((edge.target_id, depth + 1))
                elif edge.target_id == current_id and edge.source_id not in visited:
                    queue.append((edge.source_id, depth + 1))
        
        return result
    
    def get_decisions_by_person(self, person_id: str) -> list[Node]:
        """特定の人物が関わった意思決定をすべて取得する"""
        decision_ids = {
            e.target_id for e in self.edges
            if e.source_id == person_id and e.relation == "decided_by"
        }
        return [self.nodes[did] for did in decision_ids if did in self.nodes]
    
    def _save(self) -> None:
        GRAPH_FILE.write_text(json.dumps({
            "nodes": {k: asdict(v) for k, v in self.nodes.items()},
            "edges": [asdict(e) for e in self.edges]
        }, ensure_ascii=False, indent=2))

# 使用例: プロジェクトの意思決定を記録する
graph = KnowledgeGraph()

# エンティティを追加
graph.add_node(Node("user_alice", "person", "Alice（エンジニア）"))
graph.add_node(Node("proj_001", "project", "ECサイトリニューアル"))
graph.add_node(Node("dec_001", "decision", "PostgreSQL採用（MySQL→移行）"))
graph.add_node(Node("dec_002", "decision", "Next.js App Router採用"))

# 関係を追加
graph.add_edge(Edge("user_alice", "dec_001", "decided_by", {"date": "2026-04-15"}))
graph.add_edge(Edge("dec_001", "proj_001", "belongs_to"))
graph.add_edge(Edge("dec_002", "proj_001", "belongs_to"))

# クエリ: Aliceが関わった意思決定
alice_decisions = graph.get_decisions_by_person("user_alice")
print("Aliceの意思決定:", [d.label for d in alice_decisions])

解決策3：ConversationalRetrievalChainでRAG統合

LangChainの ConversationalRetrievalChain を使うと、会話履歴と外部知識（ドキュメント）を組み合わせたRAG（検索拡張生成）を実装できます。これは特にドキュメントQAやナレッジベース検索で威力を発揮します。

ステップ1: ドキュメントの準備

以下の手順でドキュメントを準備します。

# rag_setup.py
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from dotenv import load_dotenv
import os

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

# ステップ1: テキストファイルをロード
# 例：sample_docs.txt を準備
loader = TextLoader('sample_docs.txt', encoding='utf-8')
documents = loader.load()

# ステップ2: テキストを小さなチャンク（小片）に分割
text_splitter = CharacterTextSplitter(
    chunk_size=500,      # 1チャンク500文字
    chunk_overlap=50     # チャンク間で50文字重複させる
)
docs = text_splitter.split_documents(documents)

# ステップ3: OpenAI の埋め込み（embedding）を使ってベクトル化
embeddings = OpenAIEmbeddings(api_key=api_key)

# ステップ4: FAISS（ベクトルデータベース）にドキュメントを保存
vector_store = FAISS.from_documents(docs, embeddings)

# ステップ5: 検索用に保存
vector_store.save_local("faiss_index")

print("ドキュメントの準備完了。ベクトルデータベースを保存しました。")

ステップ2: ConversationalRetrievalChain の実装

# rag_conversation.py
import json
import os
from dotenv import load_dotenv
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.llms import OpenAI
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

# ステップ1: ベクトルデータベースを読み込む
embeddings = OpenAIEmbeddings(api_key=api_key)
vector_store = FAISS.load_local("faiss_index", embeddings)

# ステップ2: メモリを準備
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# ステップ3: LLM を初期化
llm = OpenAI(api_key=api_key, model="gpt-3.5-turbo", temperature=0.7)

# ステップ4: ConversationalRetrievalChain を作成
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vector_store.as_retriever(),
    memory=memory,
    verbose=True
)

# ステップ5: ユーザー入力ループ
print("RAG対応のQAチャット（メモリ機能付き）")
print("="*50)

while True:
    user_input = input("\nあなた: ").strip()
    
    if user_input.lower() in ['quit', 'exit']:
        print("終了します。")
        break
    
    if not user_input:
        continue
    
    # qa_chain で回答を生成
    result = qa_chain({"question": user_input})
    answer = result['answer']
    
    print(f"\nAI: {answer}")
    
    # 参照したドキュメントも表示（オプション）
    if "source_documents" in result:
        print("\n参考にしたドキュメント:")
        for doc in result['source_documents'][:2]:  # 最初の2つだけ表示
            print(f"  - {doc.page_content[:100]}...")

このコードの流れ：

1. ベクトルデータベース（FAISS）からドキュメントを検索
2. 過去の会話（メモリ）と検索結果を組み合わせて、LLMに入力
3. 常に最新のドキュメント知識に基づいた回答が得られる

解決策4：より高度なメモリ管理（SQLiteベース）

大規模なアプリケーションでは、JSONファイルではなくデータベースにメモリを保存するのが適切です。

# memory_sqlite.py
import sqlite3
import json
from datetime import datetime
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.llms import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

class SQLiteMemory:
    """SQLiteを使ったメモリ管理"""
    
    def __init__(self, db_path="conversation.db", user_id="default"):
        self.db_path = db_path
        self.user_id = user_id
        self._init_db()
    
    def _init_db(self):
        """テーブルを作成"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS conversation_memory (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                message TEXT NOT NULL,
                role TEXT NOT NULL,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.commit()
        conn.close()
    
    def add_message(self, message, role="user"):
        """メッセージをデータベースに追加"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO conversation_memory (user_id, message, role) VALUES (?, ?, ?)",
            (self.user_id, message, role)
        )
        conn.commit()
        conn.close()
    
    def get_history(self, limit=20):
        """会話履歴を取得"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "SELECT message, role FROM conversation_memory WHERE user_id = ? ORDER BY id DESC LIMIT ?",
            (self.user_id, limit)
        )
        rows = cursor.fetchall()
        conn.close()
        
        # 逆順にして時系列に
        return [(row[1], row[0]) for row in reversed(rows)]

# 使用例
db_memory = SQLiteMemory(user_id="user123")

# メモリから履歴を取得してLangChainメモリに組み込む
history = db_memory.get_history()
langchain_memory = ConversationBufferMemory()

# 過去のメッセージをメモリに復元
for role, message in history:
    if role == "user":
        langchain_memory.chat_memory.add_user_message(message)
    else:
        langchain_memory.chat_memory.add_ai_message(message)

# 通常通りConversationChainを実行
llm = OpenAI(api_key=api_key, model="gpt-3.5-turbo", temperature=0.7)
conversation = ConversationChain(llm=llm, memory=langchain_memory, verbose=True)

# 会話実行後、メッセージをDBに保存
user_input = "Pythonの最新トレンドを教えて"
response = conversation.predict(input=user_input)

db_memory.add_message(user_input, role="user")
db_memory.add_message(response, role="assistant")

print(response)

このアプローチの利点：

• 複数ユーザー対応: user_id でユーザーごとの会話を分離
• 永続性: SQLiteはファイルベースで、パソコン再起動後も保持
• スケーラビリティ: ファイルではなくDBなので、大量データに対応
• クエリ可能: 特定の日時や単語で検索可能

解決策5：スキーマ強制でエージェントを信頼できる状態にする

エージェントの出力を構造化して、後処理の信頼性を上げます。

# structured_agent.py
# Pydanticでエージェント出力のスキーマを強制する

import anthropic
import json
from pydantic import BaseModel, ValidationError
from typing import Literal

client = anthropic.Anthropic()

class AgentAction(BaseModel):
    action_type: Literal["search", "write", "notify", "ask_user", "done"]
    target: str
    content: str
    confidence: float  # 0.0〜1.0
    requires_confirmation: bool

class AgentResponse(BaseModel):
    reasoning: str
    next_action: AgentAction
    memory_to_save: list[str]  # 次のセッションに引き継ぐ情報

def run_structured_agent(task: str, context: str = "") -> AgentResponse | None:
    """スキーマ検証付きのエージェント実行"""
    system = f"""あなたはタスクを自律実行するエージェントです。
コンテキスト: {context}

必ず以下のJSON形式で出力してください:
{{
  "reasoning": "判断理由",
  "next_action": {{
    "action_type": "search|write|notify|ask_user|done",
    "target": "対象",
    "content": "実行内容",
    "confidence": 0.0〜1.0,
    "requires_confirmation": true|false
  }},
  "memory_to_save": ["次のセッションに引き継ぐ情報"]
}}"""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system,
        messages=[{"role": "user", "content": f"タスク: {task}"}]
    )
    
    cost = (response.usage.input_tokens * 3.00 + response.usage.output_tokens * 15.00) / 1_000_000
    print(f"エージェントコスト: ${cost:.5f}")
    
    try:
        data = json.loads(response.content[0].text)
        agent_response = AgentResponse(**data)  # Pydanticで型検証
        return agent_response
    except (json.JSONDecodeError, ValidationError) as e:
        print(f"スキーマ検証失敗: {e}")
        return None

class AgentRunner:
    """エージェントの実行を管理するクラス（後処理込み）"""
    
    def __init__(self, store: "SessionStore"):
        self.store = store
        self.execution_log: list[dict] = []
    
    def run(self, task: str, max_steps: int = 10) -> None:
        """エージェントを実行する（モデル返答後の後処理まで完了させる）"""
        context = self.store.build_context_prompt()
        
        for step in range(max_steps):
            print(f"\n=== Step {step + 1} ===")
            
            result = run_structured_agent(task, context)
            if not result:
                print("スキーマ検証失敗 → 終了")
                break
            
            print(f"判断: {result.reasoning}")
            print(f"次のアクション: {result.next_action.action_type}")
            
            # 1. メモリを更新（後処理）
            if result.memory_to_save:
                self.store.memory.long_term_facts.extend(result.memory_to_save)
                self.store.save()
                print(f"  メモリ保存: {result.memory_to_save}")
            
            # 2. 実行ログに記録（後処理）
            self.execution_log.append({
                "step": step + 1,
                "action": result.next_action.action_type,
                "target": result.next_action.target,
            })
            
            # 3. 確認が必要なアクションは一時停止
            if result.next_action.requires_confirmation:
                confirm = input(f"確認: {result.next_action.content} → 実行しますか？(y/n): ")
                if confirm != "y":
                    print("ユーザーが中断")
                    break
            
            # 4. 完了判定
            if result.next_action.action_type == "done":
                print("\nタスク完了（後処理まで完了）")
                break
        
        print(f"\n実行ログ: {len(self.execution_log)}ステップ")

3つの解決策の使い分け

シナリオ別の最適解:

┌─────────────────────────────────────────┬──────────────────────────┐
│ シナリオ                                 │ 推奨アプローチ            │
├─────────────────────────────────────────┼──────────────────────────┤
│ 個人プロジェクト・小規模チャットボット   │ JSONファイル（解決策1）   │
│ 複数ユーザー・複雑な関係性追跡が必要    │ グラフDB（解決策2）       │
│ 中規模: 複数ユーザー・複数プロジェ

---

## あわせて読みたい

- [Claude Codeで長期メモリ実装【複数セッション対応】会話履歴の活用法](/code/claude-code/)
- [Claude Code Context Window完全ガイド【2026年版】200K活用とContext超過エラーの解決法](/code/claude-code-context-window/)
- [LangChainで会話メモリを実装する3ステップ｜生成AIの履歴保存・永続化【コード付き】](/code/ai-3/)