내결함성 (Fault Tolerance)

노드 하나가 실패하는 순간은 여러 가지로 올 수 있어요. 외부 API가 느려지거나, 네트워크 오류가 잠깐 스치거나, 코드에서 처리하지 못한 예외가 터지는 식이죠. 이때 LangGraph는 세 가지를 조합해서 대응할 수 있게 해줘요.

재시도(Retries): 실패한 시도를 예외 타입과 백오프 설정에 따라 자동으로 다시 실행해요.
타임아웃(Timeouts): 한 번의 시도가 얼마나 오래 돌 수 있는지 상한을 잡아줘요.
오류 처리(Error handling): 모든 재시도가 소진된 뒤 실행되는 복구 함수를 둬요.

이 메커니즘들은 set_node_defaults 로 모든 노드에 한 번에 적용할 수도 있어요. add_node를 호출할 때마다 반복해서 적을 필요가 없죠. 그리고 셋은 고정된 순서로 조합돼요. 노드 시도가 예외를 던지면(타임아웃에서 오는 NodeTimeoutError를 포함해서요) 먼저 재시도 정책이 다시 시도할지 판단하고, 재시도가 다 소진된 다음에야 오류 핸들러가 실행됩니다. 슈퍼스텝 경계에서 실행을 깔끔하게 멈췄다가 나중에 다시 시작하려면 Graceful shutdown을 보세요.

참고: 노드별 타임아웃과 노드 레벨 오류 핸들러는 langgraph>=1.2가 필요해요.

재시도 (Retries)

재시도 정책은 실패한 노드 시도를 예외 타입과 백오프 설정에 따라 자동으로 다시 실행해 줘요. add_node에 retry_policy=를 넘기면 됩니다.

from langgraph.types import RetryPolicy

builder.add_node(
    "call_api",
    call_api,
    retry_policy=RetryPolicy(max_attempts=3),
)

기본 동작

기본적으로 retry_on은 default_retry_on을 쓰는데, 사실상 어떤 예외든 재시도해요. 단, 다음 예외와 그 하위 클래스들은 제외됩니다.

ValueError
TypeError
ArithmeticError
ImportError
LookupError
NameError
SyntaxError
RuntimeError
ReferenceError
StopIteration
StopAsyncIteration
OSError

requests나 httpx 같은 널리 쓰이는 HTTP 라이브러리의 예외는 5xx 상태 코드일 때만 재시도해요. NodeTimeoutError는 기본적으로 재시도 대상입니다.

파라미터

파라미터	타입	기본값	설명
`max_attempts`	`int`	`3`	첫 시도를 포함한 최대 시도 횟수.
`initial_interval`	`float`	`0.5`	첫 재시도 전까지의 대기 시간(초).
`backoff_factor`	`float`	`2.0`	재시도할 때마다 간격에 곱해지는 배수.
`max_interval`	`float`	`128.0`	재시도 사이의 최대 대기 시간(초).
`jitter`	`bool`	`True`	간격에 무작위 지터를 더할지 여부.
`retry_on`	`type[Exception] \| Sequence[type[Exception]] \| Callable[[Exception], bool]`	`default_retry_on`	재시도할 예외, 또는 재시도 가능 여부를 `True`로 돌려주는 콜러블.

커스텀 재시도 로직

retry_on에 콜러블이나 예외 타입을 넘길 수 있어요. 기본 동작을 확장하고 싶다면 default_retry_on을 import해서 쓰면 됩니다.

from langgraph.types import RetryPolicy, default_retry_on

def custom_retry_on(exc: BaseException) -> bool:
    if isinstance(exc, MyCustomError):
        return False
    return default_retry_on(exc)

builder.add_node(
    "call_api",
    call_api,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=custom_retry_on),
)

재시도 상태 확인하기

노드 안에서 실행 정보를 읽어 현재가 몇 번째 시도인지 확인할 수 있어요. 주 호출이 계속 실패할 때 폴백으로 전환하려는 상황에서 유용하죠.

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.types import RetryPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

def my_node(state: State, runtime: Runtime) -> State:
    if runtime.execution_info.node_attempt > 1:
        return {"result": call_fallback_api()}
    return {"result": call_primary_api()}

builder = StateGraph(State)
builder.add_node("my_node", my_node, retry_policy=RetryPolicy(max_attempts=3))
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)

execution_info가 노출하는 필드는 다음과 같아요.

속성	타입	설명
`node_attempt`	`int`	현재 시도 번호(1부터 시작). 첫 시도는 `1`, 첫 재시도는 `2`.
`node_first_attempt_time`	`float \| None`	첫 시도가 시작된 Unix 타임스탬프. 재시도 내내 일정.
`thread_id`	`str \| None`	현재 실행의 스레드 ID. 체크포인터가 없으면 `None`.
`run_id`	`str \| None`	현재 실행의 Run ID. config에 없으면 `None`.
`checkpoint_id`	`str`	현재 실행의 체크포인트 ID.
`task_id`	`str`	현재 실행의 태스크 ID.

execution_info는 재시도 정책이 없어도 쓸 수 있는데, 그때 node_attempt는 기본값 1이에요.

타임아웃 (Timeouts)

참고: langgraph>=1.2가 필요해요.

add_node의 timeout= 파라미터는 노드 시도 하나가 얼마나 오래 돌 수 있는지 상한을 잡아줘요. 초 단위 숫자, timedelta, 그리고 실행/대기 상한을 따로 두는 TimeoutPolicy 중 하나를 넘기면 됩니다.

from datetime import timedelta
from langgraph.types import TimeoutPolicy

# 단순 벽시계 상한
builder.add_node("call_model", call_model, timeout=60)
builder.add_node("call_model", call_model, timeout=timedelta(minutes=2))

# 실행/대기 상한 분리
builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30),
)

경고: 노드 타임아웃은 async 노드에만 적용돼요. timeout이 붙은 sync 노드는 컴파일 시점에 거부됩니다. 블로킹 I/O를 감싸려면 async 노드 안에서 asyncio.to_thread를 쓰세요.

실행 타임아웃 (Run timeout)

run_timeout은 한 번의 시도에 걸리는 하드한 벽시계 상한이에요. 노드가 무엇을 하든 절대 초기화되지 않습니다.

from langgraph.types import TimeoutPolicy

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120),
)

상한을 넘기면 LangGraph가 NodeTimeoutError를 던지고, 실패한 시도에서 쓴 writes를 지운 뒤, 재시도 정책이 다시 시도할지 판단하게 합니다.

대기 타임아웃 (Idle timeout)

idle_timeout은 진행 상태에 따라 초기화되는 상한이에요. run_timeout과 달리, 노드가 지정한 시간 동안 눈에 보이는 진전을 만들지 않을 때만 발동됩니다. 노드가 진행 신호를 만들면 시계가 다시 리셋되죠.

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(idle_timeout=30),
)

run_timeout과 idle_timeout을 함께 둘 수도 있어요. 둘 중 먼저 발동하는 쪽이 시도를 취소합니다.

진행 신호 (Progress signals)

기본값 refresh_on="auto"에서는 다음 중 아무것에나 반응해서 대기 시계를 리셋해요.

CONFIG_KEY_SEND를 통한 상태 쓰기
스트림 출력(비동기 스트림 청크를 yield)
하위 태스크 스케줄링
런타임 stream-writer 호출
노드 또는 그 하위 요소에서 오는 LangChain 콜백 이벤트(LLM 토큰, 툴 호출, 체인 시작/끝 등)

하트비트 모드 (Heartbeat mode)

refresh_on="heartbeat"로 설정하면 리셋 소스를 명시적인 runtime.heartbeat() 호출로만 좁힐 수 있어요. 떠들썩한 하위 요소 때문에 대기 상태 정의가 흐트러지지 않게 하려는 경우에 유용하죠.

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(idle_timeout=30, refresh_on="heartbeat"),
)

수동 하트비트 (Manual heartbeats)

진행 신호를 자연스럽게 내지 않는 오래 걸리는 작업이라면 runtime.heartbeat()를 직접 호출해서 대기 시계를 리셋할 수 있어요.

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.types import TimeoutPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    result: str

async def long_running_node(state: State, runtime: Runtime) -> State:
    for batch in fetch_batches():
        process(batch)
        runtime.heartbeat()
    return {"result": "done"}

builder = StateGraph(State)
builder.add_node(
    "long_running_node",
    long_running_node,
    timeout=TimeoutPolicy(idle_timeout=30, refresh_on="heartbeat"),
)
builder.add_edge(START, "long_running_node")
builder.add_edge("long_running_node", END)

runtime.heartbeat()는 대기 타임아웃이 걸린 시도 밖에서는 아무 일도 하지 않는 no-op이에요. 그래서 조건 없이 호출해도 안전합니다.

NodeTimeoutError

타임아웃이 발동하면 LangGraph가 NodeTimeoutError를, 어떤 상한이 걸렸는지에 대한 구조화된 컨텍스트와 함께 던집니다.

속성	타입	설명
`node`	`str`	실행이 타임아웃된 노드의 이름.
`elapsed`	`float`	타임아웃이 발동할 때까지 흐른 시간(초).
`kind`	`Literal["idle", "run"]`	어떤 타임아웃이 발동했는지.
`idle_timeout`	`float \| None`	설정된 대기 타임아웃(초), 있다면.
`run_timeout`	`float \| None`	설정된 실행 타임아웃(초), 있다면.

NodeTimeoutError는 기본적으로 재시도 가능해요. timeout과 재시도 정책을 함께 쓰면 추가 설정 없이 잘 동작합니다. 시도가 새로 시작될 때마다 타임아웃 시계도 초기화되고, 타임아웃된 시도의 writes는 다음 재시도 전에 정리되죠.

from langgraph.types import RetryPolicy, TimeoutPolicy

builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(idle_timeout=30),
    retry_policy=RetryPolicy(max_attempts=3),
)

Send로 동적 타임아웃 다루기

Send로 노드를 동적으로 분배할 때(map-reduce 패턴 같은), Send에 직접 타임아웃을 넘겨서 이번 push에 한해 대상 노드의 정적 타임아웃을 덮어쓸 수 있어요.

from langgraph.types import Send, TimeoutPolicy

def fan_out(state: OverallState):
    return [
        Send("process_item", {"item": item}, timeout=TimeoutPolicy(idle_timeout=15))
        for item in state["items"]
    ]

Send에서 타임아웃을 생략하면 대상 노드의 타임아웃(add_node 시점에 설정된 것)이 적용됩니다. 이렇게 노드에 기본 타임아웃을 두고, 개별 호출에서는 더 타이트하게 조이는 게 가능해요.

오류 처리 (Error handling)

참고: langgraph>=1.2가 필요해요.

오류 핸들러는 노드가 실패하고 모든 재시도가 소진된 뒤에 실행돼요. 현재 상태를 받아서 Command로 상태를 갱신하거나 다른 노드로 라우팅할 수 있습니다. 그래프 전체를 중단하는 대신 우아하게 복구하고 싶은 보상 흐름(Saga 패턴)에서 특히 유용하죠. add_node에 error_handler=를 넘기면 됩니다.

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict

class State(TypedDict):
    status: str

def charge_payment(state: State) -> State:
    raise RuntimeError("payment gateway timeout")

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

def finalize(state: State) -> State:
    return state

graph = (
    StateGraph(State)
    .add_node(
        "charge_payment",
        charge_payment,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
        error_handler=payment_error_handler,
    )
    .add_node("finalize", finalize)
    .add_edge(START, "charge_payment")
    .compile()
)

핸들러는 재시도 정책이 소진된 뒤에만, 재시도 정책을 아예 설정하지 않았다면 즉시 발동해요. 재시도 정책과 오류 핸들러는 서로 분리된 채로 유지됩니다. 언제 재시도할지, 언제 보상할지는 각자 독립적으로 정하면 되죠.

NodeError

오류 핸들러는 타입 어노테이션으로 주입되는 error: NodeError 파라미터를 통해 실패 컨텍스트를 받아요. 이건 runtime: Runtime을 주입받는 패턴과 같습니다.

from langgraph.errors import NodeError

def my_handler(state: State, error: NodeError) -> Command:
    print(f"Node {error.node} failed with: {error.error}")
    return Command(update={"status": "recovered"}, goto="next_step")

NodeError는 두 필드를 가진 frozen dataclass예요.

속성	타입	설명
`node`	`str`	실행에 실패한 노드의 이름.
`error`	`BaseException`	실패한 노드가 던진 예외.

error: NodeError 파라미터는 옵트인(opt-in)이에요. 실패 컨텍스트가 필요 없는 핸들러는 (state)나 (state, runtime)처럼 더 간단한 시그니처를 쓸 수 있습니다.

Command로 라우팅하기

오류 핸들러는 Command를 돌려 상태를 갱신하고 특정 노드로 라우팅할 수 있어서, Saga/보상 패턴을 가능하게 해줘요.

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict

class State(TypedDict):
    status: str

def reserve_inventory(state: State) -> State:
    return {"status": "reserved"}

def charge_payment(state: State) -> State:
    raise RuntimeError("payment timeout")

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated_after_{error.node}: {error.error}"},
        goto="finalize",
    )

def finalize(state: State) -> State:
    return state

graph = (
    StateGraph(State)
    .add_node("reserve_inventory", reserve_inventory)
    .add_node(
        "charge_payment",
        charge_payment,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
        error_handler=payment_error_handler,
    )
    .add_node("finalize", finalize)
    .add_edge(START, "reserve_inventory")
    .add_edge("reserve_inventory", "charge_payment")
    .compile()
)

charge_payment는 ConnectionError에 대해 최대 3번 재시도해요. 재시도가 소진됐거나(아니면 오류가 ConnectionError가 아니거나) 핸들러가 상태를 갱신하고 finalize로 라우팅해서 그래프를 중단하는 대신 보상하는 구조죠.

재개에 안전한 실패 (Resume-safe failures)

실패의 출처(프로비넌스)는 체크포인트에 남아요. 노드가 실패한 뒤 핸들러가 끝나기 전에 그래프가 중단되거나 프로세스가 죽어도, 그래프가 체크포인트에서 재개되면 핸들러는 동일한 NodeError 컨텍스트를 받게 됩니다.

`interrupt()`와의 동작

노드 안에서 던져진 interrupt()는 오류 핸들러로 라우팅되지 않아요. 인터럽트는 GraphBubbleUp 메커니즘으로 그래프 실행을 멈춰서 휴먼-인-더-루프 워크플로를 지원하는데, 재시도 정책과 오류 핸들러를 모두 우회합니다. 그래프는 평소처럼 일시정지되죠.

서브그래프 실패

노드가 서브그래프를 감싸고 있는데 서브그래프가 처리되지 않은 예외를 던지면, 그 예외는 부모 노드로 올라와요. 부모 노드에 오류 핸들러가 있다면 핸들러가 서브그래프의 예외를 error.error에 담아 발동합니다.

그래프 기본값 (Graph defaults)

참고: langgraph>=1.2가 필요해요.

매 add_node 호출마다 같은 retry_policy=나 error_handler=, timeout=, cache_policy=를 반복하는 대신, set_node_defaults로 그래프 전체 기본값을 한 곳에 모아 설정할 수 있어요.

from langgraph.errors import NodeError
from langgraph.types import RetryPolicy, TimeoutPolicy
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict

class State(TypedDict):
    status: str

def default_error_handler(state: State, error: NodeError) -> State:
    return {"status": f"handled: {error.error}"}

graph = (
    StateGraph(State)
    .set_node_defaults(
        retry_policy=RetryPolicy(max_attempts=3),
        error_handler=default_error_handler,
        timeout=TimeoutPolicy(run_timeout=30),
    )
    .add_node("step_a", step_a)
    .add_node("step_b", step_b)
    .add_edge(START, "step_a")
    .compile()
)

step_a와 step_b 둘 다 중복 없이 같은 재시도 정책, 오류 핸들러, 타임아웃을 공유하게 됩니다.

우선순위 (Precedence)

add_node()에 직접 넘긴 노드별 값은 항상 set_node_defaults()로 설정한 기본값을 덮어써요. 기본값은 compile() 시점에 해석되기 때문에, set_node_defaults()를 add_node()보다 먼저 호출하든 나중에 호출하든 순서는 상관없습니다.

graph = (
    StateGraph(State)
    .set_node_defaults(error_handler=default_error_handler)
    .add_node("step_a", step_a)                                     # uses default_error_handler
    .add_node("step_b", step_b, error_handler=custom_error_handler) # uses custom_error_handler
    .add_edge(START, "step_a")
    .compile()
)

기본 오류 핸들러

error_handler 기본값은 모든 그래프 실행이 외부 프로세스(예: 백그라운드 작업 행)에 매핑되고, 처리되지 않은 노드 실패를 그 프로세스에 실패로 표시하고 싶을 때 특히 값져요. 매 add_node에 error_handler=를 반복하지 않아도 되죠. 특정 단계가 자기만의 로직이 필요하면 노드별 핸들러가 여전히 우선합니다.

from langgraph.errors import NodeError
from langgraph.graph import StateGraph, START
from langgraph.types import Command, RetryPolicy
from typing_extensions import TypedDict

class State(TypedDict):
    process_id: str
    status: str

def fetch_data(state: State) -> State:
    return {"status": "fetched"}

def charge_payment(state: State) -> State:
    raise RuntimeError("payment timeout")

def finalize(state: State) -> State:
    return state

def mark_process_failed(state: State, error: NodeError) -> State:
    # Persist failure on the external process row keyed by process_id.
    return {"status": f"failed at {error.node}: {error.error}"}

def refund_payment(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated after {error.node}"},
        goto="finalize",
    )

graph = (
    StateGraph(State)
    .set_node_defaults(
        retry_policy=RetryPolicy(max_attempts=3),
        error_handler=mark_process_failed,
    )
    .add_node("fetch_data", fetch_data)  # uses mark_process_failed
    .add_node(
        "charge_payment",
        charge_payment,
        error_handler=refund_payment,  # overrides the graph-wide default
    )
    .add_node("finalize", finalize)
    .add_edge(START, "fetch_data")
    .add_edge("fetch_data", "charge_payment")
    .compile()
)

fetch_data가 재시도 후 실패하면 mark_process_failed가 실행돼요. charge_payment가 재시도 후 실패하면 노드별 핸들러가 기본값을 덮어쓰므로 refund_payment가 대신 실행됩니다. 핸들러는 Error handling에서 설명한 것과 같은 (state, error: NodeError) 시그니처를 받아요. config 값(thread_id 같은)이 필요하다면 RunnableConfig를 세 번째 인자(옵션)로 받을 수도 있습니다.

from langchain_core.runnables import RunnableConfig

def mark_process_failed(
    state: State, error: NodeError, config: RunnableConfig
) -> State:
    thread_id = config["configurable"].get("thread_id")
    return {"status": f"failed on thread {thread_id}: {error.error}"}

적용 범위 행렬 (Applicability matrix)

모든 기본값이 모든 노드 타입에 적용되는 건 아니에요. 오류 핸들러 노드(add_node(error_handler=...)로 등록된 것들)는 안전하지 않은 동작을 막기 위해 특정 기본값에서 제외됩니다.

`set_node_defaults` 파라미터	일반 노드에 적용	오류 핸들러 노드에 적용	이유
`retry_policy`	✅	✅	핸들러도 일시적 실패에는 재시도돼야 함
`timeout`	✅	✅	막힌 핸들러도 일반 노드처럼 취소돼야 함
`error_handler`	✅	❌	핸들러는 절대 자기 자신을 잡으면 안 됨
`cache_policy`	✅	❌	핸들러 결과를 캐시하는 건 안전하지 않음

범위 (Scope)

부모 그래프에 설정한 기본값은 서브그래프에 상속되지 않아요. 각 그래프가 자기 기본값을 따로 관리합니다.

함수형 API (Functional API)

timeout=과 retry_policy= 파라미터는 함수형 API의 @task와 @entrypoint에서도 그대로 쓸 수 있어요.

from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy, TimeoutPolicy

@task(
    timeout=TimeoutPolicy(idle_timeout=30),
    retry_policy=RetryPolicy(max_attempts=3),
)
async def call_api(url: str) -> str:
    response = await fetch(url)
    return response.text

@entrypoint(timeout=60)
async def my_workflow(inputs: dict) -> str:
    result = await call_api("https://api.example.com/data")
    return result

동작은 add_node와 동일해요. 타임아웃이 걸리면 NodeTimeoutError가 발생하고, 버퍼링된 writes는 정리되며, 재시도 정책이 다시 시도할지 판단합니다.

우아한 종료 (Graceful shutdown)

협력적 종료(cooperative shutdown)를 쓰면 진행 중인 그래프 실행을 현재 슈퍼스텝이 끝난 뒤 멈추고, 재개 가능한 체크포인트를 저장할 수 있어요. SIGTERM 신호를 처리하거나, 작업을 잃지 않고 리소스를 회수해야 하는 외부 슈퍼바이저를 다룰 때 유용하죠.

참고: langgraph>=1.2가 필요해요.

RunControl을 만들고 control=로 invoke나 stream에 넘겨요. 아무 스레드에서 request_drain()을 호출하면 실행을 멈추라는 신호가 전달됩니다.

from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained

control = RunControl()

# In a signal handler or supervisor:
# control.request_drain("sigterm")

try:
    result = graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    # The graph stopped early and saved a checkpoint.
    # Resume later with the same config.
    print(f"Drained: {e.reason}")

의미 (Semantics)

Drain은 협력적으로 동작하고 슈퍼스텝 사이에서만 적용돼요. 이미 실행 중인 작업을 선점하지 않습니다.

시나리오	동작
노드 실행 중	끝까지 실행. Drain은 다음 슈퍼스텝에서 적용.
재시도 중인 노드	재시도 루프가 소진되거나 성공할 때까지 실행. 그 뒤 Drain 적용.
drain과 같은 틱에 그래프가 자연 종료	정상 반환. `control.drain_requested`로 정상 실행과 구분.
슈퍼스텝이 더 남음	`GraphDrained(reason)` 발생. 체크포인트 저장 후 재개 가능.
서브그래프가 drain 요청	`GraphDrained`가 부모로 올라가 다음 슈퍼스텝 경계에서 부모도 멈춤.

Drain 후 재개

같은 thread_id로 invoke(None, config)를 호출하면 종료된 실행을 재개할 수 있어요.

result = graph.invoke(None, config)

노드 안에서 drain 상태 읽기

슈퍼스텝 경계에 도달하기 전에 노드 동작을 조정하고 싶다면 runtime 파라미터로 drain 상태에 접근할 수 있어요.

from langgraph.runtime import Runtime

async def my_node(state: State, runtime: Runtime) -> State:
    if runtime.drain_requested:
        # Skip expensive work and return a minimal result
        return {"status": "skipped", "reason": runtime.drain_reason}
    return {"status": await do_work()}

SIGTERM 훅 패턴

프로세스 종료를 처리할 때 권장되는 패턴입니다.

import signal
from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained

control = RunControl()
signal.signal(signal.SIGTERM, lambda *_: control.request_drain("sigterm"))

try:
    result = graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    log.info("graph drained: %s", e.reason)
    # Resume on next startup with the same config

request_drain()은 실행 중인 asyncio 태스크를 취소하지도, 스레드를 죽이지도 않아요. 하드한 상한이 필요하다면 drain을 우아한 타임아웃과 태스크 취소와 함께 쓰세요.

한계 (Limitations)

타임아웃은 async 전용: timeout이 붙은 sync 노드는 컴파일 시점에 거부됩니다.
노드당 핸들러 하나: 각 노드는 error_handler를 최대 하나만 가질 수 있어요.
핸들러 실패는 전파: 오류 핸들러 자체가 예외를 던지면 핸들러가 없었던 것처럼 그 예외가 전파됩니다.
set_node_defaults는 서브그래프에 상속되지 않음: 각 그래프가 자기 기본값을 따로 관리합니다.

내결함성 (Fault Tolerance)