인터럽트 (Interrupts)

인터럽트가 트리거되면 LangGraph는 영속성(persistence) 레이어를 이용해 그래프 상태를 저장하고, 실행을 재개할 때까지 무기한 기다립니다.

인터럽트는 그래프의 노드 안 어느 지점에서든 interrupt() 함수를 호출해 동작합니다. 이 함수는 호출자에게 전달되는 JSON-직렬화 가능한 값이면 무엇이든 받아들입니다. 계속할 준비가 되면 Command를 사용해 그래프를 다시 호출(resume)하는데, 이 값이 노드 안 interrupt() 호출의 반환값이 됩니다.

정적 중단점(static breakpoints, 특정 노드 앞이나 뒤에서 멈추는 방식)과 달리 인터럽트는 **동적(dynamic)**입니다. 코드 어디에든 둘 수 있고 애플리케이션 로직에 따라 조건적으로 걸 수도 있어요.

인터럽트 페이로드는 stream.interrupts로 노출됩니다. 이벤트 스트리밍(graph.stream_events(..., version="v3"))을 사용할 때 interrupt()에 넘긴 값이 stream.interrupts에 나타나고, 실행이 입력을 기다리며 멈추면 stream.interrupted가 True가 됩니다.
선택한 thread_id는 사실상 영구 커서(permanent cursor) 역할을 해요. 같은 값을 재사용하면 같은 체크포인트로 재개되고, 새 값을 쓰면 빈 상태의 완전히 새로운 스레드가 시작됩니다.

`interrupt`로 일시 중지하기

interrupt 함수는 그래프 실행을 일시 중지하고 값을 호출자에게 돌려줍니다. 노드 안에서 interrupt를 호출하면 LangGraph는 현재 그래프 상태를 저장하고, 입력과 함께 실행을 재개해 주기를 기다려요.

interrupt를 쓰려면 다음이 필요합니다.

from langgraph.types import interrupt

def approval_node(state: State):
    # 멈추고 승인을 요청합니다
    approved = interrupt("Do you approve this action?")

    # 재개하면 Command(resume=...)가 이 값을 여기로 돌려줍니다
    return {"approved": approved}

interrupt를 호출하면 이런 일이 일어납니다.

그래프 실행이 interrupt를 호출한 바로 그 지점에서 중단됩니다.
상태가 체크포인터(checkpointer)로 저장되어 나중에 실행을 재개할 수 있습니다. 운영 환경에서는 데이터베이스 기반 같은 영구 체크포인터를 쓰는 게 좋아요.
값이 호출자에게 반환됩니다. 이벤트 스트리밍(graph.stream_events(..., version="v3"))을 쓰면 stream.interrupts로, 기본 invoke() API를 쓰면 __interrupt__ 아래로 전달돼요. 값은 문자열·객체·배열 등 JSON-직렬화 가능한 무엇이든 될 수 있습니다.
실행이 멈춘 채로 남아 있습니다.

인터럽트 재개하기

인터럽트가 걸릴 수 있는 그래프를 구동하는 권장 방법은 이벤트 스트리밍입니다. stream.interrupts와 stream.interrupted로 인터럽트를 노출하고, stream.output으로 최종 상태를 드러내죠.

from langgraph.types import Command

# 최초 실행 - 인터럽트를 만나고 일시 중지됩니다
# thread_id는 영구 포인터입니다 (운영 환경에서는 안정적인 ID를 저장)
config = {"configurable": {"thread_id": "thread-1"}}

# stream.interrupted는 사람의 입력을 기다리며 멈췄을 때 True이고,
# stream.interrupts에는 interrupt()에 넘긴 페이로드가 들어 있습니다.
if stream.interrupted:
    print(stream.interrupts)
    # > (Interrupt(value='Do you approve this action?'),)

# 사람의 응답으로 재개합니다
# resume 페이로드는 노드 안 interrupt()의 반환값이 됩니다
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output

명심할 점:

재개할 때는 인터럽트가 발생했을 때 사용했던 같은 스레드 ID를 써야 합니다.
Command(resume=...)에 넘긴 값은 interrupt 호출의 반환값이 됩니다.
재개되면 노드는 interrupt를 호출한 노드의 처음부터 다시 시작하므로, 그 앞에 있던 코드도 다시 실행됩니다.
resume 값으로는 JSON-직렬화 가능한 값이면 무엇이든 넘길 수 있어요.

주의 Command(resume=...)는 invoke()/stream()/stream_events()의 입력으로 쓰도록 의도된 유일한 Command 패턴입니다. 나머지 Command 파라미터(update, goto, graph)는 노드 함수에서 반환할 때 쓰도록 설계되었어요. 다중 턴 대화를 이어갈 때 Command(update=...)를 입력으로 넘기지 말고, 일반 입력 dict를 넘기세요.

일반적인 패턴

✅ 승인 워크플로: API 호출, 데이터베이스 변경, 금융 거래 같은 중요한 작업을 실행하기 전에 일시 중지합니다.

사람-개입(HITL) 인터럽트 스트리밍

사람-개입 워크플로로 인터랙티브 에이전트를 만들 때는 이벤트 스트리밍을 사용해 메시지 청크와 상태 스냅샷을 동시에 소비하면서 인터럽트도 처리할 수 있어요. graph.stream_events(..., version="v3")가 돌려주는 타입 지정 투영(typed projection)을 실행이 끝날 때까지 루프에서 쓰면 됩니다.

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

    user_response = get_user_input(interrupt_info)
    stream_input = Command(resume=user_response)

stream.messages: 채팅 모델 출력을 콘텐츠 블록으로 제공합니다. 각 message.text를 순회하며 토큰 델타를 읽어요. 중첩 서브그래프라면 stream.subgraphs[*].messages에서 메시지 청크를 읽습니다.
stream.values: 각 단계 후 전체 상태 스냅샷.

여러 인터럽트 처리하기

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

승인 또는 거절

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

def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:

# 승인하려면
graph.stream_events(Command(resume=True), config=config, version="v3").output

# 거절하려면
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):

initial = graph.stream_events(
    {"action_details": "Transfer $500", "status": "pending"},
    config=config,
    version="v3",
)
_ = initial.output  # 스트림을 끝까지 구동합니다

상태 검토 및 편집

from langgraph.types import interrupt

def review_node(state: State):
    # 멈추고 현재 콘텐츠를 검토용으로 보여줍니다 (페이로드는 stream.interrupts로 노출)
    edited_content = interrupt({

툴 안의 인터럽트

인터럽트는 툴 함수 안에 직접 둘 수도 있어요. 그러면 툴이 호출될 때마다 승인을 위해 스스로 일시 중지하고, 실행 전에 사람이 툴 호출을 검토·편집할 수 있게 됩니다. 먼저 interrupt를 쓰는 툴을 정의하세요.

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain.messages import AnyMessage, SystemMessage, ToolMessage


class AgentState(TypedDict):

사람 입력 검증하기

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

각 재개는 get_age_node를 정확히 한 번 호출하고, interrupt() 호출을 한 번 실행한 뒤 끝납니다. 답이 유효하지 않으면 조건부 엣지가 다시 루프를 돌고, 다음 인터럽트가 갱신된 질문으로 다시 묻죠. 재개당 어떤 코드도 두 번 이상 실행되지 않습니다.

전체 예제

builder.add_conditional_edges("collect_age", route)

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

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

인터럽트 규칙

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

실행이 재개되면(요청한 입력이 주어지면) 런타임은 노드 전체를 처음부터 다시 시작합니다. interrupt를 호출한 정확한 줄부터 이어가지 않아요. 즉 interrupt 앞에서 실행됐던 코드도 다시 실행됩니다. 그래서 인터럽트가 예상대로 동작하게 하려면 몇 가지 중요한 규칙을 따라야 합니다.

`interrupt` 호출을 try/except로 감싸지 말 것

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

✅ interrupt 호출을 오류가 나기 쉬운 코드와 분리하세요.
✅ try/except 블록에서는 특정 예외 타입을 쓰세요.

노드 안에서 `interrupt` 호출 순서를 바꾸지 말 것

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

🔴 while True 검증 루프처럼 실행 간에 결정적이지 않은 로직으로 interrupt 호출을 루프시키지 마세요. 대신 조건부 엣지를 쓰세요(사람 입력 검증 참고).

`interrupt` 호출에서 복잡한 값을 반환하지 말 것

✅ interrupt에 단순한 JSON-직렬화 가능 타입을 넘기세요.
✅ 단순한 값을 가진 dict/객체를 넘기세요.

`interrupt` 앞에 호출되는 부수 효과는 멱등이어야 함

인터럽트는 호출된 노드를 다시 실행하는 방식으로 동작하므로, interrupt 앞에 호출되는 부수 효과는 (이상적으로는) 멱등(idempotent)이어야 합니다.

그 호출 뒤에 interrupt가 오면, 노드가 재개될 때 그 호출이 여러 번 다시 실행되어 처음 업데이트를 덮어쓰거나 중복 레코드를 만들 수 있어요.

✅ interrupt 앞에서는 멱등 연산을 쓰세요.
✅ 부수 효과는 interrupt 호출 뒤에 두세요.

함수로 호출되는 서브그래프와 함께 쓰기

노드 안에서 서브그래프를 호출할 때, 부모 그래프는 서브그래프가 호출되고 interrupt가 트리거된 노드의 처음부터 실행을 재개합니다. 마찬가지로 서브그래프도 interrupt를 호출한 노드의 처음부터 재개돼요.

def node_in_parent_graph(state: State):
    some_code()  # <-- 재개되면 다시 실행됩니다
    # 함수로 서브그래프를 호출합니다.
    # 이 서브그래프는 `interrupt` 호출을 포함합니다.

인터럽트 디버깅하기

그래프를 컴파일할 때 interrupt_before와 interrupt_after를 지정해 정적 중단점을 설정할 수 있어요.

참고 정적 인터럽트는 사람-개입 워크플로에는 권장되지 않습니다. 대신 interrupt 함수를 쓰세요.

5. 그래프는 첫 번째 중단점에 도달할 때까지 실행됩니다.
6. 입력으로 `None`을 넘겨 그래프를 재개합니다. 이러면 다음 중단점에 도달할 때까지 실행됩니다.

팁 인터럽트를 디버깅하려면 LangSmith를 사용하세요.

LangSmith Studio 사용하기

인터럽트 (Interrupts)