Graph API 개요

입력·출력·비공개 스키마는 각 노드가 읽는 것(입력 스키마)과 invoke가 반환하는 것(출력 스키마)을 제약합니다. stream에서 채널을 숨기지는 않습니다.

stream_mode="values"로 스트리밍하면 그래프는 기본적으로 비공개 채널을 포함한 모든 상태 채널을 방출합니다. values 스트리밍이 기본적으로 전체 상태 채널 집합을 사용하기 때문입니다(출력 스키마가 아니라). 그래서 bar 같은 비공개 채널은 invoke에서는 숨겨지지만 스트리밍 중에는 보입니다:

stream = graph.stream_events({"user_input": "My"}, version="v3")
for snapshot in stream.values:
    print(snapshot)
# {'user_input': 'My'}
# {'foo': 'My name', 'user_input': 'My'}
# {'foo': 'My name', 'user_input': 'My', 'bar': 'My name is'}        # <-- private channel
# {'foo': 'My name', 'user_input': 'My', 'graph_output': 'My name is Lance', 'bar': 'My name is'}

스트리밍 값을 특정 채널 집합(예: 출력 스키마만)으로 제한하려면 output_keys를 전달하세요:

stream = graph.stream_events(
    {"user_input": "My"},
    version="v3",
    output_keys=["graph_output"],  # [!code highlight]
)
for snapshot in stream.values:
    print(snapshot)
# {'graph_output': 'My name is Lance'}

각 단계에서 노드가 실제로 만든 채널만 필요하다면(전체 누적 상태가 아니라) stream_mode="updates"를 대신 사용하세요.

리듀서 (Reducers)

리듀서는 노드의 업데이트가 State에 어떻게 적용되는지 이해하는 열쇠입니다. State의 각 키는 독립적인 리듀서 함수를 가집니다. 리듀서 함수가 명시적으로 지정되지 않으면 해당 키에 대한 모든 업데이트가 그것을 오버라이드한다고 가정합니다. 기본 타입부터 시작해 몇 가지 다른 타입의 리듀서가 있어요:

리듀서 인자 (Reducer arguments)

모든 리듀서는 두 개의 위치 인자를 가진 이진 함수입니다:

• 왼쪽 인자: 해당 키에 대해 상태에 이미 저장된 현재 값.
• 오른쪽 인자: 노드가 반환한 해당 키에 대한 업데이트.

노드가 부분 업데이트를 반환하면 LangGraph는 갱신된 각 키에 대해 리듀서를 호출하고 반환 값을 새 상태 값으로 저장합니다:

new_value = reducer(left=current_state[key], right=node_update[key])

왼쪽 인자는 항상 누적 상태에서 옵니다. 오른쪽 인자는 항상 최신 노드 업데이트에서 옵니다. 다음 예시는 두 인자를 모두 명시적으로 이름 짓습니다:

from typing import Annotated

from typing_extensions import TypedDict


def append_strings(left: list[str], right: list[str]) -> list[str]:
    """Combine the existing state value (left) with a node update (right)."""
    return left + right


class State(TypedDict):
    tags: Annotated[list[str], append_strings]

상태가 {"tags": ["draft"]}이고 노드가 {"tags": ["review"]}를 반환한다고 가정해 봅시다. LangGraph는 다음을 호출합니다:

append_strings(left=["draft"], right=["review"])  # returns ["draft", "review"]

tags의 새 상태 값은 ["draft", "review"]입니다.

커스텀 리듀서는 왼쪽과 오른쪽 인자를 결합합니다. 기본 리듀서는 왼쪽 인자를 버리고 오른쪽만 유지합니다.

기본 리듀서 (Default reducer)

기본 리듀서는 왼쪽 인자를 무시하고 상태 값을 오른쪽 인자로 대체합니다. 다음 예시는 기본 리듀서를 사용하는 방법을 보여줍니다:

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: list[str]

이 예시에서는 어떤 키에도 리듀서 함수가 지정되지 않았습니다. 그래프 입력이 {"foo": 1, "bar": ["hi"]}라고 가정해 보겠습니다. 첫 Node가 {"foo": 2}를 반환한다고 가정합니다. 이는 상태에 대한 업데이트로 처리됩니다. Node가 전체 State 스키마를 반환할 필요는 없습니다 — 업데이트만 반환하면 되죠. 이 업데이트를 적용한 뒤 State는 {"foo": 2, "bar": ["hi"]}가 됩니다. 두 번째 노드가 {"bar": ["bye"]}를 반환하면 State는 {"foo": 2, "bar": ["bye"]}가 됩니다.

커스텀 리듀서 (Custom reducers)

커스텀 리듀서는 상태 값을 대체하는 대신 왼쪽과 오른쪽 인자를 결합합니다. 이는 목록에 업데이트를 추가하는 것처럼 값을 누적할 때 유용합니다. 다음 예시는 커스텀 리듀서를 지정하는 방법을 보여줍니다:

from operator import add
from typing import Annotated

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]

이 예시에서는 두 번째 키(bar)에 리듀서 함수(operator.add)를 지정하기 위해 Annotated 타입을 사용했습니다. 첫 번째 키는 그대로입니다. 그래프 입력이 {"foo": 1, "bar": ["hi"]}라고 가정합시다. 첫 Node가 {"foo": 2}를 반환한다고 가정합니다. 이는 상태에 대한 업데이트로 처리됩니다. Node는 전체 State 스키마를 반환할 필요가 없습니다 — 업데이트만 반환하면 되죠. 이 업데이트를 적용한 뒤 State는 {"foo": 2, "bar": ["hi"]}가 됩니다. 두 번째 노드가 {"bar": ["bye"]}를 반환하면 State는 {"foo": 2, "bar": ["hi", "bye"]}가 됩니다. 여기서 bar 키가 두 리스트를 더해 갱신됩니다.

덮어쓰기 (Overwrite)

리듀서 필드 리셋 (Resetting a reducer field)

리듀서의 흔한 혼동 지점: 병합 리듀서에서 빈 값을 반환해도 필드가 지워지지 않습니다. 리듀서가 왼쪽 인자에 오른쪽 인자를 병합하므로, 빈 업데이트가 병합되고 이전에 누적된 값은 유지됩니다.

이 패턴은 재시도 시도 사이에 지워야 하는 오류 버퍼나 재시도 카운터에 중요합니다:

from operator import add
from typing import Annotated

from typing_extensions import TypedDict


class State(TypedDict):
    errors: Annotated[list[str], add]


# node A returns {"errors": ["bad sql"]}
# node B returns {"errors": []}
# state["errors"] is still ["bad sql"]; the empty list is merged in, not cleared

병합 리듀서를 유지하면서 필드를 지우려면 업데이트를 Overwrite로 감싸세요:

from operator import add
from typing import Annotated

from langgraph.types import Overwrite
from typing_extensions import TypedDict


class State(TypedDict):
    errors: Annotated[list[str], add]


def clear_errors(state: State):
    # Bypass the merging reducer and clear the field
    return {"errors": Overwrite([])}

자세한 내용은 Overwrite로 리듀서 우회를 참고하세요.

추적되지 않는 값 (Untracked values)

UntrackedValue는 그래프 실행 중에는 존재해야 하지만 체크포인트에 절대 저장되지 않아야 하는 상태 필드에 사용됩니다. 그래프가 체크포인트에서 재개되면 추적되지 않는 값은 초기 상태로 리셋됩니다(또는 사용 불가).

이것은 다음에 유용합니다:

• 직렬화할 수 없는 데이터베이스 연결
• 재개 시 재구축해야 하는 임시 캐시
• 영속화하고 싶지 않은 큰 객체
• 매번 새로 전달해야 하는 런타임 전용 구성

import { StateSchema, UntrackedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  messages: MessagesValue,

  // Untracked: throws if multiple nodes write in same step (guard: true is default)
  dbConnection: new UntrackedValue<DatabaseConnection>(),

  // Untracked with guard: false allows multiple writes, keeps last value
  tempCache: new UntrackedValue(
    z.record(z.string(), z.unknown()),
    { guard: false }
  ),

  // Untracked without a schema (for maximum flexibility)
  runtimeConfig: new UntrackedValue(),
});

동작:

• 실행 중: 값은 일반 상태처럼 저장·접근됩니다.
• 체크포인트 시: 추적되지 않는 값은 체크포인트 데이터에서 제외됩니다.
• 재개 시: 추적되지 않는 값은 새로 시작합니다(비어 있거나 기본값).
• guard: true(기본값): 같은 단계에서 여러 노드가 쓰면 오류 발생.
• guard: false: 여러 쓰기 허용, 마지막 값이 승리.

타입 유틸리티 (Type utilities)

LangGraph는 노드와 조건부 엣지를 정의할 때 더 나은 TypeScript 타입 안전성을 위한 여러 타입 유틸리티를 제공합니다.

GraphNode

그래프 빌더 밖에서 정의한 노드 함수를 타입 지정하려면 GraphNode를 사용하세요:

import { GraphNode, StateSchema, Command } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  count: z.number().default(0),
  result: z.string(),
});

// Basic node - receives state, returns partial update
const incrementNode: GraphNode<typeof State> = (state) => {
  return { count: state.count + 1 };
};

// Async node
const fetchNode: GraphNode<typeof State> = async (state, config) => {
  const response = await fetch(`/api/data/${state.count}`);
  return { result: await response.text() };
};

// Node with Command routing - specify valid destinations
const routerNode: GraphNode<{ InputSchema: typeof State; Nodes: "process" | "done" }> = (state) => {
  if (state.count >= 10) {
    return new Command({ goto: "done" });
  }
  return new Command({
    update: { count: state.count + 1 },
    goto: "process"
  });
};

State.Node 축약 (State.Node shorthand)

각 StateSchema 인스턴스는 노드를 타입 지정하는 축약을 제공하는 Node 속성을 가집니다:

const State = new StateSchema({
  messages: MessagesValue,
  step: z.string(),
});

// These are equivalent:
const myNode1: GraphNode<typeof State> = (state) => ({ step: "done" });
const myNode2: typeof State.Node = (state) => ({ step: "done" });

ConditionalEdgeRouter

조건부 엣지의 라우팅 함수(상태 업데이트 없이 라우팅만)에는 ConditionalEdgeRouter를 사용하세요:

import { ConditionalEdgeRouter, END } from "@langchain/langgraph";

const State = new StateSchema({
  shouldContinue: z.boolean(),
  step: z.string(),
});

// Router returns node name(s) or END
const router: ConditionalEdgeRouter<{ InputSchema: typeof State; Nodes: "process" | "summarize" }> = (state) => {
  if (!state.shouldContinue) {
    return END;
  }
  return state.step === "initial" ? "process" : "summarize";
};

// Use in graph
graph.addConditionalEdges("check", router);

StateSchema.State와 StateSchema.Update

스키마에서 상태·업데이트 타입을 추출해 커스텀 타입 정의에 사용합니다:

import { StateSchema } from "@langchain/langgraph";

const MyStateSchema = new StateSchema({
  messages: MessagesValue,
  count: z.number().default(0),
});

// Extract the full state type
type MyState = typeof MyStateSchema.State;
// { messages: BaseMessage[], count: number }

// Extract the update type (partial, with reducer input types)
type MyUpdate = typeof MyStateSchema.Update;
// { messages?: Messages, count?: number }

그래프 상태에서 메시지 다루기 (Working with messages in graph state)

왜 메시지를 쓸까 (Why use messages?)

대부분의 현대 LLM 프로바이더는 메시지 목록을 입력으로 받는 채팅 모델 인터페이스를 가집니다. 특히 LangChain의 chat model 인터페이스는 메시지 객체 목록을 입력으로 받아요. 이러한 메시지는 HumanMessage(사용자 입력) 또는 AIMessage(LLM 응답) 같은 다양한 형태로 옵니다.

메시지 객체가 무엇인지 더 읽으려면 Messages 개념 가이드를 참고하세요.

그래프에서 메시지 사용 (Using messages in your graph)

많은 경우 그래프 상태에 이전 대화 이력을 메시지 목록으로 저장하는 것이 유용합니다. 이렇게 하려면 그래프 상태에 Message 객체 목록을 저장하는 키(채널)를 추가하고 리듀서 함수로 어노테이션을 답니다(아래 예시의 messages 키 참고). 리듀서 함수는 각 상태 업데이트(예: 노드가 업데이트를 보낼 때)에서 상태의 Message 객체 목록을 어떻게 갱신할지 그래프에 알려주는 데 중요합니다. 리듀서를 지정하지 않으면 모든 상태 업데이트가 메시지 목록을 가장 최근 제공된 값으로 덮어씁니다. 단순히 기존 목록에 메시지를 추가하려면 operator.add를 리듀서로 사용할 수 있어요.

하지만 그래프 상태에서 메시지를 수동으로 업데이트하고 싶을 수도 있습니다(예: human-in-the-loop). operator.add를 사용하면 그래프에 보내는 수동 상태 업데이트가 기존 메시지를 갱신하는 대신 기존 메시지 목록에 추가됩니다. 이를 피하려면 메시지 ID를 추적하고 업데이트 시 기존 메시지를 덮어쓰는 리듀서가 필요합니다. 이를 위해 사전 빌드된 add_messages 함수를 사용할 수 있어요. 새 메시지에는 단순히 기존 목록에 추가하지만, 기존 메시지의 업데이트도 올바르게 처리합니다.

직렬화 (Serialization)

메시지 ID를 추적하는 것 외에도 add_messages 함수는 messages 채널에 상태 업데이트가 수신될 때마다 메시지를 LangChain Message 객체로 역직렬화하려 시도합니다.

자세한 내용은 LangChain 직렬화/역직렬화를 참고하세요. 이렇게 하면 그래프 입력/상태 업데이트를 다음 형식으로 보낼 수 있습니다:

# this is supported
{"messages": [HumanMessage(content="message")]}

# and this is also supported
{"messages": [{"type": "human", "content": "message"}]}

add_messages를 사용하면 상태 업데이트가 항상 LangChain Messages로 역직렬화되므로, state["messages"][-1].content처럼 점 표기법으로 메시지 속성에 접근해야 합니다.

아래는 add_messages를 리듀서 함수로 사용하는 그래프의 예시입니다.

from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

MessagesState

상태에 메시지 목록이 있는 것이 매우 흔하기 때문에, 메시지를 쉽게 사용할 수 있게 해주는 사전 빌드된 상태 MessagesState가 있습니다. MessagesState는 add_messages 리듀서를 사용하고 AnyMessage 객체 목록인 단일 messages 키로 정의됩니다. 보통 메시지보다 추적할 상태가 더 많으므로, 이 상태를 서브클래스화하고 필드를 추가하는 것을 봅니다:

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

노드 (Nodes)

LangGraph에서 노드는 다음 인자를 받는 Python 함수(동기 또는 비동기)입니다:

1. state — 그래프의 state
2. config — thread_id 같은 구성 정보와 tags 같은 트레이싱 정보를 담는 RunnableConfig 객체
3. runtime — 런타임 context와 store, stream_writer, execution_info, server_info, heartbeat(유휴 타임아웃 갱신용), control(graceful shutdown용) 같은 정보를 담는 Runtime 객체

NetworkX와 비슷하게 add_node 메서드로 이 노드들을 그래프에 추가합니다:

from dataclasses import dataclass
from typing_extensions import TypedDict

from langgraph.graph import StateGraph
from langgraph.runtime import Runtime

class State(TypedDict):
    input: str
    results: str

@dataclass
class Context:
    user_id: str

builder = StateGraph(State)

def plain_node(state: State):
    return state

def node_with_runtime(state: State, runtime: Runtime[Context]):
    print("In node: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}

def node_with_execution_info(state: State, runtime: Runtime):
    print("In node with thread_id: ", runtime.execution_info.thread_id)  # [!code highlight]
    return {"results": f"Hello, {state['input']}!"}


builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)
builder.add_node("node_with_execution_info", node_with_execution_info)
...

내부적으로 함수는 RunnableLambda로 변환되어, 네이티브 트레이싱·디버깅과 함께 배치·async 지원을 함수에 추가합니다.

이름을 지정하지 않고 그래프에 노드를 추가하면 함수 이름과 동일한 기본 이름이 주어집니다.

builder.add_node(my_node)
# You can then create edges to/from this node by referencing it as `"my_node"`

재실행과 멱등성 (Re-execution and idempotency)

체크포인터로 컴파일하면 LangGraph는 노드 함수 중간이 아니라 슈퍼스텝 경계에서 체크포인트를 저장합니다. 실행이 멈추고 나중에 재개되면(예: interrupt나 retry 후) 해당 노드가 함수 시작부터 다시 실행됩니다. 일시 중지 전의 코드와 부작용이 다시 실행돼요.

멱등성. 재실행이 상태를 손상시키지 않도록 노드 로직을 설계하세요. 노드가 데이터베이스 행을 삽입한다면, 의도적이지 않는 한 두 번 실행해 중복 행이 생기지 않아야 합니다. 멱등성 키, upsert, 또는 읽기-전-쓰기 검사를 사용하세요. interrupt() 주변의 부작용은 Side effects called before interrupt must be idempotent를 참고하세요.

그래프 변경. Determinism 규칙 중 코드 변경에 관한 것은 그래프 구조에는 적용되지 않습니다. 기존 스레드의 재개를 깨뜨리지 않고 노드와 엣지를 추가·제거할 수 있어요. 재개된 런은 저장된 상태를 사용하고 현재 컴파일하는 그래프를 실행합니다.

노드 안의 태스크와 인터럽트. 노드가 tasks나 interrupt를 호출하면, 재개 시 더 엄격한 결정성 규칙이 적용됩니다. LangGraph는 완료된 task 결과를 체크포인터에서 복원하지만, 재개 지점 전의 코드에서 task나 interrupt 순서를 바꾸면 캐시된 값이 어긋날 수 있어요. Functional API entrypoint는 전체 엔트리포인트 메서드를 이런 방식으로 실행하는 단일 노드로 컴파일됩니다. Determinism, Idempotency, 노드에서 태스크 사용을 참고하세요.

노드에서 태스크 사용 (Using tasks in nodes)

노드에 여러 작업이 있다면, 로직을 여러 노드로 나누는 대신 각 작업을 task로 구현하는 것이 더 쉬울 수 있어요. 그래프가 체크포인터를 사용하면 태스크 결과는 체크포인트되므로, 스레드를 재개하면 노드 안에서 완료된 task 작업을 건너뛸 수 있습니다.

원본 — API 요청을 하는 노드:

from typing import NotRequired

import requests
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


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]  # [!code highlight]
    return {"result": result}


builder = StateGraph(State)
builder.add_node("call_api", call_api)
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

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

태스크로 — 체크포인트되는 태스크로 API 요청을 하는 노드:

from typing import NotRequired

import requests
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import task
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class State(TypedDict):
    urls: list[str]
    results: NotRequired[list[str]]


@task
def _make_request(url: str):
    """Make a request."""
    return requests.get(url).text[:100]  # [!code highlight]


def call_api(state: State):
    """Example node that makes API requests as checkpointed tasks."""
    futures = [_make_request(url) for url in state["urls"]]  # [!code highlight]
    results = [f.result() for f in futures]
    return {"results": results}


builder = StateGraph(State)
builder.add_node("call_api", call_api)
builder.add_edge(START, "call_api")
builder.add_edge("call_api", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

graph.invoke({"urls": ["https://www.example.com"]}, config)

START 노드

START 노드는 사용자 입력을 그래프로 보내는 노드를 나타내는 특수 노드입니다. 이 노드를 참조하는 주된 이유는 어떤 노드를 먼저 호출해야 하는지 결정하기 위함입니다.

from langgraph.graph import START

graph.add_edge(START, "node_a")

END 노드

END 노드는 터미널 노드를 나타내는 특수 노드입니다. 완료 후 더 이상 동작이 없는 엣지를 나타낼 때 참조됩니다.

from langgraph.graph import END

graph.add_edge("node_a", END)

노드 캐싱 (Node caching)

LangGraph는 노드에 대한 입력에 기반한 태스크/노드 캐싱을 지원합니다. 캐싱을 사용하려면:

• 그래프를 컴파일할 때 캐시를 지정(또는 엔트리포인트 지정)
• 노드에 대해 캐시 정책을 지정. 각 캐시 정책은 다음을 지원:
  • 노드 입력에 기반해 캐시 키를 생성하는 key_func. 기본은 입력을 pickle로 hash한 값.
  • ttl, 캐시의 만료 시간(초). 지정하지 않으면 캐시는 만료되지 않습니다.

예를 들어:

import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy


class State(TypedDict):
    x: int
    result: int


builder = StateGraph(State)


def expensive_node(state: State) -> dict[str, int]:
    # expensive computation
    time.sleep(2)
    return {"result": state["x"] * 2}


builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=3))
builder.set_entry_point("expensive_node")
builder.set_finish_point("expensive_node")

graph = builder.compile(cache=InMemoryCache())

print(graph.invoke({"x": 5}, stream_mode='updates'))    # [!code highlight]
# [{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates'))    # [!code highlight]
# [{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

graph.add_edge("node_a", "node_b")

graph.add_conditional_edges("node_a", routing_function)

graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})

from langgraph.graph import START

graph.add_edge(START, "node_a")

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})

from langgraph.types import Send

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state['subjects']]

graph.add_conditional_edges("node_a", continue_to_jokes)

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    if state["foo"] == "bar":
        return Command(update={"foo": "baz"}, goto="my_other_node")

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )

# WRONG - graph resumes from the latest checkpoint
# (last step that ran), appears stuck
graph.invoke(Command(update={  # [!code --]
    "messages": [{"role": "user", "content": "follow up"}]  # [!code --]
}), config)  # [!code --]

# CORRECT - plain dict restarts from __start__
graph.invoke( {  # [!code ++]
    "messages": [{"role": "user", "content": "follow up"}]  # [!code ++]
}, config)  # [!code ++]

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class State(TypedDict):
    messages: list[dict]


def human_review(state: State):
    # Pauses the graph and waits for a value
    answer = interrupt("Do you approve?")
    return {"messages": [{"role": "user", "content": answer}]}


graph = (
    StateGraph(State)
    .add_node("human_review", human_review)
    .add_edge(START, "human_review")
    .add_edge("human_review", END)
    .compile(checkpointer=InMemorySaver())
)

config = {"configurable": {"thread_id": "graph-api-resume"}}

# First run - hits the interrupt and pauses
stream = graph.stream_events({"messages": []}, config, version="v3")
_ = stream.output  # drive the stream to completion
print(stream.interrupts)

# Resume with a value - the interrupt() call returns "yes"
resumed = graph.stream_events(Command(resume="yes"), config, version="v3")
final = resumed.output

@dataclass
class ContextSchema:
    llm_provider: str = "openai"

graph = StateGraph(State, context_schema=ContextSchema)

graph.invoke(inputs, context={"llm_provider": "anthropic"})

from langgraph.runtime import Runtime

def node_a(state: State, runtime: Runtime[ContextSchema]):
    llm = get_llm(runtime.context.llm_provider)
    # ...

graph.invoke(inputs, config={"recursion_limit": 5}, context={"llm": "anthropic"})

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

def my_node(state: dict, config: RunnableConfig) -> dict:
    current_step = config["metadata"]["langgraph_step"]
    print(f"Currently on step: {current_step}")
    return state

from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps

class State(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]
    remaining_steps: RemainingSteps  # Managed value - tracks steps until limit

def reasoning_node(state: State) -> dict:
    # RemainingSteps is automatically populated by LangGraph
    remaining = state["remaining_steps"]

    # Check if we're running low on steps
    if remaining <= 2:
        return {"messages": ["Approaching limit, wrapping up..."]}

    # Normal processing
    return {"messages": ["thinking..."]}

def route_decision(state: State) -> Literal["reasoning_node", "fallback_node"]:
    """Route based on remaining steps"""
    if state["remaining_steps"] <= 2:
        return "fallback_node"
    return "reasoning_node"

def fallback_node(state: State) -> dict:
    """Handle cases where recursion limit is approaching"""
    return {"messages": ["Reached complexity limit, providing best effort answer"]}

# Build graph
builder = StateGraph(State)
builder.add_node("reasoning_node", reasoning_node)
builder.add_node("fallback_node", fallback_node)
builder.add_edge(START, "reasoning_node")
builder.add_conditional_edges("reasoning_node", route_decision)
builder.add_edge("fallback_node", END)

graph = builder.compile()

# RemainingSteps works with any recursion_limit
result = graph.invoke({"messages": []}, {"recursion_limit": 10})

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.managed import RemainingSteps
from langgraph.errors import GraphRecursionError

class State(TypedDict):
    messages: Annotated[list, lambda x, y: x + y]
    remaining_steps: RemainingSteps

# Proactive Approach (recommended) - using RemainingSteps
def agent_with_monitoring(state: State) -> dict:
    """Proactively monitor and handle recursion within the graph"""
    remaining = state["remaining_steps"]

    # Early detection - route to internal handling
    if remaining <= 2:
        return {
            "messages": ["Approaching limit, returning partial result"]
        }

    # Normal processing
    return {"messages": [f"Processing... ({remaining} steps remaining)"]}

def route_decision(state: State) -> Literal["agent", END]:
    if state["remaining_steps"] <= 2:
        return END
    return "agent"

# Build graph
builder = StateGraph(State)
builder.add_node("agent", agent_with_monitoring)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route_decision)
graph = builder.compile()

# Proactive: Graph completes gracefully
result = graph.invoke({"messages": []}, {"recursion_limit": 10})

# Reactive Approach (fallback) - catching error externally
try:
    result = graph.invoke({"messages": []}, {"recursion_limit": 10})
except GraphRecursionError as e:
    # Handle externally after graph execution fails
    result = {"messages": ["Fallback: recursion limit exceeded"]}

def inspect_metadata(state: dict, config: RunnableConfig) -> dict:
    metadata = config["metadata"]

    print(f"Step: {metadata['langgraph_step']}")
    print(f"Node: {metadata['langgraph_node']}")
    print(f"Triggers: {metadata['langgraph_triggers']}")
    print(f"Path: {metadata['langgraph_path']}")
    print(f"Checkpoint NS: {metadata['langgraph_checkpoint_ns']}")

    return state