LangChain 백엔드에 대화 상태 붙이기 — session_id, short-term memory, thread 분리를 FastAPI에서 어떻게 다루나

LangChain 백엔드에 대화 상태 붙이기 — session_id, short-term memory, thread 분리를 FastAPI에서 어떻게 다루나

지난 글에서 우리는 FastAPI로 LangChain 백엔드를 감쌌습니다.
이제 요청을 보내면 사내 도우미가 답을 하는 구조까지 왔죠.

근데 아직 한 가지가 빠져 있습니다.

대화가 이어지지 않는다는 점입니다.

예를 들어 사용자가 이렇게 묻습니다.

• “재택근무 정책 알려줘.”
• “그럼 예외는 어떻게 돼?”
• “그 내용을 한 줄로 다시 요약해줘.”

이건 사실 3개의 독립 질문이 아니죠.
앞 질문의 문맥이 계속 이어져야 합니다.

그래서 여기서부터 중요한 개념이 나옵니다.

• session_id
• thread_id
• short-term memory
• checkpointer

LangChain 공식 문서는 short-term memory를 thread-scoped memory, 즉 대화 스레드 단위의 기억으로 설명합니다. 그리고 agent에 short-term memory를 붙이려면 checkpointer를 지정해야 하며, 실행할 때 configurable.thread_id를 넘겨야 같은 대화를 이어갈 수 있다고 안내해요. 또한 이 상태는 thread별로 분리되며, step이 끝날 때마다 업데이트되고 다음 step 시작 시 다시 읽힙니다. (docs.langchain.com)

이게 진짜 중요합니다.
여기서부터 API가 “질문 하나 받는 서버”에서 “대화형 서비스”로 바뀌기 시작하거든요.

---

왜 session_id가 필요할까

지난 글의 /assistant/ask API는 이런 식이었죠.

{
  "session_id": "demo-session",
  "question": "재택근무 정책 알려줘."
}

그런데 그때는 사실 session_id를 받아만 놓고 실제로 쓰지는 않았습니다.
일부러 그랬어요. 먼저 API 껍데기를 잡는 게 중요했으니까요.

이제는 이 값을 실제로 씁니다.

핵심은 아주 단순합니다.

클라이언트의 session_id를 LangGraph/LangChain의 thread_id로 연결한다.

LangGraph persistence 문서는 checkpointer가 thread_id를 기본 키처럼 사용해 상태를 저장하고 불러온다고 설명합니다. 이 값이 없으면 저장도, 재개도 할 수 없어요. (docs.langchain.com)

즉:

• 프론트엔드에서는 session_id
• LangChain 내부에서는 thread_id

이 둘을 같은 값으로 쓰거나 매핑해서 쓰면 됩니다.

---

short-term memory는 정확히 뭘 기억하나

이 부분도 헷갈리기 쉽습니다.

short-term memory는
“이 사용자에 대한 영구 프로필” 같은 게 아닙니다.

LangChain 문서는 short-term memory를 한 thread 안에서 이어지는 대화 상태로 설명합니다. 반면 long-term memory는 thread를 넘어 사용자나 앱 수준에서 공유되는 정보라고 구분해요. (docs.langchain.com)

그러니까 지금 단계에서는 이렇게 이해하면 됩니다.

지금 붙이는 것

• 같은 대화 안의 이전 메시지
• 방금까지의 tool 호출 결과 문맥
• 이번 thread 안에서 이어지는 상태

아직 안 붙이는 것

• 사용자 선호 장기 저장
• 여러 세션을 넘나드는 개인화
• 전역 메모리/프로필 시스템

즉, 오늘 글은 **“같은 대화창 안에서 대화가 이어지는 구조”**를 만드는 글입니다.

---

왜 checkpointer가 꼭 필요할까

이건 정말 핵심입니다.

LangChain short-term memory 문서는 agent에 short-term memory를 추가하려면 checkpointer를 지정해야 한다고 아주 직접적으로 설명합니다. 예시로 InMemorySaver()를 넣고, agent.invoke(..., {"configurable": {"thread_id": "1"}})처럼 호출하라고 안내합니다. (docs.langchain.com)

쉽게 말하면 checkpointer는 이런 역할을 합니다.

• thread별 상태 저장
• 다음 요청 때 상태 복원
• tool 호출 같은 중간 step 이후 상태 갱신
• 나중에 interrupt/resume까지 확장 가능

그리고 공식 문서는 개발/테스트 단계에서는 InMemorySaver를, 운영 단계에서는 DB-backed saver 예를 들어 PostgresSaver 같은 persistent checkpointer를 권장합니다. HITL 문서도 프로토타입엔 InMemorySaver, 운영엔 AsyncPostgresSaver 같은 지속형 saver를 쓰라고 설명합니다. (docs.langchain.com)

즉 지금은 가장 쉬운 시작으로 갑니다.

• 개발용: InMemorySaver
• 운영용: 나중에 PostgresSaver 등으로 교체

---

오늘 만들 구조

이번 글에서는 지난 FastAPI 구조를 그대로 확장합니다.

project/
├─ app/
│  ├─ main.py
│  ├─ schemas.py
│  ├─ agent.py
│  └─ tools.py
├─ requirements.txt
└─ .env

차이는 하나입니다.

agent.py에 checkpointer와 thread_id를 붙인다.

그리고 /assistant/ask 엔드포인트가 session_id를 받아서
그걸 agent 호출의 thread_id로 넘기게 만들 겁니다.

---

requirements.txt

이번 글에서는 memory saver를 쓰기 때문에 LangGraph 체크포인터가 같이 필요합니다.
최근 LangChain 문서 예시도 from langgraph.checkpoint.memory import InMemorySaver를 사용합니다. (docs.langchain.com)

예시는 이렇게 잡겠습니다.

fastapi==0.112.0
uvicorn[standard]==0.30.6
langchain==1.0.3
langchain-openai==1.0.1
pydantic==2.11.7
python-dotenv==1.0.1
langgraph==1.0.2

FastAPI는 실제 앱에서 버전 pinning을 권장하고, response model과 request body를 Pydantic으로 선언하는 현재 구조와도 잘 맞습니다. (FastAPI)

---

1. 요청/응답 스키마는 그대로 두되, 의미를 더 분명히 하자

app/schemas.py

from pydantic import BaseModel, Field


class AskRequest(BaseModel):
    session_id: str = Field(description="대화 세션 ID. LangChain thread_id로 사용됨")
    question: str = Field(description="사용자 질문")


class AskResponse(BaseModel):
    answer: str = Field(description="사내 도우미의 최종 답변")

FastAPI 공식 문서는 request body를 Pydantic 모델로 선언하는 방식을 기본으로 설명하고, response model을 쓰면 직렬화와 문서화가 자동으로 따라온다고 안내합니다. (FastAPI)

여기서 포인트는 하나예요.

session_id를 이제 그냥 “클라이언트가 보내는 문자열”이 아니라
대화 상태를 이어가는 키로 쓰게 됩니다.

---

2. tools.py는 지난 글 그대로 써도 된다

app/tools.py

import json
from langchain.tools import tool


@tool
def search_policy_documents(query: str) -> str:
    """Search internal policy documents by query.

    Use this tool when the user asks about company policy, rules, benefits,
    leave, expense, onboarding, or internal process documentation.
    """
    docs = [
        {
            "title": "연차 사용 정책",
            "content": "정규직 직원은 연차를 사내 인사 시스템에서 신청할 수 있으며, 팀 리더 승인 후 확정된다.",
            "source": "policy-annual-leave",
        },
        {
            "title": "출장비 정산 정책",
            "content": "출장비는 결제 영수증과 함께 경비 시스템에 등록해야 하며, 귀사 후 7일 이내 정산이 원칙이다.",
            "source": "policy-expense-travel",
        },
        {
            "title": "재택근무 정책",
            "content": "재택근무는 팀 운영 원칙에 따라 주 2회까지 가능하며, 예외는 팀장 승인 후 가능하다.",
            "source": "policy-remote-work",
        },
    ]

    query_lower = query.lower()
    matched = []

    for doc in docs:
        haystack = f"{doc['title']} {doc['content']}".lower()
        if any(keyword in haystack for keyword in query_lower.split()):
            matched.append(doc)

    if not matched:
        return json.dumps(
            {
                "found": False,
                "results": [],
                "message": "관련 정책 문서를 찾지 못했습니다.",
            },
            ensure_ascii=False,
        )

    return json.dumps(
        {
            "found": True,
            "results": matched[:2],
        },
        ensure_ascii=False,
    )


@tool
def get_request_status(request_id: str) -> str:
    """Get the status of an internal request or ticket by request ID.

    Use this tool when the user asks for the latest state of a request,
    ticket, approval, or internal work item.
    """
    mock_requests = {
        "1024": {
            "request_id": "1024",
            "status": "승인 대기",
            "owner": "총무팀",
            "summary": "노트북 교체 요청",
        },
        "2048": {
            "request_id": "2048",
            "status": "처리 완료",
            "owner": "IT 지원팀",
            "summary": "사내 VPN 권한 부여",
        },
        "3001": {
            "request_id": "3001",
            "status": "진행 중",
            "owner": "재무팀",
            "summary": "출장비 정산 검토",
        },
    }

    data = mock_requests.get(request_id)
    if not data:
        return json.dumps(
            {
                "found": False,
                "request_id": request_id,
                "message": "해당 요청 ID를 찾지 못했습니다.",
            },
            ensure_ascii=False,
        )

    return json.dumps(
        {
            "found": True,
            "request": data,
        },
        ensure_ascii=False,
    )


@tool
def summarize_for_employee(text: str) -> str:
    """Summarize internal information for an employee in short, clear Korean.

    Use this tool when the user explicitly asks for a short summary
    or when multiple pieces of information need to be condensed.
    """
    short = text.strip().replace("\n", " ")
    if len(short) > 180:
        short = short[:180] + "..."
    return f"요약: {short}"

Tool Calling 품질은 이름, 설명, 인자 정보에 꽤 크게 좌우됩니다. LangChain 문서도 tools는 well-defined inputs/outputs와 충분한 설명이 중요하다고 설명합니다. (docs.langchain.com)

---

3. 핵심: agent.py에 checkpointer 붙이기

반응형

이제 진짜 핵심입니다.

app/agent.py

import os
from functools import lru_cache

from dotenv import load_dotenv
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

from .tools import (
    get_request_status,
    search_policy_documents,
    summarize_for_employee,
)

load_dotenv()


@lru_cache(maxsize=1)
def get_agent():
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("OPENAI_API_KEY 환경변수가 설정되어 있지 않습니다.")

    checkpointer = InMemorySaver()

    agent = create_agent(
        model="openai:gpt-4o-mini",
        tools=[
            search_policy_documents,
            get_request_status,
            summarize_for_employee,
        ],
        system_prompt=(
            "너는 사내 업무 도우미다. "
            "정책/규정/제도 질문은 정책 검색 tool을 사용해라. "
            "요청/티켓/상태 확인 질문은 요청 상태 조회 tool을 사용해라. "
            "사용자가 요약을 원하거나 정보가 길면 요약 tool을 사용해라. "
            "같은 thread 안에서는 이전 대화 문맥을 참고해라. "
            "문서나 tool 결과에 없는 내용은 추측하지 마라."
        ),
        checkpointer=checkpointer,
    )
    return agent


def ask_assistant(session_id: str, question: str) -> str:
    agent = get_agent()

    result = agent.invoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": question,
                }
            ]
        },
        {
            "configurable": {
                "thread_id": session_id,
            }
        },
    )

    return result["messages"][-1].content

이 코드에서 가장 중요한 건 딱 두 줄입니다.

checkpointer = InMemorySaver()

그리고

{"configurable": {"thread_id": session_id}}

LangChain short-term memory 문서는 바로 이 패턴을 공식 예제로 보여줍니다. agent에 checkpointer=InMemorySaver()를 주고, 실행 시 configurable.thread_id를 넘기면 그 thread 안에서 short-term memory가 이어집니다. (docs.langchain.com)

이걸 말로 풀면 이렇습니다.

• session_id="abc-123"으로 첫 질문을 보낸다
• agent가 그 대화를 thread_id="abc-123" 상태로 저장한다
• 같은 session_id로 다음 질문을 보내면
• 이전 대화 문맥을 자동으로 다시 읽고 이어서 답한다

즉, 여기서부터는 우리가 직접 messages 리스트를 서버에 따로 관리하지 않아도,
agent 런타임이 thread state로 이어서 관리해주는 구조가 됩니다.

---

4. FastAPI 엔드포인트는 session_id를 실제로 전달하게 바꾼다

app/main.py

from fastapi import FastAPI, HTTPException

from .agent import ask_assistant
from .schemas import AskRequest, AskResponse

app = FastAPI(
    title="Internal Assistant API",
    version="0.2.0",
    description="LangChain short-term memory가 붙은 사내 도우미 백엔드 예제",
)


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/assistant/ask", response_model=AskResponse)
def ask(request: AskRequest) -> AskResponse:
    try:
        answer = ask_assistant(
            session_id=request.session_id,
            question=request.question,
        )
        return AskResponse(answer=answer)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

FastAPI는 request body를 Pydantic으로 받고, response model을 선언하면 JSON 직렬화와 문서화가 따라옵니다. dependency system도 강해서 나중에 인증이나 DB 세션, 설정 주입을 붙이기 좋습니다. (FastAPI)

이제 /assistant/ask는 진짜로 stateful endpoint가 됩니다.

---

5. 실행하고 확인해보자

실행:

uvicorn app.main:app --reload

이제 같은 session_id로 여러 번 요청해보면 됩니다.

첫 번째 요청

curl -X POST "http://127.0.0.1:8000/assistant/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "team-ops-001",
    "question": "재택근무 정책 알려줘."
  }'

두 번째 요청

curl -X POST "http://127.0.0.1:8000/assistant/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "team-ops-001",
    "question": "그럼 예외는 어떻게 처리돼?"
  }'

세 번째 요청

curl -X POST "http://127.0.0.1:8000/assistant/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "team-ops-001",
    "question": "방금 내용 한 줄로 요약해줘."
  }'

같은 session_id를 썼기 때문에,
에이전트는 이 세 요청을 같은 대화 스레드로 봅니다.

반대로 session_id를 바꾸면 완전히 새 대화로 취급됩니다.
LangGraph persistence 문서가 말하는 것처럼, thread_id가 상태 저장과 복원의 핵심 키이기 때문입니다. (docs.langchain.com)

---

6. 여기서 바로 이해해야 할 것: session_id 설계

이건 실제 서비스에서 꽤 중요합니다.

처음엔 그냥 랜덤 문자열 하나 쓰면 되지만,
운영으로 가면 보통 이렇게 생각하게 됩니다.

단순한 경우

• session_id = uuid

조금 더 현실적인 경우

• session_id = user_id + conversation_id

예:

• u-102_conv-1
• u-102_conv-2

왜냐면 같은 사용자가 여러 대화를 동시에 열 수 있기 때문입니다.
short-term memory는 thread-scoped라서, thread를 어떻게 나눌지가 꽤 중요해요. LangChain memory overview도 short-term memory를 “thread-scoped conversation state”로 설명합니다. (docs.langchain.com)

즉:

• 사용자 하나 = 대화 하나
  가 아니라
• 사용자 하나 = 여러 thread 가능

이렇게 보는 게 더 현실적입니다.

---

7. 지금 구조의 한계도 분명하다

좋은 점만 보면 안 됩니다.

한계 1. InMemorySaver는 서버 재시작하면 날아간다

공식 문서도 InMemorySaver는 테스트/프로토타입용으로 설명하고, 운영에서는 DB-backed saver를 권장합니다. (docs.langchain.com)

한계 2. 멀티 인스턴스 환경에서는 상태 공유가 안 된다

서버를 여러 대 띄우면 메모리 saver로는 thread state를 공유할 수 없습니다.

한계 3. 대화가 길어질수록 비용이 늘어난다

short-term memory는 편하지만, 메시지가 계속 누적되면 토큰 비용과 지연이 커질 수 있어요. memory overview는 긴 대화에서 stale/off-topic content 때문에 성능과 비용이 나빠질 수 있다고 설명합니다. (docs.langchain.com)

즉, 지금 구조는 대화 기억의 출발점이지 완성본은 아닙니다.

---

8. 운영으로 가면 뭘 바꿔야 하나

여기서부터가 다음 단계 감각입니다.

1) InMemorySaver → Postgres 계열 saver

LangChain short-term memory와 HITL 문서는 운영에서는 persistent checkpointer, 예를 들어 PostgresSaver나 AsyncPostgresSaver를 쓰라고 권장합니다. (docs.langchain.com)

2) 긴 대화 요약/trim

대화가 길어질수록 short-term memory를 그대로 두면 느려지고 비싸집니다. memory overview는 stale content를 잊게 만들거나 줄이는 전략이 필요하다고 설명합니다. (docs.langchain.com)

3) 인증 + 권한

사내 도우미는 세션만으로 끝나면 안 됩니다.
사용자 인증과 요청 권한이 붙어야 합니다.

4) observability

어떤 thread에서 어떤 tool이 몇 번 불렸는지 추적해야 합니다.

즉, 지금 글은 “대화가 이어지게 만드는 첫 단계”이고,
운영은 그 위에 persistence와 governance를 더 올리는 구조예요.

---

9. FastAPI에서 dependency로 agent를 주입하는 것도 나중엔 좋다

지금은 단순하게 ask_assistant()를 바로 호출했죠.
그런데 FastAPI는 dependency injection이 강해서, 나중에는 설정/agent/provider를 주입형으로 바꾸기 좋습니다. 공식 dependency 문서도 이 시스템을 매우 직관적이고 강력한 DI 시스템이라고 설명합니다. (FastAPI)

예를 들면 나중엔 이렇게 갈 수 있어요.

• 설정 객체 주입
• agent 주입
• 인증 사용자 주입
• DB 세션 주입

처음엔 과하지만, 조금만 커지면 이 패턴이 꽤 편해집니다.

---

10. 여기서 바로 해보면 좋은 개선

이건 진짜 추천합니다.

개선 1

AskResponse에 thread_id를 같이 넣어보기

개선 2

같은 session_id로 연속 질문 테스트해보기

개선 3

다른 session_id로 같은 질문을 보내서 문맥이 분리되는지 보기

개선 4

요약 질문을 여러 턴 뒤에 던져보기

이 테스트를 해보면 short-term memory가 “개념”이 아니라
“아, 진짜 thread 단위로 상태를 이어가네”로 느껴집니다.

---

11. 이번 글의 전체 핵심 코드만 다시 모아보면

agent.py

from functools import lru_cache
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

@lru_cache(maxsize=1)
def get_agent():
    return create_agent(
        model="openai:gpt-4o-mini",
        tools=[...],
        system_prompt="...",
        checkpointer=InMemorySaver(),
    )

def ask_assistant(session_id: str, question: str) -> str:
    agent = get_agent()
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        {"configurable": {"thread_id": session_id}},
    )
    return result["messages"][-1].content

이게 오늘 글의 본질입니다.

checkpointer + thread_id

이 두 개가 붙는 순간,
FastAPI의 stateless request가 LangChain 쪽에서는 stateful conversation으로 바뀝니다.

---

오늘 글에서 꼭 가져가야 할 한 문장

오늘 내용을 한 줄로 줄이면 이겁니다.

FastAPI에서 대화 상태를 붙인다는 건, 클라이언트의 session_id를 LangChain의 thread_id로 연결하고 checkpointer로 thread별 상태를 저장·복원하게 만드는 일이다.

이 감각이 생기면
이제 “대화형 백엔드”가 어떻게 만들어지는지 훨씬 선명해집니다.

---

마무리

저는 이 구간이 꽤 재밌습니다.

왜냐면 여기서부터 API가 더 이상 “질문 하나 받고 답 하나 주는 함수”가 아니게 되거든요.

같은 사용자가 이어서 묻고,
그 문맥을 기억하고,
다음 턴에 반영하는 순간부터
서비스가 갑자기 살아 있는 느낌을 냅니다.

그리고 그 시작은 의외로 거창하지 않아요.

• checkpointer 하나 붙이고
• thread_id 하나 넘기는 것

그런데 그 작은 차이가
질문응답 API와 대화형 서비스 사이를 꽤 크게 갈라놓습니다.

저는 그게 좋더라고요.
되게 작은 코드 변화인데, 체감은 꽤 크거든요.

---

다음 글 예고

다음 글에서는
LangChain 백엔드 응답을 더 믿을 수 있게 만들기 — structured output과 FastAPI response model을 같이 쓰는 방법
으로 이어가겠습니다.

이제부터는 “답을 한다”를 넘어서,
프론트엔드와 다른 시스템이 다루기 좋은 형태 있는 응답으로 만드는 쪽을 본격적으로 다루게 될 거예요.

출처

• LangChain Short-term memory 문서: agent에 short-term memory를 추가하려면 checkpointer를 지정해야 하고, 실행 시 configurable.thread_id를 넘겨 thread별 대화를 이어갈 수 있다고 설명합니다. (docs.langchain.com)
• LangGraph Persistence 문서: checkpointer는 thread_id를 primary key처럼 사용해 상태를 저장하고 불러온다고 설명합니다. (docs.langchain.com)
• LangChain Human-in-the-Loop 문서: 프로토타입에는 InMemorySaver, 운영에는 AsyncPostgresSaver 같은 persistent saver를 권장합니다. (docs.langchain.com)
• FastAPI Request Body 문서: request body는 Pydantic 모델로 선언하는 방식을 기본으로 안내합니다. (FastAPI)
• FastAPI Response Model 문서: response_model을 선언하면 검증, 문서화, JSON 변환을 자동으로 처리한다고 설명합니다. (FastAPI)
• FastAPI Dependencies 문서: FastAPI는 강력하면서도 단순한 dependency injection system을 제공한다고 설명합니다. (FastAPI)

LangChain, FastAPI, Short-term Memory, Thread ID, Checkpointer, AI Agent, 생성형AI, 백엔드개발, 대화형서비스, 주니어개발자