LangGraph 챗봇에 메모리 추가하기

챗봇에 메모리 추가하기는 챗봇이 이전 대화의 맥락을 기억하여 일관된 다중 턴 대화를 가능하게 하는 방법을 다룹니다.
이전까지의 챗봇은 도구를 활용하여 사용자 질문에 답변할 수 있었지만, 이전 대화의 맥락을 유지하지 못해 연속적인 대화에 한계가 있었습니다.
LangGraph는 지속적인 체크포인팅(persistent checkpointing) 기능을 통해 이 문제를 해결합니다.

학습 목표

  • 챗봇에 지속적인 체크포인팅 기능 추가하기
  • 이전 대화 맥락을 유지하는 다중 턴 대화 구현하기
  • LangGraph 내 메모리 기능 활용법 학습하기

1. 환경 설정

환경변수 설정

환경변수 파일 .env를 생성하여 다음의 내용을 설정합니다.

OPENAI_API_KEY=본인의_OpenAI_API키
OPENAI_MODEL=gpt-5-nano
TAVILY_API_KEY=본인의_tavily_api_key

2. 코드 설명

환경변수 로딩

from dotenv import load_dotenv
import os
load_dotenv()

openai_model = os.getenv("OPENAI_MODEL", "gpt-5-nano")

도구 정의

Tavily 검색 도구를 설정합니다.

from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=2)
tool.invoke("LangGraph가 무엇인가요?")

상태(State) 정의

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

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

이 상태 정의를 통해 챗봇은:

  • 대화의 각 단계를 명확히 관리할 수 있습니다.
  • 사용자의 메시지와 모델의 응답 메시지를 체계적으로 기록하고 관리합니다.
  • 명시적인 타입과 자동 관리 기능을 활용하여 코드의 오류를 최소화하고 유지보수를 쉽게 합니다.

이러한 방식을 통해, LangGraph는 상태 기반의 복잡한 챗봇 애플리케이션에서도 효율적이고 명확한 데이터 관리와 흐름 제어를 가능하게 합니다.

LLM 모델 설정 및 도구 바인딩

from langchain_openai import ChatOpenAI 

llm = ChatOpenAI(model=openai_model)
tools = [tool]
llm_with_tools = llm.bind_tools(tools)

챗봇 노드 정의

챗봇이 도구를 이용하여 사용자 메시지에 응답하도록 설정합니다.

def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}
  • chatbot(state: State): 챗봇 노드 함수로, 상태에서 받은 메시지를 기반으로 도구를 활용하여 응답을 생성하고 상태를 업데이트합니다.

체크포인터 설정

체크포인팅을 위한 메모리 체크포인터를 생성합니다.

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
  • MemorySaver: 메모리 기반의 체크포인터로, 각 대화의 상태를 메모리에 임시로 저장하고 관리합니다.
    이를 통해 챗봇은 이전 대화 내용을 기억하고 다음 번 상호작용 시에도 맥락을 유지한 상태로 대화를 진행할 수 있습니다.
    실제 운영 환경에서는 더 영구적인 상태 관리를 위해 데이터베이스 기반 체크포인터(예: SqliteSaver 또는 PostgresSaver)를 사용하는 것이 권장됩니다.

그래프 구성 및 컴파일

LangGraph의 ToolNode와 tools_condition을 사용하여 조건부로 도구 노드를 호출합니다.

from langgraph.prebuilt import ToolNode, tools_condition

tool_node = ToolNode(tools)

workflow = StateGraph(State)

workflow.add_node("chatbot", chatbot)
workflow.add_node("tools", tool_node)

workflow.add_conditional_edges("chatbot", tools_condition)
workflow.add_edge("tools", "chatbot")
workflow.set_entry_point("chatbot")

graph = workflow.compile(checkpointer=memory)

ToolNode는 LangGraph에서 미리 만들어져 제공되는 도구 실행 전용 노드입니다.

  • LLM이 도구(tool)를 호출하겠다고 요청할 때, 실제 도구를 실행하는 역할을 합니다.
  • 도구 호출 요청 메시지를 입력으로 받아 도구를 실행하고, 그 결과를 다시 LLM에 전달할 수 있는 형태로 반환합니다.
  • 사용자가 별도의 복잡한 로직을 구현하지 않고, 간단하게 도구를 처리할 수 있도록 미리 만들어진 클래스입니다.
  • ToolNode 생성시 도구를 리스트 형태로 전달합니다.

작동 원리

ToolNode의 내부 구조는 다음과 같은 순서로 동작합니다:

  1. 챗봇(LLM) 노드가 도구 호출을 요청한 메시지를 생성합니다.
  2. ToolNode가 이 메시지를 입력으로 받아 메시지에 포함된 도구 호출(tool_calls)을 실행합니다.
  3. 도구 실행 결과를 ToolMessage 형태로 변환하여 다시 챗봇(LLM) 노드가 이해할 수 있게 반환합니다.

tools_condition은 LangGraph에서 미리 만들어 제공하는 조건부 엣지(conditional edge)를 위한 조건 함수입니다.

  • 챗봇 노드의 출력 메시지에 도구 호출 요청이 포함되어 있는지 확인합니다.
  • 도구 호출 요청이 있다면, 도구 노드(ToolNode)를 실행하도록 조건을 설정해주는 기능입니다.
  • 별도로 조건 로직을 구현하지 않아도, 도구 호출의 필요성을 쉽게 판단할 수 있도록 미리 만들어진 편의 함수입니다.

작동 원리

tools_condition 함수는 다음 로직을 수행합니다:

  1. 입력 상태에서 가장 최근 메시지를 확인합니다.
  2. 최근 메시지에 tool_calls가 포함되어 있는지 확인합니다.
  3. 만약 메시지에 하나 이상의 도구 호출 요청이 존재하면(tool_calls가 존재하면), 다음 노드를 'tools'로 설정합니다.
  4. 도구 호출 요청이 없다면 자동으로 종료(END) 노드로 설정하여 그래프를 종료합니다.

그래프 시각화

컴파일된 그래프를 이용해 시각화봅니다.

from IPython.display import Image, display

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

3. 챗봇 실행

동일한 thread_id를 사용하여 이전 대화 맥락을 유지하는 예시입니다. 이 예시에서는 두 개의 서로 다른 thread_id를 사용하여 두 개의 독립된 대화를 관리하는 방법을 보여줍니다.

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

config는 LangGraph의 그래프 실행 시 설정을 정의하는 딕셔너리입니다. 여기서 configurable 키는 실행 시 동적으로 설정할 수 있는 옵션을 포함합니다.

  • thread_id: 대화의 고유 식별자 역할을 합니다.
  • LangGraph는 상태 기반 그래프(StateGraph)를 사용하여 대화를 관리합니다.
  • thread_id는 각 대화의 상태를 구분하는 데 사용됩니다.
  • 동일한 thread_id를 사용하면 이전 대화의 맥락을 유지하며 대화를 이어갈 수 있습니다.
  • 서로 다른 thread_id를 사용하면 독립된 대화를 관리할 수 있습니다.

예를 들어:

  • thread_id가 “user123”인 경우, 해당 대화의 상태를 기반으로 응답을 생성합니다.
  • 새로운 thread_id를 지정하면 이전 대화와는 별개의 새로운 대화가 시작됩니다.

이 설정은 LangGraph의 체크포인팅(checkpointing) 기능과 결합하여 다중 턴 대화에서 맥락을 유지하거나 독립적인 대화를 관리하는 데 유용합니다.

from langchain_core.messages import HumanMessage

# 첫 번째 대화: 새로운 대화 맥락 생성
user_input1 = "LangGraph가 무엇인가요?"
state1 = {"messages": [HumanMessage(content=user_input1)]}
response1 = graph.invoke(state1, config)

# 챗봇의 첫 번째 응답 출력
print(response1["messages"][-1].content)

# 두 번째 대화: 이전 대화 맥락 유지
user_input2 = "그것을 만든 회사는 어딘가요?"
state2 = {"messages": [HumanMessage(content=user_input2)]}
response2 = graph.invoke(state2, config)

# 챗봇의 두 번째 응답 출력 (이전 대화의 맥락이 유지된 상태)
print(response2["messages"][-1].content)

# 세 번째 대화: 새로운 thread_id 사용하여 독립된 새로운 대화 맥락 생성
user_input3 = "그것을 만든 회사는 어딘가요?"
state3 = {"messages": [HumanMessage(content=user_input3)]}
response3 = graph.invoke(state3, {"configurable": {"thread_id": "user456"}})

# 새로운 대화 맥락에서의 챗봇 응답 출력 (기존 thread_id와 독립된 상태)
print(response3["messages"][-1].content)

# 기존 thread_id의 대화 상태 확인
snapshot = graph.get_state(config)
snapshot.values['messages']

# 새 thread_id의 대화 상태 확인
graph.get_state({"configurable": {"thread_id": "user456"}}).values['messages']

코드 실행 흐름

  • 첫 번째 대화는 새로운 대화 맥락을 생성하며, 질문에 대한 답변을 얻습니다.
  • 두 번째 대화는 첫 번째 대화와 동일한 thread_id를 사용하여 이전의 대화 맥락을 유지하고, 이어지는 질문에 답변을 얻습니다.
  • 세 번째 대화는 별도의 새로운 thread_id를 사용하여 첫 번째, 두 번째 대화와 독립적인 새로운 대화 맥락을 형성합니다.
  • 마지막으로 각각의 thread_id로 관리되는 대화 상태를 조회하여, 상태가 독립적으로 관리되고 있음을 확인합니다.
from pprint import pprint

# 체크포인터가 MemorySaver일 경우 예시
all_snapshots = memory.list({})  # 모든 thread_id 상태 조회

for snapshot in all_snapshots:
    print(snapshot.config['configurable']['thread_id'])
    if 'messages' in snapshot.checkpoint['channel_values']:
        pprint(snapshot.checkpoint['channel_values']['messages'])
    else:
        print("No messages found in this snapshot.")