[LangGraph 기초 4] Memory와 Checkpoint의 개념

Memory와 Checkpoint의 개념

---

Super Step 실행 모델 다시 짚기

 

 

지금까지 노드들이 순차적으로 단계별로 실행되는 것을 보았다.

이때 병렬 노드와 같은 특정 단계에서는 여러 노드가 동시에 활성화될 수 있다. (이를 Super Step이라고 부름)

State는 Graph  전체에서 공유되며, Super Step의 시작 시 Node에 제공되고, Super Step이 끝날 때 Node에 의해 업데이트된다.

 

Super Step 실행 모델을 정리하면 아래와 같다.

 

• 하나의 실행 단계에서 여러 노드가 동시에 활성화 가능
• 병렬 노드들은 동일한 State 스냅샷을 입력으로 받음
• 모든 노드 실행이 끝난 후 State가 업데이트됨

 

왜 Persistence(Memory)가 필요한가?

---

 

지금까지의 그래프는 한 번 실행하면 종료되는 구조였다.

 

• Graph 실행이 끝나면 State 소멸
• 다음 실행에서 이전 State를 기억하지 못함
• 장기 실행 Agent 구현 불가

이를 해결하기 위해 필요한 것이 바로 Persistence(영속성)이다.

 

 

Checkpointer란 무엇인가?

---

Checkpointer는 각 Super Step 종료 시 State를 저장하는 구성 요소이다.

• State의 스냅샷을 지속 저장소에 기록
• 실행 중단이나 실패 이후에도 State 복원 가능

즉, Checkpointer는 그래프에 메모리를 부여한다.

 

 

 

Thread 개념 이해하기

---

LangGraph에서 Thread란 다음을 의미한다.

• 하나의 실행 흐름에 대한 Checkpoint들의 집합
• 시간에 따른 State 변화의 전체 이력
• 동일한 Thread ID → 동일한 State 히스토리

Thread는 곧 에이전트의 기억 단위라고 볼 수 있다.

 

 

 

 

Checkpointer를 사용하는 것의 이점

---

1. 실패 복구 (Failure Recovery)

• 노드 실행 중 실패 발생
• 마지막 정상 Checkpoint에서 State 복원
• 처음부터 다시 실행할 필요가 없다. 이는 장시간 실행되는 Agent에 필수이다.
  
  

2. 과거 시점으로 롤백 가능

• 에이전트가 잘못된 방향으로 진행한 경우
• 특정 시점의 State로 되돌린 뒤 재실행

이는 디버깅과 안정성 확보에 매우 중요하다.

 

3. 그래프 비실행 상태에서도 State 유지

• 그래프가 실행 중이 아니어도 State는 존재할 수 있다.
• 이를 통해 장기 메모리(Long-term Memory) 구현이 가능하다.
  
  

4. 실행 중단 후 정확한 재개

• 노드 실행이 중단된 경우
• 동일 지점에서 다시 실행 가능
• 이후 소개할 Human-in-the-Loop의 핵심 기반

 

 

Checkpoint 구현 예제

---

이제 Memory를 추가해 Agent가 Runtime 중에도 상태를 기억하도록 만들어보자.

앞서 Conditional Edge에서 구축한 Graph는 State를 기억하지 못하는 구조이다.

여기서 이전 값은 Graph의 다음 호출에서 지속되지 않았다.

In [5]:

while True:
    user = input('b, c, or q to quit: ')
    print(user)
    input_state = State(nlist = [user])
    result = graph.invoke(input_state)
    print(result)
    if result['nlist'][-1] == "q":
        print("quit")
        break

 

OUT [5]:

b
{'nlist': ['b', 'b', 'B']}
c
{'nlist': ['c', 'c', 'C']}
q
{'nlist': ['q', 'q']}
quit

 

이 Conditional Edge Graph에 Checkpoint를 추가해 호출 간 State를 지속시킬 수 있다.

이 부분을 Checkpoint로 바꿔보자.

 

LangGraph는 기본적으로 세 가지 Checkpointer를 제공한다.

1. InMemorySaver : 메모리에 저장 / 가장 간단한 설정 / 실습 및 테스트용
2. PostgresSaver : PostgreSQL에 저장 / 프로덕션 환경에 적합
3. SQLiteSaver : SQLite DB에 저장 / 로컬/경량 환경에 적합

 

 

InMemorySaver 설정 방법

---

Saver는 각각 해당 DB에 Checkpoint를 저장한다.

Memory saver가 설정하기 가장 간단하다.

 

langGraph.checkpoint.memory에서 간단히 가져와서 instance화한 다음 thread ID를 구성한다.

이 configurable key가 있는 dictionary 구조가 실제로 그래프에 config를 전달하는 방법이다.

임의의 thread ID 1으로 thread ID를 정의하고 실행한다.

In [6]:

from langgraph.checkpoint.memory import InMemorySaver
memory = InMemorySaver()
config = {"configurable": {"thread_id": "1"}}

 

이제 Graph를 compile할 때 checkpointer를 추가하면 Graph에 Memory가 생긴다.

In [7]:

graph = builder.compile(checkpointer=memory) # Graph에 Memory 부여

 

이제 똑같이 while문을 돌면, while문 사이에서 State가 누적된 것을 볼 수 있다.

In [8]:

while True:
    user = input('b, c, or q to quit: ')
    print(user)
    input_state = State(nlist = [user])
    result = graph.invoke(input_state, config) # graph.invoke 시 thread_id를 가지는 config 인자를 포함시켜주기
    print(result)
    if result['nlist'][-1] == "q":
        print("quit")
        break

 

OUT [8]:

b
{'nlist': ['b', 'b', 'B']}
c
{'nlist': ['b', 'b', 'B', 'c', 'c', 'C']}
q
{'nlist': ['b', 'b', 'B', 'c', 'c', 'C', 'q', 'q']}
quit

 

 

 

핵심 요점

---

1. 메모리를 설정할 때, 우선 Checkkpointer를 instance화한다.

2. Memory saver 외에도 많은 옵션이 있다.

3, Graph를 구축할 때 해당 Checkpointer로 Graph를 호출한다.

4. Graph 실행 중엔는 설정 변수에 설정한 thread ID로 Graph를 호출한다.
(동일한 thread ID를 사용하면 Graph 실행 간에 State가 유지된다)