langgraph / / 2024. 11. 29. 17:23

[langgraph] 그래프에 여러 스레드 간 영속성을 추가하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/cross-thread-persistence/

이전 가이드에서는 단일 스레드에서 여러 상호작용에 걸쳐 그래프 상태를 유지하는 방법을 알아봤다. LangGraph는 또한 여러 스레드에 걸쳐 데이터를 유지할 수 있다. 예를 들어, 사용자의 이름이나 선호도와 같은 정보를 공유 메모리에 저장하고 새로운 대화 스레드에서 재사용할 수 있다.

이 가이드에서는 Store 인터페이스를 사용하여 구현된 공유 메모리를 가진 그래프를 구성하고 사용하는 방법을 보여준다.

준비

필요한 패키지를 설치한다.

pip install langchain_openai langgraph

store 정의

이 예제에서는 사용자의 선호도에 대한 정보를 검색할 수 있는 그래프를 만들 것이다. 이를 위해 InMemoryStore를 정의한다. InMemoryStore는 데이터를 메모리에 저장하고 해당 데이터를 쿼리할 수 있는 객체이다. 그런 다음 그래프를 컴파일할 때 store 객체를 전달한다. 이렇게 하면 그래프의 각 노드가 store에 접근할 수 있게 된다. 노드 함수 정의 시 store라는 키워드 인수를 정의할 수 있으며, LangGraph는 그래프를 컴파일할 때 전달한 store 객체를 자동으로 전달한다.

Store 인터페이스를 사용하여 객체를 저장할 때 두 가지를 정의해야 한다.

  1. 객체의 네임스페이스, 튜플(디렉토리와 유사)
  2. 객체 키(파일명과 유사)

우리 예제에서는 네임스페이스로 ("memories", <user_id>)를 사용하고, 각 새로운 메모리에 대해 랜덤 UUID를 키로 사용한다.

중요한 점은 사용자를 결정하기 위해 user_id를 노드 함수의 config 키워드 인수로 전달한다는 것이다.

먼저, 사용자의 기억을 이미 채운 InMemoryStore를 정의해 보자.

from langgraph.store.memory import InMemoryStore

in_memory_store = InMemoryStore()

그래프 생성

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.store.memory import InMemoryStore

load_dotenv()

in_memory_store = InMemoryStore()

import uuid
from typing import Annotated
from typing_extensions import TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.base import BaseStore


model = ChatOpenAI(model="gpt-4o-mini")


# 참고: 노드에 Store 매개변수를 전달한다.
# 이는 그래프를 컴파일할 때 사용하는 Store이다.
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    namespace = ("memories", user_id)
    memories = store.search(namespace)
    info = "\n".join([d.value["data"] for d in memories])
    system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

    # 사용자가 모델에게 기억하라고 요청하면 새로운 기억을 저장한다.
    last_message = state["messages"][-1]
    if "기억해" in last_message.content.lower():
        memory = "이름은 홍길동"
        store.put(namespace, str(uuid.uuid4()), {"data": memory})

    response = model.invoke(
        [{"type": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": response}


builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")

# 참고: 그래프를 컴파일할 때 Store 객체를 전달한다.
graph = builder.compile(checkpointer=MemorySaver(), store=in_memory_store)
# 만일 LangGraph Cloud나 LangGraph Studio를 사용하고 있다면, 그래프를 컴파일할 때 store나 checkpointer를 전달할 필요가 없다. 이는 자동으로 수행된다.

그래프 실행

이제 config에서 사용자 ID를 지정하고 모델에게 우리의 이름을 알려주자.

config = {"configurable": {"thread_id": "1", "user_id": "1"}}
input_message = {"type": "user", "content": "안녕. 기억해, 내 이름은 홍길동이야."}
for chunk in graph.stream({"messages": [input_message]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

config = {"configurable": {"thread_id": "2", "user_id": "1"}}
input_message = {"type": "user", "content": "내 이름은 뭐야?"}
for chunk in graph.stream({"messages": [input_message]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
================================ Human Message =================================

안녕. 기억해, 내 이름은 홍길동이야.
================================== Ai Message ==================================

안녕, 홍길동! 만나서 반가워. 어떻게 도와줄 수 있을까?
================================ Human Message =================================

내 이름은 뭐야?
================================== Ai Message ==================================

당신의 이름은 홍길동입니다! 도움이 필요하시면 언제든지 말씀해 주세요.

이제 인메모리 스토어를 확인하여 사용자의 기억이 실제로 저장되었는지 검증할 수 있다.

for memory in in_memory_store.search(("memories", "1")):
    print(memory.value)
{'data': '이름은 홍길동'}

이제 다른 사용자를 위해 그래프를 실행하여 첫 번째 사용자에 대한 기억이 독립적으로 유지되는지 확인해 보자.

config = {"configurable": {"thread_id": "3", "user_id": "2"}}
input_message = {"type": "user", "content": "내 이름은 뭐야?"}
for chunk in graph.stream({"messages": [input_message]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
================================ Human Message =================================

내 이름은 뭐야?
================================== Ai Message ==================================

죄송하지만, 당신의 이름을 알 수 있는 정보가 없습니다. 당신의 이름을 알려주실 수 있나요?
반응형
  • 네이버 블로그 공유
  • 네이버 밴드 공유
  • 페이스북 공유
  • 카카오스토리 공유