🤖

본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.

⚠️

본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.

이미지 로딩 중...

LangGraph Persistence 완벽 가이드 - 슬라이드 1/7
A

AI Generated

2025. 12. 12. · 11 Views

LangGraph Persistence 완벽 가이드

LangGraph의 Persistence 기능을 활용하여 대화 상태를 저장하고 관리하는 방법을 배웁니다. Thread와 Checkpoint를 이해하고, 상태 조회 및 수정 방법을 실무 중심으로 학습합니다.


목차

  1. Persistence 핵심 개념
  2. Thread와 Checkpoint 이해
  3. StateSnapshot 객체 구조
  4. get_state() 상태 조회
  5. get_state_history() 이력 조회
  6. update_state() 상태 수정

1. Persistence 핵심 개념

어느 날 김개발 씨가 챗봇 서비스를 개발하던 중 문제에 부딪혔습니다. 사용자와의 대화가 진행될 때마다 이전 대화 내용이 사라지는 것이었습니다.

"어떻게 하면 대화 기록을 유지할 수 있을까요?" 선배 박시니어 씨에게 물었더니 "Persistence를 사용해보세요"라는 답변을 들었습니다.

Persistence는 LangGraph에서 그래프의 상태를 영구적으로 저장하고 관리하는 핵심 기능입니다. 마치 게임에서 세이브 포인트를 만드는 것처럼, 대화의 진행 상황을 저장할 수 있습니다.

이를 통해 대화가 중단되더라도 언제든지 이전 상태로 돌아갈 수 있으며, 사용자별로 독립적인 대화 세션을 관리할 수 있습니다.

다음 코드를 살펴봅시다.

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# Checkpoint를 메모리에 저장하는 저장소 생성
memory = MemorySaver()

# StateGraph 생성 시 checkpointer 설정
graph = StateGraph(state_schema=MyState)
graph.add_node("agent", agent_node)
graph.add_edge("agent", END)

# Persistence 기능이 활성화된 그래프 컴파일
app = graph.compile(checkpointer=memory)

# Thread ID로 독립적인 세션 관리
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke({"messages": "안녕하세요"}, config)

김개발 씨는 입사 4개월 차 AI 엔지니어입니다. 최근 회사에서 고객 상담 챗봇 서비스를 개발하는 프로젝트를 맡았습니다.

열심히 LangGraph로 대화 로직을 구현했는데, 테스트 중에 치명적인 문제를 발견했습니다. 사용자가 "제 주문 번호는 12345입니다"라고 말한 후, 다음 질문을 할 때마다 챗봇이 주문 번호를 기억하지 못하는 것이었습니다.

매번 처음부터 다시 물어봐야 했고, 사용자 경험은 최악이었습니다. 선배 개발자 박시니어 씨가 코드를 살펴보더니 말했습니다.

"Persistence를 설정하지 않아서 그래요. 지금은 대화 상태가 메모리에만 있다가 사라지고 있어요." Persistence란 무엇일까요? 쉽게 비유하자면, Persistence는 마치 독서를 할 때 책갈피를 끼워두는 것과 같습니다.

책을 읽다가 잠시 덮어도, 나중에 다시 펼쳤을 때 책갈피가 있어서 어디까지 읽었는지 바로 알 수 있습니다. 이처럼 Persistence도 대화의 진행 상황을 저장해 두었다가, 나중에 이어서 대화할 수 있게 해줍니다.

Persistence가 없던 시절에는 어땠을까요? 초기 챗봇 시스템에서는 각 요청이 완전히 독립적이었습니다. 사용자가 "피자를 주문하고 싶어요"라고 말하면 챗봇이 응답하지만, 다음 요청인 "큰 사이즈로 주세요"를 이해하지 못했습니다.

무엇을 큰 사이즈로 해야 하는지 모르는 것이죠. 개발자들은 이 문제를 해결하기 위해 데이터베이스에 직접 상태를 저장하고, 세션 ID를 관리하고, 복잡한 로직을 작성해야 했습니다.

코드가 길어지고, 버그가 발생하기 쉬웠습니다. 더 큰 문제는 대화 히스토리를 추적하거나, 특정 시점으로 돌아가는 것이 거의 불가능했다는 점입니다.

바로 이런 문제를 해결하기 위해 LangGraph의 Persistence가 등장했습니다. Persistence를 사용하면 모든 대화 상태가 자동으로 저장됩니다. 사용자별로 독립적인 대화 스레드를 관리할 수 있고, 각 단계마다 체크포인트가 생성되어 언제든지 이전 상태로 되돌릴 수 있습니다.

무엇보다 개발자가 직접 상태 관리 로직을 작성하지 않아도 되어, 비즈니스 로직에만 집중할 수 있다는 큰 이점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다. 먼저 2번째 줄에서 MemorySaver를 import합니다.

이것은 체크포인트를 메모리에 저장하는 저장소입니다. 실제 프로덕션 환경에서는 SQLite나 PostgreSQL 같은 데이터베이스를 사용할 수 있습니다.

5번째 줄에서 MemorySaver 인스턴스를 생성합니다. 이 객체가 모든 상태 정보를 담당하게 됩니다.

11번째 줄이 핵심입니다. compile 메서드에 checkpointer 매개변수를 전달하여 Persistence 기능을 활성화합니다.

이제부터 그래프가 실행될 때마다 자동으로 상태가 저장됩니다. 14번째 줄에서는 thread_id를 설정합니다.

이것은 마치 고객별 상담 창구 번호와 같습니다. 같은 thread_id를 사용하면 이전 대화 내용을 이어갈 수 있습니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 은행 고객 상담 챗봇을 개발한다고 가정해봅시다. 고객이 "계좌 잔액을 알려주세요"라고 물으면, 챗봇은 인증 과정을 거친 후 잔액을 알려줍니다.

이때 Persistence를 활용하면 고객의 인증 상태를 유지할 수 있어, 다음 질문인 "이체하고 싶어요"에 대해 다시 인증을 요구하지 않아도 됩니다. 네이버, 카카오 같은 대형 서비스에서도 이런 패턴을 적극적으로 사용하고 있습니다.

수백만 명의 사용자가 동시에 대화하더라도, thread_id로 각각 독립적으로 관리할 수 있기 때문입니다. 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 thread_id를 설정하지 않거나, 모든 사용자에게 같은 thread_id를 사용하는 것입니다.

이렇게 하면 A 사용자의 대화 내용이 B 사용자에게 보이는 심각한 문제가 발생할 수 있습니다. 따라서 사용자마다 고유한 thread_id를 부여해야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 즉시 코드를 수정했습니다. checkpointer를 추가하고, 사용자별로 고유한 thread_id를 생성하는 로직을 추가했습니다.

다시 테스트해보니 이제 챗봇이 이전 대화 내용을 완벽하게 기억했습니다. Persistence를 제대로 이해하면 사용자 경험이 뛰어난 대화형 AI 서비스를 만들 수 있습니다.

여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - thread_id는 사용자 ID나 세션 ID를 조합하여 고유하게 생성하세요

  • 프로덕션 환경에서는 MemorySaver 대신 SqliteSaver나 PostgresSaver를 사용하세요
  • 민감한 정보는 암호화하여 저장하는 것을 권장합니다

2. Thread와 Checkpoint 이해

김개발 씨가 Persistence를 적용한 후, 박시니어 씨가 물었습니다. "그런데 Thread와 Checkpoint의 차이를 알고 있나요?" 김개발 씨는 잠시 당황했습니다.

둘 다 상태를 저장하는 것 같은데, 정확히 어떤 차이가 있는 걸까요?

Thread는 독립적인 대화 세션을 의미하며, Checkpoint는 Thread 내에서 각 단계마다 생성되는 상태 스냅샷입니다. Thread는 사용자별 대화방이고, Checkpoint는 그 대화방 안에서 발생하는 모든 상태 변화의 기록이라고 할 수 있습니다.

하나의 Thread는 여러 개의 Checkpoint를 가질 수 있으며, 이를 통해 시간 여행처럼 과거 상태로 돌아갈 수 있습니다.

다음 코드를 살펴봅시다.

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: 첫 번째 사용자의 대화
config1 = {"configurable": {"thread_id": "user_001"}}
app.invoke({"messages": "안녕하세요"}, config1)
# Checkpoint 1 생성
app.invoke({"messages": "날씨 알려주세요"}, config1)
# Checkpoint 2 생성

# Thread 2: 두 번째 사용자의 대화 (독립적)
config2 = {"configurable": {"thread_id": "user_002"}}
app.invoke({"messages": "반갑습니다"}, config2)
# Thread 2의 Checkpoint 1 생성

김개발 씨는 박시니어 씨의 질문에 대답하지 못했습니다. 그러자 박시니어 씨가 화이트보드로 가서 설명을 시작했습니다.

"Thread와 Checkpoint의 관계를 이해하는 게 정말 중요해요. 많은 개발자들이 이 부분을 헷갈려 합니다." Thread와 Checkpoint는 정확히 무엇일까요? 쉽게 비유하자면, Thread는 메신저 앱의 개별 채팅방과 같습니다.

카카오톡에서 친구 A와의 대화방, 친구 B와의 대화방이 각각 독립적으로 존재하는 것처럼, LangGraph에서도 각 사용자는 자신만의 Thread를 가집니다. 그렇다면 Checkpoint는 무엇일까요?

Checkpoint는 그 채팅방 안에서 주고받은 모든 메시지의 기록입니다. 예를 들어 오전 10시에 "안녕"이라고 보낸 것, 10시 5분에 "밥 먹었어?"라고 보낸 것, 이 모든 시점이 각각의 Checkpoint가 되는 것입니다.

왜 이런 구조가 필요할까요? 초기 챗봇 시스템에서는 단순히 "현재 상태"만 저장했습니다. 사용자가 A라고 말했다가 B라고 말하면, B 상태만 남고 A 상태는 사라졌습니다.

이것은 마치 화이트보드에 글을 쓰고 지우기를 반복하는 것과 같았습니다. 하지만 실제 서비스에서는 더 많은 것이 필요했습니다.

"사용자가 3단계 전에 뭐라고 했지?", "오류가 발생하기 직전 상태로 돌아가자", "이 사용자가 어떤 대화 흐름을 거쳤는지 분석하고 싶다" 같은 요구사항이 생겼습니다. Thread와 Checkpoint 구조가 이 모든 문제를 해결합니다. Thread를 사용하면 수백만 명의 사용자가 동시에 서비스를 이용해도 각자의 대화가 섞이지 않습니다.

user_001의 Thread와 user_002의 Thread는 완전히 독립적입니다. Checkpoint를 사용하면 각 Thread 내에서 시간 여행이 가능합니다.

특정 Checkpoint로 돌아가서 다시 시작할 수도 있고, 전체 대화 히스토리를 분석할 수도 있습니다. 마치 Git의 commit 기록처럼, 모든 변경 사항이 투명하게 기록됩니다.

위의 코드를 단계별로 살펴보겠습니다. 7번째 줄에서 첫 번째 Thread를 생성합니다. thread_id를 "user_001"로 설정했습니다.

이제 이 사용자의 모든 대화는 이 Thread에 기록됩니다. 8번째 줄에서 첫 번째 메시지를 보내면 첫 번째 Checkpoint가 자동으로 생성됩니다.

이 시점의 모든 상태가 스냅샷으로 저장됩니다. 10번째 줄에서 두 번째 메시지를 보내면 두 번째 Checkpoint가 생성됩니다.

이제 이 Thread는 2개의 Checkpoint를 가지게 됩니다. 13번째 줄이 중요합니다.

새로운 thread_id로 config를 만들면, 완전히 독립적인 새로운 Thread가 시작됩니다. user_001의 대화 내용은 전혀 영향을 받지 않습니다.

실제 현업에서는 어떻게 활용할까요? 이커머스 쇼핑 상담 챗봇을 예로 들어봅시다. 사용자 A가 "운동화를 찾고 있어요" → "나이키 제품 보여주세요" → "270 사이즈 있나요?" 순으로 질문합니다.

각 단계마다 Checkpoint가 생성됩니다. 만약 세 번째 질문 후에 사용자가 "아니다, 아디다스 제품도 보고 싶어요"라고 마음을 바꾸면, 두 번째 Checkpoint로 돌아가서 다른 브랜드로 검색을 다시 시작할 수 있습니다.

처음부터 다시 시작할 필요가 없는 것이죠. 동시에 다른 사용자 B는 자신의 Thread에서 완전히 다른 제품을 검색할 수 있습니다.

두 사용자의 대화는 절대 섞이지 않습니다. 주의해야 할 함정이 있습니다. 많은 개발자가 실수하는 부분은 thread_id를 너무 자주 바꾸는 것입니다.

같은 사용자의 대화인데도 매번 새로운 thread_id를 생성하면, 대화 맥락이 끊어집니다. thread_id는 사용자 세션 동안 일관되게 유지해야 합니다.

또 다른 실수는 Checkpoint를 수동으로 관리하려는 것입니다. LangGraph는 자동으로 Checkpoint를 생성하므로, 개발자가 직접 관리할 필요가 없습니다.

그래프의 각 노드가 실행될 때마다 자동으로 Checkpoint가 만들어집니다. 다시 화이트보드 앞으로 돌아가봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 이제 완전히 이해했습니다.

"아, Thread는 대화방이고, Checkpoint는 대화 기록이군요! 그래서 한 Thread 안에 여러 Checkpoint가 있을 수 있는 거네요." "정확해요!" 박시니어 씨가 웃으며 답했습니다.

Thread와 Checkpoint의 관계를 제대로 이해하면, 복잡한 대화 흐름도 쉽게 관리할 수 있습니다. 여러분의 프로젝트에서도 이 개념을 활용해 보세요.

실전 팁

💡 - thread_id는 사용자 세션 동안 일관되게 유지하세요

  • Checkpoint는 자동 생성되므로 수동 관리하지 마세요
  • Thread별로 대화 히스토리를 분석하면 사용자 패턴을 파악할 수 있습니다

3. StateSnapshot 객체 구조

김개발 씨가 Checkpoint를 조회하려고 하는데, 반환되는 객체가 복잡해 보였습니다. "이 StateSnapshot이라는 객체는 정확히 어떤 정보를 담고 있나요?" 박시니어 씨에게 다시 질문했습니다.

"StateSnapshot은 Checkpoint의 모든 정보를 담고 있는 컨테이너예요. 하나씩 살펴봅시다."

StateSnapshot은 특정 시점의 그래프 상태를 완전히 담고 있는 객체입니다. 단순히 데이터만 있는 것이 아니라, 어떤 노드가 다음에 실행될지, 현재 설정은 무엇인지, Checkpoint ID는 무엇인지 등 메타데이터도 함께 포함합니다.

이 객체를 이해하면 현재 대화가 어떤 상태인지, 다음에 무엇이 실행될지 완벽하게 파악할 수 있습니다.

다음 코드를 살펴봅시다.

# StateSnapshot 조회
config = {"configurable": {"thread_id": "user_123"}}
snapshot = app.get_state(config)

# StateSnapshot 객체의 주요 속성들
print(f"현재 상태 값: {snapshot.values}")
# 현재 저장된 모든 상태 데이터

print(f"다음 실행 노드: {snapshot.next}")
# 다음에 실행될 노드 목록

print(f"Checkpoint ID: {snapshot.config['configurable']['checkpoint_id']}")
# 이 스냅샷의 고유 식별자

print(f"메타데이터: {snapshot.metadata}")
# 추가 정보 (생성 시간, 단계 번호 등)

김개발 씨는 코드를 실행해보고 놀랐습니다. StateSnapshot 객체에서 출력되는 정보가 생각보다 훨씬 많았습니다.

values, next, config, metadata... 이게 다 무슨 의미일까요?

박시니어 씨가 코드를 함께 보며 설명하기 시작했습니다. "StateSnapshot은 마치 사진과 같아요.

특정 순간의 모든 것을 담아두는 거죠." StateSnapshot이란 정확히 무엇일까요? 비유하자면, StateSnapshot은 게임을 저장할 때 만들어지는 세이브 파일과 같습니다. 단순히 "현재 레벨"만 저장하는 게 아니라, 캐릭터 위치, 인벤토리 아이템, 진행 상황, 게임 설정 등 모든 정보를 담습니다.

나중에 이 세이브 파일을 불러오면 정확히 그 시점으로 돌아갈 수 있는 것이죠. LangGraph의 StateSnapshot도 마찬가지입니다.

특정 Checkpoint 시점의 모든 정보를 객체로 담아둡니다. StateSnapshot은 어떤 정보를 포함할까요? 첫 번째로 values가 있습니다.

이것은 실제 상태 데이터입니다. 예를 들어 챗봇이라면 "지금까지 나눈 대화 내용", "사용자가 선택한 옵션", "현재 진행 단계" 같은 비즈니스 데이터가 여기에 담깁니다.

두 번째로 next가 있습니다. 이것은 다음에 실행될 노드의 목록입니다.

만약 next가 빈 배열이면 그래프 실행이 완료된 상태를 의미합니다. 만약 값이 있다면, 아직 실행 중이거나 대기 중인 상태입니다.

세 번째로 config가 있습니다. 여기에는 thread_id와 checkpoint_id가 포함됩니다.

checkpoint_id는 이 스냅샷의 고유 식별자로, 마치 Git의 commit hash와 같은 역할을 합니다. 네 번째로 metadata가 있습니다.

여기에는 Checkpoint가 생성된 시간, 몇 번째 단계인지, 어떤 노드에서 생성되었는지 같은 메타 정보가 담깁니다. 왜 이렇게 많은 정보가 필요할까요? 초기 상태 관리 시스템에서는 단순히 "현재 데이터"만 저장했습니다.

하지만 실제 서비스를 운영하다 보니 문제가 생겼습니다. "이 사용자가 지금 어느 단계에 있지?", "다음에 뭘 실행해야 하지?", "언제 이 상태가 만들어졌지?" 같은 질문에 답할 수 없었습니다.

디버깅할 때도 어려웠고, 오류가 발생했을 때 원인을 찾기 힘들었습니다. StateSnapshot은 이 모든 문제를 해결합니다. values로 비즈니스 데이터를 확인하고, next로 다음 실행 흐름을 예측하고, checkpoint_id로 정확한 시점을 특정하고, metadata로 디버깅 정보를 얻을 수 있습니다.

완벽한 투명성을 제공하는 것이죠. 위의 코드를 자세히 분석해봅시다. 6번째 줄에서 snapshot.values를 출력합니다.

이것이 가장 중요한 데이터입니다. 예를 들어 챗봇 상태라면 {"messages": [...], "user_info": {...}} 같은 구조를 가질 수 있습니다.

9번째 줄의 snapshot.next는 배열로 반환됩니다. 예를 들어 ["agent_node"]라면 다음에 agent_node가 실행될 것을 의미합니다.

만약 []라면 그래프가 종료된 상태입니다. 12번째 줄에서 checkpoint_id를 추출합니다.

이것은 자동 생성되는 UUID 같은 고유 값입니다. 이 ID로 정확히 이 시점의 Checkpoint를 다시 찾을 수 있습니다.

15번째 줄의 metadata에는 추가 정보가 담깁니다. step, source, writes 같은 필드가 있어서, 이 Checkpoint가 어떻게 생성되었는지 추적할 수 있습니다.

실무에서는 어떻게 활용할까요? 고객 상담 챗봇에서 StateSnapshot을 활용한 예시를 들어봅시다. 고객이 복잡한 문의를 진행하던 중 네트워크 오류로 연결이 끊어졌습니다.

일반 시스템이라면 처음부터 다시 시작해야 하지만, StateSnapshot을 활용하면 마지막 Checkpoint를 불러와서 정확히 그 지점부터 이어갈 수 있습니다. snapshot.next를 확인하여 "아, 다음은 결제 확인 단계구나"를 알 수 있고, snapshot.values로 "고객이 선택한 상품이 뭐였지?"를 확인할 수 있습니다.

네이버 클로바나 카카오 AI 같은 서비스에서도 비슷한 방식으로 대화 상태를 관리합니다. 주의해야 할 점이 있습니다. 초보 개발자들이 자주 하는 실수는 StateSnapshot 객체를 직접 수정하려는 것입니다.

snapshot.values를 직접 바꾸고 저장하려고 하는데, 이것은 읽기 전용입니다. 상태를 수정하려면 update_state 메서드를 사용해야 합니다.

또 다른 실수는 metadata를 무시하는 것입니다. metadata에는 디버깅에 매우 유용한 정보가 많습니다.

문제가 발생했을 때 metadata를 확인하는 습관을 들이면 원인을 빠르게 찾을 수 있습니다. 김개발 씨가 고개를 끄덕였습니다. "이제 이해했어요!

StateSnapshot은 단순한 데이터 덩어리가 아니라, 그 시점의 전체 컨텍스트를 담고 있는 거네요." "맞아요!" 박시니어 씨가 답했습니다. "이걸 잘 활용하면 복잡한 대화 흐름도 완벽하게 제어할 수 있어요." StateSnapshot의 구조를 이해하면 상태 관리가 훨씬 쉬워집니다.

여러분도 실제 프로젝트에서 각 속성을 출력해보며 익혀보세요.

실전 팁

💡 - StateSnapshot은 읽기 전용이므로 직접 수정하지 마세요

  • metadata를 활용하여 디버깅 정보를 수집하세요
  • next 속성으로 그래프 실행 완료 여부를 판단할 수 있습니다

4. get state() 상태 조회

김개발 씨가 실시간으로 사용자의 대화 상태를 확인하고 싶었습니다. "현재 이 사용자가 어떤 단계에 있는지 어떻게 알 수 있나요?" 박시니어 씨가 답했습니다.

"get_state 메서드를 사용하면 됩니다. 가장 최근 Checkpoint를 즉시 조회할 수 있어요."

get_state는 특정 Thread의 가장 최근 상태를 조회하는 메서드입니다. thread_id를 전달하면 해당 Thread의 마지막 Checkpoint를 StateSnapshot 객체로 반환합니다.

실시간으로 사용자가 어떤 상태인지, 다음에 뭘 해야 하는지 즉시 확인할 수 있어, 모니터링이나 디버깅에 매우 유용합니다.

다음 코드를 살펴봅시다.

from langgraph.graph import StateGraph

app = graph.compile(checkpointer=memory)

# 특정 Thread의 현재 상태 조회
config = {"configurable": {"thread_id": "user_123"}}
current_state = app.get_state(config)

# 상태 정보 확인
if current_state.next:
    print(f"진행 중: 다음 실행 노드는 {current_state.next}")
else:
    print("완료: 그래프 실행이 종료되었습니다")

# 현재 대화 메시지 확인
messages = current_state.values.get("messages", [])
print(f"총 {len(messages)}개의 메시지가 있습니다")

김개발 씨는 운영 중인 챗봇 서비스에서 문제를 발견했습니다. 일부 사용자가 "챗봇이 응답을 멈췄어요"라고 신고했습니다.

어떻게 해야 할까요? 박시니어 씨가 말했습니다.

"이럴 때 get_state가 정말 유용해요. 사용자의 현재 상태를 실시간으로 확인할 수 있거든요." get_state란 무엇일까요? 쉽게 비유하자면, get_state는 택배 배송 조회와 같습니다.

운송장 번호(thread_id)를 입력하면, 현재 택배가 어디에 있는지, 다음 목적지는 어디인지 즉시 알 수 있습니다. LangGraph에서도 마찬가지입니다.

thread_id를 전달하면 그 Thread의 현재 상태를 즉시 조회할 수 있습니다. 복잡한 쿼리나 데이터베이스 검색 없이, 단 한 줄의 코드로 모든 정보를 얻을 수 있습니다.

왜 get_state가 필요할까요? 과거 챗봇 시스템에서는 현재 상태를 확인하는 것이 쉽지 않았습니다. 데이터베이스를 직접 쿼리하거나, 로그를 뒤져야 했습니다.

실시간 모니터링은 거의 불가능했고, 문제가 생기면 원인을 찾는 데 많은 시간이 걸렸습니다. 특히 "사용자가 지금 어느 단계에 있는지", "왜 다음 단계로 진행하지 않는지" 같은 질문에 답하기가 매우 어려웠습니다.

개발자는 추측만 할 뿐, 정확한 상태를 알 수 없었습니다. get_state가 이 모든 문제를 해결합니다. thread_id만 알면 언제든지 현재 상태를 조회할 수 있습니다.

사용자가 "챗봇이 멈췄어요"라고 신고하면, 즉시 get_state로 확인할 수 있습니다. "아, next가 비어있네요.

그래프가 정상적으로 종료된 상태입니다" 또는 "next에 특정 노드가 남아있네요. 아직 실행 중인데 응답이 늦어지고 있습니다" 같은 정확한 진단이 가능합니다.

운영 대시보드를 만들 때도 유용합니다. 모든 활성 Thread의 상태를 주기적으로 조회하여, 어떤 사용자가 어느 단계에 있는지 실시간으로 시각화할 수 있습니다.

위의 코드를 단계별로 살펴봅시다. 6번째 줄에서 config 객체를 만듭니다. thread_id를 "user_123"으로 설정했습니다.

이 사용자의 상태를 조회하겠다는 의미입니다. 7번째 줄이 핵심입니다.

**app.get_state(config)**를 호출하면, 해당 Thread의 가장 최근 Checkpoint가 StateSnapshot 객체로 반환됩니다. 단 한 줄로 모든 정보를 얻는 것이죠.

10번째 줄에서 current_state.next를 확인합니다. 이것이 비어있으면 그래프 실행이 완료된 상태입니다.

값이 있으면 아직 실행 중이거나 대기 중입니다. 이 정보만으로도 많은 것을 알 수 있습니다.

16번째 줄에서 실제 비즈니스 데이터를 꺼냅니다. current_state.values에서 원하는 필드를 가져올 수 있습니다.

예를 들어 messages를 가져와서 대화 기록을 확인할 수 있습니다. 실무에서 어떻게 활용할까요? 고객 상담 관리 시스템을 예로 들어봅시다.

상담원이 고객과 대화하는 도중, "이 고객이 이전에 무슨 문의를 했었나요?"를 확인하고 싶습니다. get_state를 호출하여 current_state.values를 확인하면, 이전 대화 내용, 고객이 선택한 옵션, 현재 진행 단계 등을 모두 볼 수 있습니다.

상담원은 이 정보를 바탕으로 더 나은 서비스를 제공할 수 있습니다. 또 다른 예시로, 자동화 테스트에서도 활용할 수 있습니다.

테스트 코드에서 특정 입력을 주고, get_state로 결과를 확인하여 예상한 상태인지 검증할 수 있습니다. 주의해야 할 함정이 있습니다. 많은 개발자가 하는 실수는 존재하지 않는 thread_id로 get_state를 호출하는 것입니다.

이 경우 에러가 발생하거나, 빈 상태가 반환될 수 있습니다. 따라서 get_state를 호출하기 전에 해당 Thread가 존재하는지 확인하는 것이 좋습니다.

또 다른 실수는 get_state를 너무 자주 호출하는 것입니다. 매 요청마다 상태를 조회하면 성능에 영향을 줄 수 있습니다.

필요할 때만 호출하고, 가능하면 결과를 캐싱하세요. 김개발 씨가 직접 코드를 작성해봤습니다. 문제가 신고된 사용자의 thread_id로 get_state를 호출했습니다.

결과를 보니 next가 비어있었습니다. "아, 챗봇이 멈춘 게 아니라 정상적으로 대화가 끝난 거였네요!" 박시니어 씨가 웃으며 말했습니다.

"그렇죠. get_state로 확인하니 금방 알 수 있었죠?

추측하지 말고 확인하는 게 중요해요." get_state를 마스터하면 시스템을 투명하게 관리할 수 있습니다. 여러분도 자주 사용하며 익숙해지세요.

실전 팁

💡 - thread_id가 존재하는지 먼저 확인하세요

  • 운영 모니터링 대시보드에 get_state를 활용하세요
  • 성능을 위해 필요할 때만 호출하고 결과를 캐싱하세요

5. get state history() 이력 조회

김개발 씨가 사용자 행동 패턴을 분석하려고 했습니다. "이 사용자가 어떤 경로로 대화를 진행했는지 전체 히스토리를 볼 수 있나요?" 박시니어 씨가 답했습니다.

"get_state_history를 사용하면 모든 Checkpoint 이력을 시간순으로 조회할 수 있어요."

get_state_history는 특정 Thread의 모든 Checkpoint 기록을 시간 역순으로 반환하는 메서드입니다. 마치 Git log처럼, 첫 Checkpoint부터 현재까지의 모든 상태 변화를 추적할 수 있습니다.

이를 통해 사용자의 전체 대화 흐름을 분석하거나, 특정 시점으로 되돌리거나, 디버깅을 수행할 수 있습니다.

다음 코드를 살펴봅시다.

# Thread의 전체 히스토리 조회
config = {"configurable": {"thread_id": "user_123"}}
history = app.get_state_history(config)

# 모든 Checkpoint를 순회하며 분석
print("=== 대화 히스토리 ===")
for i, state in enumerate(history):
    checkpoint_id = state.config["configurable"]["checkpoint_id"]
    step = state.metadata.get("step", 0)

    print(f"\n[단계 {step}] Checkpoint: {checkpoint_id[:8]}...")
    print(f"다음 노드: {state.next}")
    print(f"메시지 수: {len(state.values.get('messages', []))}")

    # 최근 3개만 출력
    if i >= 2:
        break

김개발 씨는 서비스 개선을 위해 사용자 행동 데이터를 분석하고 싶었습니다. 하지만 현재 상태만 보면 사용자가 어떤 경로로 여기까지 왔는지 알 수 없었습니다.

박시니어 씨가 말했습니다. "get_state는 현재 사진 한 장만 보여주지만, get_state_history는 전체 동영상을 보여줘요." get_state_history란 무엇일까요? 비유하자면, get_state_history는 은행 계좌의 거래 내역과 같습니다.

현재 잔액만 보는 게 아니라, 언제 입금했고 언제 출금했는지 모든 기록을 시간순으로 볼 수 있습니다. LangGraph에서도 마찬가지입니다.

특정 Thread의 첫 Checkpoint부터 현재까지, 모든 상태 변화를 추적할 수 있습니다. 각 단계에서 어떤 노드가 실행되었는지, 상태가 어떻게 변했는지 완벽하게 재구성할 수 있습니다.

왜 이런 기능이 필요할까요? 초기 챗봇 시스템에서는 현재 상태만 저장했기 때문에, 과거 기록은 사라졌습니다. "이 사용자가 왜 이 상태에 도달했지?", "어느 지점에서 문제가 시작됐지?" 같은 질문에 답할 수 없었습니다.

디버깅할 때 특히 큰 문제였습니다. 사용자가 "3단계 전에 이상한 응답을 받았어요"라고 신고해도, 개발자는 그 시점의 상태를 확인할 방법이 없었습니다.

로그를 뒤지고, 추측하고, 재현을 시도해야 했습니다. 데이터 분석도 어려웠습니다.

"사용자들이 보통 어떤 경로로 대화를 진행하지?", "어느 단계에서 이탈률이 높지?" 같은 인사이트를 얻을 수 없었습니다. get_state_history가 이 모든 문제를 해결합니다. 모든 Checkpoint를 시간순으로 조회할 수 있어, 사용자의 전체 여정을 추적할 수 있습니다.

특정 시점으로 돌아가서 "그때 상태가 어땠는지" 정확히 확인할 수 있습니다. 데이터 분석도 가능합니다.

수백, 수천 명의 사용자 히스토리를 분석하여 패턴을 찾고, 서비스를 개선할 수 있습니다. A/B 테스트를 할 때도 각 그룹의 대화 흐름을 비교할 수 있습니다.

위의 코드를 자세히 분석해봅시다. 3번째 줄에서 **app.get_state_history(config)**를 호출합니다. 이것은 제너레이터를 반환합니다.

즉, 한 번에 모든 데이터를 메모리에 올리지 않고, 필요할 때마다 하나씩 가져올 수 있습니다. 7번째 줄에서 for 루프로 각 Checkpoint를 순회합니다.

가장 최근 Checkpoint부터 과거 순으로 반환됩니다. 즉, history의 첫 번째 항목이 get_state와 동일한 최신 상태입니다.

8번째 줄에서 checkpoint_id를 추출합니다. 각 Checkpoint는 고유한 ID를 가지므로, 나중에 특정 시점으로 되돌릴 때 사용할 수 있습니다.

9번째 줄에서 step 정보를 가져옵니다. metadata에는 이 Checkpoint가 몇 번째 단계인지 기록되어 있습니다.

이를 통해 "5단계에서 문제가 발생했네요" 같은 분석이 가능합니다. 16번째 줄에서 최근 3개만 출력하고 break합니다.

모든 히스토리를 출력하면 너무 길 수 있으므로, 필요한 만큼만 조회하는 것이 좋습니다. 실무에서는 어떻게 활용할까요? 이커머스 쇼핑 상담봇에서 get_state_history를 활용한 예시입니다.

사용자가 "신발 → 운동화 → 나이키 → 사이즈 270 → 색상 검정" 순으로 대화를 진행했다가, 마지막에 "처음부터 다시 시작할래요"라고 했습니다. get_state_history로 전체 히스토리를 조회하여, 사용자가 어떤 고민을 했는지 파악할 수 있습니다.

"아, 이 고객은 나이키를 고려했다가 마음을 바꿨네요. 다른 브랜드도 추천해볼까요?" 또 다른 활용 사례는 품질 관리입니다.

무작위로 대화 샘플을 뽑아서 get_state_history로 전체 흐름을 검토합니다. "이 단계에서 챗봇 응답이 부자연스럽네요", "여기서 사용자가 혼란스러워했을 것 같아요" 같은 개선점을 찾을 수 있습니다.

주의해야 할 점이 있습니다. 많은 개발자가 하는 실수는 get_state_history를 list()로 한 번에 변환하는 것입니다. Thread가 길면 수백, 수천 개의 Checkpoint가 있을 수 있어, 메모리 문제가 발생할 수 있습니다.

제너레이터를 그대로 사용하여 필요한 만큼만 순회하세요. 또 다른 실수는 모든 요청에서 전체 히스토리를 조회하는 것입니다.

이것은 성능에 큰 영향을 줍니다. 히스토리 조회는 분석, 디버깅, 모니터링처럼 필요한 경우에만 사용하세요.

김개발 씨가 실제로 코드를 실행해봤습니다. 특정 사용자의 히스토리를 조회했더니, 총 15개의 Checkpoint가 있었습니다. 각 단계를 살펴보니 사용자의 대화 흐름이 한눈에 보였습니다.

"이 사용자는 여기서 질문을 바꿨네요. 왜 그랬을까요?" 박시니어 씨가 답했습니다.

"바로 그거예요. 히스토리를 보면 사용자의 의도를 더 잘 이해할 수 있어요.

데이터는 거짓말하지 않으니까요." get_state_history를 활용하면 사용자 경험을 깊이 이해할 수 있습니다. 여러분도 실제 서비스 데이터를 분석해보세요.

실전 팁

💡 - 제너레이터를 그대로 사용하여 메모리를 절약하세요

  • 분석과 디버깅 목적으로만 사용하고, 매 요청마다 호출하지 마세요
  • metadata의 step 정보로 문제 발생 지점을 빠르게 찾을 수 있습니다

6. update state() 상태 수정

김개발 씨가 마지막 질문을 했습니다. "사용자의 상태를 중간에 수정하고 싶으면 어떻게 하나요?

예를 들어 관리자가 대화 내용을 수정하거나, 특정 값을 초기화하고 싶을 때요." 박시니어 씨가 답했습니다. "update_state를 사용하면 됩니다.

그래프를 다시 실행하지 않고도 상태를 직접 수정할 수 있어요."

update_state는 특정 Thread의 상태를 프로그래밍 방식으로 수정하는 메서드입니다. 그래프를 실행하지 않고도 원하는 필드를 변경하거나, 특정 Checkpoint로부터 새로운 분기를 만들 수 있습니다.

관리자 개입, 오류 수정, 테스트 시나리오 설정 등 다양한 상황에서 유연하게 활용할 수 있는 강력한 기능입니다.

다음 코드를 살펴봅시다.

# 현재 상태 조회
config = {"configurable": {"thread_id": "user_123"}}
current = app.get_state(config)
print(f"수정 전 메시지: {len(current.values.get('messages', []))}")

# 상태 수정: 새로운 메시지 추가
app.update_state(
    config,
    {"messages": [{"role": "assistant", "content": "관리자가 추가한 메시지"}]},
    as_node="agent"  # 어느 노드에서 업데이트되었는지 명시
)

# 수정 후 상태 확인
updated = app.get_state(config)
print(f"수정 후 메시지: {len(updated.values.get('messages', []))}")

# 특정 Checkpoint로부터 분기 생성
history = app.get_state_history(config)
third_checkpoint = list(history)[2]
app.update_state(third_checkpoint.config, {"messages": []})

김개발 씨는 운영 중에 흥미로운 요구사항을 받았습니다. 고객 지원팀에서 "사용자가 잘못된 정보를 입력했는데, 처음부터 다시 시작하지 않고 그 부분만 수정할 수 있나요?"라고 물었습니다.

박시니어 씨가 말했습니다. "이럴 때 update_state가 정말 유용해요.

마치 문서 편집기의 '실행 취소'와 '직접 수정'을 합쳐놓은 것 같아요." update_state란 정확히 무엇일까요? 비유하자면, update_state는 포토샵의 레이어 편집과 같습니다. 전체 이미지를 다시 그리지 않고, 특정 레이어만 수정할 수 있습니다.

원하는 부분만 바꾸고, 나머지는 그대로 유지할 수 있는 것이죠. LangGraph에서도 마찬가지입니다.

그래프를 처음부터 다시 실행하지 않고, 현재 상태의 특정 필드만 수정할 수 있습니다. 또는 과거 Checkpoint로 돌아가서 거기서부터 다른 경로로 진행할 수도 있습니다.

왜 update_state가 필요할까요? 일반적인 그래프 실행은 순차적입니다. 입력을 받아서 노드를 실행하고, 결과를 받습니다.

하지만 실제 서비스 운영에서는 예외 상황이 많습니다. 사용자가 잘못된 정보를 입력했을 때, 관리자가 개입해야 할 때, 테스트를 위해 특정 상태를 설정해야 할 때, 오류를 수정해야 할 때...

이런 상황에서 매번 처음부터 다시 시작하는 것은 비효율적입니다. 특히 복잡한 대화 흐름에서는 특정 단계까지 가는 데 많은 시간이 걸릴 수 있습니다.

"10단계까지 진행한 후 오류를 발견했는데, 또 처음부터 해야 해?" 이것은 개발자에게도 사용자에게도 좋지 않습니다. update_state가 이 문제를 우아하게 해결합니다. 원하는 시점에서 원하는 필드만 수정할 수 있습니다.

관리자가 대화 내용을 수정할 수도 있고, 오류가 발생한 부분만 정정할 수도 있습니다. 테스트 환경에서 특정 상태를 빠르게 설정할 수도 있습니다.

더 강력한 기능은 시간 분기입니다. 과거 Checkpoint로 돌아가서 거기서부터 다른 선택을 할 수 있습니다.

마치 게임에서 세이브 포인트로 돌아가서 다른 선택을 하는 것처럼요. 위의 코드를 단계별로 분석해봅시다. 3번째 줄에서 먼저 현재 상태를 조회합니다.

수정 전 상태를 확인하는 것이 좋은 습관입니다. 7번째 줄이 핵심입니다.

app.update_state를 호출하여 상태를 수정합니다. 첫 번째 인자는 config(thread_id), 두 번째 인자는 수정할 데이터입니다.

10번째 줄의 as_node 매개변수가 중요합니다. 이것은 "어느 노드에서 이 업데이트가 발생했는지"를 명시합니다.

metadata에 기록되어, 나중에 히스토리를 볼 때 "관리자가 수정했구나"를 알 수 있습니다. 14번째 줄에서 수정 후 상태를 다시 조회하여 변경사항을 확인합니다.

항상 수정이 제대로 적용되었는지 검증하세요. 18-20번째 줄에서는 더 고급 기능을 보여줍니다.

get_state_history로 과거 Checkpoint를 가져와서, 그 시점의 config를 사용하여 update_state를 호출합니다. 이렇게 하면 그 Checkpoint로부터 새로운 분기가 생성됩니다.

실무에서는 어떻게 활용할까요? 고객 상담 챗봇에서 실제 일어난 사례입니다. 고객이 전화번호를 "01012345678"로 입력해야 하는데, 실수로 "010-1234-5678"로 입력했습니다.

시스템이 형식 오류로 거부했고, 고객은 처음부터 다시 시작해야 했습니다. update_state를 활용하면 이 문제를 해결할 수 있습니다.

관리자 대시보드에서 해당 사용자의 상태를 조회하고, 전화번호 필드만 올바른 형식으로 수정합니다. 고객은 다시 시작할 필요 없이 다음 단계로 바로 진행할 수 있습니다.

또 다른 활용 사례는 A/B 테스트입니다. 동일한 Checkpoint에서 시작하여 서로 다른 값으로 update_state를 호출하면, 동일한 출발점에서 다른 시나리오를 테스트할 수 있습니다.

주의해야 할 중요한 점이 있습니다. update_state는 강력한 만큼 위험할 수 있습니다. 잘못 사용하면 데이터 무결성이 깨질 수 있습니다.

예를 들어 "결제 완료" 상태인데 "주문 금액"을 0으로 바꾸면, 논리적으로 모순된 상태가 됩니다. 따라서 update_state를 사용할 때는 반드시 검증 로직을 추가하세요.

"이 수정이 논리적으로 타당한가?", "다른 필드와 일관성이 있는가?"를 확인해야 합니다. 또한 as_node를 명시하여 누가, 언제, 왜 수정했는지 기록을 남기는 것이 중요합니다.

나중에 문제가 생겼을 때 추적할 수 있어야 합니다. 김개발 씨가 직접 테스트해봤습니다. 테스트 환경에서 특정 상태를 설정하고, update_state로 다양한 시나리오를 빠르게 실험할 수 있었습니다.

"이제 테스트 속도가 훨씬 빨라졌어요!" 박시니어 씨가 웃으며 답했습니다. "맞아요.

하지만 프로덕션에서는 신중하게 사용하세요. 권한 관리도 철저히 해야 해요." update_state를 마스터하면 시스템을 훨씬 유연하게 관리할 수 있습니다.

하지만 강력한 만큼 신중하게 사용하세요.

실전 팁

💡 - update_state 전후로 항상 get_state로 검증하세요

  • as_node를 명시하여 수정 이력을 기록하세요
  • 프로덕션에서는 권한 관리와 검증 로직을 반드시 추가하세요

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#LangGraph#Persistence#Checkpoint#StateManagement#Thread#AI,LLM,Python,LangGraph

댓글 (0)

댓글을 작성하려면 로그인이 필요합니다.