AI가 드디어 기억한다 — LangGraph 메모리 완벽 가이드 [2026 실전 튜토리얼]

[1일차] 사용자: "저는 백엔드 개발자고 Python을 주로 씁니다."
        AI: "네, 알겠습니다!"
[2일차] 사용자: "제 수준에 맞게 설명해줘."
        AI: "어떤 분야를 하시는지, 어떤 언어를 쓰시는지 알려주시면..."

[1일차] 사용자: "저는 백엔드 개발자고 Python을 주로 씁니다."
        AI: "기억해두겠습니다!"
[2일차] 사용자: "제 수준에 맞게 설명해줘."
        AI: "Python 백엔드 개발자 기준으로 설명드릴게요. asyncio를 이미
             아실 테니 그 맥락에서..."

┌─────────────────────────────────────────────┐
│              LangGraph 메모리 구조              │
├─────────────────────┬───────────────────────┤
│    단기 메모리        │      장기 메모리         │
│  (Short-term)       │    (Long-term)         │
├─────────────────────┼───────────────────────┤
│ 현재 대화 스레드 내    │  여러 세션에 걸쳐 유지   │
│ 자동으로 관리됨       │  직접 저장/불러와야 함   │
│ 체크포인터가 담당     │  Store가 담당           │
│ 세션 끝나면 사라짐    │  영구 보존             │
├─────────────────────┼───────────────────────┤
│ 예시: 이번 대화 내용  │ 예시: 사용자 이름,      │
│       현재 작업 상태  │       선호 스타일,      │
│       에러 히스토리   │       이전 프로젝트      │
└─────────────────────┴───────────────────────┘

# memory_basic.py
import os
from dotenv import load_dotenv
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
load_dotenv()
# ── 상태 정의 ──────────────────────────────────────
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
    # add_messages: 메시지를 덮어쓰지 않고 누적시키는 리듀서
# ── LLM ──────────────────────────────────────────
llm = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    max_tokens=1024
)
# ── 챗봇 노드 ────────────────────────────────────
def chatbot_node(state: ChatState) -> dict:
    system = SystemMessage(content="""
당신은 친절하고 기억력이 좋은 AI 어시스턴트입니다.
이전 대화 내용을 항상 참고해서 맥락에 맞게 대답하세요.
""")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}
# ── 그래프 조립 ──────────────────────────────────
builder = StateGraph(ChatState)
builder.add_node("chatbot", chatbot_node)
builder.set_entry_point("chatbot")
builder.add_edge("chatbot", END)
# 핵심: MemorySaver 연결
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)
# ── 대화 함수 ────────────────────────────────────
def chat(user_input: str, thread_id: str = "default") -> str:
    """
    thread_id: 대화 세션을 구분하는 ID
    같은 thread_id면 이전 대화를 기억합니다.
    """
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"messages": [HumanMessage(content=user_input)]},
        config=config
    )
    return result["messages"][-1].content
# ── 테스트 ──────────────────────────────────────
if __name__ == "__main__":
    tid = "user-test-001"  # 이 ID로 대화 이력 묶임
    print("=" * 50)
    r1 = chat("안녕하세요! 저는 지수라고 해요. Python 개발자입니다.", tid)
    print(f"AI: {r1}\n")
    r2 = chat("제 이름이 뭐였죠?", tid)
    print(f"AI: {r2}\n")
    r3 = chat("제 직업도 기억하고 있나요?", tid)
    print(f"AI: {r3}\n")

pip install langgraph-checkpoint-sqlite

# memory_sqlite.py
from langgraph.checkpoint.sqlite import SqliteSaver
# 나머지 import는 위와 동일...
# MemorySaver 대신 SqliteSaver 사용
# ":memory:" → 인메모리(테스트용)
# "chat_memory.db" → 파일에 영구 저장
with SqliteSaver.from_conn_string("chat_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    # 1번 실행
    config = {"configurable": {"thread_id": "user-jisu"}}
    graph.invoke(
        {"messages": [HumanMessage("저는 Python으로 FastAPI를 주로 만들어요.")]},
        config=config
    )
    print("✅ 1번 실행 완료. 이제 프로그램을 껐다 켜보세요.")

# 나중에 다시 실행 — 이전 대화가 복원됨
with SqliteSaver.from_conn_string("chat_memory.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    # 같은 thread_id를 쓰면 이전 대화 이어짐
    config = {"configurable": {"thread_id": "user-jisu"}}
    result = graph.invoke(
        {"messages": [HumanMessage("제가 주로 어떤 프레임워크 쓴다고 했죠?")]},
        config=config
    )
    print(result["messages"][-1].content)
    # → "FastAPI라고 하셨어요!"

pip install langgraph-checkpoint-postgres psycopg psycopg-pool

# memory_postgres.py
from langgraph.checkpoint.postgres import PostgresSaver
import os
DB_URI = os.getenv("DATABASE_URL")
# 예: "postgresql://user:password@localhost:5432/mydb"
def get_graph_with_postgres():
    """PostgreSQL 체크포인터를 연결한 그래프 반환"""
    with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
        # 최초 1회만 실행 — 필요한 테이블 자동 생성
        checkpointer.setup()
        graph = builder.compile(checkpointer=checkpointer)
        return graph, checkpointer
# FastAPI와 함께 사용하는 패턴
from fastapi import FastAPI
from contextlib import asynccontextmanager
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
app = FastAPI()
graph_instance = None
@asynccontextmanager
async def lifespan(app: FastAPI):
    """서버 시작 시 DB 연결, 종료 시 해제"""
    global graph_instance
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        await checkpointer.setup()  # 테이블 초기화 (최초 1회)
        graph_instance = builder.compile(checkpointer=checkpointer)
        print("✅ PostgreSQL 체크포인터 연결 완료")
        yield  # 서버 실행 중
    # 서버 종료 시 자동으로 연결 해제
app = FastAPI(lifespan=lifespan)
@app.post("/chat/{user_id}")
async def chat_endpoint(user_id: str, message: str):
    config = {"configurable": {"thread_id": user_id}}
    result = await graph_instance.ainvoke(
        {"messages": [HumanMessage(content=message)]},
        config=config
    )
    return {"reply": result["messages"][-1].content}

# memory_store.py
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
import json
# ── 메모리 스토어 초기화 ────────────────────────────
# 개발: InMemoryStore
# 프로덕션: 별도 영구 저장소 (DB, Redis 등) 연결
store = InMemoryStore()
def save_user_preference(user_id: str, key: str, value: str):
    """사용자 선호도를 Store에 저장"""
    namespace = ("user_prefs", user_id)
    store.put(namespace, key, {"value": value})
    print(f"💾 저장: [{user_id}] {key} = {value}")
def load_user_preferences(user_id: str) -> dict:
    """사용자의 모든 선호도 불러오기"""
    namespace = ("user_prefs", user_id)
    items = store.search(namespace)
    return {item.key: item.value["value"] for item in items}
# ── 선호도를 반영하는 챗봇 노드 ──────────────────────
def smart_chatbot_node(state: ChatState, config: dict) -> dict:
    """
    사용자 ID로 저장된 선호도를 불러와 시스템 프롬프트에 반영
    """
    user_id = config.get("configurable", {}).get("user_id", "anonymous")
    prefs = load_user_preferences(user_id)
    # 선호도를 시스템 프롬프트에 주입
    pref_text = ""
    if prefs:
        pref_text = "\n\n[이 사용자에 대해 알고 있는 정보]\n"
        for k, v in prefs.items():
            pref_text += f"- {k}: {v}\n"
    system = SystemMessage(content=f"""
당신은 개인화된 AI 어시스턴트입니다.
사용자의 정보를 기반으로 맞춤형 답변을 제공하세요.{pref_text}
""")
    response = llm.invoke([system] + state["messages"])
    # 대화에서 새로운 선호도를 감지하면 자동 저장
    last_msg = state["messages"][-1].content.lower()
    if "python" in last_msg and "좋아" in last_msg:
        save_user_preference(user_id, "선호언어", "Python")
    if "개발자" in last_msg or "엔지니어" in last_msg:
        save_user_preference(user_id, "직업", "개발자")
    return {"messages": [response]}

# 사용자 정보 미리 저장
save_user_preference("user-jisu", "이름", "지수")
save_user_preference("user-jisu", "직업", "Python 백엔드 개발자")
save_user_preference("user-jisu", "선호스타일", "기술적이고 간결하게")
save_user_preference("user-jisu", "경력", "3년차")
# 저장된 정보 확인
prefs = load_user_preferences("user-jisu")
print(prefs)
# → {'이름': '지수', '직업': 'Python 백엔드 개발자', ...}
# 이제 이 사용자와 대화할 때마다 이 정보가 자동으로 반영됨
config = {
    "configurable": {
        "thread_id": "session-001",
        "user_id": "user-jisu"    # Store 조회에 사용
    }
}

# thread_id 설계 패턴
# ✅ 좋은 예 — 사용자별, 목적별로 구분
config_jisu = {"configurable": {"thread_id": "user-jisu-general"}}
config_minho = {"configurable": {"thread_id": "user-minho-general"}}
# 한 사용자가 여러 대화 스레드를 가질 수도 있음
config_jisu_work = {"configurable": {"thread_id": "user-jisu-work-project-a"}}
config_jisu_study = {"configurable": {"thread_id": "user-jisu-study-python"}}
# ❌ 나쁜 예 — 모든 사용자가 같은 thread_id 사용
# 모든 대화가 섞여버림
config_bad = {"configurable": {"thread_id": "global"}}

# FastAPI에서 사용자별 thread_id 자동 생성
from fastapi import FastAPI, Header
from typing import Optional
app = FastAPI()
@app.post("/chat")
async def chat(
    message: str,
    session_id: str,           # 클라이언트가 관리하는 세션 ID
    x_user_id: Optional[str] = Header(None)  # JWT에서 추출한 유저 ID
):
    # thread_id = 사용자ID + 세션ID 조합으로 고유성 보장
    thread_id = f"{x_user_id}-{session_id}"
    config = {"configurable": {
        "thread_id": thread_id,
        "user_id": x_user_id
    }}
    result = await graph_instance.ainvoke(
        {"messages": [HumanMessage(content=message)]},
        config=config
    )
    return {"reply": result["messages"][-1].content, "thread_id": thread_id}

# memory_summary.py — 대화 요약 패턴
def summarize_if_too_long(state: ChatState) -> dict:
    """
    메시지가 20개를 넘으면 오래된 것들을 요약으로 압축합니다.
    최근 5개 메시지 + 이전 대화 요약 형태로 유지합니다.
    """
    messages = state["messages"]
    THRESHOLD = 20  # 이 숫자를 넘으면 요약 실행
    KEEP_RECENT = 5  # 최근 N개는 원본 유지
    if len(messages) <= THRESHOLD:
        return {}  # 아직 짧으면 아무것도 안 함
    # 요약할 대상: 최근 5개를 제외한 나머지
    to_summarize = messages[:-KEEP_RECENT]
    recent = messages[-KEEP_RECENT:]
    # LLM으로 요약 생성
    summary_prompt = f"""
다음 대화를 3~5문장으로 핵심만 요약해주세요.
사용자 이름, 직업, 선호사항, 논의한 주요 주제를 반드시 포함하세요.
대화:
{chr(10).join([f"{m.type}: {m.content}" for m in to_summarize])}
"""
    summary_response = llm.invoke([HumanMessage(content=summary_prompt)])
    # 요약을 첫 번째 시스템 메시지로, 나머지는 최근 메시지만 유지
    summary_message = SystemMessage(
        content=f"[이전 대화 요약]\n{summary_response.content}"
    )
    print(f"🗜️ 메모리 압축: {len(to_summarize)}개 메시지 → 요약 1개")
    return {"messages": [summary_message] + recent}

# 그래프에 요약 노드 추가
builder = StateGraph(ChatState)
builder.add_node("chatbot", chatbot_node)
builder.add_node("summarize", summarize_if_too_long)
builder.set_entry_point("chatbot")
# 대화 후 자동으로 길이 체크
builder.add_edge("chatbot", "summarize")
builder.add_edge("summarize", END)