Durable execution

​

Requirements

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

​

Determinism and Consistent Replay

• 작업 반복 방지: node에 부작용이 있는 여러 작업(예: 로깅, 파일 쓰기 또는 네트워크 호출)이 포함된 경우 각 작업을 별도의 task로 래핑하세요. 이렇게 하면 워크플로우가 재개될 때 작업이 반복되지 않고 persistence 레이어에서 결과를 검색합니다.
• 비결정론적 작업 캡슐화: 비결정론적 결과를 생성할 수 있는 코드(예: 난수 생성)를 tasks 또는 nodes 내부에 래핑하세요. 이렇게 하면 재개 시 워크플로우가 동일한 결과로 정확히 기록된 단계 순서를 따릅니다.
• 멱등성 작업 사용: 가능한 경우 부작용(예: API 호출, 파일 쓰기)이 멱등성을 갖도록 하세요. 이는 워크플로우에서 실패 후 작업이 재시도되면 처음 실행되었을 때와 동일한 효과를 갖는다는 것을 의미합니다. 이는 데이터 쓰기를 초래하는 작업에 특히 중요합니다. task가 시작되었지만 성공적으로 완료되지 못한 경우 워크플로우 재개 시 task를 다시 실행하며 기록된 결과에 의존하여 일관성을 유지합니다. 의도하지 않은 중복을 방지하기 위해 멱등성 키를 사용하거나 기존 결과를 확인하여 원활하고 예측 가능한 워크플로우 실행을 보장하세요.

​

Durability modes

• "exit"
• "async"
• "sync"

​

"exit"

​

"async"

​

"sync"

graph.stream(
    {"input": "test"},
    durability="sync"
)

​

Using tasks in nodes

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

# Define a TypedDict to represent the state
class State(TypedDict):
    url: str
    result: NotRequired[str]

def call_api(state: State):
    """Example node that makes an API request."""
    result = requests.get(state['url']).text[:100]  # Side-effect  #
    return {
        "result": result
    }

# Create a StateGraph builder and add a node for the call_api function
builder = StateGraph(State)
builder.add_node("call_api", call_api)

# Connect the start and end nodes to the call_api node
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

# Specify a checkpointer
checkpointer = InMemorySaver()

# Compile the graph with the checkpointer
graph = builder.compile(checkpointer=checkpointer)

# Define a config with a thread ID.
thread_id = uuid.uuid4()
config = {"configurable": {"thread_id": thread_id}}

# Invoke the graph
graph.invoke({"url": "https://www.example.com"}, config)

​

Resuming Workflows

• 워크플로우 일시 중지 및 재개: interrupt 함수를 사용하여 특정 지점에서 워크플로우를 일시 중지하고 Command primitive를 사용하여 업데이트된 상태로 재개합니다. 자세한 내용은 Interrupts를 참조하세요.
• 실패로부터 복구: 예외(예: LLM 제공자 중단) 후 마지막으로 성공한 checkpoint에서 워크플로우를 자동으로 재개합니다. 이는 동일한 thread identifier로 워크플로우를 실행하고 입력 값으로 None을 제공하는 것을 포함합니다(functional API를 사용한 이 예제 참조).

​

Starting Points for Resuming Workflows

• StateGraph (Graph API)를 사용하는 경우 시작 지점은 실행이 중지된 node의 시작 부분입니다.
• node 내부에서 subgraph 호출을 하는 경우 시작 지점은 중단된 subgraph를 호출한 부모 node가 됩니다. subgraph 내부에서 시작 지점은 실행이 중지된 특정 node가 됩니다.
• Functional API를 사용하는 경우 시작 지점은 실행이 중지된 entrypoint의 시작 부분입니다.