고객 12명에게 환불 메일을 보내는 Agent 프로그램이 있다고 해 봅니다. 메일은 한 번 나가면 되돌릴 수 없으니 보내기 직전에 사람이 확인하게 하고 싶습니다. 그런데 사람이 답하는 데 몇 분이 걸릴지 몇 시간이 걸릴지 알 수 없습니다.
프로그램을 계속 켜 둔 채 기다리면 자원이 낭비되고 서버가 꺼지면 모두 사라집니다. LangGraph는 멈추는 순간의 상태를 저장소에 적어 두고 프로그램을 끕니다. 답이 오면 저장소에서 꺼내 이어 갑니다. 아래 그림에서 답을 고르고 돌려 보세요.
설명을 위한 예시 장면입니다. 저장소는 프로그램 밖에 있는 기록 장소를 그린 것이고 실제 저장 위치는 사용하는 체크포인터에 따라 다릅니다.
이 저장소를 만드는 도구가 체크포인터(checkpointer)입니다. 그래프를 compile할 때 붙여 두면 노드가 하나 끝날 때마다 State의 모습이 한 칸씩 저장됩니다. 그 한 칸을 체크포인트(checkpoint)라고 부릅니다. 게임을 하다가 저장 슬롯에 저장해 두는 것과 비슷합니다.
저장 칸들은 스레드(thread)라는 이름 아래 묶입니다. 같은 thread_id로 다시 부르면 그 이름의 마지막 칸에서 이어 갑니다. 스크롤하며 칸이 쌓이는 모습을 보세요.
graph.invoke({}, config)를 부르면 시작 직후의 모습이 첫 칸에 저장됩니다. 칸 묶음의 이름은 config의 thread_id입니다.
노드가 끝나면 그 결과가 담긴 State가 새 칸으로 저장됩니다. 이제 request 칸에 환불 메일 요청이 적혀 있습니다.
interrupt를 만나면 그래프는 멈추고 invoke는 돌아옵니다. 결과의 __interrupt__ 칸에 사람에게 물을 내용이 담기고 프로그램은 꺼도 됩니다.
같은 thread_id로 Command(resume=True)를 보내면 마지막 칸에서 이어 갑니다. 승인 뒤 send가 끝나면 마지막 칸이 생깁니다.
def approve(s: State) -> Command[Literal["send", "cancel"]]:
# 1 여기서 멈추고 물어볼 내용을 밖으로 내보낸다
answer = interrupt({"question": "이 작업을 실행할까요?", "action": s["request"]})
return Command(update={"decision": "approved" if answer else "rejected"},
goto="send" if answer else "cancel") # 2 답에 따라 길을 고른다
graph = builder.compile(checkpointer=InMemorySaver()) # 3 저장소를 붙인다
config = {"configurable": {"thread_id": "t1"}} # 4 칸 묶음의 이름
result = graph.invoke({}, config) # 멈춤: result["__interrupt__"]
graph.invoke(Command(resume=True), config) # 5 같은 이름으로 이어 가기출력은 예시입니다. 코드는 langgraph 1.2에서 직접 실행해 확인했고 노드 함수는 일부 생략했습니다.
interrupt에 넘긴 값은 호출한 쪽의 __interrupt__ 칸으로 나옵니다. 사람에게 보여 줄 질문과 대상을 여기에 담습니다.interrupt의 반환값이 됩니다. 앞에서 배운 Command로 State를 고치고 갈 길을 정합니다.InMemorySaver는 메모리에만 저장해서 프로그램을 끄면 사라집니다. 그래서 실제 서비스에서는 SQLite나 Postgres 같은 데이터베이스용 체크포인터를 씁니다.thread_id가 다르면 새로운 실행으로 취급합니다. 사용자마다 다른 이름을 써서 대화와 승인 흐름이 섞이지 않게 합니다.Command(resume=…)를 넘깁니다. 처음 시작할 때도 이어 갈 때도 같은 invoke를 씁니다.사람이 답하고 이어 가면 interrupt 줄에서 곧바로 이어지는 것처럼 보입니다. 실제로는 approve 노드가 첫 줄부터 다시 실행됩니다. 이번에는 interrupt가 멈추지 않고 사람의 답을 돌려줍니다. 직접 실행해 보니 노드가 두 번 시작됐습니다.
그래서 interrupt 앞에 메일 보내기 같은 되돌릴 수 없는 일을 두면 두 번 실행됩니다. 위치를 바꿔 보세요.
저장 칸이 남아 있으면 과거의 어느 칸으로든 돌아갈 수 있습니다. 이것을 시간 여행(time travel)이라고 합니다. 그 칸에서 그대로 다시 실행해 결과가 같은지 볼 수도 있고 State 값을 고친 뒤 실행해 다른 길로 갈라질 수도 있습니다. 갈라져도 원래 기록은 지워지지 않고 새 가지가 생깁니다.
사람이 요청 내용을 보고 "12명이 아니라 3명에게만 보내세요"라고 고칠 때 쓸 수 있습니다. 두 방식을 눌러 가지가 어떻게 달라지는지 보세요.
history = list(graph.get_state_history(config)) # 1 저장된 칸 목록(최근 것이 앞)
old = [h for h in history if h.next == ("approve",)][0] # 2 approve 직전 칸
graph.invoke(None, old.config) # 3 그 칸에서 다시 실행
new = graph.update_state(old.config, {"request": "고객 3명에게만 환불 메일 발송"})
graph.invoke(None, new) # 4 고친 값으로 갈라져 실행출력은 예시입니다. 칸의 개수와 값은 그래프에 따라 달라집니다.
next)와 그 칸을 가리키는 config가 들어 있습니다. next가 ("approve",)인 칸이 멈추기 직전입니다.None과 과거 칸의 config를 주면 그 칸에서 다시 실행합니다. 이번에도 approve에서 멈추니 다시 승인을 기다립니다.update_state는 그 칸에서 갈라진 새 칸을 만들어 돌려줍니다. 새 칸으로 실행하면 승인 질문에 고친 요청이 담겨 나옵니다.그래프를 직접 짜지 않아도 됩니다. LangChain의 HumanInTheLoopMiddleware에 승인이 필요한 도구를 적어 주면 그 도구를 부르기 직전에 멈춥니다. 이 미들웨어도 안에서 같은 interrupt를 쓰므로 체크포인터가 필요하고 Command(resume=…)로 이어 갑니다.
사람은 승인, 수정, 거절, 직접 답변 네 가지로 답할 수 있습니다. 메일 보내기나 결제처럼 되돌릴 수 없는 도구에만 걸어 두면 됩니다. 다음 글에서는 이 그래프에 실패에 강한 장치를 달고 실무형으로 다듬어 보겠습니다.