langgraph-persistence

LangGraph の永続化レイヤーは、グラフの状態をチェックポイントに保存することで、堅牢な実行を可能にします。

• チェックポインター：各スーパーステップでグラフの状態を保存／読み込みします
• スレッド ID：個別のチェックポイント系列（会話）を識別します
• ストア：ユーザーの設定や事実をスレッド間で共有するメモリです

メモリの種類は 2 つあります。

• 短期（チェックポインター）：スレッド単位の会話履歴
• 長期（ストア）：スレッド間で共有するユーザーの設定や事実

チェックポインター	ユースケース	本番環境対応
InMemorySaver	テスト、開発	いいえ
SqliteSaver	ローカル開発	部分的
PostgresSaver	本番環境	はい

---

チェックポインターのセットアップ

インメモリチェックポイントとスレッドベースの状態永続化を使用して、基本的なグラフをセットアップします。 ```python from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict, Annotated import operator

class State(TypedDict): messages: Annotated[list, operator.add]

def add_message(state: State) -> dict: return {"messages": ["Bot response"]}

checkpointer = InMemorySaver()

graph = ( StateGraph(State) .add_node("respond", add_message) .add_edge(START, "respond") .add_edge("respond", END) .compile(checkpointer=checkpointer) # Pass at compile time )

ALWAYS provide thread_id

config = {"configurable": {"thread_id": "conversation-1"}}

result1 = graph.invoke({"messages": ["Hello"]}, config) print(len(result1["messages"])) # 2

result2 = graph.invoke({"messages": ["How are you?"]}, config) print(len(result2["messages"])) # 4 (previous + new)

</python>
<typescript>
インメモリチェックポイントとスレッドベースの状態永続化を使用して、基本的なグラフをセットアップします。
```typescript
import { MemorySaver, StateGraph, StateSchema, MessagesValue, START, END } from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";

const State = new StateSchema({ messages: MessagesValue });

const addMessage = async (state: typeof State.State) => {
  return { messages: [{ role: "assistant", content: "Bot response" }] };
};

const checkpointer = new MemorySaver();

const graph = new StateGraph(State)
  .addNode("respond", addMessage)
  .addEdge(START, "respond")
  .addEdge("respond", END)
  .compile({ checkpointer });

// ALWAYS provide thread_id
const config = { configurable: { thread_id: "conversation-1" } };

const result1 = await graph.invoke({ messages: [new HumanMessage("Hello")] }, config);
console.log(result1.messages.length);  // 2

const result2 = await graph.invoke({ messages: [new HumanMessage("How are you?")] }, config);
console.log(result2.messages.length);  // 4 (previous + new)

本番環境へのデプロイ向けに、PostgreSQL を使用したチェックポイントを設定します。 ```python from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string( "postgresql://user:pass@localhost/db" ) as checkpointer: checkpointer.setup() # only needed on first use to create tables graph = builder.compile(checkpointer=checkpointer)

</python>
<typescript>
本番環境へのデプロイ向けに、PostgreSQL を使用したチェックポイントを設定します。
```typescript
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";

const checkpointer = PostgresSaver.fromConnString(
  "postgresql://user:pass@localhost/db"
);
await checkpointer.setup(); // only needed on first use to create tables

const graph = builder.compile({ checkpointer });

---

スレッド管理

異なるスレッド ID 間で状態が分離されることを示します。 ```python # Different threads maintain separate state alice_config = {"configurable": {"thread_id": "user-alice"}} bob_config = {"configurable": {"thread_id": "user-bob"}}

graph.invoke({"messages": ["Hi from Alice"]}, alice_config) graph.invoke({"messages": ["Hi from Bob"]}, bob_config)

Alice's state is isolated from Bob's

</python>
<typescript>
異なるスレッド ID 間で状態が分離されることを示します。
```typescript
// Different threads maintain separate state
const aliceConfig = { configurable: { thread_id: "user-alice" } };
const bobConfig = { configurable: { thread_id: "user-bob" } };

await graph.invoke({ messages: [new HumanMessage("Hi from Alice")] }, aliceConfig);
await graph.invoke({ messages: [new HumanMessage("Hi from Bob")] }, bobConfig);

// Alice's state is isolated from Bob's

---

状態履歴とタイムトラベル

タイムトラベル：チェックポイントの履歴を参照し、過去の状態から再実行または分岐します。 ```python config = {"configurable": {"thread_id": "session-1"}}

result = graph.invoke({"messages": ["start"]}, config)

Browse checkpoint history

states = list(graph.get_state_history(config))

Replay from a past checkpoint

past = states[-2] result = graph.invoke(None, past.config) # None = resume from checkpoint

Or fork: update state at a past checkpoint, then resume

fork_config = graph.update_state(past.config, {"messages": ["edited"]}) result = graph.invoke(None, fork_config)

</python>
<typescript>
タイムトラベル：チェックポイントの履歴を参照し、過去の状態から再実行または分岐します。
```typescript
const config = { configurable: { thread_id: "session-1" } };

const result = await graph.invoke({ messages: ["start"] }, config);

// Browse checkpoint history (async iterable, collect to array)
const states: Awaited<ReturnType<typeof graph.getState>>[] = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}

// Replay from a past checkpoint
const past = states[states.length - 2];
const replayed = await graph.invoke(null, past.config);  // null = resume from checkpoint

// Or fork: update state at a past checkpoint, then resume
const forkConfig = await graph.updateState(past.config, { messages: ["edited"] });
const forked = await graph.invoke(null, forkConfig);

実行を再開する前に、グラフの状態を手動で更新します。 ```python config = {"configurable": {"thread_id": "session-1"}}

Modify state before resuming

graph.update_state(config, {"data": "manually_updated"})

Resume with updated state

result = graph.invoke(None, config)

</python>
<typescript>
実行を再開する前に、グラフの状態を手動で更新します。
```typescript
const config = { configurable: { thread_id: "session-1" } };

// Modify state before resuming
await graph.updateState(config, { data: "manually_updated" });

// Resume with updated state
const result = await graph.invoke(null, config);

---

サブグラフのチェックポインターのスコープ

サブグラフをコンパイルするとき、checkpointer パラメーターで永続化の動作を制御します。割り込みを使用する、複数ターンのメモリを必要とする、または並列実行するサブグラフでは、これは特に重要です。

機能	checkpointer=False	None（デフォルト）	True
割り込み（HITL）	いいえ	はい	はい
複数ターンのメモリ	いいえ	いいえ	はい
複数回の呼び出し（異なるサブグラフ）	はい	はい	警告（名前空間の競合が発生する可能性があります）
複数回の呼び出し（同じサブグラフ）	はい	はい	いいえ
状態の確認	いいえ	警告（現在の呼び出しのみ）	はい

各モードの使用場面

• checkpointer=False — サブグラフで割り込みや永続化が不要な場合に使います。最もシンプルな選択肢で、チェックポイントのオーバーヘッドはありません。
• None（デフォルト／checkpointer を省略） — サブグラフで interrupt() は必要ですが、複数ターンのメモリは不要な場合に使います。呼び出しのたびに状態はリセットされますが、一時停止と再開は可能です。各呼び出しに一意の名前空間が割り当てられるため、並列実行も可能です。
• checkpointer=True — サブグラフで呼び出しをまたいで状態を保持する必要がある場合（複数ターンの会話）に使います。呼び出すたびに、前回の続きから処理が再開されます。

警告：ステートフルなサブグラフ（checkpointer=True）では、1 つのノード内で同じサブグラフインスタンスを複数回呼び出すことはできません。呼び出しごとに同じチェックポイント名前空間へ書き込むため、競合が発生します。

サブグラフに適したチェックポインターモードを選択します。 ```python # No interrupts needed — opt out of checkpointing subgraph = subgraph_builder.compile(checkpointer=False)

Need interrupts but not cross-invocation persistence (default)

subgraph = subgraph_builder.compile()

Need cross-invocation persistence (stateful)

subgraph = subgraph_builder.compile(checkpointer=True)

</python>
<typescript>
サブグラフに適したチェックポインターモードを選択します。
```typescript
// No interrupts needed — opt out of checkpointing
const subgraph = subgraphBuilder.compile({ checkpointer: false });

// Need interrupts but not cross-invocation persistence (default)
const subgraph = subgraphBuilder.compile();

// Need cross-invocation persistence (stateful)
const subgraph = subgraphBuilder.compile({ checkpointer: true });

### 並列サブグラフの名前空間設定

複数の異なるステートフルなサブグラフを並列実行する場合は、それぞれを固有のノード名を持つ独自の StateGraph でラップし、安定した名前空間の分離を確保します。

```python from langgraph.graph import MessagesState, StateGraph

def create_sub_agent(model, *, name, **kwargs): """Wrap an agent with a unique node name for namespace isolation.""" agent = create_agent(model=model, name=name, **kwargs) return ( StateGraph(MessagesState) .add_node(name, agent) # unique name -> stable namespace .add_edge("start", name) .compile() )

fruit_agent = create_sub_agent( "gpt-4.1-mini", name="fruit_agent", tools=[fruit_info], prompt="...", checkpointer=True, ) veggie_agent = create_sub_agent( "gpt-4.1-mini", name="veggie_agent", tools=[veggie_info], prompt="...", checkpointer=True, )

</python>
<typescript>
```typescript
import { StateGraph, StateSchema, MessagesValue, START } from "@langchain/langgraph";

function createSubAgent(model: string, { name, ...kwargs }: { name: string; [key: string]: any }) {
  const agent = createAgent({ model, name, ...kwargs });
  return new StateGraph(new StateSchema({ messages: MessagesValue }))
    .addNode(name, agent)  // unique name -> stable namespace
    .addEdge(START, name)
    .compile();
}

const fruitAgent = createSubAgent("gpt-4.1-mini", {
  name: "fruit_agent", tools: [fruitInfo], prompt: "...", checkpointer: true,
});
const veggieAgent = createSubAgent("gpt-4.1-mini", {
  name: "veggie_agent", tools: [veggieInfo], prompt: "...", checkpointer: true,
});

注：ノードとして追加されたサブグラフ（add_node 経由）には、すでに名前に基づく名前空間が自動的に割り当てられるため、このラッパーは不要です。

---

長期メモリ（Store）

スレッドをまたいでメモリを共有し、会話間でユーザーの好みを共有するには、Store を使用します。 ```python from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

Save user preference (available across ALL threads)

store.put(("alice", "preferences"), "language", {"preference": "short responses"})

Node with store — access via runtime

from langgraph.runtime import Runtime

def respond(state, runtime: Runtime): prefs = runtime.store.get((state["user_id"], "preferences"), "language") return {"response": f"Using preference: {prefs.value}"}

Compile with BOTH checkpointer and store

graph = builder.compile(checkpointer=checkpointer, store=store)

Both threads access same long-term memory

graph.invoke({"user_id": "alice"}, {"configurable": {"thread_id": "thread-1"}}) graph.invoke({"user_id": "alice"}, {"configurable": {"thread_id": "thread-2"}}) # Same preferences!

</python>
<typescript>
スレッドをまたいでメモリを共有し、会話間でユーザーの好みを共有するには、Store を使用します。
```typescript
import { MemoryStore } from "@langchain/langgraph";

const store = new MemoryStore();

// Save user preference (available across ALL threads)
await store.put(["alice", "preferences"], "language", { preference: "short responses" });

// Node with store — access via runtime
const respond = async (state: typeof State.State, runtime: any) => {
  const item = await runtime.store?.get(["alice", "preferences"], "language");
  return { response: `Using preference: ${item?.value?.preference}` };
};

// Compile with BOTH checkpointer and store
const graph = builder.compile({ checkpointer, store });

// Both threads access same long-term memory
await graph.invoke({ userId: "alice" }, { configurable: { thread_id: "thread-1" } });
await graph.invoke({ userId: "alice" }, { configurable: { thread_id: "thread-2" } });  // Same preferences!

基本的なストア操作：put、get、search、delete。 ```python from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

store.put(("user-123", "facts"), "location", {"city": "San Francisco"}) # Put item = store.get(("user-123", "facts"), "location") # Get results = store.search(("user-123", "facts"), filter={"city": "San Francisco"}) # Search store.delete(("user-123", "facts"), "location") # Delete

</python>
</ex-store-operations>

---

## 修正方法

<fix-thread-id-required>
<python>
状態を永続化できるよう、config には必ず thread_id を指定します。
```python
# WRONG: No thread_id - state NOT persisted!
graph.invoke({"messages": ["Hello"]})
graph.invoke({"messages": ["What did I say?"]})  # Doesn't remember!

# CORRECT: Always provide thread_id
config = {"configurable": {"thread_id": "session-1"}}
graph.invoke({"messages": ["Hello"]}, config)
graph.invoke({"messages": ["What did I say?"]}, config)  # Remembers!

状態を永続化できるよう、config には必ず thread_id を指定します。 ```typescript // WRONG: No thread_id - state NOT persisted! await graph.invoke({ messages: [new HumanMessage("Hello")] }); await graph.invoke({ messages: [new HumanMessage("What did I say?")] }); // Doesn't remember!

// CORRECT: Always provide thread_id const config = { configurable: { thread_id: "session-1" } }; await graph.invoke({ messages: [new HumanMessage("Hello")] }, config); await graph.invoke({ messages: [new HumanMessage("What did I say?")] }, config); // Remembers!

</typescript>
</fix-thread-id-required>


<fix-inmemory-not-for-production>
<python>
本番環境での永続化には、InMemorySaver ではなく PostgresSaver を使用します。
```python
# WRONG: Data lost on process restart
checkpointer = InMemorySaver()  # In-memory only!

# CORRECT: Use persistent storage for production
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()  # only needed on first use to create tables
    graph = builder.compile(checkpointer=checkpointer)

本番環境での永続化には、MemorySaver ではなく PostgresSaver を使用します。 ```typescript // WRONG: Data lost on process restart const checkpointer = new MemorySaver(); // In-memory only!

// CORRECT: Use persistent storage for production import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres"; const checkpointer = PostgresSaver.fromConnString("postgresql://..."); await checkpointer.setup(); // only needed on first use to create tables

</typescript>
</fix-inmemory-not-for-production>


<fix-update-state-with-reducers>
<python>
レデューサーを経由せずに状態値を置き換えるには、Overwrite を使用します。
```python
from langgraph.types import Overwrite

# State with reducer: items: Annotated[list, operator.add]
# Current state: {"items": ["A", "B"]}

# update_state PASSES THROUGH reducers
graph.update_state(config, {"items": ["C"]})  # Result: ["A", "B", "C"] - Appended!

# To REPLACE instead, use Overwrite
graph.update_state(config, {"items": Overwrite(["C"])})  # Result: ["C"] - Replaced

レデューサーを経由せずに状態値を置き換えるには、Overwrite を使用します。 ```typescript import { Overwrite } from "@langchain/langgraph";

// State with reducer: items uses concat reducer // Current state: { items: ["A", "B"] }

// updateState PASSES THROUGH reducers await graph.updateState(config, { items: ["C"] }); // Result: ["A", "B", "C"] - Appended!

// To REPLACE instead, use Overwrite await graph.updateState(config, { items: new Overwrite(["C"]) }); // Result: ["C"] - Replaced

</typescript>
</fix-update-state-with-reducers>

<fix-store-injection>
<python>
グラフノードでは、Runtime オブジェクトを介してストアにアクセスします。
```python
# WRONG: Store not available in node
def my_node(state):
    store.put(...)  # NameError! store not defined

# CORRECT: Access store via runtime
from langgraph.runtime import Runtime

def my_node(state, runtime: Runtime):
    runtime.store.put(...)  # Correct store instance

グラフノードでは、runtime パラメーターを介してストアにアクセスします。 ```typescript // WRONG: Store not available in node const myNode = async (state) => { store.put(...); // ReferenceError! };

// CORRECT: Access store via runtime const myNode = async (state, runtime) => { await runtime.store?.put(...); // Correct store instance };

</typescript>
</fix-store-injection>

<boundaries>
### やってはいけないこと

- 本番環境で `InMemorySaver` を使用する — 再起動時にデータが失われます。`PostgresSaver` を使用してください
- `thread_id` を指定し忘れる — 指定しないと状態は永続化されません
- `update_state` がレデューサーを迂回すると考える — レデューサーを経由します。置き換えるには `Overwrite` を使用してください
- 1 つのノード内で同じステートフルなサブグラフ（`checkpointer=True`）を並列実行する — 名前空間が競合します
- ノード内でストアに直接アクセスする — `Runtime` パラメーターを介して `runtime.store` を使用してください
</boundaries>