LangGraph #4 - Memory Chatbot

이 글에서는 LangGraph 라이브러리의 체크포인트 기능을 이용해, 사용자별로 대화 맥락을 유지할 수 있는 챗봇을 구현해보겠습니다. Anthropic Claude 모델을 활용하여 자연스럽게 대화를 이어가면서, thread_id를 통해 서로 다른 사용자의 대화를 독립적으로 관리하는 방법을 시연합니다.

1. 배경: 왜 멀티턴 메모리가 필요한가?

지금까지 단순 챗봇을 구성해본 분들은, 한 번의 질의-응답 이후 맥락이 사라지거나, 프로그램이 재시작되면 이전 대화 내용을 잊어버리는 문제를 종종 겪었을 것입니다.

• 예:
  • “내 이름이 홍길동이야” → “반갑습니다, 홍길동님”
  • …(대화 종료 후) 다시 시작하면 홍길동이라는 사실을 잊어버림.

LangGraph는 이런 문제를 해결하기 위해 **체크포인트(Checkpoint)**라는 개념을 도입했습니다. 체크포인트는 노드가 실행된 후의 상태(State)를 자동으로 저장해두고, 동일 thread_id로 그래프를 다시 호출하면 그 상태를 자동 로드하여 이어갈 수 있도록 해줍니다.

2. 코드 예시

다음 코드를 ch4_memory_chatbot.py라 하고, python ch4_memory_chatbot.py로 실행하면 됩니다.

import os

# Anthropic Claude
from langchain_anthropic import ChatAnthropic

# LangGraph 핵심
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages

# 체크포인트
from langgraph.checkpoint.memory import MemorySaver
# (DB 사용 시 SqliteSaver, PostgresSaver 등 사용 가능)

from typing import Annotated
from typing_extensions import TypedDict

###############################################################################
# 1) 상태(State) 정의
###############################################################################
class State(TypedDict):
    messages: Annotated[list, add_messages]

###############################################################################
# 2) 그래프 빌더 생성
###############################################################################
graph_builder = StateGraph(State)

###############################################################################
# 3) LLM
###############################################################################
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    anthropic_api_key=os.environ.get("ANTHROPIC_API_KEY")
)

def chatbot_node(state: State):
    """
    state["messages"]를 LLM에게 전달하고,
    새 AI 응답을 반환.
    """
    ai_resp = llm.invoke(state["messages"])
    return {"messages": [ai_resp]}

# 노드 등록
graph_builder.add_node("chatbot", chatbot_node)
# START -> chatbot
graph_builder.add_edge(START, "chatbot")

###############################################################################
# 4) 체크포인터 설정
###############################################################################
memory = MemorySaver()

###############################################################################
# 5) 그래프 컴파일
###############################################################################
# compile(...) 시 checkpointer 지정
graph = graph_builder.compile(checkpointer=memory)

################################################################################
# 그래프 시각화를 위한 ASCII 아트 생성
################################################################################
# - graph.get_graph().draw_ascii()를 호출하여 그래프 구조를 ASCII 아트로 표현
# - 노드와 엣지의 연결 관계를 시각적으로 보여줌

ascii_graph = graph.get_graph().draw_ascii()

print(ascii_graph)

###############################################################################
# 6) 메인 함수
###############################################################################
def main():
    print("==== [ch4] Memory/Checkpoint Demo ====")
    print("Type 'quit' or 'q' to exit.\n")

    # 각 유저별 thread_id를 가정
    thread_id = input("Enter thread_id (e.g., 'user1'): ").strip() or "default"

    while True:
        user_input = input(f"(thread={thread_id}) You: ")
        if user_input.lower() in ["quit", "q"]:
            print("Goodbye!")
            break

        # graph 실행:
        # (1) input: {"messages": [("user", user_input)]}
        # (2) config: {"configurable": {"thread_id": thread_id}}
        # => LangGraph가 thread_id='...' 체크포인트 상태를 로드 & 업데이트
        events = graph.stream(
            {"messages": [("user", user_input)]},
            {"configurable": {"thread_id": thread_id}},
        )

        # 실행 중 생기는 event를 출력
        for event in events:
            for node_out in event.values():
                if "messages" in node_out:
                    last_msg = node_out["messages"][-1]
                    print("Assistant:", last_msg.content)

if __name__ == "__main__":
    main()

3. 코드 흐름 분석

State(TypedDict) 정의

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

• 챗봇 대화에는 사용자 메시지와 AI(Claude) 메시지를 순서대로 저장하는 messages 리스트가 필요합니다.
• add_messages를 사용하면 새 메시지를 추가할 때마다 기존 리스트에 “append” 방식으로 쌓입니다.

그래프 빌더(StateGraph)

• graph_builder = StateGraph(State)로 그래프 설계자(Builder)를 생성합니다.
• 노드와 엣지를 등록한 뒤 compile하여 실행 가능한 그래프 객체를 만들어냅니다.

LLM 노드(chatbot_node)

def chatbot_node(state: State):
    ai_resp = llm.invoke(state["messages"])
    return {"messages": [ai_resp]}

• state["messages"]를 Anthropic Claude에 전달해 새 AI 응답을 생성하는 노드입니다.
• 반환되는 AI 메시지가 messages 리스트에 append되어 대화 내역이 누적됩니다.

체크포인터(MemorySaver)

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

• 이 설정 덕분에 LangGraph가 노드 실행 때마다 상태(State)를 memory에 저장합니다.
• 만약 동일한 thread_id로 다시 그래프를 실행하면, 이전에 저장된 상태를 자동 로드해 여러 턴 대화를 이어갈 수 있습니다.

메인 함수

• thread_id를 입력 받음: 예) user1
• 사용자 메시지를 graph.stream(...)으로 전달할 때,로 지정하면 LangGraph가 해당 스레드의 상태를 로드/업데이트합니다.

{"configurable": {"thread_id": thread_id}}

4. 실행 예시

한 터미널에서:

$ python ch4_memory_chatbot.py
+-----------+  
| __start__ |  
+-----------+  
     *        
     *        
     *        
+---------+   
| chatbot |   
+---------+   
==== [ch4] Memory/Checkpoint Demo ====
Type 'quit' or 'q' to exit.

Enter thread_id (e.g., 'user1'): userA
(thread=userA) You: 안녕
Assistant: 안녕하세요! 무엇을 도와드릴까요?
(thread=userA) You: 내 이름은 길동이야
Assistant: 길동님, 만나서 반갑습니다. ...

또 다른 터미널에서:

$ python ch4_memory_chatbot.py
==== [ch4] Memory/Checkpoint Demo ====
Type 'quit' or 'q' to exit.

Enter thread_id (e.g., 'user1'): userB
(thread=userB) You: 안녕, 난 영희야
Assistant: 영희님, 안녕하세요!

코드 복사

• userA와 userB는 서로 다른 thread_id로 대화를 유지하므로, 서로의 대화가 섞이지 않고 독립된 맥락을 이어갑니다.
• 만약 같은 thread_id로 접속하면, LangGraph는 “이전에 저장된 State가 있다”고 보고 과거 메시지를 이어서 참조합니다(프로세스를 종료하기 전까지).

5. 확장 아이디어

SqliteSaver / PostgresSaver

• MemorySaver는 파이썬 프로세스를 종료하면 저장된 상태가 사라집니다.
• SqliteSaver나 PostgresSaver를 쓰면 DB에 저장되므로,
  프로그램을 재시작해도 동일 thread_id의 대화를 이어갈 수 있습니다.

복수 노드(툴 등) 결합

• ch2, ch3처럼 툴(검색, DB 조회) 노드를 추가하면,
  사용자별로 검색 내역도 messages에 누적되어 “이전에 검색했던 내용”까지 기억하는 챗봇이 됩니다.

Human-in-the-loop

• 특정 노드 실행 전/후에 interrupt_before/interrupt_after를 설정해, “사람이 승인해야 다음 단계로 진행”처럼 구현할 수도 있습니다.
• 이때도 체크포인트가 있어야, 중간에 멈췄다가 몇 분/몇 시간 뒤에 다시 재개 가능합니다.

마무리

이상으로, LangGraph의 체크포인트 기능을 활용해 여러 턴 대화(멀티턴)와 사용자별(thread_id 기반) 대화 맥락 유지를 구현하는 방법을 살펴봤습니다.

• 핵심 포인트:
  1. State에 messages 리스트를 넣어두고,
  2. compile(checkpointer=...)로 그래프 실행 결과를 저장,
  3. graph.stream(..., {"configurable": {"thread_id": "..."}})로 이전 기록을 로드.

이를 통해 간단히 사용자별 맞춤형 기억을 유지할 수 있으며, 좀 더 고도화하면 DB 연동으로 서버 재시작에도 대화 이력이 남도록 만들 수 있습니다.