エージェントメモリシステム

エージェントメモリシステム は何をしますか？

メモリはインテリジェントエージェントの基盤です。これがないと、すべてのインタラクションがゼロから始まります。このスキルは、エージェントメモリのアーキテクチャ（短期（コンテキストウィンドウ）、長期（ベクターストア）、およびそれらを整理する認知アーキテクチャ）をカバーします。

重要な洞察：メモリとは単なる保存ではなく、検索です。100万もの事実を保存しても、適切なものを見つけられなければ意味がありません。チャンキング、埋め込み、および検索戦略によって、エージェントが覚えているか忘れるかが決まります。

この分野は断片化されており、用語も一貫していません。ここではCoALA認知アーキテクチャフレームワークを採用します。セマンティックメモリ（事実）、エピソードメモリ（経験）、および手続きメモリ（方法に関する知識）です。

原則

• メモリの品質 = 検索の品質であり、保存量ではない
• 保存のためではなく、検索のためにチャンク化する
• コンテキストの分離はメモリの敵である
• 適切な情報に適切なメモリタイプを使用する
• 古いメモリは減衰させる – すべてを永遠に保存すべきではない
• 本番環境の前に検索精度をテストする
• リアルタイムよりもバックグラウンドでのメモリ形成の方が優れている

機能

• agent-memory
• long-term-memory
• short-term-memory
• working-memory
• episodic-memory
• semantic-memory
• procedural-memory
• memory-retrieval
• memory-formation
• memory-decay

スコープ

• vector-database-operations → データエンジニア
• rag-pipeline-architecture → LLMアーキテクト
• embedding-model-selection → MLエンジニア
• knowledge-graph-design → 知識エンジニア

ツール

メモリフレームワーク

• LangMem (LangChain) - 使用場面：永続メモリを持つLangGraphエージェント。備考：セマンティック、エピソード、手続きメモリタイプ
• MemGPT / Letta - 使用場面：仮想コンテキスト管理、OS風メモリ。備考：階層型メモリ階層、自動ページング
• Mem0 - 使用場面：パーソナライゼーションのためのユーザーメモリ層。備考：ユーザーの好みと履歴向けに設計

ベクターストア

• Pinecone - 使用場面：管理型、エンタープライズスケール（数十億ベクトル）。備考：最高のクエリパフォーマンス、最高コスト
• Qdrant - 使用場面：複雑なメタデータフィルタリング、オープンソース。備考：Rustベース、優れたフィルタリング
• Weaviate - 使用場面：ハイブリッド検索、ナレッジグラフ機能。備考：GraphQLインターフェース、リレーションシップに優れる
• ChromaDB - 使用場面：プロトタイピング、小中規模アプリケーション。備考：開発者フレンドリー、10万ベクトルでp50約20ms
• pgvector - 使用場面：すでにPostgreSQLを使用している場合、セットアップが簡単。備考：100万ベクトル未満に最適、使い慣れたツール

埋め込みモデル

• OpenAI text-embedding-3-large - 使用場面：最高品質、3072次元。備考：$0.13/100万トークン
• OpenAI text-embedding-3-small - 使用場面：バランス良好、1536次元。備考：$0.02/100万トークン、5倍安い
• nomic-embed-text-v1.5 - 使用場面：オープンソース、ローカルデプロイメント。備考：768次元、良好な品質
• all-MiniLM-L6-v2 - 使用場面：軽量、高速ローカル埋め込み。備考：384次元、最低遅延

パターン

メモリタイプアーキテクチャ

異なる情報に適したメモリタイプを選択する

使用場面：エージェントメモリシステムの設計

メモリタイプアーキテクチャ（CoALAフレームワーク）

""" 目的に応じた3つのメモリタイプ：

1. セマンティックメモリ：事実と知識

   • 世界について知っていること
   • ユーザーの好み、ドメイン知識
   • プロファイル（構造化）またはコレクション（非構造化）に保存

2. エピソードメモリ：経験と出来事

   • 起こったこと（タイムスタンプ付きイベント）
   • 過去の会話、タスクの結果
   • 経験から学習するために使用

3. 手続きメモリ：物事のやり方

   • ルール、スキル、ワークフロー
   • 多くの場合、少数ショット例として実装
   • 「以前はこれをどう解決したか？」 """

LangMemの実装

from langmem import MemoryStore
from langgraph.graph import StateGraph

# メモリストアの初期化
memory = MemoryStore(
    connection_string=os.environ["POSTGRES_URL"]
)

# セマンティックメモリ：ユーザープロファイル
await memory.semantic.upsert(
    namespace="user_profile",
    key=user_id,
    content={
        "name": "Alice",
        "preferences": ["ダークモード", "簡潔な応答"],
        "expertise_level": "開発者",
    }
)

# エピソードメモリ：過去のインタラクション
await memory.episodic.add(
    namespace="conversations",
    content={
        "timestamp": datetime.now(),
        "summary": "認証問題のデバッグを支援",
        "outcome": "解決済み",
        "key_insights": ["トークンの有効期限が根本原因だった"],
    },
    metadata={"user_id": user_id, "topic": "debugging"}
)

# 手続きメモリ：学習したパターン
await memory.procedural.add(
    namespace="skills",
    content={
        "task_type": "debug_auth",
        "steps": ["トークンの有効期限を確認", "リフレッシュフローを検証"],
        "example_interaction": few_shot_example,
    }
)

ランタイムでのメモリ検索

async def prepare_context(user_id, query):
    # ユーザープロファイルを取得（セマンティック）
    profile = await memory.semantic.get(
        namespace="user_profile",
        key=user_id
    )

    # 関連する過去の経験を見つける（エピソード）
    similar_experiences = await memory.episodic.search(
        namespace="conversations",
        query=query,
        filter={"user_id": user_id},
        limit=3
    )

    # 関連するスキルを見つける（手続き）
    relevant_skills = await memory.procedural.search(
        namespace="skills",
        query=query,
        limit=2
    )

    return {
        "profile": profile,
        "past_experiences": similar_experiences,
        "relevant_skills": relevant_skills,
    }

ベクターストア選択パターン

ユースケースに適したベクターデータベースを選択する

使用場面：永続メモリストレージのセットアップ

ベクターストアの選択

""" 意思決定マトリックス：

	Pinecone	Qdrant	Weaviate	ChromaDB	pgvector
スケール	数十億	1億以上	1億以上	100万	100万
マネージド	はい	両方	両方	セルフ	セルフ
フィルタリング	基本	最良	良好	基本	SQL
ハイブリッド	いいえ	はい	最良	いいえ	はい
コスト	高	中	中	無料	無料
レイテンシ	5ms	7ms	10ms	20ms	15ms
"""

Pinecone（エンタープライズスケール）

from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("agent-memory")

# メタデータ付きでアップサート
index.upsert(
    vectors=[
        {
            "id": f"memory-{uuid4()}",
            "values": embedding,
            "metadata": {
                "user_id": user_id,
                "timestamp": datetime.now().isoformat(),
                "type": "episodic",
                "content": memory_text,
            }
        }
    ],
    namespace=namespace
)

# フィルター付きでクエリ
results = index.query(
    vector=query_embedding,
    filter={"user_id": user_id, "type": "episodic"},
    top_k=5,
    include_metadata=True
)

Qdrant（複雑なフィルタリング）

from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, Filter, FieldCondition

client = QdrantClient(url="http://localhost:6333")

# Qdrantを使った複雑なフィルタリング
results = client.search(
    collection_name="agent_memory",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="user_id", match={"value": user_id}),
            FieldCondition(key="type", match={"value": "semantic"}),
        ],
        should=[
            FieldCondition(key="topic", match={"any": ["auth", "security"]}),
        ]
    ),
    limit=5
)

ChromaDB（プロトタイピング）

import chromadb

client = chromadb.PersistentClient(path="./memory_db")
collection = client.get_or_create_collection("agent_memory")

# プロトタイプ向けにシンプルで高速
collection.add(
    ids=[str(uuid4())],
    embeddings=[embedding],
    documents=[memory_text],
    metadatas=[{"user_id": user_id, "type": "episodic"}]
)

results = collection.query(
    query_embeddings=[query_embedding],
    n_results=5,
    where={"user_id": user_id}
)

チャンキング戦略パターン

ドキュメントを検索可能なチャンクに分割する

使用場面：メモリストレージ用のドキュメント処理

チャンキング戦略

""" チャンキングのジレンマ：

• 大きすぎる：ベクトルが特異性を失う
• 小さすぎる：コンテキストを失う

最適なチャンクサイズは以下に依存する：

• ドキュメントタイプ（コード vs 散文 vs データ）
• クエリパターン（事実ベース vs 探索的）
• 埋め込みモデル（それぞれに適切な範囲がある）

一般的なガイドライン：ほとんどのユースケースで256～512トークン """

固定サイズチャンキング（ベースライン）

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 文字数
    chunk_overlap=50,    # オーバーラップにより文中での分割を防止
    separators=["\n\n", "\n", ". ", " ", ""]  # 優先順位
)

chunks = splitter.split_text(document)

セマンティックチャンキング（より高品質）

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# セマンティック類似性に基づいて分割
splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95
)

chunks = splitter.split_text(document)

構造認識チャンキング（階層を持つドキュメント）

from langchain.text_splitter import MarkdownHeaderTextSplitter

# ドキュメント構造を尊重
splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
)

chunks = splitter.split_text(markdown_doc)
# 各チャンクにはコンテキスト用のヘッダーメタデータが含まれる

コンテキストチャンキング（Anthropicのアプローチ）

# 埋め込み前に各チャンクにコンテキストを追加
# 検索失敗率を35%削減

def add_context_to_chunk(chunk, document_summary):
    context_prompt = f'''
    ドキュメントの要約：{document_summary}

    以下はこのドキュメントのチャンクです：
    {chunk}
    '''
    return context_prompt

# 生のチャンクではなく、コンテキスト化されたチャンクを埋め込む
for chunk in chunks:
    contextualized = add_context_to_chunk(chunk, summary)
    embedding = embed(contextualized)
    store(chunk, embedding)  # 元のチャンクを保存、コンテキスト化された方を埋め込み

コード固有のチャンキング

from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

# 言語認識による分割
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=200
)

# 関数/クラスの境界を尊重
chunks = python_splitter.split_text(python_code)

バックグラウンドメモリ形成

より良い品質のために非同期でメモリを処理する

使用場面：インタラクションを遅くせずに高い再現率を実現したい場合

バックグラウンドメモリ形成

""" リアルタイムのメモリ抽出は会話を遅くし、エージェントのツール呼び出しに複雑さを追加します。 会話後のバックグラウンド処理により、より高品質なメモリが生成されます。

パターン：潜在意識的なメモリ形成 """

LangGraphによるバックグラウンド処理

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver

async def background_memory_processor(thread_id: str):
    # 会話終了後またはアイドル時に実行
    conversation = await load_conversation(thread_id)

    # 時間制約なしでインサイトを抽出
    insights = await llm.invoke('''
        この会話を分析し、以下を抽出してください：
        1. ユーザーについて学んだ重要な事実
        2. 明らかになったユーザーの好み
        3. 完了または保留中のタスク
        4. ユーザーの行動パターン

        これはバックグラウンドで実行されるため、徹底的に行ってください。

        会話：
        {conversation}
    ''')

    # 長期メモリに保存
    for insight in insights:
        await memory.semantic.upsert(
            namespace="user_insights",
            key=generate_key(insight),
            content=insight,
            metadata={"source_thread": thread_id}
        )

# 会話終了時またはアイドルタイムアウト時にトリガー
@on_conversation_idle(timeout_minutes=5)
async def process_conversation(thread_id):
    await background_memory_processor(thread_id)

メモリ統合（睡眠のように）

# 定期的にメモリを統合し重複排除する

async def consolidate_memories(user_id: str):
    # ユーザーのすべてのメモリを取得
    memories = await memory.semantic.list(
        namespace="user_insights",
        filter={"user_id": user_id}
    )

    # 類似メモリを見つける（潜在的な重複）
    clusters = cluster_by_similarity(memories, threshold=0.9)

    # 類似メモリを統合
    for cluster in clusters:
        if len(cluster) > 1:
            merged = await llm.invoke(f'''
                これらの関連するメモリを1つに統合してください：
                {cluster}

                すべての重要な情報を保持してください。
            ''')
            await memory.semantic.upsert(
                namespace="user_insights",
                key=generate_key(merged),
                content=merged
            )
            # 元のメモリを削除
            for old in cluster:
                await memory.semantic.delete(old.id)

メモリ減衰パターン

古くて無関係なメモリを忘れる

使用場面：メモリが大きくなり、検索が遅くなる場合

メモリ減衰

""" すべてのメモリが永遠に生き続けるべきではない：

• 古い好みは時代遅れかもしれない
• タスクの詳細は関連性を失う
• 矛盾するメモリは検索を混乱させる

以下に基づいたインテリジェントな減衰を実装する：

• 新しさ（いつ作成/アクセスされたか？）
• 頻度（どれくらいの頻度で検索されるか？）
• 重要性（核となる事実か、それとも詳細か？） """

時間ベースの減衰

from datetime import datetime, timedelta

async def decay_old_memories(namespace: str, max_age_days: int):
    cutoff = datetime.now() - timedelta(days=max_age_days)

    old_memories = await memory.episodic.list(
        namespace=namespace,
        filter={"last_accessed": {"$lt": cutoff.isoformat()}}
    )

    for mem in old_memories:
        # ソフトデリート（アーカイブ済みとしてマーク）
        await memory.episodic.update(
            id=mem.id,
            metadata={"archived": True, "archived_at": datetime.now().isoformat()}
        )