LangChain クロスセッションストレージ —— Store

Checkpointer は「単一の会話内の記憶」の問題を解決しました。しかし、異なる会話間でデータを共有する必要がある場合——例えばユーザーの好みや学習進捗——には、使用する必要がありますStore。

Checkpointer vs Store

次元CheckpointerStore
スコープ単一の会話スレッド（thread_id）すべての会話スレッドにわたる
データ型Agent 状態のスナップショット（自動管理）任意のキーバリューデータ（手動管理）
典型的な用途マルチターン会話メモリユーザーの好み、ナレッジベース、構成
データの編成thread_id → checkpoint(namespace, key) → value

Store の基本操作

Store の使用名前空間 + キーの階層構造を使用してデータを編成します：

例

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# データの書き込み：put(namespace, key, value)
# namespace はタプル、key は文字列、value は辞書です
store.put(
    ("users", "user_001"),           # 名前空間
    "preferences",                    # キー
    {                                 # 値
        "theme": "dark",
        "language": "zh-CN",
        "level": "入門",
    }
)

store.put(
    ("users", "user_001"),
    "progress",
    {
        "completed_courses": ["HTML 基礎", "Python 基礎"],
        "total_hours": 35,
    }
)

# データの読み取り：get(namespace, key)
prefs = store.get(("users", "user_001"), "preferences")
print(f"ユーザー設定: {prefs.value}")

progress = store.get(("users", "user_001"), "progress")
print(f"学習進捗: {progress.value}")

# データの検索：search(namespace)
all_user_data = store.search(("users", "user_001"))
print(f"\nユーザーのすべてのデータ ({len(all_user_data)} 件):")
for item in all_user_data:
    print(f"  {item.key}: {item.value}")

# データの削除：delete(namespace, key)
store.delete(("users", "user_001"), "preferences")
deleted = store.get(("users", "user_001"), "preferences")
print(f"\n削除後: {deleted}")  # None

実行結果：

設定: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'}
学習進捗: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35}

用户的所有数据 (2 项):
  preferences: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'}
  progress: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35}

削除後: None

Agent で Store を使用する

Store を create_agent() に渡すと、Agent 内のすべてのツールが InjectedStore を介してアクセスできます：

例

from dotenv import load_dotenv
load_dotenv()

from typing import Annotated
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain.tools import tool, InjectedStore
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# Store を作成し、データをプリセットする
store = InMemoryStore()
store.put(("サンプル", "courses"), "catalog", {
    "Python3 基礎チュートリアル": {"price": "無料", "hours": 20, "level": "入門"},
    "Python データ分析": {"price": "会員", "hours": 30, "level": "上級"},
    "Java オブジェクト指向": {"price": "無料", "hours": 25, "level": "上級"},
})

store.put(("サンプル", "users"), "user_vip_001", {
    "name": "小明",
    "membership": "VIP",
    "joined": "2024-01-15",
})


@tool
def query_course_info(
    course_name: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """サンプル のコースの詳細情報を照会します。

    Args:
course_name: コース名
    """
    item = store.get(("サンプル", "courses"), "catalog")
    catalog = item.value if item else {}

    if course_name in catalog:
        info = catalog[course_name]
        return (
            f"《{course_name}》 - 価格：{info['price']}、"
            f"時間：{info['hours']}時間、難易度：{info['level']}"
        )
    return f"コース《{course_name}》が見つかりません"


@tool
def get_user_membership(
    user_id: str,
    store: Annotated[BaseStore, InjectedStore()],
) -> str:
    """ユーザーの会員情報を照会します。

    Args:
user_id: ユーザー ID
    """
    item = store.get(("サンプル", "users"), user_id)
    if item is None:
        return f"ユーザー {user_id} が見つかりません"

    user = item.value
    return (
        f"ユーザー {user['name']}、{user['membership']} 会員、"
        f"登録日 {user['joined']}"
    )


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[query_course_info, get_user_membership],
    store=store,
    system_prompt="あなたは サンプル のコースアドバイザーです。",
)

# コース情報を照会（データは Store から）
result = agent.invoke({
    "messages": [HumanMessage(content="Python3 基礎チュートリアルはいくらですか？")]
})
print(f"コース照会: {result['messages'][-1].content}")

# ユーザー情報を照会（データは Store から）
result = agent.invoke({
    "messages": [HumanMessage(content="ユーザー user_vip_001 の情報を調べてください")]
})
print(f"ユーザー照会: {result['messages'][-1].content}")

実行結果：

コース照会: 《Python3 基础教程》是免费的，学习时长约20小时，难度为入门级别。
ユーザー照会: 用户小明是 VIP 会员，注册日期为 2024年1月15日。

Store の永続化

InMemoryStore のデータはプログラム再起動後に失われます。本番環境では PostgresStore などの永続化ソリューションを使用できます：

例

# 開発段階
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

# 本番環境（PostgreSQL が必要）
# from langgraph.store.postgres import PostgresStore
# store = PostgresStore.from_conn_string("postgresql://...")

Store 使用のアドバイス

シナリオnamespace の例key の例説明
ユーザーの好み("users", user_id)preferencesテーマ、言語、通知設定
学習進捗("users", user_id)progress完了したコース、学習時間
ナレッジベース("kb", collection)doc_idドキュメント、FAQ、製品情報
会話サマリー("sessions", thread_id)summary長い会話の要約、Checkpointer 以外のシナリオで使用します

Checkpointer は「会話がどこまで進んだか」を担当し、Store は「ユーザーが誰か、何ができるか、何を好むか」を担当します。両者を組み合わせて使用することで、持続的な記憶を持つインテリジェントな Agent を構築できます。

その他の拡張