LangMemの3種類の長期記憶（セマンティック・エピソディック・プロシージャル）をLiteLLM経由で使い分けてみる

Python 3.13
litellm 1.83.14
langgraph 1.1.10
langmem 0.0.21
langchain-litellm 0.6.4
langchain-core 1.3.2

$ uv pip install litellm langgraph langmem langchain-litellm langchain-core
$ export OPENAI_API_KEY="sk-..."  # Embedding用
$ export ANTHROPIC_API_KEY="sk-ant-..."

"""セマンティックメモリ: ユーザーに関する事実をベクトル検索で取り出す。

LangMem の create_memory_store_manager / create_search_memory_tool を使い、
会話から自動的に事実を抽出してストアに保存し、関連時に検索する。
"""

from langchain.agents import create_agent
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langmem import create_manage_memory_tool, create_search_memory_tool

# ストアと checkpointer
# Store: 長期記憶（スレッド横断）
# Checkpointer: 短期記憶（スレッド内の会話履歴）
store = InMemoryStore(
    index={"dims": 1536, "embed": "openai:text-embedding-3-small"},
)
checkpointer = InMemorySaver()

llm = ChatLiteLLM(model="openai/gpt-5-mini")

# LangMem の標準ツール: 記憶の保存と検索
# namespace で「誰の記憶か」を指定する
NAMESPACE = ("memories", "{user_id}")
agent = create_agent(
    model=llm,
    tools=[
        create_manage_memory_tool(namespace=NAMESPACE),
        create_search_memory_tool(namespace=NAMESPACE),
    ],
    system_prompt=(
        "ユーザーに関する重要な事実は manage_memory で保存し、"
        "次回以降の会話で必要なら search_memory で検索して活用してください。"
    ),
    checkpointer=checkpointer,
    store=store,
)

def chat(user_id: str, text: str) -> str:
    config = {"configurable": {"thread_id": user_id, "user_id": user_id}}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": text}]},
        config=config,
    )
    return str(result["messages"][-1].content)

# 1日目: ユーザーが好みを伝える
print("=== Day 1 ===")
print("AI:", chat("alice", "私はマグロが好きです。アレルギーはエビです。"))

# 2日目: 別スレッドで好みを参照する（Checkpointerは短期なので参照できないが、Storeから検索される）
print("\n=== Day 2（新スレッド） ===")
config = {"configurable": {"thread_id": "alice-day2", "user_id": "alice"}}
result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "今日のおすすめのお寿司を教えてください"}
        ]
    },
    config=config,
)
print("AI:", result["messages"][-1].content)

$ python semantic_memory.py
=== Day 1 ===
AI: 情報ありがとうございます。以下を記録しました：
- 好きな食べ物：マグロ
- アレルギー：エビ（甲殻類）

今後の提案や注意点でこれを考慮します。マグロのレシピ、エビを避けた外食時の注意、代替食材の提案など、何かご希望があれば教えてください。

=== Day 2（新スレッド） ===
AI: いいですね！おすすめネタ
- 鰹（かつお）── 春〜初夏の戻り鰹は香りと脂が良い。
- 鯵（あじ）── 身がしっかりして脂がのる季節。
- 中トロ（ちゅうとろ）── 脂の旨味を楽しみたいときの定番。
- 帆立（ほたて）── 甘みが強く、炙りや生のままでも美味。
（以下略）

"""エピソディックメモリ: 過去の会話エピソードを丸ごと記憶し、類似シーンで参照する。

「以前ユーザーが似た質問をしたとき、どう答えたか」を記憶として残す。
ここでは LangMem の ReflectionExecutor を使って会話終了後に
バックグラウンドで記憶を抽出する例を示す。
"""

import litellm
from langchain_core.messages import HumanMessage
from langchain_litellm import ChatLiteLLM
from langgraph.store.memory import InMemoryStore
from langmem import ReflectionExecutor, create_memory_store_manager

litellm.modify_params = True

# 応答用は安価なGPT、記憶抽出（manager）はClaudeに任せる
llm = ChatLiteLLM(model="openai/gpt-5-mini")

store = InMemoryStore(
    index={"dims": 1536, "embed": "openai:text-embedding-3-small"},
)

# create_memory_store_manager は会話メッセージから自動で記憶を抽出する
manager = create_memory_store_manager(
    "anthropic:claude-sonnet-4-6",
    namespace=("episodes", "{user_id}"),
    instructions=(
        "ユーザーの質問の傾向や、どのように答えると喜ばれたか、"
        "といったエピソードを記憶として抽出してください。"
    ),
)

# ReflectionExecutor: 会話の後ろでバックグラウンドに記憶抽出を走らせる
reflection = ReflectionExecutor(manager, store=store)

def turn(messages: list, user_id: str) -> str:
    """1ターンの会話を実行し、その後エピソードを抽出する。"""
    response = llm.invoke(messages)
    new_messages = [*messages, response]

    # バックグラウンドでエピソード抽出を依頼
    config = {"configurable": {"user_id": user_id}}
    reflection.submit(
        {"messages": new_messages},
        config=config,
        after_seconds=0,
    )
    return str(response.content)

# サンプル会話
messages = [
    HumanMessage(content="Pythonでリスト内包表記を簡潔に書く方法を教えてください")
]
ans = turn(messages, "alice")
print("AI:", ans[:200])

# 背景で動いている記憶抽出ジョブが完了するまで待つ
# wait=True でキュー内の全ジョブの完了を待ってからシャットダウンする
reflection.shutdown(wait=True)

items = store.search(("episodes", "alice"))
print(f"\n抽出されたエピソード: {len(items)} 件")
for item in items[:3]:
    print(f"  - {item.value}")

$ python episodic_memory.py
AI: いいですね。Python のリスト内包表記（list comprehension）は短くて速く書けますが、
可読性を損なわないことが大事です。基本と実用的なテクニックを簡潔にまとめます。
（中略）

抽出されたエピソード: 3 件
  - {'kind': 'Memory', 'content': {'content': 'ユーザーはPythonに関する質問をしている（リスト内包表記の簡潔な書き方）。Pythonの実用的なコーディングテクニックに興味がある。'}}
  - {'kind': 'Memory', 'content': {'content': 'ユーザーへの回答スタイルとして、「具体的なコード例付きで網羅的に説明する」「最後に『具体的なコードを見せてくれればさらに対応する』と促す」形式が効果的だった。'}}
  - {'kind': 'Memory', 'content': {'content': 'Pythonのリスト内包表記について質問された際、基本構文・よく使うパターン・ウォルラス演算子などの発展テクニック・ベストプラクティスを網羅的に例付きで説明したところ、会話が自然に進んだ。'}}

"""プロシージャルメモリ: エージェントの「振る舞い方（プロンプト）」を記憶として更新する。

ユーザーフィードバックを受けてシステムプロンプトを段階的に改善していく。
LangMem の create_prompt_optimizer を使う。
"""

import litellm
from langchain_core.messages import AIMessage, HumanMessage
from langchain_litellm import ChatLiteLLM
from langmem import create_prompt_optimizer

litellm.modify_params = True

# 応答用はGPT、プロンプト最適化（optimizer）はClaudeに任せる
llm = ChatLiteLLM(model="openai/gpt-5-mini")
optimizer = create_prompt_optimizer(
    "anthropic:claude-sonnet-4-6",
    kind="prompt_memory",
)

initial_prompt = "あなたは丁寧なアシスタントです。質問に答えてください。"

# 過去のフィードバック付き会話
trajectories = [
    (
        [
            HumanMessage(content="プログラミングの質問があります"),
            AIMessage(content="どのような質問ですか？お気軽にどうぞ"),
        ],
        {"feedback": "もっとコード例を示してほしかった"},
    ),
    (
        [
            HumanMessage(content="JSONのパースについて"),
            AIMessage(content="Pythonでは json モジュールを使います。"),
        ],
        {"feedback": "実際のコード例があると助かる"},
    ),
]

# プロンプトをフィードバックを反映して最適化
better_prompt = optimizer.invoke(
    {"trajectories": trajectories, "prompt": initial_prompt}
)
print("=== 最適化前のプロンプト ===")
print(initial_prompt)
print("\n=== 最適化後のプロンプト ===")
print(better_prompt)

# 最適化後のプロンプトで実際に応答してみる
response = llm.invoke(
    [
        {"role": "system", "content": str(better_prompt)},
        {"role": "user", "content": "JSONのパースを教えてください"},
    ]
)
print("\n=== 最適化後のプロンプトでの応答 ===")
print(response.content)

$ python procedural_memory.py
=== 最適化前のプロンプト ===
あなたは丁寧なアシスタントです。質問に答えてください。

=== 最適化後のプロンプト ===
あなたは丁寧なアシスタントです。質問に答えてください。プログラミングに関する質問には、必ず具体的なコード例を示してください。コード例はわかりやすく、実際に動作するものを提供してください。

=== 最適化後のプロンプトでの応答 ===
JSONのパース（解析）とは、JSON形式の文字列をプログラム内で扱えるデータ構造（オブジェクト、辞書、構造体など）に変換することです。
（中略）

2) Python（標準 json モジュール）
- 文字列/ファイル読み込み
\`\`\`python
import json
json_text = '{"name": "Bob", "age": 25, "skills": ["python", "sql"]}'
try:
    obj = json.loads(json_text)
    print(obj['name'])  # Bob
except json.JSONDecodeError as e:
    print("JSON parse error:", e)
\`\`\`
（以下略：JavaScript / Go / Java / C# のサンプルも続けて生成される）