인터럽트

인터럽트 (Interrupts)

인터럽트(interrupt)는 특정 지점에서 그래프 실행을 일시 중지하고, 계속하기 전에 외부 입력을 기다리게 해줍니다. 이는 진행하기 위해 외부 입력이 필요한 human-in-the-loop 패턴을 가능하게 해요. 인터럽트가 트리거되면 LangGraph는 persistence 레이어로 그래프 상태를 저장하고, 여러분이 실행을 재개할 때까지 무기한 기다립니다.

출처: 문서

본문

인터럽트는 그래프 노드의 어느 지점에서든 interrupt() 함수를 호출해 동작합니다. 함수는 호출자에게 표면화되는 JSON 직렬화 가능한 값을 받아요. 계속할 준비가 되면 Command로 그래프를 다시 호출해 실행을 재개하며, 이것이 노드 안의 interrupt() 호출의 반환 값이 됩니다.

정적 브레이크포인트(특정 노드 전후에 일시 중지)와 달리, 인터럽트는 동적(dynamic) 입니다. 코드의 어디에나 놓일 수 있고 애플리케이션 로직에 따라 조건적일 수 있어요.

• 체크포인팅이 위치를 보존합니다: 체크포인터가 정확한 그래프 상태를 기록해서, 오류 상태에서도 나중에 재개할 수 있어요.
• thread_id가 포인터입니다: config={"configurable": {"thread_id": ...}}를 설정해 체크포인터에 어떤 상태를 로드할지 알려주세요.
• 인터럽트 페이로드는 stream.interrupts로 표면화됩니다: 이벤트 스트리밍(graph.stream_events(..., version="v3"))을 사용할 때 interrupt()에 전달한 값이 stream.interrupts에 나타나고, 런이 입력을 위해 멈추면 stream.interrupted가 True가 됩니다.

선택한 thread_id는 사실상 영속 커서입니다. 같은 값을 재사용하면 같은 체크포인트를 재개하고, 새 값을 사용하면 빈 상태로 완전히 새로운 스레드를 시작합니다.

interrupt로 일시 중지하기 (Pause using interrupt)

interrupt 함수는 그래프 실행을 일시 중지하고 값을 호출자에게 반환합니다. 노드 안에서 interrupt를 호출하면 LangGraph가 현재 그래프 상태를 저장하고, 입력으로 실행을 재개할 때까지 기다립니다.

interrupt를 사용하려면 다음이 필요합니다:

1. 그래프 상태를 영속화할 체크포인터(프로덕션에서는 내구성 있는 체크포인터 사용)
2. 런타임이 어떤 상태에서 재개할지 알도록 config의 스레드 ID
3. 멈추려는 지점에서 interrupt() 호출(페이로드는 JSON 직렬화 가능해야 함)

from langgraph.types import interrupt

def approval_node(state: State):
    # Pause and ask for approval
    approved = interrupt("Do you approve this action?")

    # When you resume, Command(resume=...) returns that value here
    return {"approved": approved}

interrupt를 호출하면 다음과 같은 일이 일어납니다:

1. 그래프 실행이 멈춥니다: interrupt가 호출된 정확한 지점에서.
2. 상태가 저장됩니다: 체크포인터로 실행을 나중에 재개할 수 있게요. 프로덕션에서는 영속 체크포인터(예: 데이터베이스 기반)여야 합니다.
3. 값이 호출자에게 반환됩니다: 이벤트 스트리밍(graph.stream_events(..., version="v3"))을 쓰면 stream.interrupts에, 기본 invoke() API에서는 __interrupt__ 아래에요. 문자열, 객체, 배열 등 어떤 JSON 직렬화 가능한 값이든 될 수 있습니다.
4. 그래프가 무기한 기다립니다: 응답으로 실행을 재개할 때까지.
5. 응답이 노드로 전달됩니다: 재개할 때 interrupt() 호출의 반환 값이 됩니다.

인터럽트 재개 (Resuming interrupts)

인터럽트가 실행을 멈춘 뒤에는 resume 값이 담긴 Command로 그래프를 다시 호출해 재개해요. resume 값이 interrupt 호출로 전달되어, 노드가 외부 입력으로 실행을 계속할 수 있게 해줍니다.

인터럽트할 수 있는 그래프를 구동하는 권장 방법은 이벤트 스트리밍입니다. stream.interrupts와 stream.interrupted로 인터럽트를 표면화하고, stream.output으로 최종 상태를 노출합니다.

from langgraph.types import Command

# Initial run - hits the interrupt and pauses
# thread_id is the persistent pointer (stores a stable ID in production)
config = {"configurable": {"thread_id": "thread-1"}}
stream = graph.stream_events({"input": "data"}, config=config, version="v3")

# Drain the stream to drive the run; stream.output awaits the final state.
final = stream.output

# stream.interrupted is True when the run paused for human input, and
# stream.interrupts contains the payloads passed to interrupt().
if stream.interrupted:
    print(stream.interrupts)
    # > (Interrupt(value='Do you approve this action?'),)

# Resume with the human's response
# The resume payload becomes the return value of interrupt() inside the node
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output

기본 `graph.invoke(...)` API도 여전히 동작하며 `result["__interrupt__"]` 아래에 인터럽트를 표면화합니다. 스트리밍 프로젝션이 필요 없을 때 사용하세요. 그 외에는 `graph.stream_events(..., version="v3")`를 선호합니다.

재개에 관한 핵심 포인트:

• 재개할 때 인터럽트가 발생했을 때 사용했던 같은 스레드 ID를 사용해야 합니다.
• Command(resume=...)에 전달한 값이 interrupt 호출의 반환 값이 됩니다.
• 재개 시 노드는 interrupt가 호출된 노드의 처음부터 다시 시작하므로, interrupt 이전의 모든 코드가 다시 실행됩니다.
• resume 값으로 어떤 JSON 직렬화 가능한 값이든 전달할 수 있습니다.

`Command(resume=...)`는 `invoke()`/`stream()`/`stream_events()`의 입력으로 의도된 **유일한** `Command` 패턴입니다. 다른 `Command` 파라미터(`update`, `goto`, `graph`)는 [노드 함수에서 반환](/oss/python/langgraph/graph-api#command)하도록 설계됐어요. 다중 턴 대화를 계속하기 위해 `Command(update=...)`를 입력으로 전달하지 마세요 — 일반 입력 dict를 전달하세요.

일반 패턴 (Common patterns)

인터럽트가 여는 핵심 기능은 실행을 멈추고 외부 입력을 기다리는 것입니다. 이는 다양한 사용 사례에 유용합니다:

• 승인 워크플로우: 중요한 작업(API 호출, DB 변경, 금융 거래) 실행 전에 일시 중지.
• 여러 인터럽트 처리: 단일 호출에서 여러 인터럽트를 재개할 때 인터럽트 ID를 resume 값과 짝짓기.
• 검토 및 편집: 계속 전에 인간이 LLM 출력이나 도구 호출을 검토·수정하게 하기.
• 도구 호출 인터럽트: 도구 호출 실행 전에 일시 중지해 검토·편집.
• 인간 입력 검증: 다음 단계로 진행 전에 인간 입력 검증을 위해 일시 중지.

human-in-the-loop(HITL) 인터럽트로 스트리밍 (Stream with human-in-the-loop interrupts)

human-in-the-loop 워크플로우로 대화형 에이전트를 만들 때 이벤트 스트리밍으로 인터럽트를 처리하면서 메시지 청크와 상태 스냅샷을 동시에 소비할 수 있어요.

graph.stream_events(..., version="v3")가 반환한 타입 있는 프로젝션을 런이 끝날 때까지 루프로 사용하세요:

• stream.messages로 AI 응답을 토큰 단위로 스트리밍
• stream.values로 단계별 상태 스냅샷 관찰
• stream.interrupted로 인터럽트 감지, stream.interrupts에서 페이로드 읽기
• Command(resume=...)로 stream_events를 다시 호출해 재개, stream.interrupted가 false가 될 때까지 반복

from langgraph.types import Command

stream_input: dict | Command = initial_input

while True:
    stream = graph.stream_events(stream_input, config=config, version="v3")

    # Stream LLM message chunks (including any in subgraphs) as they arrive.
    for message in stream.messages:
        for token in message.text:
            display_streaming_content(token)

    # After the run finishes (or pauses), check for interrupts and resume.
    if not stream.interrupted:
        final_state = stream.output
        break

    interrupt_info = stream.interrupts[0].value
    user_response = get_user_input(interrupt_info)
    stream_input = Command(resume=user_response)

• stream.messages: 콘텐츠 블록으로 된 채팅 모델 출력. 토큰 델타를 위해 각 message.text를 순회합니다. 중첩 서브그래프는 stream.subgraphs[*].messages에서 읽으세요.
• stream.values: 각 단계 후 전체 상태 스냅샷.
• stream.interrupted / stream.interrupts: 각 런 후 그래프가 멈췄는지 확인, stream.interrupts에서 페이로드 읽기.
• Command(resume=...): 다음 stream_events 입력으로 전달해 재개. 인터럽트 없이 런이 완료될 때까지 루프.

여러 인터럽트 처리 (Handling multiple interrupts)

병렬 분기가 동시에 인터럽트할 때(예: 각자 interrupt()를 호출하는 여러 노드로 fan-out), 단일 호출에서 여러 인터럽트를 재개해야 할 수 있어요. 단일 호출로 여러 인터럽트를 재개할 때는 각 인터럽트 ID를 해당 resume 값에 매핑하세요. 이렇게 하면 런타임에서 각 응답이 올바른 인터럽트와 짝을 이룹니다.

from typing import Annotated, TypedDict
import operator

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, 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"}}

# Step 1: stream events to drive the run; both parallel nodes hit interrupt() and pause
stream = graph.stream_events({"vals": []}, config, version="v3")
_ = stream.output  # drive the stream to completion
# stream.interrupts contains the pending Interrupt payloads
print(stream.interrupts)
# > (Interrupt(value='question_a', id='...'), Interrupt(value='question_b', id='...'))

# Step 2: resume all pending interrupts at once
resume_map = {
    i.id: f"answer for {i.value}" for i in stream.interrupts
}
resumed = graph.stream_events(Command(resume=resume_map), config, version="v3")

print("Final state:", resumed.output)
# Final state: {'vals': ['a:answer for question_a', 'b:answer for question_b']}

승인 또는 거부 (Approve or reject)

인터럽트의 가장 흔한 용도 중 하나는 중요한 작업 전에 멈추고 승인을 요청하는 것입니다. 예를 들어 API 호출, 데이터베이스 변경, 또는 다른 중요한 결정에 대해 인간에게 승인을 요청할 수 있어요.

from typing import Literal
from langgraph.types import interrupt, Command

def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
    # Pause execution; payload shows up on stream.interrupts (with stream_events) or result["__interrupt__"] (with invoke)
    is_approved = interrupt({
        "question": "Do you want to proceed with this action?",
        "details": state["action_details"]
    })

    # Route based on the response
    if is_approved:
        return Command(goto="proceed")  # Runs after the resume payload is provided
    else:
        return Command(goto="cancel")

그래프를 재개할 때 승인하려면 True, 거부하려면 False를 전달하세요:

# To approve
graph.stream_events(Command(resume=True), config=config, version="v3").output

# To reject
graph.stream_events(Command(resume=False), config=config, version="v3").output

완전한 예시:

from typing import Literal, Optional, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class ApprovalState(TypedDict):
    action_details: str
    status: Optional[Literal["pending", "approved", "rejected"]]


def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:
    # Expose details so the caller can render them in a UI
    decision = interrupt(
        {
            "question": "Approve this action?",
            "details": state["action_details"],
        }
    )

    # Route to the appropriate node after resume
    return Command(goto="proceed" if decision else "cancel")


def proceed_node(state: ApprovalState):
    return {"status": "approved"}


def cancel_node(state: ApprovalState):
    return {"status": "rejected"}


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_node("proceed", proceed_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("proceed", END)
builder.add_edge("cancel", END)

# Use a more durable checkpointer in production
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "approval-123"}}
initial = graph.stream_events(
    {"action_details": "Transfer $500", "status": "pending"},
    config=config,
    version="v3",
)
_ = initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'question': ..., 'details': ...}),)

# Resume with the decision; True routes to proceed, False to cancel
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
print(resumed.output["status"])

상태 검토 및 편집 (Review and edit state)

때로는 계속하기 전에 인간이 그래프 상태의 일부를 검토하고 편집하게 하고 싶을 수 있어요. LLM을 수정하거나, 누락 정보를 추가하거나, 조정을 가할 때 유용합니다.

from langgraph.types import interrupt

def review_node(state: State):
    # Pause and show the current content for review (payload surfaces on stream.interrupts)
    edited_content = interrupt({
        "instruction": "Review and edit this content",
        "content": state["generated_text"]
    })

    # Update the state with the edited version
    return {"generated_text": edited_content}

재개할 때 편집된 콘텐츠를 제공하세요:

graph.stream_events(
    Command(resume="The edited and improved text"),  # Value becomes the return from interrupt()
    config=config,
    version="v3",
).output

완전한 예시:

from typing import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class ReviewState(TypedDict):
    generated_text: str


def review_node(state: ReviewState):
    # Ask a reviewer to edit the generated content
    updated = interrupt(
        {
            "instruction": "Review and edit this content",
            "content": state["generated_text"],
        }
    )
    return {"generated_text": updated}


builder = StateGraph(ReviewState)
builder.add_node("review", review_node)
builder.add_edge(START, "review")
builder.add_edge("review", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "review-42"}}
initial = graph.stream_events(
    {"generated_text": "Initial draft"}, config=config, version="v3"
)
_ = initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'instruction': ..., 'content': ...}),)

# Resume with the edited text from the reviewer
final_state = graph.stream_events(
    Command(resume="Improved draft after review"),
    config=config,
    version="v3",
)
print(final_state.output["generated_text"])  # -> "Improved draft after review"

도구 안에서의 인터럽트 (Interrupts in tools)

도구 함수 안에 직접 인터럽트를 놓을 수도 있어요. 이렇게 하면 도구가 호출될 때마다 승인을 위해 멈추고, 도구 호출이 실행되기 전에 인간이 검토·편집할 수 있습니다.

먼저 interrupt를 사용하는 도구를 정의하세요:

from langchain.tools import tool
from langgraph.types import interrupt

@tool
def send_email(to: str, subject: str, body: str):
    """Send an email to a recipient."""

    # Pause before sending; payload surfaces on stream.interrupts when using event streaming
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?"
    })

    if response.get("action") == "approve":
        # Resume value can override inputs before executing
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        final_body = response.get("body", body)
        return f"Email sent to {final_to} with subject '{final_subject}'"
    return "Email cancelled by user"

이 접근은 승인 로직이 도구 자체에 있어서 그래프의 여러 부분에서 재사용할 수 있게 만들 때 유용해요. LLM이 자연스럽게 도구를 호출할 수 있고, 도구가 호출될 때마다 인터럽트가 실행을 멈춰 작업을 승인·편집·취소할 수 있습니다.

완전한 예시:

import sqlite3
import operator
from typing import TypedDict, Annotated, Literal
from langchain.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain.messages import AnyMessage, SystemMessage, ToolMessage


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


@tool
def send_email(to: str, subject: str, body: str):
    """Send an email to a recipient."""

    # Pause before sending; payload surfaces on stream.interrupts when using event streaming
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?",
    })

    if response.get("action") == "approve":
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        final_body = response.get("body", body)

        # Actually send the email (your implementation here)
        print(f"[send_email] to={final_to} subject={final_subject} body={final_body}")
        return f"Email sent to {final_to}"

    return "Email cancelled by user"


model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools([send_email])
tools_by_name = {"send_email": send_email}


def agent_node(state: AgentState):
    # LLM may decide to call the tool; interrupt pauses before sending
    result = model.invoke(state["messages"])
    return {"messages": [result]}

def tool_node(state: AgentState):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

def should_continue(state: AgentState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
    messages = state["messages"]
    last_message = messages[-1]

    if last_message.tool_calls:
        return "tool_node"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tool_node", tool_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tool_node", END])  # Routes to "tools" or END
builder.add_edge("tool_node", "agent")  # Loop back after tools

checkpointer = SqliteSaver(
    sqlite3.connect("tool-approval.db", check_same_thread=False)
)
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "email-workflow"}}
initial = graph.stream_events(
    {
        "messages": [
            {"role": "user", "content": "Send an email to [email protected] about the meeting"}
        ]
    },
    config=config,
    version="v3",
)
initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'action': 'send_email', ...}),)

# Resume with approval and optionally edited arguments
resumed = graph.stream_events(
    Command(resume={"action": "approve", "subject": "Updated subject"}),
    config=config,
    version="v3",
)
print(resumed.output["messages"][-1])  # -> Tool result returned by send_email

인간 입력 검증 (Validating human input)

때로는 인간의 입력을 검증하고 값이 유효하지 않으면 다시 물어봐야 합니다. 권장 접근은 노드 호출당 딱 한 번 interrupt()를 호출하고, 오류 메시지를 상태에 저장한 채 노드에서 반환한 뒤, 유효한 값이 제공될 때까지 조건부 엣지로 노드로 되돌려 보내는 것입니다.

**단일 노드 안에서 `while True` + `interrupt()` 루프를 피하세요.** 노드는 재개할 때마다 처음부터 다시 실행되므로([Rules of interrupts](#rules-of-interrupts) 참고), `interrupt()`를 여러 번 호출하는 루프는 각 재개가 이전의 모든 반복을 재생하게 합니다. 첫 번째 재개는 1회 반복을, 두 번째는 2회를, 이런 식으로요. 결과적으로 루프 본문 안의 코드가 지수적으로 재실행됩니다.

올바른 패턴:

1. 재질문 질문을 상태에 저장(예: pending_question).
2. 노드에서 interrupt()를 정확히 한 번 호출, 상태의 현재 질문을 전달.
3. 답이 유효하지 않으면 갱신된 pending_question을 반환해 다음 호출이 다시 물어보게.
4. 유효한 값이 수집될 때까지 add_conditional_edges로 노드로 다시 라우팅.

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt


class FormState(TypedDict):
    age: int | None
    pending_question: str | None


def get_age_node(state: FormState):
    question = state.get("pending_question") or "What is your age?"
    answer = interrupt(question)  # called exactly once per invocation
    if isinstance(answer, int) and answer > 0:
        return {"age": answer, "pending_question": None}
    return {"pending_question": f"'{answer}' is not a valid age. Please enter a positive number."}


def route(state: FormState):
    return END if state.get("age") is not None else "collect_age"


builder = StateGraph(FormState)
builder.add_node("collect_age", get_age_node)
builder.add_edge(START, "collect_age")
builder.add_conditional_edges("collect_age", route)

각 재개는 get_age_node를 정확히 한 번, interrupt() 호출을 한 번 실행하고 종료합니다. 답이 유효하지 않으면 조건부 엣지가 되돌려 보내고 다음 인터럽트가 갱신된 질문으로 다시 물어봅니다. 재개당 코드가 두 번 이상 실행되지 않습니다.

완전한 예시:

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class FormState(TypedDict):
    age: int | None
    pending_question: str | None


def get_age_node(state: FormState):
    question = state.get("pending_question") or "What is your age?"
    answer = interrupt(question)  # called exactly once per node invocation
    print(f"I got {answer}")  # runs exactly once per resume
    if isinstance(answer, int) and answer > 0:
        return {"age": answer, "pending_question": None}
    return {"pending_question": f"'{answer}' is not a valid age. Please enter a positive number."}


def route(state: FormState):
    # Loop back to collect_age until we have a valid age
    return END if state.get("age") is not None else "collect_age"


builder = StateGraph(FormState)
builder.add_node("collect_age", get_age_node)
builder.add_edge(START, "collect_age")
builder.add_conditional_edges("collect_age", route)

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

config = {"configurable": {"thread_id": "form-1"}}
first = graph.stream_events({"age": None, "pending_question": None}, config=config, version="v3")
_ = first.output  # drive the stream to completion
print(first.interrupts)  # -> (Interrupt(value='What is your age?', ...),)

# Provide invalid data; the node re-prompts via the conditional edge
retry = graph.stream_events(Command(resume="thirty"), config=config, version="v3")
_ = retry.output
print(retry.interrupts)  # -> (Interrupt(value="'thirty' is not a valid age...", ...),)

# Provide valid data; route() returns END and the graph finishes
final = graph.stream_events(Command(resume=30), config=config, version="v3")
print(final.output["age"])  # -> 30

인터럽트 규칙 (Rules of interrupts)

노드 안에서 interrupt를 호출하면 LangGraph는 런타임에 일시 정지하라는 신호를 보내는 예외를 일으켜 실행을 중단합니다. 이 예외는 호출 스택을 따라 올라가 런타임이 잡아내고, 그래프에 현재 상태를 저장하고 외부 입력을 기다리라고 알립니다.

실행이 재개되면(요청한 입력을 제공한 후) 런타임은 노드 전체를 처음부터 다시 시작합니다 — interrupt가 호출된 정확한 줄부터 재개하지 않아요. 이는 interrupt 이전에 실행된 코드가 다시 실행된다는 뜻입니다. 그래서 인터럽트가 예상대로 동작하도록 몇 가지 중요한 규칙을 따라야 합니다.

interrupt 호출을 try/except로 감싸지 마세요

interrupt가 호출 지점에서 실행을 멈추는 방식은 특수 예외를 던지는 것입니다. interrupt 호출을 try/except 블록으로 감싸면 이 예외를 잡아내 인터럽트가 그래프로 전달되지 않습니다.

• ✅ interrupt 호출을 오류 발생 가능 코드와 분리
• ✅ try/except 블록에서 특정 예외 타입 사용

def node_a(state: State):
    # ✅ Good: interrupting first, then handling
    # error conditions separately
    interrupt("What's your name?")
    try:
        fetch_data()  # This can fail
    except Exception as e:
        print(e)
    return state

def node_a(state: State):
    # ✅ Good: catching specific exception types
    # will not catch the interrupt exception
    try:
        name = interrupt("What's your name?")
        fetch_data()  # This can fail
    except NetworkException as e:
        print(e)
    return state

• 🔴 interrupt 호출을 bare try/except 블록으로 감싸지 마세요

def node_a(state: State):
    # ❌ Bad: wrapping interrupt in bare try/except
    # will catch the interrupt exception
    try:
        interrupt("What's your name?")
    except Exception as e:
        print(e)
    return state

노드 안에서 interrupt 호출 순서를 바꾸지 마세요

단일 노드에서 여러 인터럽트를 사용하는 것은 흔하지만, 주의해서 다루지 않으면 예상치 못한 동작이 발생할 수 있어요.

노드에 여러 인터럽트 호출이 있으면 LangGraph는 그 노드를 실행하는 태스크 특화 resume 값 목록을 유지합니다. 실행이 재개될 때마다 노드의 처음부터 시작합니다. 만나는 각 인터럽트에 대해 태스크의 resume 목록에 일치하는 값이 있는지 확인합니다. 매칭은 엄격히 인덱스 기반이므로, 노드 안의 인터럽트 호출 순서가 중요합니다.

• ✅ 노드 실행 간에 interrupt 호출을 일관되게 유지

def node_a(state: State):
    # ✅ Good: interrupt calls happen in the same order every time
    name = interrupt("What's your name?")
    age = interrupt("What's your age?")
    city = interrupt("What's your city?")

    return {
        "name": name,
        "age": age,
        "city": city
    }

• 🔴 노드 안에서 interrupt 호출을 조건부로 건너뛰지 마세요
• 🔴 실행 간에 결정적이지 않은 로직( while True 검증 루프 포함)으로 interrupt 호출을 루프시키지 마세요. 대신 조건부 엣지를 사용하세요 (인간 입력 검증 참고)

def node_a(state: State):
    # ❌ Bad: conditionally skipping interrupts changes the order
    name = interrupt("What's your name?")

    # On first run, this might skip the interrupt
    # On resume, it might not skip it - causing index mismatch
    if state.get("needs_age"):
        age = interrupt("What's your age?")

    city = interrupt("What's your city?")

    return {"name": name, "city": city}

def node_a(state: State):
    # ❌ Bad: looping based on non-deterministic data
    # The number of interrupts changes between executions
    results = []
    for item in state.get("dynamic_list", []):  # List might change between runs
        result = interrupt(f"Approve {item}?")
        results.append(result)

    return {"results": results}

interrupt 호출에 복잡한 값을 반환하지 마세요

사용하는 체크포인터에 따라 복잡한 값은 직렬화되지 못할 수 있어요(예: 함수는 직렬화할 수 없음). 그래프를 어떤 배포에도 적응시키려면 합리적으로 직렬화할 수 있는 값만 사용하는 것이 좋습니다.

• ✅ interrupt에 단순하고 JSON 직렬화 가능한 타입 전달
• ✅ 단순 값이 담긴 사전/객체 전달

def node_a(state: State):
    # ✅ Good: passing simple types that are serializable
    name = interrupt("What's your name?")
    count = interrupt(42)
    approved = interrupt(True)

    return {"name": name, "count": count, "approved": approved}

def node_a(state: State):
    # ✅ Good: passing dictionaries with simple values
    response = interrupt({
        "question": "Enter user details",
        "fields": ["name", "email", "age"],
        "current_values": state.get("user", {})
    })

    return {"user": response}

• 🔴 interrupt에 함수, 클래스 인스턴스, 기타 복잡한 객체를 전달하지 마세요

def validate_input(value):
    return len(value) > 0

def node_a(state: State):
    # ❌ Bad: passing a function to interrupt
    # The function cannot be serialized
    response = interrupt({
        "question": "What's your name?",
        "validator": validate_input  # This will fail
    })
    return {"name": response}

class DataProcessor:
    def __init__(self, config):
        self.config = config

def node_a(state: State):
    processor = DataProcessor({"mode": "strict"})

    # ❌ Bad: passing a class instance to interrupt
    # The instance cannot be serialized
    response = interrupt({
        "question": "Enter data to process",
        "processor": processor  # This will fail
    })
    return {"result": response}

interrupt 이전에 호출되는 부작용은 멱등해야 합니다

인터럽트는 호출된 노드를 다시 실행하는 방식으로 동작하므로, interrupt 이전에 호출되는 부작용은 (이상적으로는) 멱등(idempotent)해야 해요. 멱등성이란 같은 연산을 여러 번 적용해도 초기 실행 이후의 결과가 바뀌지 않는다는 뜻입니다.

예를 들어 노드 안에 레코드를 갱신하는 API 호출이 있을 수 있어요. interrupt가 그 호출 뒤에 오면, 노드가 재개될 때 여러 번 다시 실행되어 초기 갱신을 덮어쓰거나 중복 레코드를 만들 수 있습니다.

• ✅ interrupt 이전에 멱등 연산 사용
• ✅ 부작용을 interrupt 호출 뒤에 배치
• ✅ 가능하면 부작용을 별도 노드로 분리

def node_a(state: State):
    # ✅ Good: using upsert operation which is idempotent
    # Running this multiple times will have the same result
    db.upsert_user(
        user_id=state["user_id"],
        status="pending_approval"
    )

    approved = interrupt("Approve this change?")

    return {"approved": approved}

def node_a(state: State):
    # ✅ Good: placing side effect after the interrupt
    # This ensures it only runs once after approval is received
    approved = interrupt("Approve this change?")

    if approved:
        db.create_audit_log(
            user_id=state["user_id"],
            action="approved"
        )

    return {"approved": approved}

def approval_node(state: State):
    # ✅ Good: only handling the interrupt in this node
    approved = interrupt("Approve this change?")

    return {"approved": approved}

def notification_node(state: State):
    # ✅ Good: side effect happens in a separate node
    # This runs after approval, so it only executes once
    if (state.approved):
        send_notification(
            user_id=state["user_id"],
            status="approved"
        )

    return state

• 🔴 interrupt 이전에 비멱등 연산을 수행하지 마세요
• 🔴 존재 여부를 확인하지 않고 새 레코드를 만들지 마세요

def node_a(state: State):
    # ❌ Bad: creating a new record before interrupt
    # This will create duplicate records on each resume
    audit_id = db.create_audit_log({
        "user_id": state["user_id"],
        "action": "pending_approval",
        "timestamp": datetime.now()
    })

    approved = interrupt("Approve this change?")

    return {"approved": approved, "audit_id": audit_id}

def node_a(state: State):
    # ❌ Bad: appending to a list before interrupt
    # This will add duplicate entries on each resume
    db.append_to_history(state["user_id"], "approval_requested")

    approved = interrupt("Approve this change?")

    return {"approved": approved}

함수로 호출되는 서브그래프와 함께 사용 (Using with subgraphs called as functions)

노드 안에서 서브그래프를 호출하면, 부모 그래프는 서브그래프가 호출되고 interrupt가 트리거된 노드의 시작에서 실행을 재개합니다. 마찬가지로 서브그래프도 interrupt가 호출된 노드의 시작에서 재개됩니다.

def node_in_parent_graph(state: State):
    some_code()  # <-- This will re-execute when resumed
    # Invoke a subgraph as a function.
    # The subgraph contains an `interrupt` call.
    subgraph_result = subgraph.invoke(some_input)
    # ...

def node_in_subgraph(state: State):
    some_other_code()  # <-- This will also re-execute when resumed
    result = interrupt("What's your name?")
    # ...

인터럽트로 디버깅 (Debugging with interrupts)

그래프를 디버깅·테스트하려면 정적 인터럽트를 브레이크포인트로 사용해 그래프 실행을 한 노드씩 진행할 수 있어요. 정적 인터럽트는 노드가 실행되기 전이나 후에 정의된 지점에서 트리거됩니다. 그래프를 컴파일할 때 interrupt_before와 interrupt_after를 지정해 설정할 수 있어요.

정적 인터럽트는 human-in-the-loop 워크플로우에는 **권장되지 않습니다**. 대신 [`interrupt`](https://reference.langchain.com/python/langgraph/types/interrupt) 함수를 사용하세요.

컴파일 시점 설정:

graph = builder.compile(
    interrupt_before=["node_a"],  # [!code highlight]
    interrupt_after=["node_b", "node_c"],  # [!code highlight]
    checkpointer=checkpointer,
)

# Pass a thread ID to the graph
config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# Run the graph until the breakpoint
graph.invoke(inputs, config=config)  # [!code highlight]

# Resume the graph
graph.invoke(None, config=config)  # [!code highlight]

1. 브레이크포인트는 compile 시점에 설정됩니다.
2. interrupt_before는 노드가 실행되기 전에 실행을 멈출 노드를 지정합니다.
3. interrupt_after는 노드가 실행된 후에 실행을 멈출 노드를 지정합니다.
4. 브레이크포인트를 활성화하려면 체크포인터가 필요합니다.
5. 그래프는 첫 브레이크포인트에 도달할 때까지 실행됩니다.
6. 입력에 None을 전달해 그래프를 재개합니다. 이는 다음 브레이크포인트에 도달할 때까지 그래프를 실행합니다.

런타임 설정:

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

# Run the graph until the breakpoint
graph.invoke(
    inputs,
    interrupt_before=["node_a"],  # [!code highlight]
    interrupt_after=["node_b", "node_c"],  # [!code highlight]
    config=config,
)

# Resume the graph
graph.invoke(None, config=config)  # [!code highlight]

1. graph.invoke를 interrupt_before와 interrupt_after 파라미터와 함께 호출합니다. 이는 런타임 구성이며 호출마다 변경할 수 있습니다.
2. interrupt_before는 노드 실행 전에 멈출 노드를 지정합니다.
3. interrupt_after는 노드 실행 후에 멈출 노드를 지정합니다.
4. 그래프는 첫 브레이크포인트에 도달할 때까지 실행됩니다.
5. 입력에 None을 전달해 그래프를 재개합니다. 이는 다음 브레이크포인트에 도달할 때까지 그래프를 실행합니다.

인터럽트를 디버그하려면 [LangSmith](/langsmith/observability)를 사용하세요.

LangSmith Studio 사용 (Using LangSmith Studio)

LangSmith Studio를 사용해 그래프를 실행하기 전에 UI에서 그래프에 정적 인터럽트를 설정할 수 있어요. 또한 UI로 실행의 어느 지점에서든 그래프 상태를 검사할 수 있습니다.

더 알아보기 (Learn more)

• event streaming — 인터럽트를 stream.interrupts로 소비.
• checkpointers — 인터럽트용 상태 영속화.
• persistence — 스레드 영속화.