LangGraph Interrupt와 사람 승인
노드 내부에서 interrupt()로 중단하고, 같은 thread_id에 Command(resume=...)를 전달하여 재개한다.
사람 승인의 필요성
설비 값을 변경하는 단계는 사람이 확인한 뒤에 실행해야 한다. 노드 내부에서 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 State(TypedDict):
target: float
status: str
def approve(state: State):
# 여기서 멈추고 묻는 내용을 넘긴다
ok = interrupt({"ask": "설정값 변경 승인", "target": state["target"]})
return {"status": "승인" if ok else "거절"}
builder = StateGraph(State)
builder.add_node("approve", approve)
builder.add_edge(START, "approve")
builder.add_edge("approve", END)
graph = builder.compile(checkpointer=InMemorySaver())
실행 재개
config = {"configurable": {"thread_id": "job-1"}}
result = graph.invoke({"target": 72.5}, config)
print(result["__interrupt__"][0].value) # 묻는 내용
# 넘긴 답이 interrupt()의 반환값이 된다
result = graph.invoke(Command(resume=True), config)
print(result["status"]) # 승인
주의 사항
- checkpointer와
thread_id: 중단된 상태를 보관할 저장소와 재개할 실행의 식별자이다. 두 가지 모두 필요하다. - 재개 지점: 노드는 중단된 줄이 아니라 함수의 처음부터 다시 실행된다. 외부에 영향을 주는 동작은
interrupt()뒤에 배치한다. - 전달하는 값: JSON으로 직렬화할 수 있어야 한다.
try/except:interrupt()는 예외를 발생시켜 실행을 중단한다.except로 이 예외를 포착하면 실행이 중단되지 않는다.
