エージェントシリーズ（六）：メモリ管理——エージェントに重要なことを覚えさせる

記憶は、Agent を「ツール」から「助手」へと変える

記憶のない Agent は、毎回会話がゼロから始まる。あなたが彼に「李雷だよ、Python 工程師だよ、手を動かすのが好きだよ」と教えても、次回の会話では全く覚えていない

これはバグではなく、アーキテクチャの欠如だ

LLM 自体には持続的な記憶がない。毎回の呼び出しは状態を持たない。Agent に何かを覚えてもらうためには、アーキテクチャレベルで明示的に記憶を格納し、管理し、読み取る記憶を必要とする。これが記憶管理モジュールが解決すべき課題だ

本稿では、Agent の記憶を四つの次元から解体する：記憶タイプの分類モデル、コンテキスト管理の三つの戦略、LangGraph が提供する二つの記憶メカニズム（checkpointer と store）、および超長会話の自動圧縮方案

四つの記憶タイプ：認知科学からエンジニアリング実現

認知科学における人間の記憶の分類を参考に、エージェントの記憶も四層に分けられ、LangGraphではそれぞれ対応する実装方法があります：

┌──────────────────────────────────────────────────────────┐
│                     记忆层级                              │
├──────────────────────┬───────────────────────────────────┤
│ 感觉记忆 (Sensory)   │ 当前 Turn 的 in-flight 消息       │
│                      │ 生命周期：单次 LLM 调用            │
├──────────────────────┼───────────────────────────────────┤
│ 工作记忆 (Working)   │ 对话历史 MessageHistory（有限 K 轮）│
│                      │ 实现：messages 列表注入 Prompt     │
├──────────────────────┼───────────────────────────────────┤
│ 情景记忆 (Episodic)  │ 向量化/摘要化的历史片段            │
│                      │ 实现：摘要压缩 + VectorStore 检索  │
├──────────────────────┼───────────────────────────────────┤
│ 语义记忆 (Semantic)  │ 长期存储的用户偏好、事实           │
│                      │ 实现：LangGraph store (KV Store)  │
└──────────────────────┴───────────────────────────────────┘

感覚記憶：現在のターンメッセージ

最も短い記憶。一度のLLMコールの入力と出力、使い捨てです：

q = "Python 中 len([1, 2, 3]) 等于多少"
answer = llm.invoke([HumanMessage(q)])
# answer.content → "len([1, 2, 3]) 的结果等于 3。"
# 这次 invoke 结束后，这个 answer 就消失了

感覚記憶を「管理」するための機構は不要——それはLLMコールそのものです。

作業記憶：限られた会話履歴

これまでの数ラウンドの会話メッセージをプロンプトに組み込み、最も直接的な記憶の実装です。効果はすぐに現れます：

history = [
    HumanMessage("我叫李雷，是一名 Python 工程师"),
    AIMessage("你好，李雷！很高兴认识你。"),
    HumanMessage("我最近在学习 LangGraph"),
    AIMessage("LangGraph 很强大，特别适合构建有状态的 Agent。"),
]
test_q = "我之前告诉你我叫什么名字？"

実測比較：

有历史 → "是的，你之前告诉我你的名字是李雷，并且你是一名 Python 工程师..."
无历史 → "抱歉，我无法回忆起您之前告诉我的名字，因为作为一个 AI，我没有
          持久的记忆功能来存储个人数据..."

差は非常に直感的です。作業記憶の制限は、トークンコストが会話の長さに比例して増加することです。切り詰めや要約を組み合わせて管理する必要があります。

情景記憶：要約された歴史の断片

会話の履歴が長くなると、その全てを直接プロンプトに突っ込むコストが高い。情景記憶のアプローチはまず圧縮し、次に保存である：

long_history = history * 4  # 16 条消息
summary = llm.invoke([
    SystemMessage("将以下对话压缩为 60 字以内的摘要，保留关键信息"),
    HumanMessage(str([m.content for m in long_history])),
])
# → "李雷，Python工程师，积极学习LangGraph，赞其强大，适合构建有状态Agent。"

16条のメッセージを28文字に圧縮し、次のラウンドでは要約で元の履歴の代わりにする。これにより、トークン消費が大幅に削減される。

意味記憶：跨会話のユーザーファクト

最も持続性のある記憶層。会話の終了に伴って消えず、ユーザーの長期的な事実（名前、職業、好みなど）を専用に保存する：

# 把用户信息存入 KV Store，下一次会话直接读取
user_profile = {
    "name": "李雷",
    "role": "Python 工程师",
    "interests": ["LangGraph", "Agent 开发"],
    "level": "中级",
}
# 基于这些信息，Agent 能给出个性化的回答
# "下一步学习方向" → 推荐 LangGraph 进阶用法而不是 Python 入门

文脈管理の三つの戦略：切り詰め / 要約 / 検索

会話の履歴がますます長くなり、文脈ウィンドウが対応しきれない場合どうする？三つの戦略にはそれぞれトレードオフがある：

戦略一：切り詰め（Truncation）

最も単純で強引な方法——最近N条のメッセージを残し、残りはすべて捨てる：

# 保留最近 4 条消息
truncated = history[-4:]
resp = llm.invoke(truncated + [HumanMessage(test_q)])

実際のテストで8つのトピック（16メッセージ）の歴史を使用し、最近4つまで切り詰めて「Pythonのリストは何ですか」と尋ねます：

截断后最早可见：'解释一下 Python 装饰器'（第 5 个主题，"列表"在第 1 个）
回答：Python 中的列表是一种内置的数据结构... （靠 LLM 自身知识回答）
⚠ 丢失了我们"学过"列表这个事实，LLM 只是在用通用知识回答

適用シナリオ：歴史の連続性を要求しないシナリオ、または歴史自体が重要でない純粋な質問応答型のエージェント。

戦略2：要約（Summarization）

LLMを使用して長い歴史を1つの要約に圧縮し、以降の対話では要約を原始歴史の代わりに使用します：

summary_resp = llm.invoke([
    SystemMessage("将对话历史压缩为一段摘要（不超过 80 字），保留所有已介绍的主题名称"),
    HumanMessage("\n".join([f"{m.type}: {m.content}" for m in history])),
])
# → "Python列表可变有序，元组不可变省内存，字典键值对映射，集合唯一元素，
#    函数封装逻辑，类面向对象，装饰器函数包装，生成器惰性计算。"
# 16 条消息 → 66 字摘要

同じく「Pythonのリストは何ですか」と尋ねても、要約方案の回答は「これが私たちの議論したトピックである」と示すことができ、単なる一般的な知識ではありません。

3つの戦略の比較：

戦略	トークン消費	情報保持	複雑さ	適用シーン
切断	最低	最近のみ	非常に低い	純粋な質問応答、歴史は重要でない
要約	低	全体の骨格	中	教育、相談、長期的な会話
検索	最低	正確に関連	高（ベクトルライブラリが必要）	知識ベース、多分野のエージェント

戦略三：検索（Retrieval）

は現在の問題の意味に関連する過去の断片のみを引き出す——最も効果的で最も複雑な方案である：

# 简化演示：按关键词过滤（生产中用向量相似度）
relevant = [m for m in history if "列表" in m.content or "list" in m.content.lower()]
# 16 条 → 3 条相关历史
resp = llm.invoke(relevant + [HumanMessage(test_q)])

適用シーン：知識ベース型 Agent、ユーザーに大量の履歴がある個別のアシスタント。

LangGraph checkpointer：セッション内状態の持続化

LangGraph の MemorySaver（checkpointer）は thread_id を用いて異なるセッションを区別し、同一セッション内で自動的に会話履歴を蓄積する：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

checkpointer = MemorySaver()
agent = create_react_agent(model=llm, tools=[get_weather], checkpointer=checkpointer)

# 同一 thread_id = 同一个会话，状态持久化
config_a = {"configurable": {"thread_id": "weather_001"}}

跨輪引用実測

セッション内連続三轮の天気問い合わせ：

[Turn 1] 用户: 北京今天天气怎么样？
         Agent: 北京今天的天气是晴，温度为 25°C，东北风 3 级，空气良好。
         (当前状态中消息数: 4)

[Turn 2] 用户: 那上海呢？          ← "那" 字没有明确指代，需要上文理解
         Agent: 上海今天的天气是多云，22°C，东南风 2 级，轻度雾霾。
         (当前状态中消息数: 8)

[Turn 3] 用户: 这两个城市哪个更适合今天出行？   ← 需要前两轮的查询结果
         Agent: 考虑到上海的雾霾情况，建议您在北京出行。
         (当前状态中消息数: 10)

ターン2とターン3の回答はすべて前の履歴に依存しており、checkpointerが自動的に跨りターンの文脈管理を完了しました。

セッション分離

異なるthread_id間は完全に独立しています：

[新会话 - thread_id: weather_002]
用户: 我刚才问的是哪个城市？
Agent: 您刚才问的是"哪个城市"，但是没有提供具体的城市名称。
      如果您需要查询某个城市的天气，请告诉我具体的城市名称。
→ 新 thread_id 没有任何历史，完全不知道刚才问过什么

セッションは

同じthread_idに隔った時間後に戻ってきても、履歴は残ります：

[会话 A 继续 - 同一 thread_id]
用户: 刚才查的两个城市，再查一下深圳对比一下
Agent: 深圳今天的天气是阵雨，27°C，西南风 2 级，同时还有雷暴预警。
       （进行了北京/上海/深圳三城比较）
→ 记住了前面查过北京和上海

MemorySaverはメモリ実装で、プロセスの再起動後データは失われます。生産環境ではSqliteSaver（ローカルファイル）やPostgresSaver（データベース）を使って代替します。

LangGraph InMemoryStore：跨セッションの長期記憶

checkpointerは単一セッション内の記憶を解決します。跨セッションの長期記憶が必要です。store：

checkpointer  →  绑定 thread_id，会话生命周期内有效
store         →  绑定 user_id，跨会话永久存在

の核心API

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 写入：(namespace, key, value)
store.put(("user_facts", user_id), key, {"fact": "李雷，后端工程师"})

# 读取：搜索某个 namespace 下的所有条目
facts = store.search(("user_facts", user_id))
for item in facts:
    print(item.value["fact"])

# 精确读取
item = store.get(("user_facts", user_id), specific_key)

会話間記憶のテスト

会話Aで、エージェントは自動的に会話からユーザー情報を抽出し、ストアに格納します：

[会话 A] 用户说了三句话 → 自动提取并存储：
  • 李雷，后端工程师
  • Python, Go, LangGraph, Agent 开发
  • 动手实践，不喜欢纯看文档

全く異なる会話B（新しいthread_id、しかし同じuser_id）で「あなたは私を知っていますか？」と尋ねます：

[会话 B - 全新 thread_id]
用户: 你好，你认识我吗？
Agent: 你好！根据你提供的信息，我认识你。你是李雷，一位后端工程师，
       擅长使用 Python、Go、LangGraph 和 Agent 进行开发。
       你更喜欢动手实践，而不是仅仅阅读文档。有什么可以帮助你的吗？
→ 虽然是全新 thread_id，但 store 中的用户信息跨会话持久

異なるuser_idのデータは完全に分離され、互いに干渉しません。

checkpointerとstoreの比較

# 短期记忆：checkpointer — 绑定 thread_id，会话内有效
app = graph.compile(checkpointer=MemorySaver())
result = app.invoke(input, config={"configurable": {"thread_id": "abc"}})

# 长期记忆：store — 绑定 user_id，跨会话有效
store = InMemoryStore()
app = graph.compile(store=store, checkpointer=MemorySaver())
# 在 node 中操作 store
store.put(("user_facts", user_id), key, {"fact": "..."})
stored = store.search(("user_facts", user_id))

本番環境でInMemoryStoreをPostgresStoreまたはRedisStoreに置き換えるだけで、本物の永続化が可能です。

自動要約圧縮：RemoveMessage + 摘要交代

メッセージ数が閾値を超えた時、自動圧縮がトリガーされる——これは無限に長い会話の中でエージェントが覚醒し続けるための核心的なメカニズムである。

図構造

[chat 节点]
    │
    ├─ 消息数 ≤ 阈值 → END
    └─ 消息数 > 阈值 → [compress 节点] → END

compress ノード：RemoveMessage が古いメッセージを削除

def compress_node(state: SummaryState) -> dict:
    messages = state["messages"]
    to_compress = messages[:-2]   # 保留最新 2 条，其余全部压缩
    keep = messages[-2:]

    # 旧消息 → 新摘要
    new_summary = llm.invoke([
        SystemMessage("将以下内容压缩为 120 字以内的摘要"),
        HumanMessage(existing_summary + old_messages_text),
    ]).content

    # RemoveMessage：告诉 add_messages reducer 删除这些消息
    remove_ops = [RemoveMessage(id=m.id) for m in to_compress]
    return {"messages": remove_ops, "summary": new_summary}

RemoveMessage は LangGraph専用のメッセージ削除オペレーターであり、add_messages reducer がそれを見ると、対応するidのメッセージを状態から削除する。

実測効果

11ラウンドの会話で、閾値を8つのメッセージに設定：

[Turn 1-4]  消息数:  2/4/6/8  | 摘要: ○ 无

  [压缩触发] 10 条 → 压缩 8 条，保留 2 条
  [新摘要]   Python列表常用方法包括查找、排序、添加删除等。
             `dict.get()` 避免 `KeyError`，返回默认值。
             `*args` 接收任意位置参数，`**kwargs` 接收任意关键字参数...

[Turn 5]    消息数:  2  | 摘要: ✓ 已压缩  ← 压缩后从 2 条重新开始

  [压缩再次触发] 10 条 → 压缩 8 条，保留 2 条

[Turn 11]  最终汇总：
           "根据我们之前的讨论，以下是您掌握的 Python 知识点汇总：
            1. Python 列表推导式...
            2. 集合推导式...  ← 通过摘要链条传承，第 1 轮的内容还在
            3. Lambda 函数..."

主要な結果：11ラウンドの会話は常に2-8つのactiveメッセージしか持たない、しかし要約チェーンを通じて、すべての歴史知識は継承されてきました。

要約状態設計

class SummaryState(TypedDict):
    messages: Annotated[list, add_messages]  # add_messages 处理 RemoveMessage
    summary: Optional[str]                   # 累积的历史摘要，注入 system prompt

def chat_node(state: SummaryState) -> dict:
    summary = state.get("summary") or ""
    system_parts = ["你是助手。"]
    if summary:
        system_parts.append(f"\n\n【历史摘要】{summary}")  # 摘要注入 system prompt
    resp = llm.invoke([SystemMessage("".join(system_parts))] + state["messages"])
    return {"messages": [resp]}

記憶管理設計リスト

完全なエージェント記憶システムを実装する際に考慮すべき点：

短期記憶（チェックポイントャー）

適切なチェックポイントャー後方エンドを選択（開発用MemorySaver、本番用SqliteSaver/PostgresSaver）
各ユーザー/セッションにユニークなthread_id
歴史切断閾値を設定し、Tokenが無限に増加しないようにする

長期記憶（ストア）

ユーザーデータをnamespaceで整理(类型, user_id) は ("user_facts", uid)
メモリを抽出する際に信頼度フィルタリングを行い、無意味なノイズを格納しないようにする必要があります
環境を PostgresStore / RedisStore に置き換える

コンテキストの圧縮

圧縮の閾値を確定する（推奨 8-20 メッセージ、シナリオに応じて）
要約のプロンプトでどのような情報を保持するか明確に指定する（トピック名、重要な決定、ユーザーの好み）
要約の連鎖をテストする：第 N ラウンドの圧縮された要約が第 1 ラウンドの重要な情報を含んでいるかどうか
RemoveMessage を使ってメッセージリスト全体を置き換えるのではなく（後者は checkpointer がある場合に問題があるため）

メモリの読み書きのタイミング

記憶の読み込み：chat_node の始めに、system prompt を注入する
記憶の書き込み：chat_node の終わりに、ユーザーの新しい情報を抽出する
各ラウンドごとに書き込まないように（抽出信頼度や内容の長さをフィルタリングする）

この章のまとめ

几つかの核心的なポイント：

四つの記憶タイプはそれぞれ役割分担がある：感覚記憶は LLM の呼び出し自体；作業記憶は会話履歴；情景記憶は圧縮された要約；意味記憶は跨会話の KV Store
checkpointer は会話内を管理し、store は跨会話を管理：thread_id は会話の次元で、user_id はユーザーの次元で、両方の次元は分けて管理する
要約圧縮は長い会話の核心的な解決策です：RemoveMessage + 要約注入で、Agentが無限に長い会話の中でTokenを制御できるようにします
セッション分離は基本的です：異なるthread_idの歴史は互いに干渉せず、異なるuser_idの長期記憶も互いに干渉しません
InMemoryStoreからPostgresStoreに変更するだけです：アーキテクチャは変わらず、バックエンドはプラグアンドプレイ可能です

次回：知識ベース統合——RAGがAgentツールとしての役割とRAGパイプラインの本質的な違い、複数の知識ベースへのルーティング、Agentがいつ検索するか、何を検索するか、何回検索するかを決定する方法。

参考文献

私の個人のホームページより多くの有用な知識と面白い製品を見つけるようお歓迎します

おすすめ購読元

掘金