langgraph-human-in-the-loop

LangGraph のヒューマンイン・ザ・ループパターンでは、グラフの実行を一時停止してデータをユーザーに提示し、入力を受け取って再開できます。

• interrupt(value) — 実行を一時停止し、呼び出し元に値を提示します
• Command(resume=value) — 実行を再開し、値を interrupt() に渡します
• チェックポインター — 一時停止中の状態を保存するために必要です
• スレッド ID — どの一時停止中の実行を再開するかを識別するために必要です

---

要件

割り込みを機能させるには、次の3つが必要です。

1. チェックポインター — checkpointer=InMemorySaver()（開発環境）または PostgresSaver（本番環境）を指定してコンパイルします
2. スレッド ID — すべての invoke／stream 呼び出しに {"configurable": {"thread_id": "..."}} を渡します
3. JSON シリアライズ可能なペイロード — interrupt() に渡す値は JSON シリアライズ可能である必要があります

---

基本的な割り込みと再開

interrupt(value) はグラフを一時停止します。値は結果の __interrupt__ に表示されます。Command(resume=value) で実行を再開すると、再開値が interrupt() の戻り値になります。

重要：グラフが再開すると、ノードは最初から実行し直されます。interrupt() より前のコードはすべて再実行されます。

人間による確認のために実行を一時停止し、Command で再開します。 ```python from langgraph.types import interrupt, Command from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict

class State(TypedDict): approved: bool

def approval_node(state: State): # Pause and ask for approval approved = interrupt("Do you approve this action?") # When resumed, Command(resume=...) returns that value here return {"approved": approved}

checkpointer = InMemorySaver() graph = ( StateGraph(State) .add_node("approval", approval_node) .add_edge(START, "approval") .add_edge("approval", END) .compile(checkpointer=checkpointer) )

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

Initial run — hits interrupt and pauses

result = graph.invoke({"approved": False}, config) print(result["interrupt"])

[Interrupt(value='Do you approve this action?')]

Resume with the human's response

result = graph.invoke(Command(resume=True), config) print(result["approved"]) # True

</python>
<typescript>
人間による確認のために実行を一時停止し、Command で再開します。
```typescript
import { interrupt, Command, MemorySaver, StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  approved: z.boolean().default(false),
});

const approvalNode = async (state: typeof State.State) => {
  // Pause and ask for approval
  const approved = interrupt("Do you approve this action?");
  // When resumed, Command({ resume }) returns that value here
  return { approved };
};

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

const config = { configurable: { thread_id: "thread-1" } };

// Initial run — hits interrupt and pauses
let result = await graph.invoke({ approved: false }, config);
console.log(result.__interrupt__);
// [{ value: 'Do you approve this action?', ... }]

// Resume with the human's response
result = await graph.invoke(new Command({ resume: true }), config);
console.log(result.approved);  // true

---

承認ワークフロー

よく使われるパターンは、割り込みで下書きを提示し、人間の判断に応じて処理を振り分ける方法です。

人間の確認のために割り込み、その判断に応じて送信するか終了するかを振り分けます。 ```python from langgraph.types import interrupt, Command from langgraph.graph import StateGraph, START, END from typing import Literal from typing_extensions import TypedDict

class EmailAgentState(TypedDict): email_content: str draft_response: str classification: dict

def human_review(state: EmailAgentState) -> Command[Literal["send_reply", "end"]]: """Pause for human review using interrupt and route based on decision.""" classification = state.get("classification", {})

# interrupt() must come first — any code before it will re-run on resume
human_decision = interrupt({
    "email_id": state.get("email_content", ""),
    "draft_response": state.get("draft_response", ""),
    "urgency": classification.get("urgency"),
    "action": "Please review and approve/edit this response"
})

# Process the human's decision
if human_decision.get("approved"):
    return Command(
        update={"draft_response": human_decision.get("edited_response", state.get("draft_response", ""))},
        goto="send_reply"
    )
else:
    # Rejection — human will handle directly
    return Command(update={}, goto=END)

</python>
<typescript>
人間の確認のために割り込み、その判断に応じて送信するか終了するかを振り分けます。
```typescript
import { interrupt, Command, END, GraphNode } from "@langchain/langgraph";

const humanReview: GraphNode<typeof EmailAgentState> = async (state) => {
  const classification = state.classification!;

  // interrupt() must come first — any code before it will re-run on resume
  const humanDecision = interrupt({
    emailId: state.emailContent,
    draftResponse: state.responseText,
    urgency: classification.urgency,
    action: "Please review and approve/edit this response",
  });

  // Process the human's decision
  if (humanDecision.approved) {
    return new Command({
      update: { responseText: humanDecision.editedResponse || state.responseText },
      goto: "sendReply",
    });
  } else {
    return new Command({ update: {}, goto: END });
  }
};

---

検証ループ

interrupt() をループ内で使い、人間の入力を検証し、無効な場合は再入力を求めます。

人間の入力をループ内で検証し、有効になるまで再入力を求めます。 ```python from langgraph.types import interrupt

def get_age_node(state): prompt = "What is your age?"

while True:
    answer = interrupt(prompt)

    # Validate the input
    if isinstance(answer, int) and answer > 0:
        break
    else:
        # Invalid input — ask again with a more specific prompt
        prompt = f"'{answer}' is not a valid age. Please enter a positive number."

return {"age": answer}

各 `Command(resume=...)` 呼び出しで次の回答が渡されます。無効な場合は、より明確なメッセージを添えて再び割り込みます。
```python
config = {"configurable": {"thread_id": "form-1"}}
first = graph.invoke({"age": None}, config)
# __interrupt__: "What is your age?"

retry = graph.invoke(Command(resume="thirty"), config)
# __interrupt__: "'thirty' is not a valid age..."

final = graph.invoke(Command(resume=30), config)
print(final["age"])  # 30

人間の入力をループ内で検証し、有効になるまで再入力を求めます。 ```typescript import { interrupt } from "@langchain/langgraph";

const getAgeNode = (state: typeof State.State) => { let prompt = "What is your age?";

while (true) { const answer = interrupt(prompt);

// Validate the input
if (typeof answer === "number" && answer > 0) {
  return { age: answer };
} else {
  // Invalid input — ask again with a more specific prompt
  prompt = `'${answer}' is not a valid age. Please enter a positive number.`;
}

} };

</typescript>
</ex-validation-loop>

---

## 複数の割り込み

並列ブランチがそれぞれ `interrupt()` を呼び出す場合、各割り込み ID と再開値を対応付けて、1回の呼び出しですべてを再開します。

<ex-multiple-interrupts>
<python>
割り込み ID と値を対応付けて、複数の並列割り込みを再開します。
```python
from typing import Annotated, TypedDict
import operator
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, END, StateGraph
from langgraph.types import Command, interrupt

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

def node_a(state):
    answer = interrupt("question_a")
    return {"vals": [f"a:{answer}"]}

def node_b(state):
    answer = interrupt("question_b")
    return {"vals": [f"b:{answer}"]}

graph = (
    StateGraph(State)
    .add_node("a", node_a)
    .add_node("b", node_b)
    .add_edge(START, "a")
    .add_edge(START, "b")
    .add_edge("a", END)
    .add_edge("b", END)
    .compile(checkpointer=InMemorySaver())
)

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

# Both parallel nodes hit interrupt() and pause
result = graph.invoke({"vals": []}, config)
# result["__interrupt__"] contains both Interrupt objects with IDs

# Resume all pending interrupts at once using a map of id -> value
resume_map = {
    i.id: f"answer for {i.value}"
    for i in result["__interrupt__"]
}
result = graph.invoke(Command(resume=resume_map), config)
# result["vals"] = ["a:answer for question_a", "b:answer for question_b"]

割り込み ID と値を対応付けて、複数の並列割り込みを再開します。 ```typescript import { Command, END, MemorySaver, START, StateGraph, interrupt, isInterrupted, INTERRUPT, Annotation } from "@langchain/langgraph";

const State = Annotation.Root({ vals: Annotation<string[]>({ reducer: (left, right) => left.concat(Array.isArray(right) ? right : [right]), default: () => [], }), });

function nodeA(_state: typeof State.State) { const answer = interrupt("question_a") as string; return { vals: [a:${answer}] }; }

function nodeB(_state: typeof State.State) { const answer = interrupt("question_b") as string; return { vals: [b:${answer}] }; }

const graph = new StateGraph(State) .addNode("a", nodeA) .addNode("b", nodeB) .addEdge(START, "a") .addEdge(START, "b") .addEdge("a", END) .addEdge("b", END) .compile({ checkpointer: new MemorySaver() });

const config = { configurable: { thread_id: "1" } };

const interruptedResult = await graph.invoke({ vals: [] }, config);

// Resume all pending interrupts at once const resumeMap: Record<string, string> = {}; if (isInterrupted(interruptedResult)) { for (const i of interruptedResult[INTERRUPT]) { if (i.id != null) { resumeMap[i.id] = answer for ${i.value}; } } } const result = await graph.invoke(new Command({ resume: resumeMap }), config); // result.vals = ["a:answer for question_a", "b:answer for question_b"]

</typescript>
</ex-multiple-interrupts>

ユーザーが修正可能なエラーでは、`interrupt()` を使って実行を一時停止し、不足しているデータを収集します。このスキルではこのパターンを扱います。4段階のエラー処理戦略（RetryPolicy、Command によるエラーループなど）の全体については、**fundamentals** スキルを参照してください。

---

## 割り込み前の副作用は冪等にする必要があります

グラフが再開すると、ノードは**最初から**実行し直されます。`interrupt()` より前のコードはすべて再実行されます。サブグラフでは、親ノードとサブグラフノードの**両方**が再実行されます。

<idempotency-rules>

**推奨事項：**
- `interrupt()` の前には、insert ではなく **upsert** 操作を使います
- **作成前に確認する**パターンを使います
- 可能であれば、副作用は `interrupt()` の**後**に配置します
- 副作用を専用のノードに分離します

**避けること：**
- `interrupt()` の前に新しいレコードを作成します。再開のたびに重複します
- `interrupt()` の前にリストへ追加します。再開のたびに項目が重複します

</idempotency-rules>

<ex-idempotent-patterns>
<python>
割り込み前の冪等な操作と、冪等でない操作（誤った例）。
```python
# GOOD: Upsert is idempotent — safe before interrupt
def node_a(state: State):
    db.upsert_user(user_id=state["user_id"], status="pending_approval")
    approved = interrupt("Approve this change?")
    return {"approved": approved}

# GOOD: Side effect AFTER interrupt — only runs once
def node_a(state: State):
    approved = interrupt("Approve this change?")
    if approved:
        db.create_audit_log(user_id=state["user_id"], action="approved")
    return {"approved": approved}

# BAD: Insert creates duplicates on each resume!
def node_a(state: State):
    audit_id = db.create_audit_log({  # Runs again on resume!
        "user_id": state["user_id"],
        "action": "pending_approval",
    })
    approved = interrupt("Approve this change?")
    return {"approved": approved}

割り込み前の冪等な操作と、冪等でない操作（誤った例）。 ```typescript // GOOD: Upsert is idempotent — safe before interrupt const nodeA = async (state: typeof State.State) => { await db.upsertUser({ userId: state.userId, status: "pending_approval" }); const approved = interrupt("Approve this change?"); return { approved }; };

// GOOD: Side effect AFTER interrupt — only runs once const nodeA = async (state: typeof State.State) => { const approved = interrupt("Approve this change?"); if (approved) { await db.createAuditLog({ userId: state.userId, action: "approved" }); } return { approved }; };

// BAD: Insert creates duplicates on each resume! const nodeA = async (state: typeof State.State) => { await db.createAuditLog({ // Runs again on resume! userId: state.userId, action: "pending_approval", }); const approved = interrupt("Approve this change?"); return { approved }; };

</typescript>
</ex-idempotent-patterns>

<subgraph-interrupt-re-execution>
### 再開時のサブグラフの再実行

サブグラフに `interrupt()` が含まれている場合、再開時には、サブグラフを呼び出した親ノードと `interrupt()` を呼び出したサブグラフのノードの**両方**が再実行されます。

<python>
```python
def node_in_parent_graph(state: State):
    some_code()  # <-- Re-executes on resume
    subgraph_result = subgraph.invoke(some_input)
    # ...

def node_in_subgraph(state: State):
    some_other_code()  # <-- Also re-executes on resume
    result = interrupt("What's your name?")
    # ...

```typescript async function nodeInParentGraph(state: State) { someCode(); // <-- Re-executes on resume const subgraphResult = await subgraph.invoke(someInput); // ... }

async function nodeInSubgraph(state: State) { someOtherCode(); // <-- Also re-executes on resume const result = interrupt("What's your name?"); // ... }

</typescript>
</subgraph-interrupt-re-execution>

---

## Command(resume) に関する警告

`Command(resume=...)` は、`invoke()`／`stream()` の入力として使うことを意図した唯一の Command パターンです。入力として `Command(update=...)` を渡してはいけません。最新のチェックポイントから再開されるため、グラフが停止したように見えます。アンチパターンの詳しい説明については、fundamentals スキルを参照してください。

---

## 修正方法

<fix-checkpointer-required-for-interrupts>
<python>
interrupt 機能にはチェックポインターが必要です。
```python
# WRONG
graph = builder.compile()

# CORRECT
graph = builder.compile(checkpointer=InMemorySaver())

interrupt 機能にはチェックポインターが必要です。 ```typescript // WRONG const graph = builder.compile();

// CORRECT const graph = builder.compile({ checkpointer: new MemorySaver() });

</typescript>
</fix-checkpointer-required-for-interrupts>

<fix-resume-with-command>
<python>
interrupt から再開するには Command を使用してください（通常の dict ではグラフが再起動します）。
```python
# WRONG
graph.invoke({"resume_data": "approve"}, config)

# CORRECT
graph.invoke(Command(resume="approve"), config)

interrupt から再開するには Command を使用してください（通常のオブジェクトではグラフが再起動します）。 ```typescript // WRONG await graph.invoke({ resumeData: "approve" }, config);

// CORRECT await graph.invoke(new Command({ resume: "approve" }), config);

</typescript>
</fix-resume-with-command>

<boundaries>
### してはいけないこと

- チェックポインターなしで割り込みを使用する — 失敗します
- 同じ thread_id を使わずに再開する — 再開ではなく新しいスレッドが作成されます
- `Command(update=...)` を invoke の入力として渡す — グラフが停止したように見えます（通常の dict を使用してください）
- `interrupt()` の前に冪等でない副作用を実行する — 再開時に重複が発生します
- `interrupt()` より前のコードは一度しか実行されないと想定する — 再開のたびに再実行されます
</boundaries>