[langgraph] 메모리 추가하기

원문 출처: https://langchain-ai.github.io/langgraph/tutorials/get-started/3-add-memory/

이제 챗봇이 도구를 사용해 사용자 질문에 답할 수 있지만, 이전 대화의 맥락을 기억하지 못합니다. 이로 인해 여러 턴에 걸친 자연스러운 대화가 어렵습니다.

LangGraph는 지속적 체크포인팅(persistent checkpointing) 기능으로 이 문제를 해결합니다. 그래프를 컴파일할 때 checkpointer를 제공하고, 그래프를 호출할 때 thread_id를 지정하면 LangGraph가 각 단계마다 상태를 자동으로 저장합니다. 같은 thread_id로 다시 호출하면 저장된 상태를 불러와 대화를 이어갈 수 있습니다.

체크포인팅은 단순한 챗 메모리보다 훨씬 강력합니다. 언제든 복잡한 상태를 저장/복원할 수 있어 오류 복구, human-in-the-loop, 타임트래블 등 다양한 워크플로우에 활용할 수 있습니다. 여기서는 멀티턴 대화를 위한 체크포인팅만 다룹니다.

이 튜토리얼은 도구 추가하기를 기반으로 합니다.

1. `MemorySaver` 체크포인터 생성

메모리 기반 체크포인터를 생성합니다:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

튜토리얼에서는 인메모리 체크포인터를 사용하지만, 실제 서비스에서는 SqliteSaver나 PostgresSaver 등 데이터베이스 연동을 권장합니다.

2. 그래프 컴파일

체크포인터를 지정해 그래프를 컴파일하면, 각 노드 실행 후 상태가 자동 저장됩니다:

graph = graph_builder.compile(checkpointer=memory)

그래프 시각화 예시:

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # 추가 종속성이 필요하며 선택 사항입니다
    pass

3. 챗봇과 대화하기

이제 챗봇과 대화해봅시다!

대화의 키로 사용할 thread를 정합니다.

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

챗봇 호출:

user_input = "Hi there! My name is Will."
# config는 stream() 또는 invoke()의 두 번째 인수입니다!
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print()

출력 예시:

================================ Human Message =================================
Hi there! My name is Will.
================================== Ai Message ==================================
Hello Will! It's nice to meet you. How can I assist you today? Is there anything specific you'd like to know or discuss?

config는 반드시 입력 딕셔너리({'messages': []})와 별개로 두 번째 인수로 전달해야 합니다.

4. 후속 질문하기

user_input = "Remember my name?"
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print()

출력 예시:

================================ Human Message =================================
Remember my name?
================================== Ai Message ==================================
Of course, I remember your name, Will. I always try to pay attention to important details that users share with me. Is there anything else you'd like to talk about or any questions you have? I'm here to help with a wide range of topics or tasks.

외부 리스트 없이도 메모리가 동작합니다! 체크포인터가 모든 상태를 관리합니다.

다른 thread_id로 시도해보세요:

# thread_id만 "2"로 변경
user_input = "Remember my name?"
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    {"configurable": {"thread_id": "2"}},
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print()

출력 예시:

================================ Human Message =================================
Remember my name?
================================== Ai Message ==================================
I apologize, but I don't have any previous context or memory of your name. As an AI assistant, I don't retain information from past conversations. Each interaction starts fresh. Could you please tell me your name so I can address you properly in this conversation?

thread_id만 바꿔도 완전히 새로운 대화가 시작됩니다.

5. 상태 확인하기

여러 thread에서 체크포인트가 쌓였습니다. 특정 config의 상태를 언제든 확인할 수 있습니다:

snapshot = graph.get_state(config)
snapshot

출력 예시:

StateSnapshot(values={'messages': [HumanMessage(content='Hi there! My name is Will.', ...), AIMessage(content="Hello Will! It's nice to meet you. ..."), HumanMessage(content='Remember my name?', ...), AIMessage(content="Of course, I remember your name, Will. ...")]}, ...)

snapshot.next를 통해 다음 실행될 노드도 확인할 수 있습니다.

축하합니다! 이제 챗봇이 LangGraph의 체크포인팅 시스템 덕분에 대화 상태를 세션별로 유지할 수 있습니다. 이 기능은 자연스럽고 맥락 있는 상호작용을 가능하게 하며, 단순한 챗 메모리보다 훨씬 강력합니다.

아래는 튜토리얼 전체 코드 예시입니다:

from typing import Annotated
from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

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

graph_builder = StateGraph(State)

tool = TavilySearch(max_results=2)
tools = [tool]
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)

tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

다음 튜토리얼에서는 human-in-the-loop 기능을 추가해, 챗봇이 진행 전 사용자 확인이나 안내가 필요한 상황을 처리하는 방법을 다룹니다.