메인 콘텐츠로 건너뛰기
Alpha Notice: These docs cover the v1-alpha release. Content is incomplete and subject to change.For the latest stable version, see the current LangGraph Python or LangGraph JavaScript docs.
**지속 가능한 실행(Durable execution)**은 프로세스나 워크플로우가 주요 지점에서 진행 상황을 저장하여 일시 중지한 후 정확히 중단된 지점부터 다시 시작할 수 있도록 하는 기술입니다. 이는 사용자가 프로세스를 검사, 검증 또는 수정한 후 계속할 수 있는 human-in-the-loop가 필요한 시나리오와, 중단이나 오류가 발생할 수 있는 장시간 실행 작업(예: LLM 호출 시간 초과)에서 특히 유용합니다. 완료된 작업을 보존함으로써 지속 가능한 실행은 상당한 시간(예: 일주일 후)이 지난 후에도 이전 단계를 재처리하지 않고 프로세스를 재개할 수 있게 합니다. LangGraph의 내장 지속성 레이어는 워크플로우에 지속 가능한 실행을 제공하여 각 실행 단계의 상태가 영구 저장소에 저장되도록 보장합니다. 이 기능은 시스템 장애나 human-in-the-loop 상호작용으로 워크플로우가 중단되더라도 마지막으로 기록된 상태에서 재개할 수 있음을 보장합니다.
checkpointer와 함께 LangGraph를 사용하고 있다면 이미 지속 가능한 실행이 활성화되어 있습니다. 중단이나 장애가 발생한 후에도 언제든지 워크플로우를 일시 중지하고 재개할 수 있습니다. 지속 가능한 실행을 최대한 활용하려면 워크플로우가 결정론적이고 일관된 재실행이 가능하도록 설계되었는지 확인하고, 모든 부작용이나 비결정론적 연산을 task 안에 래핑하세요. StateGraph (Graph API)와 Functional API 모두에서 task를 사용할 수 있습니다.

​
요구사항

LangGraph에서 지속 가능한 실행을 활용하려면 다음이 필요합니다:
  1. 워크플로우 진행 상황을 저장할 checkpointer를 지정하여 워크플로우에서 지속성을 활성화합니다.
  2. 워크플로우를 실행할 때 스레드 식별자를 지정합니다. 이는 워크플로우의 특정 인스턴스에 대한 실행 기록을 추적합니다.
  3. 워크플로우가 재개될 때 이러한 연산이 특정 실행에 대해 반복되지 않고 대신 지속성 레이어에서 결과를 검색하도록, 비결정론적 연산(예: 난수 생성)이나 부작용이 있는 연산(예: 파일 쓰기, API 호출)을 task 안에 래핑합니다. 자세한 내용은 결정론과 일관된 재실행을 참조하세요.

​
결정론과 일관된 재실행

워크플로우 실행을 재개할 때, 코드는 실행이 중지된 동일한 코드 라인에서 재개되는 것이 아닙니다. 대신 중단된 지점부터 이어갈 적절한 시작 지점을 식별합니다. 즉, 워크플로우는 시작 지점부터 중지된 지점까지 모든 단계를 재실행합니다. 따라서 지속 가능한 실행을 위한 워크플로우를 작성할 때는 비결정론적 연산(예: 난수 생성)과 부작용이 있는 연산(예: 파일 쓰기, API 호출)을 task 또는 node 안에 래핑해야 합니다. 워크플로우가 결정론적이고 일관되게 재실행될 수 있도록 다음 가이드라인을 따르세요:
  • 작업 반복 방지: node에 부작용이 있는 여러 연산(예: 로깅, 파일 쓰기, 네트워크 호출)이 포함된 경우, 각 연산을 별도의 task로 래핑하세요. 이렇게 하면 워크플로우가 재개될 때 연산이 반복되지 않고 지속성 레이어에서 결과를 검색합니다.
  • 비결정론적 연산 캡슐화: 비결정론적 결과를 생성할 수 있는 코드(예: 난수 생성)를 task 또는 node 안에 래핑하세요. 이렇게 하면 재개 시 워크플로우가 동일한 결과로 정확히 기록된 단계 순서를 따르게 됩니다.
  • 멱등 연산 사용: 가능한 경우 부작용(예: API 호출, 파일 쓰기)이 멱등이 되도록 하세요. 즉, 워크플로우에서 장애 후 연산이 재시도되면 처음 실행했을 때와 동일한 효과를 갖게 됩니다. 이는 데이터 쓰기를 발생시키는 연산에 특히 중요합니다. task가 시작되었지만 성공적으로 완료되지 못한 경우, 워크플로우 재개 시 기록된 결과를 활용하여 일관성을 유지하면서 task를 다시 실행합니다. 의도하지 않은 중복을 방지하려면 멱등성 키를 사용하거나 기존 결과를 확인하여 원활하고 예측 가능한 워크플로우 실행을 보장하세요.
피해야 할 함정의 몇 가지 예는 functional API의 일반적인 함정 섹션을 참조하세요. 이 섹션은 이러한 문제를 방지하기 위해 task를 사용하여 코드를 구조화하는 방법을 보여줍니다. 동일한 원칙이 StateGraph (Graph API)에도 적용됩니다.

​
지속성 모드

LangGraph는 애플리케이션의 요구사항에 따라 성능과 데이터 일관성의 균형을 맞출 수 있는 세 가지 지속성 모드를 지원합니다. 지속성이 낮은 것부터 높은 순서로 지속성 모드는 다음과 같습니다: 더 높은 지속성 모드는 워크플로우 실행에 더 많은 오버헤드를 추가합니다.
v0.6.0에서 추가됨 지속성 정책 관리를 위해 checkpoint_during(v0.6.0에서 deprecated) 대신 durability 매개변수를 사용하세요:
  • durability="async"는 checkpoint_during=True를 대체합니다
  • durability="exit"는 checkpoint_during=False를 대체합니다
지속성 정책 관리를 위한 매핑은 다음과 같습니다:
  • checkpoint_during=True -> durability="async"
  • checkpoint_during=False -> durability="exit"

​
"exit"

변경 사항은 그래프 실행이 완료될 때(성공적으로 또는 오류와 함께)만 저장됩니다. 이는 장시간 실행되는 그래프에 최상의 성능을 제공하지만 중간 상태가 저장되지 않으므로 실행 중간의 장애에서 복구하거나 그래프 실행을 중단할 수 없습니다.

​
"async"

변경 사항은 다음 단계가 실행되는 동안 비동기적으로 저장됩니다. 이는 우수한 성능과 지속성을 제공하지만, 실행 중 프로세스가 충돌하면 체크포인트가 기록되지 않을 수 있는 작은 위험이 있습니다.

​
"sync"

변경 사항은 다음 단계가 시작되기 전에 동기적으로 저장됩니다. 이는 실행을 계속하기 전에 모든 체크포인트가 기록되도록 보장하여 일부 성능 오버헤드를 감수하고 높은 지속성을 제공합니다. 그래프 실행 메서드를 호출할 때 지속성 모드를 지정할 수 있습니다: 
graph.stream(
    {"input": "test"},
    durability="sync"
)

​
노드에서 task 사용하기

node에 여러 연산이 포함된 경우, 연산을 개별 노드로 리팩토링하는 것보다 각 연산을 task로 변환하는 것이 더 쉬울 수 있습니다.
  • Original
  • With task
from typing import NotRequired
from typing_extensions import TypedDict
import uuid

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
import requests

# 상태를 나타내는 TypedDict 정의
class State(TypedDict):
    url: str
    result: NotRequired[str]

def call_api(state: State):
    """API 요청을 수행하는 예제 노드."""
    result = requests.get(state['url']).text[:100]  # 부작용
    return {
        "result": result
    }

# StateGraph 빌더를 생성하고 call_api 함수에 대한 노드 추가
builder = StateGraph(State)
builder.add_node("call_api", call_api)

# start와 end 노드를 call_api 노드에 연결
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

# checkpointer 지정
checkpointer = InMemorySaver()

# checkpointer와 함께 그래프 컴파일
graph = builder.compile(checkpointer=checkpointer)

# 스레드 ID가 포함된 config 정의
thread_id = uuid.uuid4()
config = {"configurable": {"thread_id": thread_id}}

# 그래프 호출
graph.invoke({"url": "https://www.example.com"}, config)

​
워크플로우 재개하기

워크플로우에서 지속 가능한 실행을 활성화하면 다음 시나리오에서 실행을 재개할 수 있습니다:
  • 워크플로우 일시 중지 및 재개: interrupt 함수를 사용하여 특정 지점에서 워크플로우를 일시 중지하고 Command 프리미티브를 사용하여 업데이트된 상태로 재개합니다. 자세한 내용은 Interrupts를 참조하세요.
  • 장애에서 복구: 예외(예: LLM 공급자 중단) 후 마지막 성공한 체크포인트에서 워크플로우를 자동으로 재개합니다. 이는 동일한 스레드 식별자로 워크플로우를 실행하되 입력 값으로 None을 제공하는 것을 포함합니다(functional API의 이 예제 참조).

​
워크플로우 재개를 위한 시작 지점

  • StateGraph (Graph API)를 사용하는 경우, 시작 지점은 실행이 중지된 node의 시작 부분입니다.
  • 노드 내에서 서브그래프 호출을 수행하는 경우, 시작 지점은 중단된 서브그래프를 호출한 부모 노드가 됩니다. 서브그래프 내부에서 시작 지점은 실행이 중지된 특정 node가 됩니다.
  • Functional API를 사용하는 경우, 시작 지점은 실행이 중지된 entrypoint의 시작 부분입니다.
Edit the source of this page on GitHub
⌘I