LangGraph

제어 가능한 AI 에이전트(Agent)를 구축하기 위한 로우 레벨(low-level) 오케스트레이션 프레임워크입니다. 개발자가 강력하고 적응성 높은 AI 에이전트를 구축하도록 지원하는 것을 목표로 합니다.

에이전트 오케스트레이션에 특화되어 사용자 정의 아키텍처, 장기 기억(long-term memory), 인간 개입(human-in-the-loop) 등을 통해 복잡한 작업을 안정적으로 처리할 수 있도록 지원합니다.

저스틴) 장기 기억할 것은 검색 엔진에서 얻은 데이터와 사용자가 제공한 데이터여야 합니다. LLM이 갖고 있는 데이터는 필요 없고, 검색 엔진에서 얻은 것은 다시 검색 엔진에서 얻으면 되니, 결국 장기 기억할 것은 사용자 데이터만이 됩니다. 사용자 데이터도 LLM 이 보유한 데이터에서 가능한 것은 장기 기억할 필요 없습니다.

LangGraph 퀵스타트

이 튜토리얼은 LangGraph를 사용하여 지원 챗봇(support chatbot)을 구축하는 과정을 단계별로 안내합니다. 목표는 기본적인 챗봇부터 시작하여 점진적으로 다음과 같은 고급 기능을 추가하는 것입니다.

웹 검색을 통한 일반적인 질문 답변
대화 상태 유지 (메모리 기능) 저스틴) 챗 이력 수준이 아니다.
복잡한 질문에 대한 인간 검토 요청 (Human-in-the-loop) 저스틴) 사람은 대화 방식보다 UI 방식을 편해할 수 있습니다.
- 실행 중단 및 재개 메커니즘은 상호작용적이고 신뢰성 있는 에이전트 구축에 중요합니다.
챗봇 행동 제어를 위한 사용자 정의 상태(Custom State) 사용
- 기본 메시지 목록 외에 애플리케이션별 상태를 정의하고 관리하는 방법은 복잡한 로직 구현에 필요합니다.
대화 경로 되감기 및 탐색 (시간 여행, Time Travel)
- 체크포인트 기록을 사용하여 그래프의 이전 상태로 돌아가 실행을 재개하는 방법. 단순한 대화 기록 저장을 넘어, LangGraph의 상태 지속성 및 메모리 관리의 핵심 원리입니다.
- 체크포인팅을 활용한 고급 기능으로, 디버깅, 실험, 사용자 경험 향상에 유용합니다.

LangGraph를 사용하여 지원 챗봇(support chatbot)을 만드는 방법을 배웁니다. 이 튜토리얼을 통해 챗봇에 다음과 같은 기능을 단계별로 추가하게 됩니다.

✅ 웹 검색: 챗봇이 모르는 질문에 답하기 위해 웹을 검색하는 기능.
✅ 대화 상태 유지: 이전 대화 내용을 기억하여 자연스러운 대화를 이어가는 기능 (메모리).
✅ 인간 검토: 챗봇이 스스로 처리하기 어렵거나 중요한 결정이 필요한 경우, 사람에게 검토를 요청하는 기능 (Human-in-the-loop).
✅ 사용자 정의 상태: 챗봇의 동작을 세밀하게 제어하기 위해 기본 대화 기록 외에 추가적인 정보를 상태에 저장하고 사용하는 기능.
✅ 시간 여행: 대화의 특정 시점으로 돌아가서 다른 경로를 탐색하거나 수정하는 기능.

Setup (설정)

이 섹션은 튜토리얼 코드를 실행하기 전에 필요한 사전 준비 작업을 안내합니다.

%%capture --no-stderr
# 터미널 출력을 최소화하여 노트북을 깔끔하게 유지 (Jupyter Notebook 명령어)

%pip install -U langgraph langsmith langchain_anthropic
# 필요한 파이썬 패키지를 설치하거나 최신 버전으로 업그레이드합니다.
# - langgraph: 핵심 LangGraph 라이브러리
# - langsmith: (선택 사항이지만 권장됨) LangGraph 실행 과정을 추적하고 디버깅하는 데 사용되는 LangSmith 플랫폼 연동 라이브러리
# - langchain_anthropic: 튜토리얼에서 사용할 언어 모델(Anthropic의 Claude)을 사용하기 위한 라이브러리

import getpass
import os
# 운영체제 기능(환경 변수 접근)과 비밀번호를 안전하게 입력받기 위한 라이브러리를 가져옵니다.

def _set_env(var: str):
    # 환경 변수를 설정하는 도우미 함수 정의
    if not os.environ.get(var):
        # 해당 이름(var)의 환경 변수가 이미 설정되어 있지 않으면,
        os.environ[var] = getpass.getpass(f"{var}: ")
        # 사용자에게 해당 변수 이름(예: ANTHROPIC_API_KEY:)을 보여주고,
        # 비밀번호처럼 입력값이 보이지 않게 안전하게 입력받아 환경 변수로 설정합니다.

_set_env("ANTHROPIC_API_KEY")
# 위에서 정의한 함수를 호출하여 'ANTHROPIC_API_KEY' 환경 변수를 설정합니다.
# 이 키는 Anthropic의 언어 모델(Claude)을 사용하기 위해 필요합니다.
# 코드를 실행하면 이 부분에서 API 키를 입력하라는 프롬프트가 나타납니다.

%%capture --no-stderr

# 터미널 출력을 최소화하여 노트북을 깔끔하게 유지 (Jupyter Notebook 명령어)

%pip install -U langgraph langsmith langchain_anthropic

# 필요한 파이썬 패키지를 설치하거나 최신 버전으로 업그레이드합니다.

# - langgraph: 핵심 LangGraph 라이브러리

# - langsmith: (선택 사항이지만 권장됨) LangGraph 실행 과정을 추적하고 디버깅하는 데 사용되는 LangSmith 플랫폼 연동 라이브러리

# - langchain_anthropic: 튜토리얼에서 사용할 언어 모델(Anthropic의 Claude)을 사용하기 위한 라이브러리

import getpass

import os

# 운영체제 기능(환경 변수 접근)과 비밀번호를 안전하게 입력받기 위한 라이브러리를 가져옵니다.

def _set_env(var: str):

# 환경 변수를 설정하는 도우미 함수 정의

if not os.environ.get(var):

# 해당 이름(var)의 환경 변수가 이미 설정되어 있지 않으면,

os.environ[var] = getpass.getpass(f"{var}: ")

# 사용자에게 해당 변수 이름(예: ANTHROPIC_API_KEY:)을 보여주고,

# 비밀번호처럼 입력값이 보이지 않게 안전하게 입력받아 환경 변수로 설정합니다.

_set_env("ANTHROPIC_API_KEY")

# 위에서 정의한 함수를 호출하여 'ANTHROPIC_API_KEY' 환경 변수를 설정합니다.

# 이 키는 Anthropic의 언어 모델(Claude)을 사용하기 위해 필요합니다.

# 코드를 실행하면 이 부분에서 API 키를 입력하라는 프롬프트가 나타납니다.

LangSmith 설정 안내:

LangSmith 가입 권장: 튜토리얼 중간중간 LangSmith 링크가 제공되는데, LangSmith에 가입하면 LangGraph 애플리케이션의 실행 과정(Trace)을 시각적으로 확인하고, 문제가 발생했을 때 원인을 찾거나(디버깅), 성능을 테스트하고 개선하는 데 매우 유용합니다. LangSmith는 LangGraph 개발 경험을 향상시키는 데 도움을 줍니다.

Part 1: Build a Basic Chatbot (기본 챗봇 구축)

“ 섹션을 설명드리겠습니다. 이 섹션에서는 LangGraph의 가장 기본적인 구성 요소들을 사용하여 간단한 챗봇을 만듭니다.

목표:

LangGraph를 이용해 간단한 챗봇 구조를 정의하고 실행하는 방법을 배웁니다.
LangGraph의 핵심 개념인 StateGraph, State, Node, Edge, START, END를 이해합니다. 저스틴) 워크플로어이니 UML의 액티비티 다이어그램이나 플로어 차트를 생각하면 됩니다. 상태라는 단어가 나오듯이 상태를 중심으로 하는 UML 상태머신을 생각하면 더 도움이 됩니다. 그런데 이런 것들은 절차가 정해진 것들인데 절차보다 단일 단계를 실행하는 ReAct와 어떻게 연결할 지 궁급해집니다.

코드 설명:

State 정의 (State 클래스):

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    # 'messages'라는 키를 가진 상태를 정의합니다. 이 키의 값은 리스트(list) 타입입니다.
    messages: Annotated[list, add_messages]
    # Annotated[list, add_messages] 부분은 특별한 의미를 가집니다:
    # - list: 'messages' 키의 값은 파이썬 리스트여야 함을 명시합니다.
    # - add_messages: 이 상태 키('messages')가 업데이트될 때 어떻게 처리될지를 정의하는 함수(리듀서)입니다.
    #   'add_messages'는 LangGraph에서 제공하는 내장 함수로, 기존 리스트에 새로운 메시지를 덮어쓰는 대신 추가(append)하도록 합니다.
    #   만약 이 Annotation이 없다면, 새로운 메시지가 들어올 때마다 기존 메시지 리스트가 통째로 교체됩니다.

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

from langgraph.graph.message import add_messages

class State(TypedDict):

# 'messages'라는 키를 가진 상태를 정의합니다. 이 키의 값은 리스트(list) 타입입니다.

messages: Annotated[list, add_messages]

# Annotated[list, add_messages] 부분은 특별한 의미를 가집니다:

# - list: 'messages' 키의 값은 파이썬 리스트여야 함을 명시합니다.

# - add_messages: 이 상태 키('messages')가 업데이트될 때 어떻게 처리될지를 정의하는 함수(리듀서)입니다.

# 'add_messages'는 LangGraph에서 제공하는 내장 함수로, 기존 리스트에 새로운 메시지를 덮어쓰는 대신 추가(append)하도록 합니다.

# 만약 이 Annotation이 없다면, 새로운 메시지가 들어올 때마다 기존 메시지 리스트가 통째로 교체됩니다.

State: 그래프의 상태(state)를 정의하는 파이썬 딕셔너리 형태의 클래스입니다. 그래프의 각 노드는 이 State 객체를 입력받고, 처리 결과를 이 State 형태로 반환하여 그래프의 상태를 업데이트합니다.
TypedDict: 파이썬의 타입 힌트를 사용하여 딕셔너리의 키와 값 타입을 명시적으로 정의합니다.
Annotated: 타입 힌트에 추가적인 메타데이터(여기서는 상태 업데이트 방식인 add_messages)를 붙일 수 있게 합니다.

StateGraph 생성 (graph_builder = StateGraph(State)):

graph_builder = StateGraph(State)
# 위에서 정의한 'State' 클래스를 기반으로 StateGraph 객체를 만듭니다.
# 이 'graph_builder' 객체를 사용하여 앞으로 노드(Node)와 엣지(Edge)를 추가하여 그래프 구조를 정의하게 됩니다.

graph_builder = StateGraph(State)

# 위에서 정의한 'State' 클래스를 기반으로 StateGraph 객체를 만듭니다.

# 이 'graph_builder' 객체를 사용하여 앞으로 노드(Node)와 엣지(Edge)를 추가하여 그래프 구조를 정의하게 됩니다.

StateGraph: LangGraph에서 그래프 구조를 정의하고 관리하는 핵심 클래스입니다. 그래프를 상태 머신(state machine)으로 간주합니다. 저스틴) 상태를 표현하려면 여러 변수들 값이 필요할 수 있으니 Dictionary가 적당합니다.

챗봇 노드 정의 (chatbot 함수 및 add_node):

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
# 사용할 언어 모델(LLM)을 초기화합니다. 여기서는 Anthropic의 Claude 3.5 Sonnet 모델을 사용합니다.

def chatbot(state: State):
    # 'chatbot'이라는 이름의 함수를 정의합니다. 이 함수가 그래프의 '노드' 역할을 합니다.
    # 입력으로 현재 그래프의 'State' 객체를 받습니다.
    return {"messages": [llm.invoke(state["messages"])]}
    # state 딕셔너리에서 'messages' 키(현재까지의 대화 기록)를 가져와 LLM에게 전달합니다.
    # LLM의 응답(AIMessage 객체)을 리스트에 담아 'messages' 키 아래 딕셔너리로 반환합니다.
    # 이 반환값은 그래프의 'State'를 업데이트하는 데 사용됩니다.
    # 'State' 정의에서 'add_messages'를 사용했으므로, 이 새로운 메시지는 기존 메시지 리스트에 추가됩니다.

graph_builder.add_node("chatbot", chatbot)
# 'graph_builder'에 노드를 추가합니다.
# - 첫 번째 인자 "chatbot": 이 노드의 고유한 이름입니다.
# - 두 번째 인자 chatbot: 이 노드가 실행될 때 호출될 함수(위에서 정의한 chatbot 함수)입니다.

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

# 사용할 언어 모델(LLM)을 초기화합니다. 여기서는 Anthropic의 Claude 3.5 Sonnet 모델을 사용합니다.

def chatbot(state: State):

# 'chatbot'이라는 이름의 함수를 정의합니다. 이 함수가 그래프의 '노드' 역할을 합니다.

# 입력으로 현재 그래프의 'State' 객체를 받습니다.

return {"messages": [llm.invoke(state["messages"])]}

# state 딕셔너리에서 'messages' 키(현재까지의 대화 기록)를 가져와 LLM에게 전달합니다.

# LLM의 응답(AIMessage 객체)을 리스트에 담아 'messages' 키 아래 딕셔너리로 반환합니다.

# 이 반환값은 그래프의 'State'를 업데이트하는 데 사용됩니다.

# 'State' 정의에서 'add_messages'를 사용했으므로, 이 새로운 메시지는 기존 메시지 리스트에 추가됩니다.

graph_builder.add_node("chatbot", chatbot)

# 'graph_builder'에 노드를 추가합니다.

# - 첫 번째 인자 "chatbot": 이 노드의 고유한 이름입니다.

# - 두 번째 인자 chatbot: 이 노드가 실행될 때 호출될 함수(위에서 정의한 chatbot 함수)입니다.

Node: 그래프 내에서 특정 작업을 수행하는 단위입니다. 주로 파이썬 함수로 정의됩니다. 입력으로 현재 상태(State)를 받고, 상태를 업데이트할 딕셔너리를 반환합니다.

진입점(Entry Point) 설정 (add_edge(START, "chatbot")):

graph_builder.add_edge(START, "chatbot")
# 그래프 실행이 어디서 시작될지를 정의합니다.
# START는 LangGraph에서 제공하는 특별한 식별자로, 그래프의 시작점을 의미합니다.
# 이 코드는 그래프가 시작되면 "chatbot" 노드부터 실행하라는 의미입니다.

graph_builder.add_edge(START, "chatbot")

# 그래프 실행이 어디서 시작될지를 정의합니다.

# START는 LangGraph에서 제공하는 특별한 식별자로, 그래프의 시작점을 의미합니다.

# 이 코드는 그래프가 시작되면 "chatbot" 노드부터 실행하라는 의미입니다.

Edge: 노드 간의 전환(transition) 또는 흐름을 정의합니다.
START: 그래프 실행의 시작을 나타내는 특수 노드 이름입니다. 저스틴) 사용자가 에이전트에게 지시하면서 에이전트가 할 일을 시작합니다.

종료점(Finish Point) 설정 (add_edge("chatbot", END)):

graph_builder.add_edge("chatbot", END)
# "chatbot" 노드가 실행된 후, 그래프 실행이 종료될 수 있음을 정의합니다.
# END는 LangGraph에서 제공하는 특별한 식별자로, 그래프의 종료점을 의미합니다.
# 이 기본 챗봇에서는 'chatbot' 노드 실행 후 바로 종료됩니다.

graph_builder.add_edge("chatbot", END)

# "chatbot" 노드가 실행된 후, 그래프 실행이 종료될 수 있음을 정의합니다.

# END는 LangGraph에서 제공하는 특별한 식별자로, 그래프의 종료점을 의미합니다.

# 이 기본 챗봇에서는 'chatbot' 노드 실행 후 바로 종료됩니다.

END: 그래프 실행의 종료를 나타내는 특수 노드 이름입니다. 특정 노드가 실행된 후 END로 엣지가 연결되면, 더 이상 진행할 작업이 없으면 그래프 실행이 멈춥니다. 저스틴) 사용자 지시(목표)를 달성하고 마치거나 목표 달성하지 못하고 마치거나 중간에 사용자가 멈추라고 할 수 있습니다.

그래프 컴파일 (graph = graph_builder.compile()):

graph = graph_builder.compile()
# 지금까지 정의한 노드와 엣지를 바탕으로 실행 가능한 그래프 객체('CompiledGraph')를 생성합니다.
# 이 'graph' 객체를 사용하여 실제로 챗봇을 실행할 수 있습니다.

graph = graph_builder.compile()

# 지금까지 정의한 노드와 엣지를 바탕으로 실행 가능한 그래프 객체('CompiledGraph')를 생성합니다.

# 이 'graph' 객체를 사용하여 실제로 챗봇을 실행할 수 있습니다.

compile(): StateGraph에 정의된 노드와 엣지 구성을 바탕으로 실제 실행 가능한 객체를 만듭니다.

그래프 시각화 (선택 사항):

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
    # 생성된 그래프의 구조를 Mermaid 형식의 PNG 이미지로 시각화하여 보여줍니다. (Jupyter 환경)
    # 그래프의 노드와 엣지가 어떻게 연결되어 있는지 한눈에 파악하는 데 도움이 됩니다.
except Exception:
    # 시각화에 필요한 추가 라이브러리가 설치되지 않았을 경우 오류가 발생할 수 있으므로 try-except로 감쌉니다.
    pass

from IPython.display import Image, display

try:

display(Image(graph.get_graph().draw_mermaid_png()))

# 생성된 그래프의 구조를 Mermaid 형식의 PNG 이미지로 시각화하여 보여줍니다. (Jupyter 환경)

# 그래프의 노드와 엣지가 어떻게 연결되어 있는지 한눈에 파악하는 데 도움이 됩니다.

except Exception:

# 시각화에 필요한 추가 라이브러리가 설치되지 않았을 경우 오류가 발생할 수 있으므로 try-except로 감쌉니다.

pass

이 시각화를 통해 START에서 chatbot 노드로, 그리고 chatbot 노드에서 END로 이어지는 단순한 흐름을 확인할 수 있습니다.

챗봇 실행 루프:

def stream_graph_updates(user_input: str):
    # 사용자 입력을 받아 그래프를 스트리밍 모드로 실행하고, 각 단계의 업데이트를 출력하는 함수
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        # graph.stream(): 그래프를 실행하고 중간 결과(이벤트)를 실시간으로 받아옵니다.
        # 입력: 사용자의 메시지를 'messages' 키 아래 리스트 형태로 전달합니다.
        for value in event.values():
            # 각 이벤트(주로 노드 실행 결과)에서 실제 값(value)을 추출합니다.
            print("Assistant:", value["messages"][-1].content)
            # 업데이트된 'messages' 리스트의 마지막 메시지(AI의 응답) 내용을 출력합니다.

while True:
    # 사용자와 계속 상호작용하기 위한 무한 루프
    try:
        user_input = input("User: ") # 사용자 입력 받기
        if user_input.lower() in ["quit", "exit", "q"]: # 종료 명령어 확인
            print("Goodbye!")
            break
        stream_graph_updates(user_input) # 입력받은 내용으로 챗봇 실행 및 결과 출력
    except:
        # input() 함수 사용이 불가능한 환경(예: 특정 노트북 환경)을 위한 예외 처리
        user_input = "What do you know about LangGraph?"
        print("User: " + user_input)
        stream_graph_updates(user_input)
        break

def stream_graph_updates(user_input: str):

# 사용자 입력을 받아 그래프를 스트리밍 모드로 실행하고, 각 단계의 업데이트를 출력하는 함수

for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):

# graph.stream(): 그래프를 실행하고 중간 결과(이벤트)를 실시간으로 받아옵니다.

# 입력: 사용자의 메시지를 'messages' 키 아래 리스트 형태로 전달합니다.

for value in event.values():

# 각 이벤트(주로 노드 실행 결과)에서 실제 값(value)을 추출합니다.

print("Assistant:", value["messages"][-1].content)

# 업데이트된 'messages' 리스트의 마지막 메시지(AI의 응답) 내용을 출력합니다.

while True:

# 사용자와 계속 상호작용하기 위한 무한 루프

try:

user_input = input("User: ") # 사용자 입력 받기

if user_input.lower() in ["quit", "exit", "q"]: # 종료 명령어 확인

print("Goodbye!")

break

stream_graph_updates(user_input) # 입력받은 내용으로 챗봇 실행 및 결과 출력

except:

# input() 함수 사용이 불가능한 환경(예: 특정 노트북 환경)을 위한 예외 처리

user_input = "What do you know about LangGraph?"

print("User: " + user_input)

stream_graph_updates(user_input)

break

graph.stream(): 그래프를 실행하고 각 단계(노드 실행)의 결과를 스트리밍으로 받아볼 수 있습니다. 사용자에게 실시간 응답을 보여주는 데 유용합니다.
입력 형식: 그래프를 실행할 때는 항상 State에서 정의한 키(messages)를 포함하는 딕셔너리 형태로 초기 상태를 전달해야 합니다.

Part 2: 🛠️ Enhancing the Chatbot with Tools (도구로 챗봇 강화) – 웹 검색 도구

도구 사용을 위한 사전 준비 (Requirements):

%%capture --no-stderr
%pip install -U tavily-python langchain_community
# Tavily 검색 엔진을 사용하기 위한 'tavily-python' 패키지와
# LangChain 커뮤니티 도구를 포함하는 'langchain_community'를 설치/업그레이드합니다.

_set_env("TAVILY_API_KEY")
# Tavily 검색 API를 사용하기 위한 API 키를 환경 변수로 설정합니다. (Setup 섹션의 함수 재사용)

%%capture --no-stderr

%pip install -U tavily-python langchain_community

# Tavily 검색 엔진을 사용하기 위한 'tavily-python' 패키지와

# LangChain 커뮤니티 도구를 포함하는 'langchain_community'를 설치/업그레이드합니다.

_set_env("TAVILY_API_KEY")

# Tavily 검색 API를 사용하기 위한 API 키를 환경 변수로 설정합니다. (Setup 섹션의 함수 재사용)

Tavily는 LLM 애플리케이션에 최적화된 검색 API입니다. 저스틴) Tavily는 My AI Smateasy에서 AI 검색엔진으로 채택한 것입니다.

도구 정의 및 테스트 (TavilySearchResults):

from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=2)
# Tavily 검색 도구 객체를 생성합니다. 최대 2개의 검색 결과를 가져오도록 설정합니다.
tools = [tool]
# 사용할 도구들을 리스트에 담습니다. 현재는 Tavily 검색 도구 하나만 있습니다.
tool.invoke("What's a 'node' in LangGraph?")
# 도구가 제대로 작동하는지 테스트해봅니다. 'invoke' 메서드로 도구를 직접 실행합니다.
# 실행 결과로 검색된 웹페이지 URL과 요약 내용이 담긴 리스트가 출력됩니다.

from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=2)

# Tavily 검색 도구 객체를 생성합니다. 최대 2개의 검색 결과를 가져오도록 설정합니다.

tools = [tool]

# 사용할 도구들을 리스트에 담습니다. 현재는 Tavily 검색 도구 하나만 있습니다.

tool.invoke("What's a 'node' in LangGraph?")

# 도구가 제대로 작동하는지 테스트해봅니다. 'invoke' 메서드로 도구를 직접 실행합니다.

# 실행 결과로 검색된 웹페이지 URL과 요약 내용이 담긴 리스트가 출력됩니다.

TavilySearchResults: LangChain에서 제공하는 Tavily 검색 도구 클래스입니다.
tools: 에이전트(챗봇)가 사용할 수 있는 도구들의 목록입니다. 이 목록을 나중에 LLM과 그래프 노드에 전달합니다.

State 및 GraphBuilder 재정의 (도구 연동 준비):

# Part 1과 동일한 State 정의
class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
# Modification: tell the LLM which tools it can call
llm_with_tools = llm.bind_tools(tools)
# LLM 객체에 'bind_tools' 메서드를 사용하여 사용할 도구(tools 리스트)를 알려줍니다.
# 이렇게 하면 LLM은 필요할 때 도구를 호출해야 한다는 것을 인지하고,
# 도구 호출에 필요한 특정 형식(예: JSON)으로 응답을 생성할 수 있게 됩니다.

def chatbot(state: State):
    # 챗봇 노드 함수는 거의 동일하지만, 이제 'llm_with_tools'를 사용합니다.
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

# Part 1과 동일한 State 정의

class State(TypedDict):

messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

# Modification: tell the LLM which tools it can call

llm_with_tools = llm.bind_tools(tools)

# LLM 객체에 'bind_tools' 메서드를 사용하여 사용할 도구(tools 리스트)를 알려줍니다.

# 이렇게 하면 LLM은 필요할 때 도구를 호출해야 한다는 것을 인지하고,

# 도구 호출에 필요한 특정 형식(예: JSON)으로 응답을 생성할 수 있게 됩니다.

def chatbot(state: State):

# 챗봇 노드 함수는 거의 동일하지만, 이제 'llm_with_tools'를 사용합니다.

return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

bind_tools(tools): LLM이 tools 리스트에 있는 도구들의 존재와 사용법(이름, 설명, 필요한 인자 등)을 알게 합니다. LLM은 이제 단순히 텍스트 응답뿐만 아니라, “이 도구를 이런 인자로 호출해야 해”라는 형식의 응답(AIMessage 내 tool_calls 속성)을 생성할 수 있습니다.

도구 실행 노드 정의 (BasicToolNode – 직접 구현 예시):

튜토리얼에서는 이해를 돕기 위해 먼저 ToolNode의 기능을 직접 구현하는 BasicToolNode 클래스를 보여줍니다. (나중에는 LangGraph의 내장 ToolNode를 사용합니다.)

import json
from langchain_core.messages import ToolMessage

class BasicToolNode:
    """A node that runs the tools requested in the last AIMessage."""
    def __init__(self, tools: list) -&gt; None:
        # 생성자: 도구 리스트를 받아 이름으로 쉽게 찾을 수 있도록 딕셔너리 형태로 저장합니다.
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        # 노드가 호출될 때 실행되는 메서드 (State 딕셔너리를 입력으로 받음)
        if messages := inputs.get("messages", []):
            message = messages[-1] # 상태에서 가장 마지막 메시지(AIMessage)를 가져옵니다.
        else:
            raise ValueError("No message found in input")

        outputs = [] # 도구 실행 결과를 담을 리스트
        # LLM이 도구 호출을 요청했는지 확인 (message.tool_calls 확인)
        if hasattr(message, 'tool_calls'):
            for tool_call in message.tool_calls:
                # 요청된 도구 이름(tool_call["name"])으로 도구 객체를 찾습니다.
                # 해당 도구의 invoke 메서드를 호출하여 도구를 실행합니다 (인자는 tool_call["args"]).
                tool_result = self.tools_by_name[tool_call["name"]].invoke(
                    tool_call["args"]
                )
                # 도구 실행 결과를 ToolMessage 형태로 만들어 outputs 리스트에 추가합니다.
                # ToolMessage는 어떤 도구 호출(tool_call_id)에 대한 결과인지 명시합니다.
                outputs.append(
                    ToolMessage(
                        content=json.dumps(tool_result), # 결과는 JSON 문자열로 변환
                        name=tool_call["name"],
                        tool_call_id=tool_call["id"],
                    )
                )
        # 도구 실행 결과(ToolMessage 리스트)를 'messages' 키 아래 딕셔너리로 반환하여 상태를 업데이트합니다.
        return {"messages": outputs}

tool_node = BasicToolNode(tools=[tool]) # 위에서 정의한 Tavily 도구로 BasicToolNode 객체 생성
graph_builder.add_node("tools", tool_node) # 그래프 빌더에 "tools"라는 이름으로 이 노드를 추가합니다.

import json

from langchain_core.messages import ToolMessage

class BasicToolNode:

"""A node that runs the tools requested in the last AIMessage."""

def __init__(self, tools: list) -> None:

# 생성자: 도구 리스트를 받아 이름으로 쉽게 찾을 수 있도록 딕셔너리 형태로 저장합니다.

self.tools_by_name = {tool.name: tool for tool in tools}

def __call__(self, inputs: dict):

# 노드가 호출될 때 실행되는 메서드 (State 딕셔너리를 입력으로 받음)

if messages := inputs.get("messages", []):

message = messages[-1] # 상태에서 가장 마지막 메시지(AIMessage)를 가져옵니다.

else:

raise ValueError("No message found in input")

outputs = [] # 도구 실행 결과를 담을 리스트

# LLM이 도구 호출을 요청했는지 확인 (message.tool_calls 확인)

if hasattr(message, 'tool_calls'):

for tool_call in message.tool_calls:

# 요청된 도구 이름(tool_call["name"])으로 도구 객체를 찾습니다.

# 해당 도구의 invoke 메서드를 호출하여 도구를 실행합니다 (인자는 tool_call["args"]).

tool_result = self.tools_by_name[tool_call["name"]].invoke(

tool_call["args"]

)

# 도구 실행 결과를 ToolMessage 형태로 만들어 outputs 리스트에 추가합니다.

# ToolMessage는 어떤 도구 호출(tool_call_id)에 대한 결과인지 명시합니다.

outputs.append(

ToolMessage(

content=json.dumps(tool_result), # 결과는 JSON 문자열로 변환

name=tool_call["name"],

tool_call_id=tool_call["id"],

)

# 도구 실행 결과(ToolMessage 리스트)를 'messages' 키 아래 딕셔너리로 반환하여 상태를 업데이트합니다.

return {"messages": outputs}

tool_node = BasicToolNode(tools=[tool]) # 위에서 정의한 Tavily 도구로 BasicToolNode 객체 생성

graph_builder.add_node("tools", tool_node) # 그래프 빌더에 "tools"라는 이름으로 이 노드를 추가합니다.

ToolNode의 역할: LLM이 도구 사용을 요청하면(즉, 마지막 메시지에 tool_calls가 있으면), 해당 도구를 실제로 실행하고 그 결과를 ToolMessage 형태로 상태에 추가하는 역할을 합니다. 저스틴) 도구는 자체적으로 구현해야 하는 내부 도구들과 MCP 같은 표준을 지키는 외부 도구들을 사용합니다. MCP도 LLM에 사용되려면 여기서 도구를 정의할 때 필요한 정보들을 제공해야 합니다.

조건부 엣지 정의 (route_tools 함수 및 add_conditional_edges):

LLM의 응답에 따라 다음 단계를 결정하는 라우터 함수를 정의합니다.

def route_tools(state: State):
    """
    Use in the conditional_edge to route to the ToolNode if the last message
    has tool calls. Otherwise, route to the end.
    """
    # 상태에서 마지막 메시지(AI의 응답)를 가져옵니다.
    if isinstance(state, list): # 상태가 리스트 형식일 경우 (에러 처리 등)
        ai_message = state[-1]
    elif messages := state.get("messages", []): # 상태가 딕셔너리 형식일 경우
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to tool_edge: {state}")

    # 마지막 메시지에 'tool_calls' 속성이 있고, 그 안에 내용이 있는지 확인합니다.
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) &gt; 0:
        # 도구 호출이 있으면 "tools" 문자열을 반환합니다. (즉, "tools" 노드로 가라는 의미)
        return "tools"
    # 도구 호출이 없으면 END를 반환합니다. (즉, 그래프 실행을 종료하라는 의미)
    return END

# 'chatbot' 노드 다음에 실행될 엣지를 조건부로 설정합니다.
graph_builder.add_conditional_edges(
    "chatbot", # 시작 노드: "chatbot" 노드가 실행된 후에 이 조건을 검사합니다.
    route_tools, # 조건 함수: 위에서 정의한 route_tools 함수를 사용하여 다음 노드를 결정합니다.
    {
        # route_tools 함수의 반환값("tools" 또는 END)에 따라 실제로 이동할 노드 이름을 매핑합니다.
        "tools": "tools", # route_tools가 "tools"를 반환하면 -&gt; "tools" 노드로 이동
        END: END          # route_tools가 END를 반환하면 -&gt; 그래프 종료(END)
    },
)

def route_tools(state: State):

"""

Use in the conditional_edge to route to the ToolNode if the last message

has tool calls. Otherwise, route to the end.

"""

# 상태에서 마지막 메시지(AI의 응답)를 가져옵니다.

if isinstance(state, list): # 상태가 리스트 형식일 경우 (에러 처리 등)

ai_message = state[-1]

elif messages := state.get("messages", []): # 상태가 딕셔너리 형식일 경우

ai_message = messages[-1]

else:

raise ValueError(f"No messages found in input state to tool_edge: {state}")

# 마지막 메시지에 'tool_calls' 속성이 있고, 그 안에 내용이 있는지 확인합니다.

if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:

# 도구 호출이 있으면 "tools" 문자열을 반환합니다. (즉, "tools" 노드로 가라는 의미)

return "tools"

# 도구 호출이 없으면 END를 반환합니다. (즉, 그래프 실행을 종료하라는 의미)

return END

# 'chatbot' 노드 다음에 실행될 엣지를 조건부로 설정합니다.

graph_builder.add_conditional_edges(

"chatbot", # 시작 노드: "chatbot" 노드가 실행된 후에 이 조건을 검사합니다.

route_tools, # 조건 함수: 위에서 정의한 route_tools 함수를 사용하여 다음 노드를 결정합니다.

{

# route_tools 함수의 반환값("tools" 또는 END)에 따라 실제로 이동할 노드 이름을 매핑합니다.

"tools": "tools", # route_tools가 "tools"를 반환하면 -> "tools" 노드로 이동

END: END # route_tools가 END를 반환하면 -> 그래프 종료(END)

)

Conditional Edges: 특정 노드 실행 후, 현재 상태(State)를 바탕으로 다음에 실행할 노드를 동적으로 결정하는 엣지입니다. 조건 함수(여기서는 route_tools)의 반환값에 따라 다른 노드로 분기할 수 있습니다. 저스틴) 조건문은 절차를 계획해서 실행하는 에이전트에 필요한 것입니다. ReAct는 한 단계만 계획해서 실행하고 변화를 보고 다음 상태를 결정하는 것입니다. 아직까지 ReAct를 어떻게 연결할지는 안 나왔네요.

엣지 추가 및 그래프 컴파일:

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
# "tools" 노드(도구 실행 노드)가 실행된 후에는 항상 "chatbot" 노드로 돌아가도록 엣지를 추가합니다.
# 도구 실행 결과를 받은 LLM이 다음 행동(답변 생성, 추가 도구 사용 등)을 결정하게 됩니다.

graph_builder.add_edge(START, "chatbot") # 시작점은 여전히 "chatbot" 노드
graph = graph_builder.compile() # 그래프 컴파일

# Any time a tool is called, we return to the chatbot to decide the next step

graph_builder.add_edge("tools", "chatbot")

# "tools" 노드(도구 실행 노드)가 실행된 후에는 항상 "chatbot" 노드로 돌아가도록 엣지를 추가합니다.

# 도구 실행 결과를 받은 LLM이 다음 행동(답변 생성, 추가 도구 사용 등)을 결정하게 됩니다.

graph_builder.add_edge(START, "chatbot") # 시작점은 여전히 "chatbot" 노드

graph = graph_builder.compile() # 그래프 컴파일

이제 그래프는 다음과 같은 루프(Loop) 구조를 가집니다:
- START -> chatbot
- chatbot -> (조건 검사)
  - 도구 호출 O -> tools -> chatbot (다시 LLM에게)
  - 도구 호출 X -> END (종료)

그래프 시각화 및 실행:
- 그래프 시각화를 통해 새로 추가된 tools 노드와 chatbot 노드 사이의 양방향 흐름(조건부 엣지 포함)을 확인할 수 있습니다.
- 챗봇 실행 코드는 Part 1과 동일한 stream_graph_updates 함수를 사용합니다. 이제 챗봇에게 학습 데이터에 없는 질문(예: “LangGraph에 대해 무엇을 아나요?”)을 하면, LLM이 tavily_search_results_json 도구를 호출하고(tool_calls 생성), tools 노드가 실행되어 검색 결과를 ToolMessage로 반환하며, 다시 chatbot 노드가 이 결과를 바탕으로 최종 답변을 생성하는 과정을 볼 수 있습니다.

Part 3: Adding Memory to the Chatbot (챗봇에 메모리 추가) – 체크포인팅(Checkpointing)

체크포인팅 (Checkpointing)

LangGraph에서 ‘메모리’는 단순히 이전 메시지를 저장하는 것 이상입니다. 그래프의 전체 상태(State)를 각 단계(노드 실행 후)마다 저장하는 것을 의미합니다.이렇게 저장된 상태를 체크포인트(Checkpoint)라고 합니다.
체크포인트를 사용하면, 나중에 동일한 대화(식별자 사용)를 다시 시작할 때 마지막 체크포인트부터 이어서 실행할 수 있습니다. 이를 통해 챗봇은 이전 대화 내용을 ‘기억’하는 것처럼 보이게 됩니다. 저스틴) 기억하지 않고 기억하는 것처럼 보이게만 하네요. 장기기억하는게 아닙니다.
체크포인팅은 단순히 대화 기록뿐만 아니라, Part 5에서 배울 사용자 정의 상태(custom state) 등 그래프의 모든 상태 정보를 저장하므로 매우 강력합니다.

체크포인터 생성 (MemorySaver):

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
# 인메모리(in-memory) 체크포인터 객체를 생성합니다.
# 이 체크포인터는 모든 대화 상태(체크포인트)를 프로그램 실행 중인 메모리에 저장합니다.
# 장점: 설정이 간편하고 빠릅니다.
# 단점: 프로그램이 종료되면 모든 저장 내용이 사라집니다.
# 튜토리얼에서는 간편함을 위해 사용하지만, 실제 프로덕션 환경에서는
# 데이터베이스 기반 체크포인터(SqliteSaver, PostgresSaver 등)를 사용하여 영구적으로 저장하는 것이 일반적입니다.

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

# 인메모리(in-memory) 체크포인터 객체를 생성합니다.

# 이 체크포인터는 모든 대화 상태(체크포인트)를 프로그램 실행 중인 메모리에 저장합니다.

# 장점: 설정이 간편하고 빠릅니다.

# 단점: 프로그램이 종료되면 모든 저장 내용이 사라집니다.

# 튜토리얼에서는 간편함을 위해 사용하지만, 실제 프로덕션 환경에서는

# 데이터베이스 기반 체크포인터(SqliteSaver, PostgresSaver 등)를 사용하여 영구적으로 저장하는 것이 일반적입니다.

Checkpointer: 그래프의 상태를 저장하고 불러오는 역할을 담당하는 객체입니다. 다양한 종류(인메모리, SQLite, PostgreSQL 등)가 있습니다. 저스틴) 장기 기억한다는 것은 영속성을 제공하는 데이터베이스에 체크포인트를 저장하는 것으로 하네요. 시맨틱 커널의 context variables가 state에 해당하고 그 상태를 저장해 놓는게 checkpointer게 되겠네요. 장기 기억도 중요하지만 어떤 맥락에서의 기억 체인도 중요함을 보여주네요.

그래프 정의 (Part 2 코드 재사용 + Prebuilt 컴포넌트 사용):

# State 정의는 동일
class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

# 도구 설정 및 LLM 바인딩도 동일
tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
llm_with_tools = llm.bind_tools(tools)

# 챗봇 노드 정의도 동일
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

# BasicToolNode 대신 LangGraph의 내장 ToolNode 사용
from langgraph.prebuilt import ToolNode, tools_condition
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# route_tools 대신 LangGraph의 내장 tools_condition 사용
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition, # prebuilt 조건 함수 사용
    # 매핑은 prebuilt 함수 내부 규칙에 따라 자동으로 처리되므로 명시할 필요 없을 수 있음 (튜토리얼 코드에는 생략됨)
    # prebuilt는 보통 {"tools": "tools", END: END} 와 유사하게 작동
)

# 엣지 설정 동일
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

# State 정의는 동일

class State(TypedDict):

messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

# 도구 설정 및 LLM 바인딩도 동일

tool = TavilySearchResults(max_results=2)

tools = [tool]

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

llm_with_tools = llm.bind_tools(tools)

# 챗봇 노드 정의도 동일

def chatbot(state: State):

return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

# BasicToolNode 대신 LangGraph의 내장 ToolNode 사용

from langgraph.prebuilt import ToolNode, tools_condition

tool_node = ToolNode(tools=[tool])

graph_builder.add_node("tools", tool_node)

# route_tools 대신 LangGraph의 내장 tools_condition 사용

graph_builder.add_conditional_edges(

"chatbot",

tools_condition, # prebuilt 조건 함수 사용

# 매핑은 prebuilt 함수 내부 규칙에 따라 자동으로 처리되므로 명시할 필요 없을 수 있음 (튜토리얼 코드에는 생략됨)

# prebuilt는 보통 {"tools": "tools", END: END} 와 유사하게 작동

)

# 엣지 설정 동일

graph_builder.add_edge("tools", "chatbot")

graph_builder.add_edge(START, "chatbot")

이 파트에서는 Part 2에서 직접 구현했던 BasicToolNode와 route_tools 대신, LangGraph에서 제공하는 ToolNode와 tools_condition을 사용합니다. 기능은 동일하지만 코드가 더 간결해지고, ToolNode는 병렬 도구 실행 같은 추가적인 최적화 기능을 제공할 수 있습니다.

체크포인터와 함께 그래프 컴파일:

graph = graph_builder.compile(checkpointer=memory)
# 그래프를 컴파일할 때 'checkpointer' 인자에 위에서 생성한 'memory' 객체를 전달합니다.
# 이렇게 하면 <span style="color: #3366ff;">그래프가 실행될 때마다 각 단계 이후의 상태가 'memory' 체크포인터에 자동으로 저장됩니다.
</span>

graph = graph_builder.compile(checkpointer=memory)

# 그래프를 컴파일할 때 'checkpointer' 인자에 위에서 생성한 'memory' 객체를 전달합니다.

# 이렇게 하면 그래프가 실행될 때마다 각 단계 이후의 상태가 'memory' 체크포인터에 자동으로 저장됩니다.

compile() 메서드에 checkpointer를 전달하는 것이 핵심입니다. 이제 이 graph 객체는 상태 저장 및 로드 기능을 갖게 됩니다.

그래프 시각화 (선택 사항):
- 그래프 구조 자체는 Part 2와 동일합니다. 체크포인터 추가는 그래프의 노드나 엣지 구조를 바꾸지 않고, 실행 시 상태 저장 방식을 변경하는 것입니다.

메모리 기능 테스트 (config 와 thread_id 사용):

config = {"configurable": {"thread_id": "1"}}
# 그래프를 실행할 때 전달할 설정(config) 딕셔너리를 만듭니다.
# 'configurable' 키 아래에 'thread_id'를 지정합니다.
# 'thread_id'는 특정 대화 세션(conversation thread)을 식별하는 고유한 키입니다.
# 체크포인터는 이 'thread_id'를 사용하여 해당 대화의 상태를 저장하고 불러옵니다. "1"은 임의의 ID입니다.

user_input = "Hi there! My name is Will."

# stream() 또는 invoke() 메서드를 호출할 때,
# 첫 번째 인자로는 입력 데이터(딕셔너리), 두 번째 인자로 'config'를 전달합니다.
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]}, # 첫 번째 인자: 입력
    config, # 두 번째 인자: 설정 (thread_id 포함)
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print() # 결과 출력

# --- 첫 번째 호출 후 ---

user_input = "Remember my name?"

# 동일한 'config' (즉, 동일한 thread_id="1")를 사용하여 다시 호출합니다.
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]}, # 새 입력
    config, # 동일한 config 사용!
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print() # 결과 출력 (챗봇이 이름을 기억하는 것을 확인)

config = {"configurable": {"thread_id": "1"}}

# 그래프를 실행할 때 전달할 설정(config) 딕셔너리를 만듭니다.

# 'configurable' 키 아래에 'thread_id'를 지정합니다.

# 'thread_id'는 특정 대화 세션(conversation thread)을 식별하는 고유한 키입니다.

# 체크포인터는 이 'thread_id'를 사용하여 해당 대화의 상태를 저장하고 불러옵니다. "1"은 임의의 ID입니다.

user_input = "Hi there! My name is Will."

# stream() 또는 invoke() 메서드를 호출할 때,

# 첫 번째 인자로는 입력 데이터(딕셔너리), 두 번째 인자로 'config'를 전달합니다.

events = graph.stream(

{"messages": [{"role": "user", "content": user_input}]}, # 첫 번째 인자: 입력

config, # 두 번째 인자: 설정 (thread_id 포함)

stream_mode="values",

)

for event in events:

event["messages"][-1].pretty_print() # 결과 출력

# --- 첫 번째 호출 후 ---

user_input = "Remember my name?"

# 동일한 'config' (즉, 동일한 thread_id="1")를 사용하여 다시 호출합니다.

events = graph.stream(

{"messages": [{"role": "user", "content": user_input}]}, # 새 입력

config, # 동일한 config 사용!

stream_mode="values",

)

for event in events:

event["messages"][-1].pretty_print() # 결과 출력 (챗봇이 이름을 기억하는 것을 확인)

핵심: invoke나 stream 메서드를 호출할 때 두 번째 인자로 config 딕셔너리를 전달하는 것입니다. 이 config 안에 있는 thread_id를 기준으로 체크포인터가 해당 대화의 마지막 상태를 불러와서 실행을 이어갑니다.
따라서 두 번째 호출에서 챗봇은 첫 번째 호출의 내용(“My name is Will”)을 기억하고 “Will”이라는 이름을 사용하여 응답할 수 있습니다.

다른 스레드 ID로 테스트:

# thread_id를 "2"로 변경하여 새로운 config 생성
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]}, # 동일한 질문 ("Remember my name?")
    {"configurable": {"thread_id": "2"}}, # 다른 thread_id 사용!
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print() # 결과 출력 (챗봇이 이름을 기억하지 못하는 것을 확인)

# thread_id를 "2"로 변경하여 새로운 config 생성

events = graph.stream(

{"messages": [{"role": "user", "content": user_input}]}, # 동일한 질문 ("Remember my name?")

{"configurable": {"thread_id": "2"}}, # 다른 thread_id 사용!

stream_mode="values",

)

for event in events:

event["messages"][-1].pretty_print() # 결과 출력 (챗봇이 이름을 기억하지 못하는 것을 확인)

thread_id를 “1”이 아닌 “2”로 변경하여 실행하면, 체크포인터는 thread_id="2"에 해당하는 저장된 상태를 찾으려 하지만 없으므로, 처음부터 대화를 시작합니다. 따라서 챗봇은 이름을 기억하지 못합니다. 이는 thread_id가 대화 세션을 구분하는 역할을 한다는 것을 보여줍니다.

저장된 상태 확인 (get_state):

snapshot = graph.get_state(config) # thread_id="1"에 해당하는 config 사용
# 'get_state' 메서드를 사용하면 특정 config(thread_id)에 대한 현재 저장된 상태 스냅샷을 가져올 수 있습니다.
snapshot
# 출력된 StateSnapshot 객체를 보면 'values' 안에 지금까지의 'messages' (HumanMessage, AIMessage)가 모두 누적되어 저장된 것을 <br />볼 수 있습니다. <br /><span style="color: #339966;">저스틴) 이렇게 하면 메시지를 모아 놓은 것 뿐이 됩니다. 특정 시점 대화를 모아 그 상태를 LLM이 체크포인트화해야 합니다.<br /></span>있게 해야 합니다. 
# 또한, 'next' 필드는 다음에 실행될 노드를 나타냅니다(대화가 끝났으므로 여기서는 비어 있음).
# 'config' 필드에는 사용된 thread_id와 내부적으로 생성된 checkpoint_id 등이 포함됩니다.

snapshot = graph.get_state(config) # thread_id="1"에 해당하는 config 사용

# 'get_state' 메서드를 사용하면 특정 config(thread_id)에 대한 현재 저장된 상태 스냅샷을 가져올 수 있습니다.

snapshot

# 출력된 StateSnapshot 객체를 보면 'values' 안에 지금까지의 'messages' (HumanMessage, AIMessage)가 모두 누적되어 저장된 것을 볼 수 있습니다. 저스틴) 이렇게 하면 메시지를 모아 놓은 것 뿐이 됩니다. 특정 시점 대화를 모아 그 상태를 LLM이 체크포인트화해야 합니다. 있게 해야 합니다.

# 또한, 'next' 필드는 다음에 실행될 노드를 나타냅니다(대화가 끝났으므로 여기서는 비어 있음).

# 'config' 필드에는 사용된 thread_id와 내부적으로 생성된 checkpoint_id 등이 포함됩니다.

get_state(config): 특정 대화 스레드의 현재 상태를 직접 확인하고 싶을 때 유용합니다.

Part 4: Human-in-the-loop (인간 개입)

이 파트에서는 에이전트(챗봇)의 실행 흐름 중간에 사람의 입력이나 승인을 받을 수 있도록 하는 방법을 배웁니다. 저스틴) 사람과 계속 상호작용해야 합니다. 필요에 따라 대화나 UI로 상호작용해야 합니다. My AI Smarteasy의 파일럿 에이전트인 “My AI”가 중간 중간에 사용자 의견을 묻는 팝업 화면이 나오는 이유입니다.

핵심 개념: Human-in-the-loop (HITL)

AI 에이전트는 때때로 부정확하거나 예상치 못한 행동을 할 수 있습니다. 중요한 작업을 수행하기 전에 사람의 검토를 받거나, 에이전트가 막혔을 때 사람에게 도움을 요청하는 기능은 신뢰성을 높이는 데 중요합니다.
LangGraph는 체크포인팅 시스템을 기반으로 이러한 HITL 워크플로우를 자연스럽게 지원합니다. 실행이 중단되면 현재 상태가 체크포인트에 저장되고, 나중에 사람의 입력과 함께 실행을 재개할 수 있습니다.

코드 설명:

HITL을 위한 도구 정의 (human_assistance):

from langgraph.types import Command, interrupt
from langchain_core.tools import tool

@tool # LangChain의 @tool 데코레이터를 사용하여 함수를 LangChain 도구로 만듭니다.
def human_assistance(query: str) -&gt; str:
    """Request assistance from a human.""" # 도구 설명 (LLM이 이 설명을 보고 도구를 선택)
    # 이 함수(도구)가 호출되면 interrupt 함수를 실행합니다.
    human_response = interrupt({"query": query})
    # interrupt 함수는 그래프 실행을 '일시 중지'시킵니다.
    # 중지 시, interrupt에 전달된 딕셔너리({"query": query})는 외부(호출자)에 전달되어
    # 사람에게 어떤 정보가 필요한지 알려주는 데 사용될 수 있습니다.
    # 실행이 재개될 때, 외부에서 전달된 값(사람의 응답)이 human_response 변수에 저장됩니다.
    # 여기서는 재개 시 {"data": "사람의 응답 내용"} 형태의 딕셔너리가 전달될 것으로 예상합니다.
    return human_response["data"] # 사람의 응답 내용("data" 키의 값)을 반환합니다.

from langgraph.types import Command, interrupt

from langchain_core.tools import tool

@tool # LangChain의 @tool 데코레이터를 사용하여 함수를 LangChain 도구로 만듭니다.

def human_assistance(query: str) -> str:

"""Request assistance from a human.""" # 도구 설명 (LLM이 이 설명을 보고 도구를 선택)

# 이 함수(도구)가 호출되면 interrupt 함수를 실행합니다.

human_response = interrupt({"query": query})

# interrupt 함수는 그래프 실행을 '일시 중지'시킵니다.

# 중지 시, interrupt에 전달된 딕셔너리({"query": query})는 외부(호출자)에 전달되어

# 사람에게 어떤 정보가 필요한지 알려주는 데 사용될 수 있습니다.

# 실행이 재개될 때, 외부에서 전달된 값(사람의 응답)이 human_response 변수에 저장됩니다.

# 여기서는 재개 시 {"data": "사람의 응답 내용"} 형태의 딕셔너리가 전달될 것으로 예상합니다.

return human_response["data"] # 사람의 응답 내용("data" 키의 값)을 반환합니다.

@tool: 함수를 LangChain 도구로 쉽게 만들어주는 데코레이터입니다. 함수의 독스트링(docstring)이 도구의 설명으로 사용됩니다.
interrupt(payload): 이 함수가 호출되면 LangGraph 실행이 즉시 멈춥니다. payload(여기서는 {"query": query})는 실행을 멈춘 이유나 필요한 정보를 외부에 전달하는 데 사용됩니다. 그래프의 현재 상태는 체크포인터에 저장됩니다.
반환 값: interrupt 함수는 실행이 재개될 때 외부에서 Command(resume=...) 객체를 통해 전달된 값을 반환합니다.

그래프 설정 (Part 3 기반 + 새 도구 추가):

# State 정의는 동일
class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

# 기존 Tavily 도구에 새로 만든 human_assistance 도구를 추가합니다.
tool = TavilySearchResults(max_results=2)
tools = [tool, human_assistance] # 이제 도구가 2개

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")
llm_with_tools = llm.bind_tools(tools) # LLM에게 2개의 도구를 알려줍니다.

def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    # 중요: 병렬 도구 호출 비활성화 가정
    # interrupt가 포함된 도구가 있을 때, 만약 LLM이 여러 도구를 동시에 호출하도록 요청하면,
    # interrupt 발생 후 재개 시 다른 도구 호출이 반복 실행될 수 있는 문제가 생길 수 있습니다.
    # 이 예제에서는 설명을 간단히 하기 위해 LLM이 한 번에 하나의 도구만 호출한다고 가정합니다.
    # (assert 문으로 확인 - 실제로는 모델 설정이나 프롬프트 조정이 필요할 수 있음)
    assert len(message.tool_calls) &lt;= 1
    return {"messages": [message]}

graph_builder.add_node("chatbot", chatbot)

# ToolNode에 두 가지 도구를 모두 전달합니다.
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)

# 조건부 엣지 및 다른 엣지 설정은 동일
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

# 체크포인터와 함께 컴파일
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# State 정의는 동일

class State(TypedDict):

messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

# 기존 Tavily 도구에 새로 만든 human_assistance 도구를 추가합니다.

tool = TavilySearchResults(max_results=2)

tools = [tool, human_assistance] # 이제 도구가 2개

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

llm_with_tools = llm.bind_tools(tools) # LLM에게 2개의 도구를 알려줍니다.

def chatbot(state: State):

message = llm_with_tools.invoke(state["messages"])

# 중요: 병렬 도구 호출 비활성화 가정

# interrupt가 포함된 도구가 있을 때, 만약 LLM이 여러 도구를 동시에 호출하도록 요청하면,

# interrupt 발생 후 재개 시 다른 도구 호출이 반복 실행될 수 있는 문제가 생길 수 있습니다.

# 이 예제에서는 설명을 간단히 하기 위해 LLM이 한 번에 하나의 도구만 호출한다고 가정합니다.

# (assert 문으로 확인 - 실제로는 모델 설정이나 프롬프트 조정이 필요할 수 있음)

assert len(message.tool_calls) <= 1

return {"messages": [message]}

graph_builder.add_node("chatbot", chatbot)

# ToolNode에 두 가지 도구를 모두 전달합니다.

tool_node = ToolNode(tools=tools)

graph_builder.add_node("tools", tool_node)

# 조건부 엣지 및 다른 엣지 설정은 동일

graph_builder.add_conditional_edges("chatbot", tools_condition)

graph_builder.add_edge("tools", "chatbot")

graph_builder.add_edge(START, "chatbot")

# 체크포인터와 함께 컴파일

memory = MemorySaver()

graph = graph_builder.compile(checkpointer=memory)

핵심 변경 사항은 tools 리스트에 human_assistance 도구를 추가하고, ToolNode에도 이 도구를 포함시킨 것입니다.
병렬 도구 호출 관련 주의사항은 interrupt 사용 시 고려해야 할 중요한 점입니다.

그래프 시각화:
- 구조는 Part 2, 3과 동일하게 보이지만, tools 노드는 이제 human_assistance 도구도 처리할 수 있습니다.

Human-in-the-loop 실행 테스트:

user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}

events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)
# 이벤트 스트림 처리 (출력 확인)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"

config = {"configurable": {"thread_id": "1"}}

events = graph.stream(

{"messages": [{"role": "user", "content": user_input}]},

config,

stream_mode="values",

)

# 이벤트 스트림 처리 (출력 확인)

for event in events:

if "messages" in event:

event["messages"][-1].pretty_print()

사용자가 명시적으로 “assistance”를 요청했으므로, LLM은 human_assistance 도구를 호출하는 tool_calls를 생성할 가능성이 높습니다.
ToolNode가 human_assistance 도구를 실행하면, 그 안의 interrupt() 함수가 호출되어 그래프 실행이 일시 중지됩니다. 터미널이나 노트북 출력에는 LLM이 human_assistance 도구를 호출했다는 메시지까지만 나오고 더 이상 진행되지 않습니다.

중단 상태 확인:

snapshot = graph.get_state(config)
snapshot.next
# 출력이 ('tools',) 와 같이 나옵니다.
# 이는 그래프가 'tools' 노드를 실행하는 도중에 멈췄음을 의미합니다.

snapshot = graph.get_state(config)

snapshot.next

# 출력이 ('tools',) 와 같이 나옵니다.

# 이는 그래프가 'tools' 노드를 실행하는 도중에 멈췄음을 의미합니다.

get_state()를 통해 현재 상태를 확인하면, 다음에 실행될 노드가 tools임을 알 수 있습니다. 즉, tools 노드 내부의 human_assistance 함수 안에서 interrupt가 발생하여 멈춘 것입니다.

실행 재개 및 사람 입력 전달 (Command 사용):

human_response = (
    "We, the experts are here to help! We'd recommend you check out LangGraph to build your agent."
    " It's much more reliable and extensible than simple autonomous agents."
)

# Command 객체를 사용하여 실행 재개 명령과 데이터를 함께 전달합니다.
human_command = Command(resume={"data": human_response})
# resume 키에 딕셔너리를 전달합니다. 이 딕셔너리의 내용은 interrupt 함수가 호출된 곳으로 반환됩니다.
# human_assistance 함수에서는 human_response["data"] 로 접근할 수 있도록 {"data": ...} 형태로 전달합니다.

# 그래프를 다시 stream() 메서드로 호출하되, 입력 데이터 대신 Command 객체를 전달합니다.
# config는 동일한 thread_id를 사용해야 합니다.
events = graph.stream(human_command, config, stream_mode="values")
# 이벤트 스트림 처리 (출력 확인)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

human_response = (

"We, the experts are here to help! We'd recommend you check out LangGraph to build your agent."

" It's much more reliable and extensible than simple autonomous agents."

)

# Command 객체를 사용하여 실행 재개 명령과 데이터를 함께 전달합니다.

human_command = Command(resume={"data": human_response})

# resume 키에 딕셔너리를 전달합니다. 이 딕셔너리의 내용은 interrupt 함수가 호출된 곳으로 반환됩니다.

# human_assistance 함수에서는 human_response["data"] 로 접근할 수 있도록 {"data": ...} 형태로 전달합니다.

# 그래프를 다시 stream() 메서드로 호출하되, 입력 데이터 대신 Command 객체를 전달합니다.

# config는 동일한 thread_id를 사용해야 합니다.

events = graph.stream(human_command, config, stream_mode="values")

# 이벤트 스트림 처리 (출력 확인)

for event in events:

if "messages" in event:

event["messages"][-1].pretty_print()

Command(resume=...): 중단된 그래프 실행을 재개시키는 데 사용됩니다. resume 키에 전달된 값은 interrupt 함수가 반환할 값이 됩니다.
그래프의 stream 또는 invoke 메서드에 입력 데이터 대신 Command 객체를 전달하고, **동일한 config**를 사용하면, LangGraph는 체크포인트에서 상태를 로드하고 중단된 지점부터 실행을 재개합니다.
재개 후 실행 흐름:
1. human_assistance 함수 내 interrupt가 human_command의 resume 값({"data": ...})을 반환합니다.
2. human_assistance 함수는 "data" 키의 값을 반환합니다.
3. ToolNode는 이 반환값을 ToolMessage로 만들어 상태에 추가합니다.
4. 제어는 다시 chatbot 노드로 넘어가고, LLM은 ToolMessage 내용을 바탕으로 최종 응답을 생성합니다.

Part 5: Customizing State (상태 사용자 정의)

애플리케이션의 특정 요구에 맞는 추가적인 정보(필드)를 그래프의 상태(State)에 정의하고 활용하는 방법을 배웁니다.

목표:

TypedDict를 사용하여 그래프의 State에 사용자 정의 필드를 추가하는 방법을 배웁니다.
도구(Tool) 내에서 이러한 사용자 정의 상태 필드를 업데이트하는 방법을 배웁니다 (Command(update=...) 사용).
상태에 저장된 추가 정보가 그래프 내 다른 노드나 외부 시스템에서 어떻게 활용될 수 있는지 이해합니다.
graph.update_state()를 사용하여 외부에서 직접 상태를 수정하는 방법을 알아봅니다.

핵심 개념: 사용자 정의 상태 (Custom State)

지금까지 State에는 messages 필드만 있었습니다. 하지만 더 복잡한 애플리케이션에서는 대화 기록 외에도 다양한 정보를 그래프의 상태로 관리해야 할 수 있습니다. 예를 들어, 사용자의 이름, 특정 작업의 진행 상태, 검색 결과에서 추출한 구조화된 데이터 등이 필요할 수 있습니다.
LangGraph에서는 TypedDict를 확장하여 필요한 만큼 필드를 State에 추가할 수 있습니다.
이렇게 추가된 상태 필드도 체크포인팅 시스템에 의해 자동으로 저장되고 관리됩니다.

코드 설명:

사용자 정의 필드가 추가된 State 정의:

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages] # 기존 메시지 리스트
    # 사용자 정의 필드 추가
    name: str     # 어떤 개체(entity)의 이름을 저장할 필드 (문자열 타입)
    birthday: str # 해당 개체의 생년월일(또는 출시일 등)을 저장할 필드 (문자열 타입)

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph.message import add_messages

class State(TypedDict):

messages: Annotated[list, add_messages] # 기존 메시지 리스트

# 사용자 정의 필드 추가

name: str # 어떤 개체(entity)의 이름을 저장할 필드 (문자열 타입)

birthday: str # 해당 개체의 생년월일(또는 출시일 등)을 저장할 필드 (문자열 타입)

기존 State 클래스에 name과 birthday라는 두 개의 str 타입 필드를 추가했습니다. 이제 이 그래프의 상태는 messages, name, birthday 세 가지 정보를 포함하게 됩니다.

상태 업데이트 기능이 포함된 도구 재정의 (human_assistance):

이전 파트의 human_assistance 도구를 수정하여, 사람의 검토를 받은 후 name과 birthday 상태 필드를 직접 업데이트하도록 변경합니다.

from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId, tool
from langgraph.types import Command, interrupt

@tool
def human_assistance(
    # 도구 인자에 name과 birthday 추가됨. LLM이 이 도구를 호출할 때 이 정보들을 채워 넣도록 유도.
    name: str,
    birthday: str,
    # tool_call_id: 도구 메시지를 생성할 때 어떤 도구 호출에 대한 응답인지 연결하기 위해 필요.
    # InjectedToolCallId: 이 인자는 LLM에게는 보이지 않도록(도구 스키마에 포함되지 않도록) 설정. LangChain이 내부적으로 주입해 줌.
    tool_call_id: Annotated[str, InjectedToolCallId]
) -&gt; str: # 반환 타입이 str이지만, 실제로는 Command 객체를 반환하여 상태 업데이트
    """Request assistance from a human."""
    human_response = interrupt(
        { # interrupt 시 사람에게 보여줄 정보 (질문과 함께 LLM이 제안한 이름/생일)
            "question": "Is this correct?",
            "name": name,
            "birthday": birthday,
        },
    )

    # 사람이 제공한 응답(human_response)을 기반으로 검증된 이름과 생일을 결정
    if human_response.get("correct", "").lower().startswith("y"): # 사람이 'yes'라고 답한 경우
        verified_name = name # LLM이 제안한 값 그대로 사용
        verified_birthday = birthday
        response = "Correct" # ToolMessage에 들어갈 내용
    else: # 사람이 수정한 경우 (예: {"name": "수정된 이름", "birthday": "수정된 생일"})
        verified_name = human_response.get("name", name) # 사람이 제공한 이름 사용 (없으면 원래값)
        verified_birthday = human_response.get("birthday", birthday) # 사람이 제공한 생일 사용 (없으면 원래값)
        response = f"Made a correction: {human_response}" # 수정 내용 포함

    # 상태 업데이트 내용을 딕셔너리로 정의
    state_update = {
        "name": verified_name,
        "birthday": verified_birthday,
        # 도구 실행 결과로 ToolMessage도 messages 리스트에 추가
        "messages": [ToolMessage(response, tool_call_id=tool_call_id)],
    }
    # Command 객체를 반환하여 상태 업데이트를 지시합니다.
    # update 키에 상태 업데이트 내용을 담은 딕셔너리를 전달합니다.
    return Command(update=state_update)

from langchain_core.messages import ToolMessage

from langchain_core.tools import InjectedToolCallId, tool

from langgraph.types import Command, interrupt

@tool

def human_assistance(

# 도구 인자에 name과 birthday 추가됨. LLM이 이 도구를 호출할 때 이 정보들을 채워 넣도록 유도.

name: str,

birthday: str,

# tool_call_id: 도구 메시지를 생성할 때 어떤 도구 호출에 대한 응답인지 연결하기 위해 필요.

# InjectedToolCallId: 이 인자는 LLM에게는 보이지 않도록(도구 스키마에 포함되지 않도록) 설정. LangChain이 내부적으로 주입해 줌.

tool_call_id: Annotated[str, InjectedToolCallId]

) -> str: # 반환 타입이 str이지만, 실제로는 Command 객체를 반환하여 상태 업데이트

"""Request assistance from a human."""

human_response = interrupt(

{ # interrupt 시 사람에게 보여줄 정보 (질문과 함께 LLM이 제안한 이름/생일)

"question": "Is this correct?",

"name": name,

"birthday": birthday,

)

# 사람이 제공한 응답(human_response)을 기반으로 검증된 이름과 생일을 결정

if human_response.get("correct", "").lower().startswith("y"): # 사람이 'yes'라고 답한 경우

verified_name = name # LLM이 제안한 값 그대로 사용

verified_birthday = birthday

response = "Correct" # ToolMessage에 들어갈 내용

else: # 사람이 수정한 경우 (예: {"name": "수정된 이름", "birthday": "수정된 생일"})

verified_name = human_response.get("name", name) # 사람이 제공한 이름 사용 (없으면 원래값)

verified_birthday = human_response.get("birthday", birthday) # 사람이 제공한 생일 사용 (없으면 원래값)

response = f"Made a correction: {human_response}" # 수정 내용 포함

# 상태 업데이트 내용을 딕셔너리로 정의

state_update = {

"name": verified_name,

"birthday": verified_birthday,

# 도구 실행 결과로 ToolMessage도 messages 리스트에 추가

"messages": [ToolMessage(response, tool_call_id=tool_call_id)],

}

# Command 객체를 반환하여 상태 업데이트를 지시합니다.

# update 키에 상태 업데이트 내용을 담은 딕셔너리를 전달합니다.

return Command(update=state_update)

Command(update=...): 도구 실행 결과로 단순히 문자열이나 객체를 반환하는 대신, Command 객체를 반환하면서 update 키에 상태를 업데이트할 내용을 딕셔너리로 전달할 수 있습니다. LangGraph는 이 Command를 받으면 해당 내용을 현재 상태에 병합(merge)하여 업데이트합니다. (messages는 add_messages 리듀서에 의해 추가되고, name과 birthday는 새 값으로 덮어쓰기됩니다.)
InjectedToolCallId: ToolMessage를 생성할 때 원본 tool_call과 연결하기 위한 ID가 필요합니다. 이 ID를 LLM이 직접 생성하거나 알 필요는 없으므로, @tool 데코레이터와 InjectedToolCallId 어노테이션을 함께 사용하면 LangChain이 실행 시점에 해당 ID를 함수 인자로 자동으로 주입해 줍니다.

그래프 설정 및 컴파일 (나머지는 동일):
- State 정의만 변경되었고, 나머지 그래프 구성(노드, 엣지, 체크포인터)은 이전과 동일하게 설정하고 컴파일합니다.

사용자 정의 상태 활용 테스트:

user_input = (
    "Can you look up when LangGraph was released? "
    "When you have the answer, use the human_assistance tool for review."
)
config = {"configurable": {"thread_id": "1"}}

# 그래프 실행 (stream)
events = graph.stream(...)
# ... 이벤트 처리 ...

user_input = (

"Can you look up when LangGraph was released? "

"When you have the answer, use the human_assistance tool for review."

)

config = {"configurable": {"thread_id": "1"}}

# 그래프 실행 (stream)

events = graph.stream(...)

# ... 이벤트 처리 ...

사용자는 LangGraph 출시일을 찾아보고, 찾으면 human_assistance 도구를 사용하여 검토해달라고 요청합니다.
LLM은 먼저 tavily_search를 호출하여 정보를 찾고, 그 결과를 바탕으로 human_assistance 도구를 호출할 때 name과 birthday 인자를 채워서 호출할 것입니다. (LLM이 잘못된 정보를 찾거나 추측할 수도 있습니다.)
human_assistance 도구 내에서 interrupt가 발생하여 실행이 멈춥니다.

사람의 개입 및 상태 업데이트 확인:

# LLM이 잘못된 정보를 찾았다고 가정하고, 사람이 올바른 정보로 수정하여 재개 명령
human_command = Command(
    resume={ # interrupt에서 반환될 값
        "name": "LangGraph",
        "birthday": "Jan 17, 2024",
        # 'correct' 키가 없거나 'y'로 시작하지 않으므로 else 구문 실행됨
    },
)

# 그래프 재개 (stream)
events = graph.stream(human_command, config, ...)
# ... 이벤트 처리 ...

# 재개 후 상태 확인
snapshot = graph.get_state(config)
{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
# 출력 결과: {'name': 'LangGraph', 'birthday': 'Jan 17, 2024'}

# LLM이 잘못된 정보를 찾았다고 가정하고, 사람이 올바른 정보로 수정하여 재개 명령

human_command = Command(

resume={ # interrupt에서 반환될 값

"name": "LangGraph",

"birthday": "Jan 17, 2024",

# 'correct' 키가 없거나 'y'로 시작하지 않으므로 else 구문 실행됨

)

# 그래프 재개 (stream)

events = graph.stream(human_command, config, ...)

# ... 이벤트 처리 ...

# 재개 후 상태 확인

snapshot = graph.get_state(config)

{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}

# 출력 결과: {'name': 'LangGraph', 'birthday': 'Jan 17, 2024'}

사람이 Command(resume=...)으로 올바른 이름과 생일 정보를 전달하며 실행을 재개합니다.
human_assistance 도구는 이 정보를 받아 Command(update=...)를 반환합니다.
LangGraph는 이 update 내용을 상태에 반영합니다.
get_state()를 통해 확인하면 name과 birthday 필드가 사람에 의해 수정된 값으로 업데이트된 것을 볼 수 있습니다.

외부에서 상태 수동 업데이트 (graph.update_state):

graph.update_state(config, {"name": "LangGraph (library)"})
# 'update_state' 메서드를 사용하면 그래프 실행 외부에서도 특정 스레드(config)의 상태를 직접 수정할 수 있습니다.
# 여기서는 'name' 필드의 값을 변경합니다.

snapshot = graph.get_state(config)
{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
# 출력 결과: {'name': 'LangGraph (library)', 'birthday': 'Jan 17, 2024'}
# 변경된 값이 상태에 반영된 것을 확인할 수 있습니다.

graph.update_state(config, {"name": "LangGraph (library)"})

# 'update_state' 메서드를 사용하면 그래프 실행 외부에서도 특정 스레드(config)의 상태를 직접 수정할 수 있습니다.

# 여기서는 'name' 필드의 값을 변경합니다.

snapshot = graph.get_state(config)

{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}

# 출력 결과: {'name': 'LangGraph (library)', 'birthday': 'Jan 17, 2024'}

# 변경된 값이 상태에 반영된 것을 확인할 수 있습니다.

update_state(config, values): 특정 대화 스레드(config)의 상태(values 딕셔너리에 지정된 필드)를 강제로 업데이트합니다. 이는 디버깅이나 특정 시나리오에서 상태를 직접 조작해야 할 때 유용합니다. 이 업데이트도 LangSmith 추적에 기록됩니다.

Part 6: Time Travel (시간 여행)

체크포인팅 시스템을 활용하여 과거의 특정 시점으로 돌아가 실행을 재개하는 방법을 배웁니다.

LangGraph의 체크포인팅이 단순히 마지막 상태만 저장하는 것이 아니라, 실행 히스토리 전체를 기록한다는 것을 이해합니다.

이 “시간 여행” 기능이 디버깅, 실험, 사용자 경험 개선(예: 실수 되돌리기, 다른 경로 탐색) 등에 어떻게 활용될 수 있는지 이해합니다.

저스틴) 꽤 오랜 기간 대화한 내용이라면 이렇게 기억하는 것은 너무 많은 양을 기억하는거 아닌가?

기본 그래프 설정 (Part 3의 도구 + 메모리 챗봇 재사용):

이 파트에서는 시간 여행 기능을 명확히 보여주기 위해, Part 3에서 만들었던 비교적 간단한 도구 사용 및 메모리 기능이 있는 챗봇 그래프 코드를 다시 사용합니다. (State에는 messages 필드만 있음)

# Part 3 코드와 동일 (State, Tools, LLM, Nodes, Edges, Checkpointer, Compile)
# ... (코드 생략) ...
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# Part 3 코드와 동일 (State, Tools, LLM, Nodes, Edges, Checkpointer, Compile)

# ... (코드 생략) ...

memory = MemorySaver()

graph = graph_builder.compile(checkpointer=memory)

여러 단계의 대화 실행 (히스토리 생성):

챗봇과 여러 턴에 걸쳐 대화를 진행하여 체크포인트 히스토리를 만듭니다.

config = {"configurable": {"thread_id": "1"}}

# 첫 번째 사용자 입력 및 그래프 실행
user_input_1 = "I'm learning LangGraph. Could you do some research on it for me?"
events = graph.stream({"messages": [...]}, config, stream_mode="values")
# ... (이벤트 처리 및 출력) ...
# 이 과정에서 여러 체크포인트 생성됨 (START -&gt; chatbot -&gt; tools -&gt; chatbot -&gt; END)

# 두 번째 사용자 입력 및 그래프 실행 (동일한 config 사용)
user_input_2 = "Ya that's helpful. Maybe I'll build an autonomous agent with it!"
events = graph.stream({"messages": [...]}, config, stream_mode="values")
# ... (이벤트 처리 및 출력) ...
# 추가적인 체크포인트 생성됨 (chatbot -&gt; tools -&gt; chatbot -&gt; END)

config = {"configurable": {"thread_id": "1"}}

# 첫 번째 사용자 입력 및 그래프 실행

user_input_1 = "I'm learning LangGraph. Could you do some research on it for me?"

events = graph.stream({"messages": [...]}, config, stream_mode="values")

# ... (이벤트 처리 및 출력) ...

# 이 과정에서 여러 체크포인트 생성됨 (START -> chatbot -> tools -> chatbot -> END)

# 두 번째 사용자 입력 및 그래프 실행 (동일한 config 사용)

user_input_2 = "Ya that's helpful. Maybe I'll build an autonomous agent with it!"

events = graph.stream({"messages": [...]}, config, stream_mode="values")

# ... (이벤트 처리 및 출력) ...

# 추가적인 체크포인트 생성됨 (chatbot -> tools -> chatbot -> END)

thread_id="1"에 대해 두 번의 stream 호출을 통해 여러 노드가 실행되었고, 각 노드 실행 후의 상태가 memory 체크포인터에 순서대로 기록되었습니다.

상태 히스토리 조회 (get_state_history):

to_replay = None
# get_state_history(config) 메서드를 호출하여 해당 스레드(thread_id="1")의 모든 상태 스냅샷(StateSnapshot)을 가져옵니다.
# 결과는 가장 최근 상태부터 오래된 순서대로 이터레이터(iterator) 형태로 반환됩니다.
for state in graph.get_state_history(config):
    # 각 상태 스냅샷의 정보 출력 (메시지 개수, 다음에 실행될 노드 이름)
    print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)
    print("-" * 80)
    # 예시: 특정 조건(메시지 개수가 6개인 상태)에 맞는 상태를 'to_replay' 변수에 저장
    if len(state.values["messages"]) == 6:
        to_replay = state

to_replay = None

# get_state_history(config) 메서드를 호출하여 해당 스레드(thread_id="1")의 모든 상태 스냅샷(StateSnapshot)을 가져옵니다.

# 결과는 가장 최근 상태부터 오래된 순서대로 이터레이터(iterator) 형태로 반환됩니다.

for state in graph.get_state_history(config):

# 각 상태 스냅샷의 정보 출력 (메시지 개수, 다음에 실행될 노드 이름)

print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)

print("-" * 80)

# 예시: 특정 조건(메시지 개수가 6개인 상태)에 맞는 상태를 'to_replay' 변수에 저장

if len(state.values["messages"]) == 6:

to_replay = state

graph.get_state_history(config): 특정 config(주로 thread_id 기준)에 대한 모든 과거 체크포인트(StateSnapshot 객체)를 최신순으로 반환합니다.
StateSnapshot 객체에는 해당 시점의 상태 값(values), 다음에 실행될 노드(next), 그리고 해당 스냅샷을 식별하는 config(여기에는 고유한 checkpoint_id 포함) 정보가 들어있습니다.
튜토리얼에서는 임의의 조건(메시지 6개)을 만족하는 과거 상태(to_replay)를 선택합니다. 이 상태는 두 번째 stream 호출 중 chatbot 노드가 도구 호출(tool_calls)을 생성한 직후, tools 노드가 실행되기 전의 상태입니다. (Next: ('tools',) 로 확인 가능)

선택한 과거 시점 정보 확인:

print(to_replay.next)
# 출력: ('tools',) - 이 상태 다음에는 'tools' 노드가 실행될 예정이었음을 의미
print(to_replay.config)
# 출력: {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1efd43e3-...'}}
# 이 상태 스냅샷에 해당하는 고유한 checkpoint_id를 확인합니다.

print(to_replay.next)

# 출력: ('tools',) - 이 상태 다음에는 'tools' 노드가 실행될 예정이었음을 의미

print(to_replay.config)

# 출력: {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1efd43e3-...'}}

# 이 상태 스냅샷에 해당하는 고유한 checkpoint_id를 확인합니다.

to_replay.config에 포함된 checkpoint_id는 우리가 돌아가고 싶은 정확한 과거 시점을 가리킵니다.

과거 시점에서 실행 재개 (시간 여행!):

# graph.stream() 메서드를 호출할 때, 입력 데이터 대신 'None'을 전달하고,
# config 인자로는 과거 상태 스냅샷에서 가져온 'to_replay.config' (특정 checkpoint_id 포함)를 전달합니다.
for event in graph.stream(None, to_replay.config, stream_mode="values"):
    if "messages" in event:
        event["messages"][-1].pretty_print()

# graph.stream() 메서드를 호출할 때, 입력 데이터 대신 'None'을 전달하고,

# config 인자로는 과거 상태 스냅샷에서 가져온 'to_replay.config' (특정 checkpoint_id 포함)를 전달합니다.

for event in graph.stream(None, to_replay.config, stream_mode="values"):

if "messages" in event:

event["messages"][-1].pretty_print()

핵심: stream (또는 invoke) 메서드의 두 번째 인자로 특정 checkpoint_id가 포함된 config 객체를 전달하는 것입니다.
LangGraph는 이 checkpoint_id를 보고 해당 시점의 상태를 체크포인터에서 로드합니다.
그리고 to_replay.next에 지정된 노드(tools 노드)부터 실행을 재개합니다.
실행 결과 출력: 첫 번째로 출력되는 것이 Tool Message (검색 결과)인 것을 볼 수 있습니다. 이는 chatbot 노드를 건너뛰고 바로 tools 노드가 실행되었음을 의미합니다. 즉, 성공적으로 과거 시점에서 실행을 재개한 것입니다.