LangGraph 에이전트 실행 중단 및 체크포인터를 사용한 인간 승인 구현하기

대부분의 에이전트 데모는 사람의 개입 없이 시작부터 끝까지 실행됩니다. 이는 괜찮지만, 에이전트가 되돌릴 수 없는 행동을 하려고 할 때 문제가 됩니다. 환불 처리, 이메일 전송, 기록 삭제 등입니다. 이때 에이전트가 멈추고 사용자에게 질문한 다음, 그 후에만 계속 진행되도록 해야 합니다.

LangGraph는 이를 깔끔하게 구현할 두 가지 기본 기능(primitive)을 제공합니다: 노드를 일시 중지하는 interrupt()와 실행 상태를 영속화하여 다른 프로세스에서도 나중에 재개할 수 있게 하는 체크포인터(checkpointer)입니다. 이 튜토리얼에서는 제가 만든 RelayG라는 작은 지원 트리아지 에이전트의 구조를 사용하여 이 기술을 설명하겠습니다. 여기서 특정 임계값을 초과하는 환불은 사람이 승인해야만 처리될 수 있습니다.

목표는 제 프로젝트를 복사하는 것이 아니라, 메커니즘 자체를 이해하여 어떤 그래프에도 적용할 수 있도록 하는 것입니다.

정신 모델 (The mental model)

LangGraph 에이전트는 상태 기계(state machine)입니다. 노드들은 공유된 상태 딕셔너리(shared state dict)를 읽고 부분적인 업데이트 값을 반환합니다. 엣지(Edges)는 다음에 무엇을 실행할지 결정합니다. 일반적으로 graph.invoke(input, config)를 호출하면 모든 노드가 끝까지 실행됩니다.

interrupt()가 이를 깨뜨립니다. 노드가 이 함수를 호출하면 LangGraph는 다음 작업을 수행합니다:

1. 현재 상태를 체크포인터에 저장하고,
2. 실행을 중지하며,
3. 인터럽트 페이로드(interrupt payload)를 호출자에게 반환합니다.

interrupt() 호출 이후에는 아무것도 실행되지 않습니다. 나중에 특수한 Command(resume=...) 값을 사용하여 그래프를 다시 호출함으로써 재개할 수 있습니다. LangGraph는 저장된 상태를 다시 로드하고, 같은 노드로 재진입하며, 이때 interrupt()는 사용자가 재개할 때 제공한 값(value)을 반환합니다. 실행은 그 지점부터 계속됩니다.

이것이 안정적으로 작동하게 만드는 두 가지 요소가 있습니다. 바로 영구적인 메모리 역할을 하는 체크포인터와, 특정 실행을 저장된 상태에 연결하는 열쇠(key)인 thread_id입니다.

1단계: 상태 및 그래프 만들기

타입이 지정된 상태(typed state)와 간단한 세 개의 노드로 구성된 그래프를 시작합니다: 티켓 분류, 정책 확인, 조치 수행.

import operator
from typing import Annotated, TypedDict
from langgraph.graph import START, END, StateGraph
...

참고로 actions에 있는 Annotated[list[dict], operator.add] 리듀서를 주목하세요. 이것은 LangGraph에게 덮어쓰지 않고 업데이트를 누적(append)하도록 지시하며, 이는 노드가 실행되고 일시 중지된 후 재개되어 다시 실행될 때 중요합니다.

여기서 알아야 할 한 가지가 있습니다: compile()은 체크포인터를 받습니다. 이것이 없으면, interrupt()는 상태를 저장할 곳이 없어 재개할 수 없습니다. 이것이 가장 흔한 실수입니다.

2단계: interrupt()를 사용하여 노드 내부에서 일시 중지하기

act 노드는 인간의 게이트(human gate)가 존재하는 곳입니다. 정책이 환불 승인이 필요하다고 말하면, 우리는 결정 내용을 설명하는 페이로드와 함께 interrupt()를 호출하고 기다립니다.

from langgraph.types import interrupt

def act(state: TicketState) -> dict:
...

핵심 원리는 interrupt()를 두 번 반환하는 함수로 이해하는 것입니다. 노드가 처음 실행될 때는 interrupt()가 전혀 반환되지 않습니다. 대신 제어권을 호출자에게 다시 던집니다. 두 번째, 즉 재개된 후에는 interrupt()가 재개 값(resume value)을 반환하고 나머지 노드는 정상적으로 실행됩니다.

interrupt() 이전에 있는 모든 것이 재개 시 다시 실행되므로, 되돌릴 수 없는 부작용은 인터럽트 전에 두지 말고 후에 배치하세요.

3단계: 영구적인 체크포인터

데모를 위해서는 MemorySaver를 사용할 수 있지만, 이것은 프로세스가 죽을 때 상태를 잃어버리므로 의미가 없습니다. 실제 승인은 몇 시간이 걸릴 수도 있습니다. 따라서 상태가 재시작에도 살아남도록 SQLite 체크포인터를 사용하세요.

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

...

check_same_thread=False는 LangGraph가 연결을 열었던 스레드와 다른 스레드에서 해당 연결을 건드릴 수 있기 때문에 중요합니다.

4단계: 실행, 일시 중지 감지, 재개하기

모든 실행에는 설정(config)에 thread_id가 필요합니다. 이 ID는 이 티켓의 저장된 상태 주소입니다. 같은 ID를 사용하여 재개하세요.

from langgraph.types import Command

config = {

그래프가 일시 중지되면 반환 값에 `__interrupt__` 키가 포함됩니다. `result["__interrupt__"][0].value`를 읽어 `interrupt()`에 전달했던 페이로드를 얻을 수 있습니다. 이것이 검토자에게 보여줄 내용입니다.

재개하려면 동일한 `config`와 `Command(resume=verdict)`를 입력으로 사용하여 그래프를 다시 호출합니다. 전달하는 딕셔너리는 노드 내부에서 `interrupt()`의 반환 값이 됩니다. SQLite가 상태를 영속화했기 때문에, 일시 중지와 재개 사이에 전체 프로그램을 다시 시작하고, 체크포인터를 다시 로드하며, 동일한 `thread_id`로 호출하면 정확히 멈췄던 지점부터 이어서 진행됩니다. 재분류도 없고, 컨텍스트 손실도 없습니다.

## 한 가지 솔직한 주의사항

`interrupt()`가 두 번째 실행되기 전까지 노드 내의 모든 것이 실행됩니다. LangGraph는 노드를 처음부터 다시 재생합니다. 함수 중간에서 멈추지 않습니다. 따라서 인터럽트 호출 전에 Slack에 게시하거나 카드에 요금을 청구하는 것과 같은 부수 효과(side effect)를 발생시키는 작업을 했다면, 그것이 두 번 발생할 수 있습니다. `interrupt()` 위쪽의 코드를 순수하게 유지하고, 읽기 작업과 계산을 그곳에 수행하며, 되돌릴 수 없는(irreversible) 동작은 인터럽트가 반환된 후에 배치해야 합니다. 이것은 설계 규칙이며, 설정으로 우회할 수 있는 버그가 아닙니다.

## 마무리

이것이 전체 기술입니다. 체크포인터로 컴파일하고, 결정 지점에서 `interrupt()`를 호출하며, 결과에서 `__interrupt__`를 감지하고, 동일한 `thread_id`로 `Command(resume=...)`를 사용하여 재개합니다. 상태 머신과 영속적인 상태가 무거운 작업을 처리하므로, 실행이 인간의 커피 브레이크나 서버 재시작 동안에도 장소를 잃지 않고 지속될 수 있습니다.

정책 계층, 목업 및 실제 분류기(classifier), 그리고 모든 작업에 대한 감사 추적(audit trail)을 포함하여 엔드투엔드로 연결된 것을 보고 싶다면, 전체 지원-트리아지 버전은 [github.com/AgentPostmortem/relayg](https://github.com/AgentPostmortem/relayg)에서 확인할 수 있습니다. 이를 클론하고 API 키 없이 데모를 실행한 다음, 승인을 위해 환불이 일시 중지되었다가 재개되는 것을 지켜보세요.