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 で再開します。

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

人による確認のため実行を一時停止し、Command で再開します。

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

---

承認ワークフロー

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

人による確認のために割り込み、その判断に基づいて送信または終了に振り分けます。

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)

人による確認のために割り込み、その判断に基づいて送信または終了に振り分けます。

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() をループ内で使用して人の入力を検証し、無効な場合は再度プロンプトを表示します。

人の入力をループ内で検証し、有効になるまで再度プロンプトを表示します。

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=...) を呼び出すたびに、次の回答が渡されます。無効な場合、ループはより明確なメッセージで再度割り込みます。

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

人の入力をループ内で検証し、有効になるまで再度プロンプトを表示します。

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.`;
    }
  }
};

---

複数の割り込み

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

割り込み ID と値を対応付けて、複数の並列割り込みを再開します。

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 と値を対応付けて、複数の並列割り込みを再開します。

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"]

ユーザーが修正できるエラーには、interrupt() を使って一時停止し、不足しているデータを収集します。このスキルではこのパターンを扱います。4 層からなるエラー処理戦略（RetryPolicy、Command によるエラーループなど）の詳細については、fundamentals スキルを参照してください。

---

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

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

推奨事項：

• interrupt() の前には、insert ではなく upsert 操作を使用します
• 作成前に確認するパターンを使用します
• 可能であれば、副作用を interrupt() の後に配置します
• 副作用をそれぞれ独立したノードに分けます

避けること：

• interrupt() の前にレコードを新規作成すること — 再開のたびに重複します
• interrupt() の前にリストへ追加すること — 再開のたびに項目が重複します

割り込み前の冪等な操作と、冪等でない操作（誤り）です。

# 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}

割り込み前の冪等な操作と、冪等でない操作（誤り）です。

// 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 };
};

### 再開時のサブグラフの再実行

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

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?")
    # ...

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?");
  // ...
}

---

Command(resume) に関する警告

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

---

修正方法

interrupt 機能にはチェックポインターが必要です。

# WRONG
graph = builder.compile()

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

interrupt 機能にはチェックポインターが必要です。

// WRONG
const graph = builder.compile();

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

interrupt から再開するには Command を使用します（通常の dict を使うとグラフが最初から実行されます）。

# WRONG
graph.invoke({"resume_data": "approve"}, config)

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

interrupt から再開するには Command を使用します（通常のオブジェクトを使うとグラフが最初から実行されます）。

// WRONG
await graph.invoke({ resumeData: "approve" }, config);

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

### やってはいけないこと

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