LangGraph 병렬 노드 실패 복구: 성공한 노드는 건너뛰는 Python 예제
체크포인터를 연결하고, 실패 원인을 해결한 뒤 같은 thread_id로 graph.invoke(None, config)를 호출하면 돼요. 같은 병렬 실행 단계에서 성공한 노드의 결과는 pending writes로 보존되어 재개할 때 다시 계산하지 않아요. 단, 재실행을 피할 수 있는 대상은 완료 결과가 저장된 노드예요. 실패한 노드 내부에서 이미 수행한 작업까지 자동으로 보호되는 것은 아니에요. 체크포인터와 pending writes
병렬 실패와 재개를 확인하는 전체 코드
# 준비: python -m pip install -U langgraph
# 이 코드를 parallel_resume.py로 저장한 뒤 실행:
# python parallel_resume.py
# 외부 API와 계정은 사용하지 않습니다.
# 공식 API에 맞춘 예제이며, 실측 실행 로그가 아닙니다.
from importlib.metadata import version
from threading import Lock
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy
class State(TypedDict):
job: str
left_result: str
right_result: str
summary: str
calls = {"left": 0, "right": 0, "join": 0}
lock = Lock()
service = {"available": False}
def count(name: str) -> None:
with lock:
calls[name] += 1
def left(state: State) -> dict:
count("left")
return {"left_result": state["job"] + ":left"}
def right(state: State) -> dict:
count("right")
if not service["available"]:
raise ConnectionError("의도적으로 만든 분기 장애")
return {"right_result": state["job"] + ":right"}
def join(state: State) -> dict:
count("join")
return {
"summary": state["left_result"] + " | " + state["right_result"]
}
builder = StateGraph(State)
builder.add_node("left", left)
builder.add_node(
"right",
right,
retry_policy=RetryPolicy(
max_attempts=2,
initial_interval=0.1,
backoff_factor=1.0,
jitter=False,
retry_on=ConnectionError,
),
)
builder.add_node("join", join)
builder.add_edge(START, "left")
builder.add_edge(START, "right")
builder.add_edge(["left", "right"], "join")
builder.add_edge("join", END)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "parallel-recovery-demo"}}
print("langgraph:", version("langgraph"))
try:
graph.invoke({"job": "sample"}, config, durability="sync")
except ConnectionError as exc:
print("중단:", exc)
else:
raise AssertionError("첫 실행은 실패해야 합니다.")
print("중단 후 호출 횟수:", calls)
assert calls == {"left": 1, "right": 2, "join": 0}, calls
snapshot = graph.get_state(config)
print("기록된 작업 오류:", [(t.name, t.error) for t in snapshot.tasks])
# 장애 원인이 해결됐다고 가정합니다.
service["available"] = True
# 새 입력 대신 None, 최초 실행과 같은 thread_id를 전달합니다.
result = graph.invoke(None, config, durability="sync")
print("재개 후 호출 횟수:", calls)
print("최종 결과:", result)
# 성공한 left는 그대로 1회, 실패했던 right만 1회 더 실행됩니다.
assert calls == {"left": 1, "right": 3, "join": 1}, calls
assert result["left_result"] == "sample:left", result
assert result["right_result"] == "sample:right", result
assert result["summary"] == "sample:left | sample:right", result
assert graph.get_state(config).next == ()
print("검증 통과")전체 단계의 실패와 개별 노드 결과 저장은 달라요
LangGraph의 super-step은 함께 실행하도록 예약된 노드들이 처리되는 한 단계예요. 병렬 분기 하나에서 예외가 발생하면 그 단계는 정상 완료되지 않지만, 체크포인터는 성공한 형제 노드의 출력을 별도로 보존해요. 단계 전체의 상태 확정과 개별 작업 결과 보존을 구분하면, 실패했는데도 일부 작업을 건너뛰는 이유가 이해돼요. 병렬 노드의 예외 처리
그래서 성공 여부를 직접 state에 표시하고 모든 노드를 처음부터 돌리는 분기문부터 만들 필요는 없어요. 먼저 LangGraph가 저장한 작업 결과를 이용해 재개하도록 구성하세요. 반대로 한 노드 안에서 여러 함수를 동시에 실행했다면, 그 함수들이 각각 그래프의 독립 노드가 된 것은 아니에요. 복구 경계가 필요한 작업은 노드로 나누는 설계가 적합해요. 노드 크기와 복구 경계
호출 횟수와 최종 결과를 함께 검사하는 이유
‘병렬 실패와 재개를 확인하는 전체 코드’는 자동 재시도를 소진한 상황을 의도적으로 만들어요. RetryPolicy의 max_attempts에는 최초 실행도 포함되므로, 2로 지정하면 최초 시도와 재시도 한 번까지 허용해요. 예외 종류도 ConnectionError로 명시해서 기본 재시도 대상에 관한 가정을 없앴어요. 재시도 정책
최종 문자열만 확인하면 성공 노드가 중복 호출됐는지 놓칠 수 있어요. 같은 값을 두 번 계산해도 결과는 똑같을 수 있거든요. 코드의 calls는 그래프 상태 밖에서 실제 함수 진입을 세고, assert는 호출 횟수와 반환값을 각각 검사해요. 이 카운터는 예제 관찰용이며 재개 여부를 결정하는 저장소로 사용하지 않아요.
두 분기는 서로 다른 상태 키에 결과를 써요. 공통 리스트에 결과를 모으는 구현으로 바꾼다면 병렬 업데이트를 합치는 reducer도 설계해야 해요. 또한 병렬 실행의 출력 순서를 업무 순서로 해석하면 안 돼요. 정렬이 필요하다면 결과에 정렬 기준을 포함하세요. 병렬 업데이트와 reducer
재개 요청을 새로운 실행 요청으로 바꾸지 마세요
오류 복구 시에는 처음 넣었던 입력 딕셔너리를 다시 보내지 않고 None을 전달해요. 오류 이후 재개와 interrupt()에 대한 응답도 구분해야 해요. Command(resume=...)는 중단된 사용자 입력 자리에 값을 전달하는 용도이고, 일반적인 예외 복구에서는 원인을 해결한 다음 저장된 실행을 이어가요. 공식 Functional API 문서도 오류 재개를 None과 같은 thread_id의 조합으로 설명해요. 오류 이후 재개
thread_id는 로그에 붙이는 이름에 그치지 않아요. 체크포인터가 복구할 실행을 찾는 식별자예요. 재개할 때 새 UUID를 발급하거나, 같은 문자열을 쓰면서 빈 InMemorySaver를 새로 만들면 기존 기록을 이용할 수 없어요. 예제는 같은 프로세스에서 같은 저장소 객체를 유지하는 조건으로 구성했어요. 체크포인터의 thread
실패 원인에 맞춰 복구하세요. 예를 들어 외부 서비스의 일시적인 장애라면 서비스가 회복된 뒤 재개하고, 잘못된 데이터가 원인이라면 해당 데이터를 어떻게 수정할지 먼저 결정해야 해요. 예외를 잡고 빈 결과를 반환하는 방식으로 바꾸면 런타임은 그 노드를 정상 반환한 것으로 취급할 수 있어요. ‘나중에 다시 해야 하는 실패’와 ‘빈 결과도 유효한 성공’을 코드에서 구별하는 편이 좋아요.
성공한 형제 노드와 실패한 노드 내부는 보호 범위가 달라요
가령 한 노드가 ‘파일 저장 → 알림 전송 → 결과 반환’을 수행하다가 알림 전송에서 실패했다고 해볼게요. 재개하면 그 노드는 처음부터 실행되므로 파일 저장도 다시 수행할 수 있어요. 성공한 형제 노드를 건너뛰는 기능이 실패한 함수의 중간 줄부터 실행해 주는 기능은 아니에요. 실패한 노드의 재실행 범위
이 경우 파일 저장과 알림 전송을 별도 노드로 분리하는 방법을 고려하세요. 다만 외부 작업은 끝났는데 완료 결과를 기록하기 전에 실행이 끊기는 간격도 남아요. 알림 요청에 업무 식별자를 붙여 중복 처리를 막거나, 기존 처리 결과를 조회한 뒤 실행하도록 만드는 식의 멱등성 설계가 필요해요. 공식 문서도 완료되지 않은 작업의 재실행에 대비해 멱등성 키나 기존 결과 확인을 권해요. 부분 실패와 멱등성
복구 코드에서 과거 checkpoint_id를 임의로 골라 넣는 것도 피하세요. 과거 체크포인트 이후의 작업을 다시 실행하는 replay는 최신 실패 지점의 미완료 작업을 이어가는 목적과 달라요. 완료된 작업까지 재계산하려는 경우에만 과거 시점을 명시적으로 선택하세요. 체크포인트 replay
프로세스 재시작까지 복구하려면 저장소를 바꿔야 해요
InMemorySaver는 RAM에 저장하므로 프로그램이 종료되면 체크포인트도 사라져요. 이 예제를 실행하다 종료한 뒤 스크립트를 다시 실행하는 것은 장애 복구 검증이 아니에요. 재시작 복구를 확인하려면 파일 기반 SQLite나 PostgreSQL 같은 지속성 있는 체크포인터를 사용해야 해요. 메모리 저장소의 재시작 한계
검증 범위도 나눠 잡으세요. 먼저 제공한 예제로 노드 예외 후 재개를 확인하고, 다음에는 사용하는 지속성 저장소에서 프로세스를 재시작한 뒤 같은 실행을 복구하는 테스트를 추가하는 순서가 좋아요. 이때 예제의 전역 카운터와 장애 플래그 역시 재시작하면 초기화되므로, 실제 작업 호출은 별도 로그나 테스트용 기록으로 관찰해야 해요.
체크포인트 기록 시점도 확인할 항목이에요. 공식 문서의 sync 모드는 다음 단계로 넘어가기 전에 기록을 마치고, async 모드는 다음 단계 실행과 기록을 겹쳐 수행해요. 예제에 sync를 넣었더라도 메모리 저장소가 디스크 저장소로 바뀌지는 않아요. 운영 검증에서는 저장 위치와 기록 시점을 따로 확인하세요. Durability 모드
