LangGraph interrupt 재개 때 API가 두 번 호출되는 이유와 수정 예제
LangGraph에서 interrupt를 재개하면 멈췄던 노드가 처음부터 다시 실행돼요. 같은 노드에서 interrupt보다 먼저 API를 호출했다면 그 호출도 반복됩니다. 승인 후 실행할 작업은 승인 뒤로 옮기거나 별도 실행 노드로 분리하고, 장애로 인한 재시도에는 같은 작업의 중복 처리를 막는 멱등성 키를 적용하세요. LangGraph Interrupts
로컬 재현 코드
# 준비: python -m pip install -U langgraph
# 파일명: interrupt_demo.py
# 실행: python interrupt_demo.py
# 외부 API나 LLM을 호출하지 않는 순차 실행 예제입니다.
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
class State(TypedDict, total=False):
operation_id: str
title: str
approved: bool
document_id: str
class MockAPI:
def __init__(self):
self.calls = 0
self.documents = []
self.results = {}
def create(self, title, key=None):
self.calls += 1
print(f'API request={self.calls}, key={key!r}')
if key is not None and key in self.results:
old_title, document_id = self.results[key]
if old_title != title:
raise ValueError('같은 키에 다른 요청 내용이 들어왔어요')
return document_id
document_id = f'doc-{len(self.documents) + 1}'
self.documents.append((document_id, title))
if key is not None:
self.results[key] = (title, document_id)
return document_id
def build(api, fixed):
builder = StateGraph(State)
if not fixed:
def create_then_approve(state):
document_id = api.create(state['title'])
approved = interrupt({'approve_creation': state['title']})
return {'approved': approved is True,
'document_id': document_id}
builder.add_node('create_then_approve', create_then_approve)
builder.add_edge(START, 'create_then_approve')
builder.add_edge('create_then_approve', END)
else:
def approve(state):
decision = interrupt({'approve_creation': state['title']})
return {'approved': decision is True}
def create(state):
document_id = api.create(
state['title'], key=state['operation_id']
)
return {'document_id': document_id}
builder.add_node('approve', approve)
builder.add_node('create', create)
builder.add_edge(START, 'approve')
builder.add_conditional_edges(
'approve',
lambda state: 'create' if state['approved'] else END,
{'create': 'create', END: END},
)
builder.add_edge('create', END)
return builder.compile(checkpointer=InMemorySaver())
def run_case(fixed, decision):
api = MockAPI()
graph = build(api, fixed)
config = {'configurable': {'thread_id': 'demo-thread'}}
inputs = {'operation_id': 'document:create:request-001',
'title': '검토용 문서'}
paused = graph.invoke(inputs, config)
assert paused.get('__interrupt__')
before = 0 if fixed else 1
assert (api.calls, len(api.documents)) == (before, before)
result = graph.invoke(Command(resume=decision), config)
expected = (1 if decision else 0) if fixed else 2
assert (api.calls, len(api.documents)) == (expected, expected)
print(f'fixed={fixed}, approved={decision}: '
f'calls={api.calls}, documents={len(api.documents)}')
if fixed and decision:
# API 계층에 같은 요청을 직접 한 번 더 보내 봅니다.
document_id = api.create(inputs['title'], inputs['operation_id'])
assert document_id == result['document_id']
assert (api.calls, len(api.documents)) == (2, 1)
print('같은 키 재요청: calls=2, documents=1')
if __name__ == '__main__':
run_case(fixed=False, decision=True)
run_case(fixed=True, decision=True)
run_case(fixed=True, decision=False)
print('모든 assert 통과')체크포인트가 외부 작업까지 되돌리지는 않아요
체크포인터가 보관하는 것은 스레드의 그래프 상태예요. 외부 서비스의 처리 결과와는 저장 경계가 달라요. 예를 들어 문서 생성 API가 성공한 다음 승인 대기에 들어갔다면, 그래프를 재개한다고 이미 생성된 문서가 사라지는 것은 아니에요. LangGraph Persistence
승인을 받기 위해 생성 결과가 먼저 필요한 경우도 있죠. 이때는 생성 노드를 완료한 뒤 검토 노드에서 interrupt를 호출하는 구성을 고려하세요. Functional API를 사용한다면 생성 작업을 task로 감싸 저장된 결과를 재사용하는 방식도 있어요. 다만 작업이 성공적으로 완료되지 못한 것으로 기록되면 재실행될 수 있으므로 멱등성 설계는 여전히 필요해요. Functional API
호출 횟수와 생성 건수를 따로 확인하세요
로컬 재현 코드의 MockAPI는 HTTP 서버 대신 함수와 메모리 목록으로 외부 서비스를 흉내 내요. assert에 적은 수치는 측정 결과가 아니라 이 예제의 통과 조건이에요. 마지막 직접 재요청은 그래프 장애 복구를 재현하는 테스트가 아니라, 같은 키를 받은 API가 기존 결과를 돌려주는지 확인하는 테스트예요.
멱등성 키를 붙여도 요청 자체는 다시 들어올 수 있어요. 그래서 로그에서는 요청 횟수와 실제 생성 건수를 구분해야 합니다. 같은 작업이 재요청됐을 때 기존 결과를 반환하도록 설계하되, 키는 노드 실행마다 새로 만들지 말고 작업에 연결해 유지하세요. 공식 문서도 재실행에 대비해 멱등성 키나 기존 결과 확인을 권장해요. Functional API
실서비스에 적용할 때는 예제의 딕셔너리를 그대로 복사하지 마세요. 같은 키의 동시 요청을 하나의 처리로 묶고, 처리 결과를 영속적으로 저장하도록 API 쪽을 설계해야 해요. 같은 키인데 요청 내용이 달라지면 오류로 처리하는 정책도 함께 정하세요. 사용자 수정으로 새로운 문서를 생성하는 작업이라면 새 작업 ID를 부여하는 편이 명확해요.
멀티 에이전트라면 부모 노드도 살펴보세요
노드 안에서 서브그래프를 함수처럼 호출하고 그 안에서 interrupt가 발생하면, 재개 시 부모의 호출 노드도 처음부터 실행돼요. 중복 요청을 찾을 때는 하위 에이전트뿐 아니라 부모 노드의 서브그래프 호출 앞에 있는 API 작업도 확인하세요. LangGraph Interrupts
이 예제는 한 프로세스에서 실행하는 학습용이에요. InMemorySaver는 프로세스가 종료되면 체크포인트를 잃으므로, 재시작 후 승인 대기를 이어가야 하는 서비스에는 영속 체크포인터가 필요해요. 모의 API의 목록과 키 저장소 역시 메모리에만 남습니다. LangGraph Persistence
실제 코드에서는 중복되는 API 호출 한 곳을 골라 노드 이름, 작업 ID, 요청 횟수, 생성 결과 ID를 함께 기록해 보세요. 승인 재개로 호출이 반복되는지, 동일 요청이 별도 결과를 만드는지 구분하면 수정할 경계가 드러나요.