LangGraph 병렬 노드 실패 복구: 성공한 노드는 건너뛰는 Python 예제
Python LangGraph에서 병렬 노드 하나만 실패했다면, checkpointer에 실행 기록을 저장하고 같은 thread_id로 입력값 None을 전달해 재개하면 돼요. 같은 super-step에서 이미 완료되어 저장된 노드의 결과는 pending writes로 보존되므로 해당 노드를 다시 실행하지 않아요. 단, 실패한 노드 내부의 작업까지 모두 보존된다는 뜻은 아니에요. 외부 API 없이 이 경계를 호출 횟수로 확인하는 예제를 준비했어요. 공식 Checkpointers 문서
병렬 실패와 재개를 검증하는 전체 코드
# 준비: python -m pip install -U langgraph
# 파일명: resume_parallel.py
# 실행: python resume_parallel.py
# 공식 API를 바탕으로 구성한 예제이며, 실측 실행 로그는 아니에요.
import asyncio
from importlib.metadata import version
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
class State(TypedDict, total=False):
number: int
left: int
right: int
total: int
async def main():
calls = {"left": 0, "right": 0, "join": 0}
async def left_node(state: State):
calls["left"] += 1
return {"left": state["number"] * 2}
async def right_node(state: State):
calls["right"] += 1
if calls["right"] == 1:
# 성공 분기가 먼저 완료되도록 잠시 양보하는 데모용 지연이에요.
# 운영 코드의 동기화 수단으로 사용하는 값은 아니에요.
await asyncio.sleep(0.1)
raise ValueError("demo: right failed once")
return {"right": state["number"] + 10}
async def join_node(state: State):
calls["join"] += 1
return {"total": state["left"] + state["right"]}
builder = StateGraph(State)
builder.add_node("left_node", left_node)
builder.add_node("right_node", right_node)
builder.add_node("join_node", join_node)
builder.add_edge(START, "left_node")
builder.add_edge(START, "right_node")
builder.add_edge(["left_node", "right_node"], "join_node")
builder.add_edge("join_node", END)
saver = InMemorySaver()
graph = builder.compile(checkpointer=saver)
config = {"configurable": {"thread_id": "parallel-recovery-demo"}}
print("langgraph:", version("langgraph"))
try:
await graph.ainvoke({"number": 3}, config)
except ValueError as exc:
if str(exc) != "demo: right failed once":
raise
print("caught:", exc)
else:
raise AssertionError("첫 실행은 의도한 예외로 중단되어야 해요.")
print("after failure:", calls)
assert calls == {"left": 1, "right": 1, "join": 0}, calls
# saver, graph, thread_id를 유지하고 새 입력 대신 None으로 재개해요.
result = await graph.ainvoke(None, config)
print("after resume:", calls)
print("result:", result)
assert calls == {"left": 1, "right": 2, "join": 1}, calls
assert result["left"] == 6, result
assert result["right"] == 13, result
assert result["total"] == 19, result
snapshot = await graph.aget_state(config)
assert snapshot.next == (), snapshot.next
print("PASS: 완료된 분기를 다시 호출하지 않고 복구했어요.")
if __name__ == "__main__":
asyncio.run(main())전체 단계가 실패해도 완료된 결과는 남아요
super-step은 그 단계에 예약된 노드들이 실행되는 한 차례의 실행 단위예요. 병렬 분기에서는 여러 노드가 같은 단계에 속해요. 전체 상태의 체크포인트는 단계 경계에 만들어지지만, 개별 노드의 완료 결과도 별도로 저장돼요. 그래서 한 분기의 예외 때문에 단계가 끝나지 못해도 다른 분기의 계산 결과를 복구에 사용할 수 있어요. 체크포인트와 노드별 쓰기
여기서 완료의 기준은 업무 로그에 ‘성공’이 찍혔는지가 아니에요. 노드가 결과를 반환하고 그 쓰기가 저장되어야 해요. 예를 들어 파일 변환을 끝냈어도 반환 직전에 예외가 나면, 그 노드는 여전히 실패한 실행이에요. 복구 설계를 할 때는 업무상 완료 시점과 그래프에 결과가 남는 시점을 구분하는 편이 좋아요.
이 예제의 대상은 예외를 받아 중단된 실행이에요. 운영체제가 프로세스를 강제 종료하거나 저장소 쓰기 자체가 실패한 상황까지 같은 조건으로 묶으면 안 돼요. 우선 정상적으로 저장된 완료 결과를 재사용하는 경계를 확인한 다음, 프로세스 재시작 복구를 별도 시험으로 추가하세요.
최종 값뿐 아니라 실행 횟수를 확인하세요
병렬 실패와 재개를 검증하는 전체 코드에는 결과와 호출 횟수를 각각 검사하는 assert가 들어 있어요. 최종 합계만 맞으면 성공한 분기를 두 번 계산한 구현도 통과할 수 있거든요. 함수 진입 횟수를 함께 확인해야 ‘값이 복원됐다’와 ‘계산을 다시 했다’를 구분할 수 있어요.
calls는 그래프 상태에 넣지 않은 관찰용 변수예요. 복구 과정에서 상태가 복원되더라도 실제 함수 진입 횟수는 따로 남겨 보려는 구성이에요. 다만 이 변수는 프로세스가 살아 있는 동안만 유효하므로 운영용 재시도 기록으로 사용하지는 마세요.
예제에서는 두 분기가 서로 다른 상태 키에 결과를 써요. 공유 리스트에 결과를 모으는 기능까지 넣으면 병렬 업데이트를 병합하는 reducer 설계가 섞여서, 실패 복구 문제를 따로 보기 어려워져요. 자신의 그래프로 옮길 때도 먼저 분기별 결과를 분리해서 검증하는 방법을 권해요. 상태 업데이트와 reducer
재개, 자동 재시도, 과거 재생은 목적이 달라요
예제의 첫 호출은 예외를 밖으로 전달하고, 두 번째 호출이 저장된 실행을 이어 가요. 동기 노드로 구성한 그래프라면 graph.invoke(None, config), 이 예제처럼 비동기 노드라면 await graph.ainvoke(None, config)를 사용해요. None은 원래 요청을 다시 제출하는 대신 남은 실행을 이어 가기 위한 입력이에요. 공식 오류 복구 예제도 같은 thread 설정과 None을 사용해요. 오류 이후 재개 예제
잠깐의 장애를 호출 내부에서 처리하려면 노드에 RetryPolicy를 붙일 수 있어요. 이때 max_attempts는 첫 시도까지 포함하고, 어떤 예외를 재시도할지는 retry_on이 결정해요. 예를 들어 max_attempts=3은 최초 실행 후 최대 두 번 더 시도한다는 뜻이에요. ValueError는 기본 재시도 대상에서 제외되므로, 예제의 예외를 그대로 두고 정책만 추가한 뒤 자동 복구를 기대하면 안 돼요. 재시도 정책과 기본 제외 예외
반면 과거 checkpoint_id를 선택하는 것은 재생하려는 시점을 직접 고르는 작업이에요. 공식 시간 여행 예제는 generate_topic 다음, write_joke 이전의 체크포인트를 선택하면 write_joke를 다시 실행한다고 설명해요. 장애 복구가 목적이라면 임의의 과거 기록을 골라 되돌아가기보다 최신 실패 상태를 이어 가세요. 과거 체크포인트 재생
사람의 승인 등을 기다리는 interrupt()는 또 다른 경우예요. 여기서는 Command(resume=...)로 대기 중인 호출에 답을 전달해요. 일반 예외 복구 코드에 승인용 resume 값을 추가할 필요는 없어요. interrupt 재개 예제
실패한 함수의 중간 줄부터 이어지지는 않아요
실패한 노드는 함수 시작부터 다시 실행돼요. 가령 한 노드가 ‘주문 저장 → 알림 문구 생성’을 수행하다 두 번째 작업에서 실패했다면, 복구할 때 주문 저장 코드에도 다시 도달할 수 있어요. 다른 병렬 노드의 결과 보존과, 실패한 함수 안의 부수 효과 중복 방지는 별개의 문제예요. 노드 재실행과 멱등성
이 경우에는 주문 저장에 업무상 고유한 요청 키를 사용해 중복 생성을 막는 설계를 먼저 고려하세요. 큰 노드를 더 작은 노드로 나누거나, 내부 작업을 체크포인트가 적용되는 task로 분리하는 방법도 있어요. 공식 문서는 노드 내부의 완료된 task 결과도 복원할 수 있다고 설명해요. 다만 외부 시스템의 처리 완료와 결과 저장 사이에 실패하는 경우까지 생각하면, 저장 작업 자체의 중복 방지 설계는 여전히 필요해요. 노드 안에서 task 사용하기
또 하나의 실수는 실패를 잡아서 빈 결과를 반환하는 거예요. 예를 들어 except 블록에서 {'right': 0}을 반환하도록 바꾸면 이 예제의 ‘실패 후 복구’ 조건 자체가 사라져요. 대체값으로 계속 진행하는 것이 업무적으로 맞는지, 아니면 중단 후 재개해야 하는지 먼저 결정하세요. 두 동작의 테스트도 분리하는 편이 명확해요.
프로세스 재시작은 저장소까지 포함해 검증해요
InMemorySaver는 이 예제를 간단하게 실행하기 위한 선택이에요. 체크포인트가 RAM에 있기 때문에 프로세스를 종료하면 사라져요. 같은 thread_id를 문자열로 다시 적어도 없어진 기록을 복원할 수는 없어요. 공식 문서는 재시작을 견뎌야 하는 경우 영속 checkpointer를 사용하도록 안내해요. 메모리 저장소의 재시작 제한
다음 단계로는 파일 기반 SQLite 같은 영속 저장소를 연결하고, 최초 실행 프로그램과 재개 프로그램을 나누어 보세요. 이 비동기 예제를 확장한다면 AsyncSqliteSaver가 대응하는 구현이에요. 재개 프로그램은 동일한 저장소와 thread_id를 사용해야 해요. 이때 예제의 ‘첫 호출이면 실패’라는 메모리 변수 조건은 제거하고, 테스트용 오류를 해소한 상태로 실행하세요. 변수가 초기화되는 현상을 복구 기능의 실패로 오해하지 않도록 하는 거예요. 비동기 체크포인터 선택
저장 시점도 검토할 대상이에요. 공식 문서의 sync 모드는 다음 단계 전에 체크포인트 저장을 기다리고, async 모드는 다음 단계와 저장을 겹쳐 수행해요. 프로세스 중단 시험에서는 사용한 durability 설정을 함께 기록하세요. 단, sync를 선택해도 RAM 저장소가 디스크 저장소로 바뀌지는 않아요. 체크포인트 내구성 설정
