🦜
LangGraphのメモリ機能まとめ（サンプルコード付き）
LangGraphのメモリ機能について、整理しました。
sqliteによる永続化、また、非同期処理についてもまとめています。
概要
LangGraphのメモリ機能は、会話履歴やエージェントの状態を永続化し、セッションの内外で文脈を保持する仕組みです。
大きく分けて、短期メモリ（スレッド内メモリ） と 長期メモリ（スレッド間メモリ） の 2 種類を提供しています。
短期メモリ（スレッド内メモリ）
短期メモリは 1 つの会話セッション（スレッド）内でのみ保持されます。
StateGraph 上で定義した状態オブジェクトに、ユーザー入力と AI 応答を逐次的に追加します。
チェックポイント機能（checkpoint）が自動的に呼び出されデータベースに状態が永続化されるため、突然プロセスが落ちてもスレッドを再開できます。
長期メモリ（スレッド間メモリ）
長期メモリは、異なる会話スレッドをまたいで情報を共有・再利用できる仕組みです。
内部的にはドキュメントストアを提供し、put・get・search の基本操作でメモリを管理します。
ネームスペース機能により、ユーザー別や用途別にメモリを切り分けることも可能です。
セットアップ
インストール
pip install langchain langchain-community langchain-openai langgraph
環境変数の設定
OPENAI_API_KEY=""
サンプルコード
記憶なしの例
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
# 環境変数の読み込み
load_dotenv()
# LLM セットアップ
llm = ChatOpenAI(temperature=0, model_name="gpt-4.1")
# ReAct Agent の作成
agent = create_react_agent(
model=llm,
tools=[],
)
# 会話例
chat_sample = [
"こんにちは！私はUMENERUです。",
"私の名前は？"
]
# エージェントとの対話
# while True: # 実際に入力して対話する場合
# user_input = input("User: ")
# if user_input.lower() in ["exit", "quit", "q", "終了"]:
# break
for user_input in chat_sample:
print(f"[User] {user_input}")
response = agent.invoke(
{
"messages": [{"role": "user", "content": user_input}],
}
)
print(f"[Agent] {response['messages'][-1].content}")
print("-" * 40)
[User] こんにちは！私はUMENERUです。
[Agent] こんにちは、UMENERUさん！お会いできてうれしいです。今日はどんなことをお手伝いしましょうか？
----------------------------------------
[User] 私の名前は？
[Agent] あなたの名前は、私のデータベースやこの会話の中ではまだ教えていただいていません。そのため、分かりません。もしよろしければ、教えていただけますか？
----------------------------------------
会話を記憶していないので、私の名前を回答できません。
create_react_agentとは
create_react_agentは、LangGraphのlanggraph.prebuiltモジュールに含まれる関数であり、ReAct（Reasoning and Acting）のフレームワークに沿ったエージェントを即座に生成するためのユーティリティです。
具体的には、与えられた言語モデル（LLM）とツール群を組み合わせて、「モデルが思考する（Reasoning）→ツールを呼び出す（Acting）→結果を観察→再度思考」というループを自動的に実行できるグラフ（CompiledStateGraph）を構築します。
短期メモリ(InMemorySaver)
InMemorySaverでは、実行中にメモリ上にデータを保持します。
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver # ★
# 環境変数の読み込み
load_dotenv()
# LLM セットアップ
llm = ChatOpenAI(temperature=0, model_name="gpt-4.1")
# ★ チェックポイントセーバーのセットアップ
checkpointer = InMemorySaver()
# ReAct Agent の作成
agent = create_react_agent(
model=llm,
tools=[],
checkpointer=checkpointer, # ★ チェックポイントセーバーを追加
)
# 会話例
chat_sample = [
"こんにちは！私はUMENERUです。",
"私の名前は？"
]
# エージェントとの対話
session_id = "test_session" # セッション毎の thread_id
config = {"configurable": {"thread_id": session_id}}
for user_input in chat_sample:
print(f"[User] {user_input}")
response = agent.invoke(
{
"messages": [{"role": "user", "content": user_input}],
},
config=config # ★ セッションごとの設定を渡す
)
print(f"[Agent] {response['messages'][-1].content}")
print("-" * 40)
[User] こんにちは！私はUMENERUです。
[Agent] こんにちは、UMENERUさん！お会いできてうれしいです😊
今日はどんなことをお手伝いしましょうか？
----------------------------------------
[User] 私の名前は？
[Agent] あなたの名前は「UMENERU」さんですね！
他にも知ってほしいことがあれば、ぜひ教えてください😊
----------------------------------------
checkpoint に情報が保存されるので、名前を覚えておくことができます。
しかし、InMemorySaver を使用しているため、プロセスが落ちると情報は失われます。
短期メモリ（SqliteSaver）
SQLiteにデータを保持し、Checkpoint情報を永続化します。
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.sqlite import SqliteSaver # ★
# 環境変数の読み込み
load_dotenv()
# LLM セットアップ
llm = ChatOpenAI(temperature=0, model_name="gpt-4.1")
# ★ チェックポイントセーバーのセットアップ
# コンテキストマネージャーとして使用
with SqliteSaver.from_conn_string("test_checkpoints.db") as checkpointer:
# ReAct Agent の作成
agent = create_react_agent(
model=llm,
tools=[],
checkpointer=checkpointer, # ★ チェックポイントセーバーを追加
)
# 会話例
chat_sample = [
"こんにちは！私はUMENERUです。",
"私の名前は？"
]
# エージェントとの対話
session_id = "test_session" # セッション毎の thread_id
config = {"configurable": {"thread_id": session_id}}
for user_input in chat_sample:
print(f"[User] {user_input}")
response = agent.invoke(
{
"messages": [{"role": "user", "content": user_input}],
},
config=config
)
print(f"[Agent] {response['messages'][-1].content}")
print("-" * 40)
なぜコンテキストマネージャーとして使用する必要があるの？
→ DB 接続を確立し、スレッドの開始と終了を管理するためです。
短期メモリ（AsyncSqliteSaver）
SqliteSaverで、非同期処理を行いたい場合です。
import asyncio # ★ 非同期処理のために追加
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver # ★
# 環境変数の読み込み
load_dotenv()
# LLM セットアップ
llm = ChatOpenAI(temperature=0, model_name="gpt-4.1")
# ★ チェックポイントセーバーのセットアップ
# コンテキストマネージャーとして使用
async with AsyncSqliteSaver.from_conn_string("async_checkpoints.db") as checkpointer:
# ReAct Agent の作成
agent = create_react_agent(
model=llm,
tools=[],
checkpointer=checkpointer, # ★ チェックポイントセーバーを追加
)
# 会話例
chat_sample = [
"こんにちは！私はUMENERUです。",
"私の名前は？"
]
# エージェントとの対話
session_id = "test_session" # セッション毎の thread_id
config = {"configurable": {"thread_id": session_id}}
for user_input in chat_sample:
print(f"[User] {user_input}")
response = await agent.ainvoke( # ★ 非同期メソッドを使用
{
"messages": [{"role": "user", "content": user_input}],
},
config=config
)
print(f"[Agent] {response['messages'][-1].content}")
print("-" * 40)
長期メモリ(InMemoryStore)
InMemoryStoreでは、実行中にメモリ上にデータを保持します。
import uuid
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore # ★
# 環境変数の読み込み
load_dotenv()
# LLM セットアップ
llm = ChatOpenAI(temperature=0, model="gpt-4.1")
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
# ★ メモリストアのセットアップ
store = InMemoryStore(
index={
"dims": 1536, # 埋め込みの次元数
"embed": embedding,
}
)
def prompt(state):
# メモリストアを検索
memories = store.search(
("memories",),
query=state["messages"][-1].content,
limit=5 # 検索結果の上限
)
# メモリ内容を文字列として結合
memory_texts = [memory.value["content"] for memory in memories]
memories_str = "\n".join(memory_texts) if memory_texts else "関連するメモリはありません"
system_msg = f"""あなたは、親切なアシスタントです。
以下のユーザーの会話の記憶をもとに、ユーザーの質問に答えてください。
## 記憶
<memories>
{memories_str}
</memories>
"""
return [{"role": "system", "content": system_msg}, *state["messages"]]
# ReAct Agent の作成
agent = create_react_agent(
model=llm,
prompt=prompt, # ★ カスタムプロンプト
tools=[],
store=store, # ★ メモリストアを追加
)
# 会話例
chat_sample = [
"こんにちは！私はUMENERUです。",
"私の名前は？",
"私はリンゴジュースが好きです。",
"私はもっと好きなのは、お米ジュースです。",
"私が一番好きな飲み物は？"
]
# エージェントとの対話
for user_input in chat_sample:
print(f"[User] {user_input}")
response = agent.invoke(
{
"messages": [{"role": "user", "content": user_input}],
}
)
print(f"[Agent] {response['messages'][-1].content}")
print("-" * 40)
# メモリに保存（今回はユーザー入力をそのまま保存）
store.put(("memories",), str(uuid.uuid4()), {"content": f"{user_input}"})
# メモリの内容を表示
memories = store.search(("memories",), limit=5)
print("\n<現在のメモリ>")
for memory in memories:
print(f"- {memory.value['content']}")
print("=" * 40)
[User] こんにちは！私はUMENERUです。
[Agent] こんにちは、UMENERUさん！お話しできてうれしいです。今日はどんなご用件でしょうか？何かお手伝いできることがあれば、何でも教えてくださいね。
----------------------------------------
<現在のメモリ>
- こんにちは！私はUMENERUです。
========================================
[User] 私の名前は？
[Agent] あなたの名前は「UMENERU」です。
----------------------------------------
<現在のメモリ>
- こんにちは！私はUMENERUです。
- 私の名前は？
========================================
[User] 私はリンゴジュースが好きです。
[Agent] 教えてくれてありがとうございます、UMENERUさん！リンゴジュース、美味しいですよね。ほかにも好きな飲み物はありますか？
----------------------------------------
<現在のメモリ>
- こんにちは！私はUMENERUです。
- 私の名前は？
- 私はリンゴジュースが好きです。
========================================
[User] 私はもっと好きなのは、お米ジュースです。
[Agent] お米ジュースがもっと好きなんですね！UMENERUさんはリンゴジュースも好きとおっしゃっていましたが、お米ジュースの方がさらにお気に入りなんですね。お米ジュースはどんなところが好きですか？
----------------------------------------
<現在のメモリ>
- こんにちは！私はUMENERUです。
- 私の名前は？
- 私はリンゴジュースが好きです。
- 私はもっと好きなのは、お米ジュースです。
========================================
[User] 私が一番好きな飲み物は？
[Agent] UMENERUさんが一番好きな飲み物は「お米ジュース」です。
----------------------------------------
<現在のメモリ>
- こんにちは！私はUMENERUです。
- 私の名前は？
- 私はリンゴジュースが好きです。
- 私はもっと好きなのは、お米ジュースです。
- 私が一番好きな飲み物は？
========================================
情報を store に保存し、検索しながら会話を進めることができます。
しかし、InMemoryStore を使用しているため、プロセスが落ちると情報は失われます。
今回はユーザー入力をそのまま保存しましたが、実際には会話を整理し重要な情報だけを保存したい場合が多いでしょう。
それを行うのが LangMem です。(この記事では解説しません)
長期メモリ（SqliteStore）
SQLiteにデータを保持し、Store情報を永続化します。
import uuid
import sqlite3 # ★ SQLite を使用
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.prebuilt import create_react_agent
from langgraph.store.sqlite import SqliteStore # ★
# 環境変数の読み込み
load_dotenv()
# LLM セットアップ
llm = ChatOpenAI(temperature=0, model="gpt-4.1")
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
# ★ メモリストアのセットアップ
conn = sqlite3.connect(
"sqlite_memories.db",
check_same_thread=False,
isolation_level=None
)
store = SqliteStore(
conn=conn,
index={
"dims": 1536,
"embed": embedding,
}
)
store.setup() # スキーマのセットアップ
def prompt(state):
# メモリストアを検索
memories = store.search(
("memories",),
query=state["messages"][-1].content,
limit=5
)
memory_texts = [memory.value["content"] for memory in memories]
memories_str = "\n".join(memory_texts) if memory_texts else "関連するメモリはありません"
system_msg = f"""あなたは、親切なアシスタントです。
以下のユーザーの会話の記憶をもとに、ユーザーの質問に答えてください。
## 記憶
<memories>
{memories_str}
</memories>
"""
return [{"role": "system", "content": system_msg}, *state["messages"]]
# ReAct Agent の作成
agent = create_react_agent(
model=llm,
prompt=prompt,
tools=[],
store=store, # ★ メモリストアを追加
)
# 会話例
chat_sample = [
"こんにちは！私はUMENERUです。",
"私の名前は？",
"私はリンゴジュースが好きです。",
"私はもっと好きなのは、お米ジュースです。",
"私が一番好きな飲み物は？"
]
# エージェントとの対話
for user_input in chat_sample:
print(f"[User] {user_input}")
response = agent.invoke(
{
"messages": [{"role": "user", "content": user_input}],
}
)
print(f"[Agent] {response['messages'][-1].content}")
print("-" * 40)
# メモリに保存
store.put(("memories",), str(uuid.uuid4()), {"content": f"{user_input}"})
# メモリの内容を表示
memories = store.search(("memories",), limit=5)
print("\n<現在のメモリ>")
for memory in memories:
print(f"- {memory.value['content']}")
print("=" * 40)
長期メモリ（AsyncSqliteStore）
SqliteStoreで、非同期処理を行いたい場合です。
import uuid
import aiosqlite # ★ 非同期版 SQLite
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langgraph.prebuilt import create_react_agent
from langgraph.store.sqlite.aio import AsyncSqliteStore # ★
# 環境変数の読み込み
load_dotenv()
# LLM セットアップ
llm = ChatOpenAI(temperature=0, model="gpt-4.1")
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
# ★ メモリストアのセットアップ
conn = await aiosqlite.connect(
"async_memories.db",
check_same_thread=False,
isolation_level=None
)
store = AsyncSqliteStore(
conn=conn,
index={
"dims": 1536,
"embed": embedding,
}
)
await store.setup() # スキーマのセットアップ
async def prompt(state):
memories = await store.asearch(
("memories",),
query=state["messages"][-1].content,
limit=5
)
memory_texts = [memory.value["content"] for memory in memories]
memories_str = "\n".join(memory_texts) if memory_texts else "関連するメモリはありません"
system_msg = f"""あなたは、親切なアシスタントです。
以下のユーザーの会話の記憶をもとに、ユーザーの質問に答えてください。
## 記憶
<memories>
{memories_str}
</memories>
"""
return [{"role": "system", "content": system_msg}, *state["messages"]]
# ReAct Agent の作成
agent = create_react_agent(
model=llm,
prompt=prompt,
tools=[],
store=store, # ★ メモリストアを追加
)
# 会話例
chat_sample = [
"こんにちは！私はUMENERUです。",
"私の名前は？",
"私はリンゴジュースが好きです。",
"私はもっと好きなのは、お米ジュースです。",
"私が一番好きな飲み物は？"
]
# エージェントとの対話
for user_input in chat_sample:
print(f"[User] {user_input}")
response = await agent.ainvoke(
{
"messages": [{"role": "user", "content": user_input}],
}
)
print(f"[Agent] {response['messages'][-1].content}")
print("-" * 40)
# メモリに保存
await store.aput(("memories",), str(uuid.uuid4()), {"content": f"{user_input}"})
# メモリの内容を表示
memories = await store.asearch(("memories",), limit=5)
print("\n<現在のメモリ>")
for memory in memories:
print(f"- {memory.value['content']}")
print("=" * 40)
まとめ
-
LangGraph のメモリ機能
会話履歴やエージェントの状態を永続化し、セッション内外で文脈を保持できる。短期メモリ（スレッド内）と長期メモリ（スレッド間）の 2 つのレイヤーを提供する。 -
短期メモリ（スレッド内メモリ）
- InMemorySaver：プロセス内のメモリ上に状態を保持。プロセス終了で情報は消失。
- SqliteSaver：SQLite データベースにチェックポイントを永続化。プロセス落ち後も再開可能。
-
AsyncSqliteSaver：非同期処理対応バージョン。
ainvokeメソッドで非同期にチェックポイントを保存。
-
長期メモリ（スレッド間メモリ）
- InMemoryStore：実行中にメモリ上で情報を保持。プロセス終了で情報は消失。
-
SqliteStore：SQLite にドキュメントストアを永続化。
put／get／searchでメモリ操作し、ネームスペースで用途別に切り分け可能。 -
AsyncSqliteStore：非同期処理対応バージョン。
asearch／aputなどを用いてスレッド間メモリを非同期操作。
-
各ストア・セーバーの使い分け
- 簡易プロトタイプやテストでは InMemory 系。
- 本番環境や長時間動作には永続化対応（Sqlite 系）を推奨。
- 非同期ワークフローでは AsyncSqlite 系を利用し、性能や並行性を確保。
Discussion