[LangGraph 기초 5] Interrupt가 필요한 이유 (Human-in-the-Loop)

Interrupt가 필요한 이유

---

 

 

지금까지의 실습에서는 그래프를 Python 루프에서 반복 호출하는 방식으로 실행해 왔다.
하지만 실제 애플리케이션에서는 그래프가 외부 입력을 기다려야 하는 상황이 자주 발생한다.

현실에서 발생할 수 있는 대표적인 경우는 다음과 같다.

• 툴 실행 전 인간 승인 필요
• 데이터베이스 쓰기 전 관리자 확인
• 외부 시스템 응답을 오래 기다려야 하는 경우

이번에서는 LangGraph 기능인 Interrupt를 통해 이를 어떻게 해결하는지 살펴보자.

 

 

Interrupt의 개념

---

Interrupt는 그래프 실행을 일시 중단시키는 메커니즘이다.

• 노드 실행 중 Interrupt 발생
• 그래프 실행 즉시 중단
• Python 런타임으로 제어 반환
• 외부 입력을 받은 후 다시 재개

이 기능은 Checkpointing 덕분에 가능하다.

 

이때 Interrupt는 Checkpointer와 함께 기능한다.

 

• Interrupt는 Checkpointer가 있어야 State 저장 가능
• Interrupt 발생 시 State 스냅샷 저장
• Resume 시 정확한 상태 복원

즉, Interrupt는 Checkpointer 위에서 동작하는 기능이다.

 

 

 

 

Interrupt 예제

---

In [1]:

from IPython.display import Image, display
import operator
from typing import Annotated, List, Literal, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

 

In [2]:

from langgraph.checkpoint.memory import InMemorySaver
memory = InMemorySaver()
config = {"configurable": {"thread_id": "1"}}

 

In [3]:

# State 상태 정의 (TypeDict 타입 사용)
# 하나의 매개변수를 가짐 (nlist라는 문자열 목록 + 리듀서 도입으로 병렬 실행에서 State 누적)

class State(TypedDict):
    nlist: Annotated[List[str], operator.add]

 

In [4]:

# LangGraph의 built-in command type을 Conditional Edge 구현 방법
# Node A 수정 ()
# 

def node_a(state: State) -> Command[Literal["b", "c", END]]:
    print("Entered 'a' node") # print문으로 Node A 수행시마다 디버깅
    select = state["nlist"][-1]
    if select == "b":
        next_node = "b"
    elif select == "c":
        next_node = "c"
    elif select == "q":
        next_node = END
    else: # Interrupt를 구현하기 위해 수정된 부분
        admin = interrupt(f"Unexpected input '{select}'") # langgraph.types에서 interrupt 가져오기
        print(admin)
        if admin == "continue":
            next_node = "b"
        else:
            next_node = END
            select = "q"


    return Command(
        update = State(nlist=[select]),
        goto = next_node
    )

def node_b(state: State) -> State:
    return State(nlist=["B"])

def node_c(state: State) -> State:
    return State(nlist=["C"])

 

In [5]:

builder = StateGraph(State)

# Node 추가
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_node("c", node_c)

# Edge 추가
builder.add_edge(START, "a")
builder.add_edge("b", END)
builder.add_edge("c", END)

# StateGraph 생성
graph = builder.compile(checkpointer=memory) # Graph에 Memory 부여

# Display
display(Image(graph.get_graph().draw_mermaid_png()))

 

 

 

In [6]:

while True:
    user = input('b, c, or q to quit: ')
    input_state = State(nlist = [user])
    result = graph.invoke(input_state, config) # graph.invoke 시 thread_id를 가지는 config 인자를 포함시켜주기
    
    print(result)
    break

    if result['nlist'][-1] == "q":
        print("quit")
        break

# result에는 __interrupt__ key가 포함됨. 이는 interrupt를 포함하는 list를 저장하며, 값은 발생한 메시지.
# 추가로 id도 할당된다.
# 이는 interrupt handler 구현을 통해 제어할 수 있다.

 

 

Entered 'a' node
{'nlist': ['p'], '__interrupt__': [Interrupt(value="Unexpected input 'p'", id='786fed5959257b30537a51630367e188')]}

 

 

In [7 ]:

while True:
    user = input('b, c, or q to quit: ')
    input_state = State(nlist = [user])
    result = graph.invoke(input_state, config) # graph.invoke 시 thread_id를 가지는 config 인자를 포함시켜주기

    if '__interrupt__' in result:
        print(f"Interrupt: {result}") # if문에서 result에 interrupt가 포함되어 있으면 간단한 출력문을 실행한다.
        msg = result['__interrupt__'][-1].value # 그리고 msg를 추출한다
        print(msg) # msg를 출력한다.
        human = input(f"\n{msg}: ") # 사람에게 input을 요청한다.

        human_response = Command( # Command type을 사용한다. 이는 resume 속성을 제공한다.
            resume = human
        )
        result = graph.invoke(human_response, config) # human response를 통해 다시 Graph를 호출한다.
        # 이때 Graph가 Memory를 가지려면 이전 실행에서 상태를 복원하려는 동일한 thread_id(config)로 호출되어야 한다.

    if result['nlist'][-1] == 'q':
        print("quit")
        break

# 여기서 'p'를 누르면 interrupt 상태로 사람의 응답을 기다린다.
# 그 후 'continue'를 누르면 다시 Graph를 호출한다. 초기 Node인 A로 간다.
# 이후 'q'를 누르면 종료된다.

 

 

Out [7]:

Entered 'a' node
Interrupt: {'nlist': ['p', 'p'], '__interrupt__': [Interrupt(value="Unexpected input 'p'", id='487f8eb00407e32a65476f74b173f44a')]}
Unexpected input 'p'
Entered 'a' node
continue
Entered 'a' node
quit