🐍

LangGraphにおけるAIエージェントの記憶管理：短期記憶と長期記憶を解説

に公開
LangGraph
AIエージェント
tech

今日のAIエージェントは、単一の質問に答えるだけでなく、複雑なタスクを連続的にこなしたり、ユーザーとの長期的な関係性を築いたりすることが求められています。そのためには、エージェントが過去の対話履歴や獲得した知識を適切に「記憶」し、必要に応じて「記憶から取り出す」能力が不可欠です。

LangGraphは、この記憶管理の課題を解決するために、人間の記憶のメカニズムに着想を得て、AIエージェントのメモリを「短期記憶」と「長期記憶」という二つのコンセプトで管理しています。会話セッション内の履歴を管理する「短期記憶（checkpointer）」と、複数のセッションを横断して永続的な知識を保持する「長期記憶（Store）」という二層構造のメモリシステムです。前回の記事では、複数回のやり取りを記憶して会話するための機能としてcheckpointerに触れました。

この記事では、LangGraphの短期記憶を司るcheckpointerの詳しい挙動から、複数の会話セッションを超えて情報を保持する「長期記憶」の概念、そしてその具体的な実装方法までを、初級者から中級者の読者向けに解説します。

実装方法だけ知りたいよって方は、「長期記憶の実装」までにスキップしてください。

メモリ機能：短期記憶と長期記憶

短期記憶と長期記憶[1]というと人の記憶が思い浮かぶと思います。

  • 短期記憶：保持期間が数十秒程度の記憶である。保持時間だけではなく、一度に保持される情報の容量の大きさにも限界があることが特徴とされる。
  • 長期記憶: 短期記憶に含まれる情報の多くは忘却され、その一部が長期記憶として保持される。この保持情報が長期記憶として安定化する過程は記憶の固定化と呼ばれる。長期記憶は保持時間が長く、数分から一生にわたって保持される記憶である。

LangGraphもAIエージェントに情報を保持する方法として、人と同じ短期記憶・長期記憶というコンセプトを導入しました。

  • 短期記憶：Threadスコープメモリとも呼ばれ、セッション内の会話履歴を記憶するための機能です。
  • 長期記憶：複数のセッション履歴を記憶するための機能です。複数の短期記憶・会話履歴を保持します。

短期記憶は同じthread_id（= 会話セッションを識別するID）内の会話を記憶し、言語モデルが参照することで複数回のやり取りを可能とします。それぞれの会話はthread_idで管理されているため、別のthread_idで管理される情報にはアクセスできません。しかし、システムによっては別のthread_idで管理される情報にアクセスしたい時があると思います。それを可能にする機能が長期記憶です。

図を見て誤解しないでほしいのですが、長期記憶は複数の短期記憶・会話履歴をただ集めるだけでなく、セッション間で再利用したい情報を抽出して集積することもできます。「長期記憶：ProfileとCollection」でもう少し解説します。

短期記憶

checkpointerの使用法は前回の入門記事で解説しました。しかし、詳しい挙動に関しては、解説していなかったので、ここで紹介したいと思います。

各ステップ・ノードでグラフのstate(状態)が更新されます。

checkpointerはこの更新された状態をcheckpointsとして逐次保存します。このcheckpointsをthreadに追加することで会話履歴の一貫性と再現性が保証されます。

イメージとして、各ステップでの処理データ・メッセージ・メタデータを一枚のメモ(checkpoints)に書き込み、そのメモの束を順番にノート(thread)に貼り付けることで会話を再現しています。そのため、メモの情報を再編集[2]したりすることもできます。

get_state()やget_state_history()を利用することで各ステップのcheckpointsに保存されている詳細を確認することができます。

サンプルコード

コードは公式サイトOverview - Persistenceからコピーしています。

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}


workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": ""}, config)

# get the list of state snapshots
config = {"configurable": {"thread_id": "1"}}
list(graph.get_state_history(config))

# output
[
    StateSnapshot(
        values={'foo': 'b', 'bar': ['a', 'b']},
        next=(),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}},
        metadata={'source': 'loop', 'writes': {'node_b': {'foo': 'b', 'bar': ['b']}}, 'step': 2},
        created_at='2024-08-29T19:19:38.821749+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
        tasks=(),
    ),
    StateSnapshot(...) ...
]

長期記憶

StoreはJSON互換の形式で保存され、namespace、key、valueの3要素で構成されます。

  • namespace: tuple[str]で階層的な分類が可能

    タプルの形式ではなく、stringで渡すとイテラブルとして解釈される

    タプルの形式ではなく、stringで渡すとイテラブルとして解釈されるので注意が必要です。
    例えば、以下のようにstringを渡してみると、namespaceの表記が変になってしまいます。

    memory_store = InMemoryStore()

namespace = 'Client'

# 正しい表記　 
# namespace = ('Client,')

key = 'abcd'
value = {'operation' : 'failed'}

memory_store.put(namespace, key, value)
print(f'namespace: {memory_store.search(namespace)[0].namespace}')

    # 出力
namespace: ('C', 'l', 'i', 'e', 'n', 't')

# 正しい表記の場合
# namespace: ('Client',)

    そのため、タプル形式で指定してください。

  • key: str

  • value: dict[str, Any]

Storeは「仮想ファイルシステム」をイメージするとわかりやすいかもしれません。

  • namespace: フォルダ構造（例：("Client", "Account") → Client/Account/）
  • key: ファイル名（例："operation"）
  • value: ファイルの中身（JSON形式のデータ）

長期記憶の種類

LangGraphの長期記憶は、保存される情報の性質に基づいて以下の3つのタイプに分類されます。これにより、エージェントは異なる種類の情報を効率的に管理し、必要に応じて活用することができます。

タイプ 保存される情報 使用例
Semantic 事実情報の保持 ユーザープロファイル
Episodic 過去の経験、行動記録 タスクを遂行の過程、日時レポート生成
Procedural 動作・タスク遂行ののルール ワークフロー改善ルールの蓄積

LangGraphのstore（InMemoryStoreなど）の内部的な実装においては、これらの記憶タイプを直接区別するための特別なAPIや機能は現在時点(2025年9月)、提供されていません。したがって、これらの記憶タイプは、LangGraphのstoreを効果的に活用するための「設計パターン」や「概念的なガイドライン」として理解することが重要です。

ProfileとCollection

Semantic メモリは2種類の管理方法があります。保存の形式と捉えても構いません。

  • Profile: 設定されたテンプレートに対し、会話履歴を参照して情報をアップデートします。
  • Collection: 会話履歴や情報をドキュメントとして、追記する形で情報をアップデートします。
    Profile
    Collection

イメージとしてProfileは、あらかじめ集めたい個人情報(名前、年齢、出身地)をテンプレートとしてモデルに渡して、会話の中にこれらに関する情報が出てきたら穴埋めする形で情報をアップデートします。一方Collectionは、ToDoリストを想像してもらえるとわかりやすいと思います。会話に出てきた遂行したいタスクをどんどん追加するようなイメージです。

これらの実装に関しては、別の記事にまとめようと思います。

記憶の保存方法

実際に記憶を保存・書き込む方法は2種類用意されています：in the hot pathとin the background。

  • in the hot path:リアルタイムでアップデートするため、次のプロセスですぐに利用することができます。逐次アップデートのため、パフォーマンスが落ちる可能性がある。
  • in the background:プロセスと記憶の処理を分けることで、待ち時間を短縮することができます。しかし、保存の周期をどうするかという問題があります。

詳しく知りたい方は "in the hot path" や "in the background"を覗いてみてください。[3]

長期記憶の実装

この実装では、LangGraphの短期記憶と長期記憶を組み合わせたマルチセッション対応のAIエージェントを構築します。前回の入門記事で使用したコードを極力再利用する形で拡張しています。

システム構成：

インポート文とモデル・ツールを定義
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.tools import tool

# Define a model
llm = ChatGoogleGenerativeAI(model='gemini-2.5-flash', api_key=api_key)  

# Define tools
@tool
def add(a: int, b: int):
  "add two numbers together"
  return a + b

@tool
def multiply(a: int, b: int):
  "multiply two numbers together"
  return a * b

次に短期記憶と長期記憶を保存するためのオブジェクトを作成します。

from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver

# short-term memory(短期記憶)
checkpointer = InMemorySaver()
# long-term memory(長期記憶)
store = InMemoryStore()

AIエージェントにモデル・ツール・各記憶のオブジェクトを渡します。前回checkpointerを渡したようにstoreをcreate_react_agent渡します。

from langgraph.prebuilt import create_react_agent
# create an agent bound with tools, checkpointer, and store
agent = create_react_agent(llm, tools = [add, multiply], checkpointer=checkpointer, store=store)

次にエージェントを呼び出したときの動作を書いていきます。

def call_agent(user_input, thread_id: int):
  """
  Function to invoke an agent response with given a user_input and thread_id.
  Its response is saved in a store for long-term memory.
  """
　短期記憶用のconfigを設定、セッション管理のためのthread_idを指定する
  config = {
      'configurable': {
          'thread_id': thread_id
          }
  }

   try: 
    response = agent.invoke({'messages': [('human', user_input)]}, config=config)

  except Exception as e:
    print(e)

  print(f" {response['messages'][-1].content}")

  # long-term memory(長期記憶)に保存
  namespace = ('chat_session', str(thread_id))
  key = 'logs'
  value = str(response['messages'])
  agent.store.put(namespace, key, value)

後で何度も呼び出すので関数化してありますが、前記事と同様にconfigにthread_idを指定して、セッションの内の会話を保存できるようにします。エージェント起動するときにconfigを渡して、最後の結果を表示しています。ここまでは前回同様にセッション内の情報を記憶するための短期記憶に関するコードです。

最後の4行のコードで長期記憶するための3要素を定義して、putメソッドを使用して情報をstoreオブジェクトに保存します。

では、実際にデータを渡して、データが保存されているか確認していきます。

# sample run to test a short-term memory (within session memory)
user1 = [('Add 2 + 4', 1), ('Multiply that by 4', 1)]
user2 = [('Multiply 4 by 7', 2), ('Add that by 7', 2)] 
users = [user1, user2]

for user in users:
  for i, query in enumerate(user):
    print(f'\nRun with thread_id: {query[1]}')
    print(f'Number of query: {i+1}/{len(user)}')
    call_agent(query[0], query[1])

# 出力
Run with thread_id: 1
Number of query: 1/2
AI Response: The sum of 2 and 4 is 6.

Run with thread_id: 1
Number of query: 2/2
AI Response: The product of 6 and 4 is 24.

Run with thread_id: 2
Number of query: 1/2
AI Response: The answer is 28.

Run with thread_id: 2
Number of query: 2/2
AI Response: The answer is 35.

短期記憶はしっかり働いていてthatの意味を正確に推測できています。

次に、storeの中身を確認します。.list_namespace()のメソッドはstoreにあるすべてのnamespaceをリストとして返してくれます。.search()を使用することで指定したnamespaceにあるkey、valueやメタ情報を取得できます。出力をそのまますべて保存しているため、valueの部分は省略して表示しています。

search()メソッドの補足

.search()はnamespace_prefixという引数取ります。これはフォルダ構造と同じようにnamespaceの階層を上から順番に指定しないといけません。

memory_store = InMemoryStore()

def save_memory(namespace_key_value: list):
    namespace, key, value = *namespace_key_value,
    memory_store.put(namespace, key, value)

items = [(("Client",'1'),'abcd',{"operation" : "failed"}),
         (("Client",'2'),'bdke',{"operation" : "sucess"}),
         (("Client",'1','2025-03-01'),'nekl',{"operation" : "failed"}),
         (("Client",'1', '2025-03-01', 'session_1'),'meln',{"operation" : "sucess"}),
         (("Test","Client",'2'),'dfnet',{"test" : "pass"})]

for item in items:
    save_memory(item)

例えば、上のnamespaceに対して、.search('1')を指定してしまうと、空のリストが返ります。

namespace = ('1',)
print(f'Searched namespace: {namespace}')
print(f'Namecpase: {memory_store.search(namespace)}')

Searched namespace: ('1',)
Namecpase: []

また、フォルダとは違い'Client'と指定すると、'Client'で始まるすべてのnamespaceが返ります。

namespace=('Client',)
print(f'\nSearched namespace: {namespace}')
for i in range(len(memory_store.search(namespace))):
  print(f'Namespace: {memory_store.search(namespace)[i].namespace}')

Searched namespace: ('Client',)
Namespace: ('Client', '1')
Namespace: ('Client', '2')
Namespace: ('Client', '1', '2025-03-01')
Namespace: ('Client', '1', '2025-03-01', 'session_1')

そのため、namespaceは構造を考えて設定しないと使いにくいものになります。まずは、フォルダ構造を作る時のように考えてみると良いでしょう。

for namespace in store.list_namespaces():
  print(f'\nNamespace: {namespace}')
  for k, v in agent.store.search((namespace))[0].dict().items():
    print(f'  {k}: {v}')

# 出力
Namespace: ('chat_session', '1')
  namespace: ['chat_session', '1']
  key: logs
  value: [HumanMessage(content='Add 2 + 4', additional_kwargs={},　...]
  created_at: 2025-09-06T09:32:58.879770+00:00
  updated_at: 2025-09-06T09:32:58.879773+00:00
  score: None

Namespace: ('chat_session', '2')
  namespace: ['chat_session', '2']
  key: logs
  value: [HumanMessage(content='Multiply 4 by 7', additional_kwargs={},　...]
  created_at: 2025-09-06T09:33:01.130056+00:00
  updated_at: 2025-09-06T09:33:01.130059+00:00
  score: None

実際の運用では、このように生の情報をそのまま渡してしまうとトークン数の増加によるコスト増加、パフォーマスンの低下が考えられるので、putに渡すvalueは「出力のみ保存」や「メタデータだけ保存」したり用途に応じて、変更します。

軽量化の例
  • 出力のみ
value = response['messages'][-1].content
  • 必要なメタデータのみ
value = {"answer": response['messages'][-1].content, "timestamp": str(datetime.now())}

では、最後に長期記憶を利用してみます。
.getは.searchに似たメソッドでnamespaceとkeyを指定することで情報を取得できます。今回はログとして各セッションの全会話履歴をリストにします。

# prepare histories of conversations in each thread_id
previous_results = []
for namespace in agent.store.list_namespaces():
  previous_results.append(agent.store.get(namespace, 'logs').value)

プロンプトに会話履歴を渡して参照できるようにして、答えを求めてもらいます。

config = {
    'configurable': {
        'thread_id': 3
    }
}

# make a prompt
history_text = "\n".join(previous_results)
user_input = f"Using the histories of conversations: {history_text}. Please sum up the latest two answers of products."

# invoke a response
response = agent.invoke({'messages':('human', user_input)}, config=config)

print(response['messages'][-1].content)

# 出力
The sum of the last two products is 52.

このように複数のセッション間の情報を利用して答えを導くことができました。

補足・注意点

create_react_agentにstoreを渡して、情報の書き込みや読み込みを行い、プロンプトには一度読み出してから渡す形で実装しました。調査時点(2025年9月, langgraph== 0.6.6 )では呼び出さずに利用する方法は、見当たりませんでした。

今後のアップデートなど、最新の動作については公式ドキュメントやリリースノートを随時確認してください。

結論

本記事を通じて、LangGraphが提供するAIエージェントの記憶管理機能、すなわち短期記憶（checkpointer）と長期記憶（Store）の重要性と、その具体的な活用方法について深く掘り下げてきました。

短期記憶は、thread_idによって管理される会話セッション内のコンテキストを維持し、エージェントが「今、何が話されているか」を正確に把握し、一貫性のある応答を生成するために不可欠です。これにより、ユーザーは途切れることのないスムーズな対話体験を得られます。

そして、長期記憶は、namespaceとkeyを用いて構造化された情報を永続的に保存し、複数の会話セッションを横断して利用することを可能にします。これにより、エージェントはより高度な「記憶」に基づく行動を実現できます。

  • ユーザープロファイルの構築と活用: ユーザーの氏名、年齢、過去の行動パターンなどを記憶し、個々に最適化されたサービス提供。
  • タスクの継続と進捗管理: 長期にわたるプロジェクトや複雑なタスクにおいて、中断した時点から再開し、進捗を記憶。
  • 知識ベースの自動構築: 会話から得られた知見やユーザーからのフィードバックを蓄積し、エージェント自身の知識を継続的に更新。

LangGraphの短期記憶と長期記憶を効果的に組み合わせることで、AIエージェントは単なる情報処理ツールではなく、ユーザーの信頼できるパートナーとして、より複雑な問題解決やパーソナライズされた支援を提供できるようになります。ぜひ、これらの記憶機能を活用し、エージェントの可能性を最大限に引き出してください。

脚注

  1. 記憶の分類 ↩︎

  2. Time-travel - LangGraph ↩︎

  3. Writing memories - LangGraph ↩︎

Discussion