day5_practice/interrupt_basic.py

그래프를 멈추고 사람을 기다리기

에이전트가 출장 신청서 제출처럼 되돌리기 어려운 작업을 하기 직전에 멈추고, 사람이 yes/no를 누르면 그 자리에서 이어서 실행하는 실습입니다. Human-in-the-loop(HITL)라고 부릅니다.

백엔드로 치면

결재 워크플로의 "승인 대기" 상태와 같습니다. 요청이 들어오면 처리하다가 결재가 필요한 지점에서 현재 상태를 DB에 저장하고 응답을 끝냅니다. 나중에 승인 API가 호출되면 DB에서 상태를 꺼내 멈춘 지점부터 이어서 처리합니다. LangGraph에서는 멈추기가 interrupt(), 상태 저장이 checkpointer, 결재 건 ID가 thread_id, 승인 API 호출이 Command(resume=...)입니다.

1. 그래프 구조와 invoke 두 번

노드는 일렬로 3개입니다. 특별한 건 가운데 approval 노드 하나뿐이고, 한 번의 실행이 invoke 두 번으로 쪼개집니다.

1차 invoke: 여기서 멈추고 반환 2차 invoke: approval부터 다시 끝까지 START propose pending_action 채움 approval ⏸ interrupt() execute decision 보고 메시지 END checkpoints.sqlite thread_id=hitl_demo 멈출 때 State 저장 재개할 때 불러옴 사람 input() → yes / no Command(resume=…) graph.get_state(config) 어디서 멈췄나, 뭘 물어봤나 조회 읽기

두 invoke 사이에 프로세스가 끝나도, 서버가 재시작돼도 상관없습니다. 멈춘 위치와 State가 SQLite 파일에 남아 있어서 같은 thread_id로 재개하면 이어집니다.

2. main()이 실행되는 순서

1

체크포인터를 붙여 compile, thread_id 정하기

with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "hitl_demo"}}

노드가 끝날 때마다 State가 이 SQLite 파일에 저장됩니다. thread_id는 "어느 대화(결재 건)의 저장분인가"를 가리키는 키입니다.

2

1차 invoke: propose → approval에서 멈춤

graph.invoke({"messages": [HumanMessage("작업 시작")]}, config=config)

propose가 pending_action을 채우고, approval 노드에서 interrupt({...})를 만나면 그래프가 멈춥니다. 그 시점 State를 저장하고 invoke가 정상적으로 반환됩니다. 에러가 아닙니다. execute는 아직 실행되지 않았습니다.

3

멈춘 상태 들여다보기

state = graph.get_state(config)
state.next                          # ('approval',)  다음에 실행될 노드
state.tasks[0].interrupts[0].value  # interrupt()에 넣은 dict

저장된 체크포인트를 읽어서 "어디서 멈췄고 사람에게 뭘 물어봤는지"를 꺼냅니다. 실제 서비스라면 이 값으로 승인 화면을 그립니다.

4

사람에게 묻기

user_decision = input("승인하시겠습니까? (yes/no): ").strip()

여기서는 터미널 입력이지만, 실제로는 웹 화면의 승인 버튼이나 메신저 버튼입니다. 몇 시간 뒤에 눌러도 됩니다.

5

2차 invoke: Command(resume=...)로 재개

result = graph.invoke(Command(resume=user_decision), config=config)

새 입력 대신 Command(resume=값)을 넣으면 "멈춘 곳부터 이어서, 이 값을 interrupt()의 답으로 써라"는 뜻이 됩니다. approval 노드가 다시 실행되고, 이번에는 interrupt()가 멈추지 않고 "yes"를 리턴합니다. 이어서 execute → END입니다.

3. 핵심: interrupt()는 두 번 불립니다

재개는 "interrupt 다음 줄부터"가 아니라 "approval 노드의 첫 줄부터"입니다. 노드 함수가 처음부터 다시 실행되고, 두 번째에는 interrupt()가 저장된 resume 값을 돌려줄 뿐입니다.

1차 invoke 때 
def human_approval(state):
    decision = interrupt({...})
    # ↑ 여기서 멈춤. 아래 줄은 실행 안 됨
    return {"decision": decision}
2차 invoke 때 (resume="yes") 
def human_approval(state):   # 처음부터 다시
    decision = interrupt({...})
    # ↑ 이번엔 "yes"를 리턴
    return {"decision": "yes"}

직접 확인해 봤습니다. 노드 첫 줄과 interrupt 다음 줄에 기록을 남기고 두 번 invoke하면 결과가 ['enter', 'enter', 'after interrupt -> yes']입니다. 노드에 두 번 들어오고, interrupt 다음 줄은 한 번만 실행됩니다.

그래서: interrupt보다 앞에 DB 쓰기나 메일 발송 같은 부작용 코드를 두면 두 번 실행됩니다. 부작용은 interrupt 뒤나 다음 노드(execute)에 둡니다. 이 파일이 실행을 별도 노드로 뺀 이유입니다.

4. 코드 블록별 역할

class State(TypedDict):
    messages: Annotated[list, add_messages]
    pending_action: str
    decision: str

State

messages는 누적(add_messages), 나머지 둘은 덮어쓰기입니다. pending_action은 "승인받을 작업", decision은 "사람의 답"입니다.

def propose_action(state):
    return {"pending_action":
        "출장 신청서 제출: 부산 2일, 예산 800,000원"}

propose: 위험 작업 제안

실제로는 LLM이 "이 도구를 이 인자로 부르겠다"고 정하는 단계입니다. 실습이라 문자열을 고정했고, LLM 호출은 없습니다.

def human_approval(state):
    decision = interrupt({
        "type": "approval",
        "action": state["pending_action"],
        "question": "이 작업을 실행할까요? (yes/no)",
    })
    return {"decision": decision}

approval: 멈추는 지점

interrupt()에 넣은 dict는 바깥(사람 쪽)에 보여 줄 질문지입니다. 형식은 자유입니다.

리턴값은 나중에 Command(resume=...)로 들어온 답입니다.

def execute(state):
    if state["decision"] == "yes":
        return {"messages": [AIMessage("실행됨: ...")]}
    return {"messages": [AIMessage("실행 거부됨")]}

execute: 실제 실행

사람의 답에 따라 실행하거나 거부합니다. 정확히 "yes"일 때만 실행되므로 "Yes", "y", "네"는 전부 거부로 처리됩니다.

builder.add_edge(START, "propose")
builder.add_edge("propose", "approval")
builder.add_edge("approval", "execute")
builder.add_edge("execute", END)

엣지: 일렬

조건부 엣지가 없습니다. 멈춤과 재개는 엣지가 아니라 interrupt()와 체크포인터가 처리합니다.

5. 실제로 돌린 결과

day5_practice에서 yes와 no를 한 번씩 넣어 실행했습니다.

yes 입력 
[1차 invoke]
  멈춘 노드: ('approval',)
  요청 내용: {'type': 'approval',
    'action': '출장 신청서 제출: ...',
    'question': '이 작업을 실행할까요? (yes/no)'}

승인하시겠습니까? (yes/no): yes

[2차 invoke]
  최종: 실행됨: 출장 신청서 제출: 부산 2일, 예산 800,000원
no 입력 
[1차 invoke]
  멈춘 노드: ('approval',)
  요청 내용: { ...같음... }

승인하시겠습니까? (yes/no): no

[2차 invoke]
  최종: 실행 거부됨

6. 체크포인터가 없으면?

코드 주석에는 "없으면 interrupt 시점에 오류가 납니다"라고 되어 있지만, 지금 설치된 LangGraph 1.2.19에서 직접 해 보니 조금 다릅니다.

단계체크포인터 있음체크포인터 없음 (builder.compile())
1차 invokeapproval에서 멈추고 State 저장에러 없이 멈춤. 반환값에 __interrupt__만 들어 있음. 저장은 안 됨
2차 invoke (resume)저장분을 불러와 이어서 실행RuntimeError: Cannot use Command(resume=...) without checkpointer

멈추는 건 되지만 이어 갈 곳이 없습니다. 재개하려면 "어디까지 했는지"가 어딘가 저장돼 있어야 하므로 체크포인터가 필수입니다.

7. 알아둘 점