LangGraph - Persistence 영속성

 

 

 

Persistence 영속성

LangGraph는 체크포인트를 통해 구현된 내장형 영속성 레이어를 제공합니다. 체크포인트를 사용하여 그래프를 컴파일할 때, 체크포인트는 각 Super-steps 마다 그래프 상태의 체크포인트를 저장합니다. 이러한 체크포인트는 스레드에 저장되며, 그래프 실행 후에도 액세스 할 수 있습니다. 스레드는 실행 후에도 그래프의 상태에 액세스 할 수 있도록 허용하기 때문에, 인간 개입(human-in-the-loop), 메모리, 시간 여행, 고장 허용성 등 다양한 강력한 기능이 가능합니다.

 

 

LangGraph API는 체크포인트를 자동으로 처리하고 체크포인트를 수동으로 구현하거나 구성할 필요가 없고 API는 모든 영속성 인프라를 배경에서 자동으로 처리합니다.

---

 

👉🏻 Threads

스레드는 체크포인터에 의해 저장된 각 체크포인트에 할당되는 고유 ID 또는 스레드 식별자입니다. 이는 실행 시퀀스의 누적된 상태를 포함합니다. 실행이 수행될 때, 어시스턴트의 기본 그래프 상태가 스레드에 영구 저장됩니다.
체크포인터와 함께 그래프를 호출할 때, config의 구성 가능한 부분에 thread_id를 지정해야 합니다:

{“configurable”: {‘thread_id’: “1”}}
스레드의 현재 및 과거 상태를 조회할 수 있습니다. 상태를 저장하려면 실행을 실행하기 전에 스레드를 생성해야 합니다. LangGraph Platform API는 스레드 및 스레드 상태를 생성하고 관리하기 위한 여러 엔드포인트를 제공합니다.

 

 

👉🏻 Checkpoints

 

특정 시점에서의 스레드의 상태를 체크포인트라고 합니다. 체크포인트는 각 슈퍼 스텝에서 저장된 그래프 상태의 스냅샷이며, 다음 주요 속성을 가진 StateSnapshot 객체로 표현됩니다.

 

config : 이 체크포인트와 연관된 구성 정보
metadata : 이 체크포인트와 연관된 메타데이터
values : 이 시점에서의 상태 채널의 값
next : 그래프에서 다음에 실행할 노드 이름의 튜플
tasks : 다음에 실행할 작업에 대한 정보를 포함하는 PregelTask 객체의 튜플. 이전에 해당 단계가 시도된 경우 오류 정보가 포함됩니다. 노드 내에서 그래프가 동적으로 중단된 경우 tasks에는 중단과 관련된 추가 데이터가 포함됩니다.

 

체크포인트는 영구 저장되며 나중에 스레드의 상태를 복원하는 데 사용할 수 있습니다.

 

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

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

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}


workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

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

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": ""}, config)

 

위 예제를 보면 그래프를 실행한 후 정확히 4개의 체크포인트를 확인할 수 있습니다.
1. START가 다음 실행될 노드로 지정된 빈 체크포인트
2. 사용자 입력 {‘foo’: ‘’, ‘bar’: []}과 다음 실행될 노드로 node_a가 지정된 체크포인트
3. node_a의 출력 {‘foo’: ‘a’, ‘bar’: [‘a’]}와 다음 실행 노드로 node_b를 가진 체크포인트
4. node_b의 출력 {‘foo’: ‘b’, ‘bar’: [‘a’, ‘b’]}와 다음 실행 노드가 없는 체크포인트

참고로, bar 값에는 두 노드 모두의 출력이 포함되어 있습니다. 이는 bar에 리듀서가 있기 때문입니다.

 

 

👉🏻 Get state

저장된 그래프 상태와 상호작용할 때 스레드 식별자를 반드시 지정해야 합니다. 그래프의 최신 상태를 확인하려면 graph.get_state(config)를 호출하면 됩니다. 이 메서드는 config에 지정된 스레드 ID와 연관된 최신 체크포인트 또는 config에 체크포인트 ID가 지정된 경우 해당 스레드와 연관된 체크포인트에 해당하는 StateSnapshot 객체를 반환합니다.

 

# get the latest state snapshot
print(graph.get_state(config))

# get a state snapshot for a specific checkpoint_id
config = {"configurable": {"thread_id": "1", "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"}}
print(graph.get_state(config))

 

StateSnapshot(values={'foo': 'b', 'bar': ['a', 'b']}, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f081c39-2be0-64a0-8002-ccac8bfd5a1a'}}, metadata={'source': 'loop', 'step': 2, 'parents': {}, 'thread_id': '1'}, created_at='2025-08-25T14:55:42.901772+00:00', parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f081c39-2bdf-6d02-8001-82c5af8ae4a8'}}, tasks=(), interrupts=())
StateSnapshot(values={}, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}}, metadata=None, created_at=None, parent_config=None, tasks=(), interrupts=())

 

 

👉🏻 Get state history

특정 스레드의 그래프 실행 이력을 확인하려면 graph.get_state_history(config)를 호출합니다. 이 메서드는 config에 제공된 스레드 ID와 연관된 StateSnapshot 객체 목록을 반환합니다. 중요한 점은 MemorySaver()로 체크포인트를 설정해야 하고, 체크포인트가 시간순으로 정렬되며, 가장 최근 체크포인트/StateSnapshot이 목록의 맨 앞에 위치한다는 것입니다.

# Compile
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke({"user_input":"My"}, config)

# 체크포인터 확인
print(result)
print(list(graph.get_state_history(config)))

 

 

👉🏻 InMemoryStore

• Memory Store (장기 메모리 저장소) 역할
• 노드에서 사용자 프로필, 사실 정보, 지식 조각 같은 걸 장기적으로 저장하고 검색할 때 씁니다.
• 특징
  • 벡터스토어/키-값 DB처럼 쓰이는 데이터 저장소
  • 체크포인터와는 별개 → 그래프 실행 상태가 아니라 사용자 지식·맥락 같은 장기 기억을 관리
  • 마찬가지로 휘발성(in-memory)이라 세션이 끝나면 사라짐

from langgraph.memory import Memory
from langgraph.checkpoint.memory
import InMemoryStore
memory_store = InMemoryStore()
memory = Memory(store=memory_store)

 

 

예시 코드

checkpointer = InMemorySaver()
long_term_store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=long_term_store)

# Invoke the graph
user_id = "1"
config = {"configurable": {"thread_id": "1", "user_id": user_id}}

# First let's just say hi to the AI
for update in graph.stream(
    {
        "user_input": "My"
    }, 
    config=config, 
    stream_mode="updates",
):
    print(update)

{'node1': {'foo': 'My name'}}
{'node2': {'bar': 'My name is'}}
{'node3': {'graph_output': 'My name is Lance'}}

 

예시 코드를 보면 Long-Term Memory 설정을 위해 InMemorySaver(스레드 기반 단기 메모리 저장(checkpointer)), InMemoryStore(장기 메모리 저장소(store))를 활용하여 사용자별 컨텍스트나 추가 정보를 스레드를 넘어 비슷한 컨텍스트에서 반복 사용할 수 있도록 저장합니다.

stream_mode="updates"는 노드별 상태 변경만 스트리밍으로 전달해 줍니다.

따라서, 그래프의 각 노드의 리턴된 값을 반환 받을 수 있습니다.

 

 

그 외에도 sqlite, postgres db, EncryptedSerializer를 활용하여 체크포인트 저장기로 선택적으로 모든 지속 상태를 암호화하는 기능도 있습니다.

 

 

---

마치며

오늘 포스팅에서는 LangGraph의 영속성 개념인 체크포인트 기반 상태 저장, 스레드 기반 관리, 단기 메모리, 장기 메모리 등을 통해 Graph의 상태 재현, 워크플로우 복원성, 대화 기억 유지를 효율적으로 활용하는 방법을 알아봤습니다.