Python LangGraph Store로 새 대화에도 사용자 설정 이어 쓰기

새 대화에서 thread_id가 바뀌어도 사용자 설정을 유지하려면, 설정을 Store의 사용자별 namespace에 저장하고 매 실행마다 같은 user_id로 조회하면 돼요. 대화 상태는 체크포인터에 맡기고, 언어·응답 형식처럼 대화를 넘어 적용할 값만 따로 저장하는 구조예요. 외부 LLM이나 API 키 없이 실행하는 예제로 설정 공유와 대화 분리를 확인해 봐요. LangGraph 영속성 문서

체크포인터와 Store의 역할 비교

설정 저장과 스레드 분리를 확인하는 전체 코드

# preferences_demo.py
# 설치: python -m pip install -U langgraph
# 실행: python preferences_demo.py

from dataclasses import dataclass
from operator import add
from typing import Annotated
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, END, StateGraph
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore


DEFAULTS = {"language": "en", "response_format": "paragraph"}
ALLOWED = {
    "language": {"ko", "en"},
    "response_format": {"bullets", "paragraph"},
}


@dataclass(frozen=True)
class Context:
    user_id: str


class State(TypedDict, total=False):
    turns: Annotated[list[str], add]
    prefs_patch: dict[str, str] | None
    applied: dict[str, str]


def apply_preferences(state: State, runtime: Runtime[Context]):
    store = runtime.store
    if store is None:
        raise RuntimeError("compile(store=...) 설정이 필요합니다.")
    if not runtime.context.user_id:
        raise ValueError("user_id가 필요합니다.")

    namespace = ("users", runtime.context.user_id, "preferences")
    item = store.get(namespace, "profile")
    prefs = {**DEFAULTS, **(item.value if item is not None else {})}

    patch = state.get("prefs_patch")
    if patch is not None:
        for field, value in patch.items():
            if field not in ALLOWED or value not in ALLOWED[field]:
                raise ValueError(f"허용하지 않은 설정: {field}={value}")
        prefs.update(patch)
        store.put(namespace, "profile", prefs)

    # 실제 모델을 붙인다면 이 prefs를 응답 생성에 사용합니다.
    return {"applied": prefs, "prefs_patch": None}


store = InMemoryStore()
checkpointer = InMemorySaver()
builder = StateGraph(State, context_schema=Context)
builder.add_node("apply_preferences", apply_preferences)
builder.add_edge(START, "apply_preferences")
builder.add_edge("apply_preferences", END)
graph = builder.compile(store=store, checkpointer=checkpointer)


def run(user_id, thread_id, text, patch=None):
    return graph.invoke(
        {"turns": [text], "prefs_patch": patch},
        {"configurable": {"thread_id": thread_id}},
        context=Context(user_id=user_id),
    )


# 1. 첫 대화에서 설정을 저장합니다.
alice_1 = run(
    "alice", "alice-chat-1", "첫 대화",
    {"language": "ko", "response_format": "bullets"},
)
assert alice_1["applied"] == {
    "language": "ko", "response_format": "bullets"
}

# 2. 같은 사용자의 새 대화: 설정은 공유하고 입력 기록은 분리합니다.
alice_2 = run("alice", "alice-chat-2", "새 대화")
assert alice_2["applied"] == alice_1["applied"]
assert alice_2["turns"] == ["새 대화"]

# 3. 다른 사용자: 저장된 설정이 없으므로 기본값을 사용합니다.
bob_1 = run("bob", "bob-chat-1", "밥의 대화")
assert bob_1["applied"] == DEFAULTS
assert bob_1["turns"] == ["밥의 대화"]
assert store.get(("users", "bob", "preferences"), "profile") is None

# 4. 일부 설정만 바꾸면 저장된 나머지 설정은 유지합니다.
alice_3 = run(
    "alice", "alice-chat-3", "형식 변경",
    {"response_format": "paragraph"},
)
assert alice_3["applied"] == {
    "language": "ko", "response_format": "paragraph"
}

# 5. 기존 대화로 돌아가도 매번 Store에서 최신 설정을 읽습니다.
alice_1_again = run("alice", "alice-chat-1", "이어서 질문")
assert alice_1_again["applied"] == alice_3["applied"]
assert alice_1_again["turns"] == ["첫 대화", "이어서 질문"]
assert alice_1_again["prefs_patch"] is None

print("모든 검증 통과: 설정 공유, 사용자 분리, 대화 분리, 부분 수정")

설정의 주소에는 대화 ID를 넣지 않아요

설정의 저장 주소를 먼저 정하면 구현이 단순해져요. 예제에서는 namespace를 ('users', user_id, 'preferences'), key를 'profile'로 정했어요. 여기서 'users', 'preferences', 'profile'은 LangGraph의 예약어가 아니라 애플리케이션이 선택한 이름이에요. namespace는 문자열 튜플이고, 그 안에서 key로 저장 항목을 구분해요. Store 사용 문서

이 주소에 thread_id를 추가하면 새 대화가 시작될 때 설정을 찾는 주소도 달라져요. 반대로 user_id를 빼면 모든 사용자가 같은 설정을 읽게 되겠죠. 여러 조직을 지원하는 서비스라면 ('tenants', tenant_id, 'users', user_id, 'preferences')처럼 조직까지 구분하는 설계를 권해요.

코드의 Context는 실행에 필요한 사용자 정보를 전달하는 용도예요. user_id를 Context에 넣는 것만으로 저장이 일어나는 것은 아니에요. 노드가 runtime.context.user_id를 읽어 주소를 만들고, runtime.store로 실제 저장·조회를 수행해요. 이 연결 방식은 공식 Runtime 사용 예제를 따랐어요. 노드에서 Store 접근하기

파일 하나로 공유 범위를 확인해요

Python 3.11 이상을 실습 환경으로 잡고, 새 가상 환경에서 python -m pip install -U langgraph를 실행해 주세요. ‘설정 저장과 스레드 분리를 확인하는 전체 코드’를 preferences_demo.py로 저장한 뒤 python preferences_demo.py로 실행하면 돼요. 예제는 공식 문서에서 확인한 API로 구성했으며, 실제 실행 로그를 제시한 것은 아니에요.

assert는 단순히 답변 문자열이 비슷한지 비교하지 않아요. Store에서 읽은 설정 딕셔너리와 스레드 안에 누적된 입력 목록을 각각 검사해요. 두 값이 함께 맞아야 ‘사용자 설정은 공유되고 대화 상태는 분리된다’는 조건을 충족해요. 검사가 모두 통과하면 코드 마지막의 ‘모든 검증 통과’ 문장이 출력되도록 구성했어요.

대화 상태를 눈으로 확인하기 쉽게 messages 대신 turns라는 문자열 목록을 사용했어요. Annotated와 add는 새 입력을 기존 목록에 이어 붙이는 reducer예요. 실제 챗봇에 옮길 때는 메시지 전용 상태와 reducer를 사용하되, 설정을 읽는 부분은 별도로 유지하면 돼요. reducer를 지정하지 않은 상태 필드는 기본적으로 새 값으로 교체돼요. Graph API의 상태와 reducer

실패한 assert를 해석할 때는 대상부터 나누세요. 설정 비교가 실패하면 저장 주소와 Store 인스턴스의 수명을 확인하고, turns 비교가 실패하면 thread_id와 reducer를 확인하는 식이에요. 읽어 온 설정은 맞는데 실제 모델의 답변만 다르다면, 그때는 설정을 모델 입력에 반영하는 코드를 점검할 차례예요.

부분 수정은 기존 설정과 합쳐서 저장해요

예제는 설정 변경 요청을 prefs_patch로 받고, 읽어 온 설정에 변경분을 합친 뒤 profile 전체를 다시 저장해요. 같은 namespace와 key의 put은 항목을 저장하거나 덮어쓰는 동작이므로, 일부 필드만 보냈을 때 나머지를 유지하는 정책은 애플리케이션 코드가 담당해야 해요. Store 저장 계약

가령 이미 한국어·목록형 응답을 선택한 사용자가 형식만 문단형으로 바꿨다면 언어는 그대로 남아야 해요. 이 사례를 부분 수정 검증에 넣었어요. 기본값, 저장된 값, 이번 변경분 순서로 합치므로 이번 변경이 우선하고, 저장된 설정에 없는 항목만 기본값으로 채워져요.

prefs_patch를 처리한 뒤 None으로 되돌리는 것도 의도한 설계예요. 변경 요청까지 스레드 상태에 계속 남겨 두면, 후속 호출이 그 필드를 생략했을 때 예전 요청을 다시 처리할 여지가 있어요. 코드에서는 호출마다 변경 여부를 명시하고, 노드도 처리한 요청을 비워 두도록 했어요. 반면 applied는 이번 실행에서 사용한 설정의 기록이므로 남겨요.

이 예제는 순차 호출용이에요. 두 요청이 동시에 같은 profile을 읽고 각각 다른 필드를 수정한 다음 전체를 저장하면, 나중 저장이 앞선 변경을 덮을 수 있는 구조예요. 동시 편집이 필요한 서비스에서는 사용자별 변경 요청을 직렬화하거나, 저장 계층에 버전 확인을 추가하는 방안을 권해요. 조회 후 저장하는 두 호출만으로 충돌 제어까지 구현됐다고 보지는 마세요.

사용자 구분과 접근 권한은 따로 설계해요

namespace는 데이터를 찾는 주소로 사용하세요. 이 예제에는 로그인이나 권한 검사가 없으므로, Context에 어떤 user_id를 넣어도 그 주소로 접근을 시도해요. 서비스에 연결할 때는 요청 본문에서 받은 사용자 ID를 그대로 신뢰하지 말고, 서버에서 확인한 로그인 사용자로 Context를 만드는 구성을 권해요.

thread_id의 소유권도 별도로 확인해야 해요. 체크포인터는 thread_id를 기준으로 상태를 불러오므로, Store의 namespace만 사용자별로 나눴다고 대화 상태까지 사용자별 접근 통제가 끝나는 것은 아니에요. 예제에서는 충돌을 피하려고 사용자 이름을 포함한 스레드 ID를 썼지만, ID의 모양 자체는 권한 검사를 대신하지 않아요. 체크포인터와 스레드

설정 필드도 자유로운 지시문보다 허용값이 있는 데이터로 시작하는 편이 좋아요. 예제는 language와 response_format만 받고 각 값도 검사해요. 실제 모델을 붙일 때는 이 값을 애플리케이션이 정한 언어·형식 지시로 변환해 전달하세요. 저장된 설정이 자동으로 프롬프트에 들어가거나 모델의 출력을 강제하는 구조는 이 코드에 포함되어 있지 않아요.

새 대화와 서버 재시작은 다른 검증이에요

이번 예제의 범위는 같은 프로세스에서 같은 Store를 사용하는 여러 대화예요. InMemoryStore는 메모리에 데이터를 보관하므로 프로세스가 종료되면 설정을 잃어요. 요청 처리 함수 안에서 매번 새 인스턴스를 만드는 구조도 피해야 해요. 재시작 후 보존이 필요하다면 데이터베이스 기반 Store로 바꿔야 해요. InMemoryStore 참조 문서

RAM 모듈 사진은 여기서 말하는 메모리의 실물 참고예요. 부품 사진으로 LangGraph의 저장 동작이나 실행 결과를 확인할 수는 없어요. 이 예제에서 보존 범위를 결정하는 것은 RAM의 모양이나 용량이 아니라, 프로세스 내부에만 데이터를 두는 저장소 구현이에요. RAM 모듈 실물 자료

공식 문서의 PostgreSQL 구성은 langgraph-checkpoint-postgres 패키지의 PostgresStore를 사용하며, 처음 사용할 때 store.setup()으로 필요한 구조를 준비해요. 대화 기록도 재시작 뒤 이어야 한다면 체크포인터 역시 PostgresSaver 같은 영속 구현으로 바꾸고 별도로 초기화해야 해요. Store만 바꾸는 것으로 두 요구가 모두 해결되지는 않아요. 데이터베이스 기반 메모리 구성

영속 저장소로 옮긴 다음에는 검증을 두 프로세스로 나눠 보세요. 첫 실행에서 설정을 저장하고 종료한 뒤, 두 번째 실행에서는 초기 설정을 다시 쓰는 코드를 빼고 조회만 해야 해요. 같은 사용자·새 스레드에서 이전 값이 돌아오는지 확인하면 재시작 보존을 검사할 수 있어요. 매번 시작할 때 기본값을 put하는 코드를 남겨 두면, 보존된 데이터를 스스로 덮어써 놓고 저장소 문제로 오해하기 쉬워요.

• LangGraph Persistence
• LangGraph Stores
• LangGraph Memory
• LangGraph Graph API overview
• LangGraph Checkpointers
• LangGraph InMemoryStore 참조
• Wikimedia Commons — File:512 MB DDR2 DIMM RAM Module.jpg