Part 3: LangGraph 메모리 관리 - 대화 기억하기 🧠

Part 3: LangGraph 메모리 관리 - 대화 기억하기 🧠

이번 글에서는 LangGraph에 메모리 기능을 추가하여 이전 대화 내용을 기억하는 챗봇을 구현하는 방법을 설명한다.

1. 메모리 기능 구현을 위한 코드 설정 🛠️

import os
from dotenv import load_dotenv
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver  # 메모리 기능 추가

load_dotenv()

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

2. 메모리 저장소 설정 💾

def setup_graph():
    # 메모리 저장소 초기화
    memory = MemorySaver()

    # 그래프 설정
    graph_builder = StateGraph(State)

    # AI 모델과 도구 설정
    tool = TavilySearchResults(max_results=2)
    tools = [tool]
    llm = ChatOpenAI(
        model="gpt-4",
        temperature=0.7,
        streaming=True
    )
    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)
    graph_builder.add_node("tools", ToolNode(tools=[tool]))
    graph_builder.add_conditional_edges("chatbot", tools_condition)
    graph_builder.add_edge("tools", "chatbot")
    graph_builder.add_edge(START, "chatbot")

    # 메모리 기능을 포함하여 컴파일
    return graph_builder.compile(checkpointer=memory)

3. 대화별 구분을 위한 테스트 함수 🧪

def test_chatbot(graph, question: str, thread_id: str = "default"):
    print("\n" + "="*50)
    print(f"😀 사용자: {question}")
    print(f"🧵 대화 ID: {thread_id}")
    print("="*50)

    try:
        # thread_id를 포함한 설정 추가
        config = {"configurable": {"thread_id": thread_id}}
        for event in graph.stream(
            {"messages": [("human", question)]},
            config  # 설정 추가
        ):
            for value in event.values():
                if "messages" in value:
                    message = value["messages"][-1]
                    if hasattr(message, "content"):
                        print(f"\n🤖 AI: {message.content}")
    except Exception as e:
        print(f"\n❌ 오류 발생: {str(e)}")

4. 다중 대화 테스트 실행 ▶️

def main():
    graph = setup_graph()
    print("✅ 챗봇 준비 완료!\n")

    # 첫 번째 대화 (thread_id: conversation_1)
    print("\n🗣️ 첫 번째 대화 시작")
    thread_1 = "conversation_1"
    questions_1 = [
        "내 이름은 철수야",
        "내 이름이 뭐였지?",
        "나는 학생이야",
        "내가 뭐라고 했었지?"
    ]

    for question in questions_1:
        test_chatbot(graph, question, thread_1)
        print("\n" + "-"*50)

    # 두 번째 대화 (thread_id: conversation_2)
    print("\n🗣️ 두 번째 대화 시작")
    thread_2 = "conversation_2"
    questions_2 = [
        "안녕! 내 이름은 영희야",
        "내 이름이 뭐였지?",
        "나는 선생님이야",
        "내가 뭐라고 했었지?"
    ]

    for question in questions_2:
        test_chatbot(graph, question, thread_2)
        print("\n" + "-"*50)

5. 각 대화의 상태 확인 방법 📊

def check_conversation_state(graph, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = graph.get_state(config)

    print(f"\n📝 대화 {thread_id}의 현재 상태:")
    for message in state.values["messages"]:
        if hasattr(message, "content"):
            print(f"- {message.content}")

6. 실행 결과 예시 📝

🗣️ 첫 번째 대화 시작
==================================================
😀 사용자: 내 이름은 철수야
🧵 대화 ID: conversation_1
==================================================
🤖 AI: 안녕하세요, 철수님! 만나서 반갑습니다.

==================================================
😀 사용자: 내 이름이 뭐였지?
🧵 대화 ID: conversation_1
==================================================
🤖 AI: 네, 철수님이세요! 앞서 말씀해주신 것을 기억하고 있습니다.

🗣️ 두 번째 대화 시작
==================================================
😀 사용자: 안녕! 내 이름은 영희야
🧵 대화 ID: conversation_2
==================================================
🤖 AI: 안녕하세요, 영희님! 반갑습니다.

7. 주요 포인트 정리 ⭐

1. 메모리 관리
   • MemorySaver로 대화 내용을 저장한다
   • thread_id로 각 대화를 구분한다
   • 상태 정보를 자동으로 체크포인트한다
2. 대화 구분
   • 각 대화는 독립적으로 관리된다
   • 서로 다른 thread_id는 서로의 내용을 볼 수 없다
3. 상태 유지
   • 대화가 끝나도 상태가 유지된다
   • 나중에 같은 thread_id로 대화 재개 가능

8. 활용 예시 💡

1. 다중 사용자 지원
2. user_threads = { "user1": "thread_1", "user2": "thread_2" }
3. 대화 맥락 활용
4. # 이전 대화 내용 참조 test_chatbot(graph, "이전에 내가 한 말 기억나?", thread_id)

이렇게 LangGraph의 메모리 기능을 활용하여 맥락을 유지하는 챗봇을 구현할 수 있다. 다음 글에서는 사람과 AI의 협업을 위한 중단점 설정 방법을 알아본다. 🚀

 

참고 : https://langchain-ai.github.io/langgraph/tutorials/introduction/#part-3-adding-memory-to-the-chatbot

Learn the basics

참고 : https://github.com/junyong1111/LangGraph/tree/main/example3

LangGraph/example3 at main · junyong1111/LangGraph