재시작해도 안 끊기는 LangGraph PG 연동법: FastAPI 무손실 복원 완벽 가이드

💡 3초 핵심 요약 (Key Takeaways)

새벽에 터지는 psycopg.OperationalError: the connection is closed의 원인은 튜토리얼용 MemorySaver나 요청별 임시 커넥션을 실서버에 적용했기 때문입니다. 본 가이드에서는 AsyncPostgresSaver + psycopg_pool + FastAPI Lifespan을 결합해 서버가 불시에 재시작되어도 대화 세션을 100% 무손실 복원하는 프로덕션 레벨 아키텍처를 구축합니다.

1. 왜 MemorySaver로는 안 되는가 — 호텔 프론트 비유

체크포인터는 게임의 자동 세이브 포인트와 같습니다. 대화가 한 턴 진행될 때마다 고유한 세이브 슬롯(thread_id)에 상태를 영구 저장하는 원리입니다.

MemorySaver는 이 세이브 데이터를 서버 프로세스의 RAM에만 보관합니다. 이는 호텔 프론트 직원이 투숙객의 특이사항을 포스트잇에만 적어두는 것과 같습니다. 서버 재배포나 파드 오토스케일링으로 직원이 교대되는 순간 메모지는 휴지통으로 직행하며, 손님의 이전 요청 맥락은 통째로 증발합니다.

📌 PostgreSQL 체크포인터의 3대 핵심 이점

서버 재시작 및 장애 크래시 발생 후에도 멀티턴 대화 맥락 100% 연속 유지
Human-in-the-loop(사람 개입 및 결재 대기) 상태를 일수 단위로 안전하게 보관
checkpoint_id 지정을 통한 과거 특정 시점 롤백 및 타임트래블(Time-travel) 기능 지원

AsyncPostgresSaver.setup()`을 최초 1회 호출하면 다음 4개 관리 테이블이 자동으로 생성되어 데이터 무결성을 보장합니다.

테이블명	역할 및 데이터 보존 전략
checkpoints	실행 단계(Super-step)별 스냅샷, `thread_id`, 메타데이터 기록
checkpoint_blobs	대용량 상태 바이너리 데이터를 분리 저장하여 색인 및 쿼리 성능 확보
checkpoint_writes	노드별 중간 쓰기 상태를 기록해 장애 복구 시 중복 연산 방지
checkpoint_migrations	LangGraph 프레임워크 업그레이드 시 스키마 버전 호환성 관리

2. psycopg 커넥션 풀 — 누락 시 장애를 부르는 3대 설정

FastAPI와 같은 비동기 프레임워크에서 요청마다 동기식 커넥션을 새로 맺는 것은 이벤트 루프를 블로킹하는 주원인입니다. 따라서 psycopg_pool.AsyncConnectionPool을 활용해 커넥션을 사전에 프로비저닝하고 재사용해야 합니다.

필수 파라미터	누락 시 발생 장애 (Root Cause)
autocommit=True	DDL 테이블 생성 미반영 및 체크포인트 쓰기 트랜잭션 Lock 발생
row_factory=dict_row	내부 조회가 `row["col"]` 형태이므로 기본 튜플 반환 시 `TypeError`
prepare_threshold=0	PgBouncer/Supabase 트랜잭션 풀링 환경에서 Prepared Statement 충돌

커넥션 풀 생성 시 프로덕션 환경 권장 코드는 다음과 같습니다.

from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

DATABASE_URL = "postgresql://user:password@host:5432/agent_db"

pool = AsyncConnectionPool(
    conninfo=DATABASE_URL,
    min_size=5,          # 서버 기동 시 사전 확보할 커넥션 수
    max_size=20,         # 피크 트래픽 대응 상한 커넥션
    max_idle=300,        # 5분 이상 유휴 커넥션 자동 회수 (Keepalive 방어)
    max_lifetime=1800,   # 30분 주기로 세션 재생성 (클라우드 만료 방지)
    kwargs={
        "autocommit": True,
        "row_factory": dict_row,
        "prepare_threshold": 0,
    },
)

⚠️ 풀 사이즈 산정 공식 풀 크기는 무조건 크게 설정하지 않고, DB 서버의 max_connections(기본 100)와 배포된 API 서버 인스턴스 수를 고려해 (max_connections - 예약 커넥션) / Pod 수 이하로 제한해야 합니다.

3. FastAPI Lifespan으로 싱글톤 그래프 구축

요청 핸들러 내부에서 매번 체크포인터를 인스턴스화하는 것은 심각한 안티패턴입니다. 서버 수명주기(Lifespan) 시작 시 커넥션 풀을 열고 컴파일된 그래프를 app.state에 싱글톤으로 등록해 재사용해야 합니다.

🔄 전체 아키텍처 실행 라이프사이클 FastAPI 기동 ➔ AsyncConnectionPool 오픈 ➔ AsyncPostgresSaver 생성 ➔ setup() 1회 호출 ➔ graph.compile() 싱글톤 등록 ➔ /chat 엔드포인트 공유 서빙 ➔ 서버 종료 시 풀 안전 해제

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from pydantic import BaseModel
from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph import StateGraph, MessagesState

DATABASE_URL = "postgresql://user:password@host:5432/agent_db"

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 1. 서버 시작 시 커넥션 풀 오픈
    async with AsyncConnectionPool(
        conninfo=DATABASE_URL,
        min_size=5,
        max_size=20,
        max_idle=300,
        max_lifetime=1800,
        kwargs={"autocommit": True, "row_factory": dict_row, "prepare_threshold": 0},
    ) as pool:
        checkpointer = AsyncPostgresSaver(pool)
        
        # 2. 필수 테이블 스키마 자동 프로비저닝 (최초 1회 동작)
        await checkpointer.setup()

        # 3. 그래프 컴파일 및 싱글톤 인스턴스 주입
        builder = StateGraph(MessagesState)
        # ... 노드 및 엣지 라우팅 설정 추가
        graph = builder.compile(checkpointer=checkpointer)

        app.state.graph = graph
        app.state.checkpointer = checkpointer

        yield  # 요청 수신 시작

        # 4. 서버 종료 시 커넥션 풀은 async with에 의해 안전하게 클로즈됨

app = FastAPI(lifespan=lifespan)

class ChatRequest(BaseModel):
    thread_id: str
    message: str

@app.post("/chat")
async def chat(req: ChatRequest, request: Request):
    graph = request.app.state.graph
    config = {"configurable": {"thread_id": req.thread_id}}

    # thread_id만 전달하면 이전 대화 맥락이 자동 복원되어 실행됨
    result = await graph.ainvoke(
        {"messages": [{"role": "user", "content": req.message}]},
        config=config,
    )
    return {"reply": result["messages"][-1].content}

@app.get("/history/{thread_id}")
async def history(thread_id: str, request: Request):
    checkpointer = request.app.state.checkpointer
    config = {"configurable": {"thread_id": thread_id}}

    # 특정 세션의 최신 스냅샷 조회
    snapshot = await checkpointer.aget_tuple(config)
    if snapshot is None:
        return {"detail": "해당 thread_id의 기록이 없습니다."}

    return {
        "checkpoint_id": snapshot.config["configurable"]["checkpoint_id"],
        "state": snapshot.checkpoint,
    }

4. 실무에서 자주 발생하는 3대 런타임 에러 대응법

발생 에러	원인 분석	조치 방안
OperationalError: connection closed	클라우드 방화벽/DB가 유휴 커넥션을 강제 드롭함	`max_idle` 및 `max_lifetime`을 단축하여 선제적 재생성
TypeError: tuple indices must be integers	psycopg 기본 결과 반환이 튜플 형식이어서 컬럼 키 접근 실패	풀 `kwargs`에 `row_factory=dict_row` 설정
prepared statement already exists	PgBouncer 트랜잭션 풀러 환경에서 구문 캐시가 충돌	`prepare_threshold=0` 적용해 캐싱 비활성화

🔒 프로덕션 필수 보안 환경 변수

체크포인터 직렬화 및 역직렬화 과정에서의 임의 코드 실행 공격을 차단하기 위해 아래 환경변수를 컨테이너 배포 시 반드시 주입하세요.

export LANGGRAPH_STRICT_MSGPACK=true

LangGraph의 전반적인 상태 설계 및 StateGraph 구조가 익숙하지 않다면 LangGraph 그래프 빌더 기본기와 StateGraph 구성법을 먼저 참고하시기 바랍니다. 아울러 FastAPI 스트리밍 아키텍처에 대해 깊이 알고 싶다면 FastAPI 비동기 라이프사이클과 스트리밍 최적화 가이드도 함께 확인해보세요.

💡 배포 전 최종 검증 5대 체크리스트

✔️ DB URL 검증: 실제 운영 DB 환경 변수와 시크릿이 정상 바인딩되었는가?
✔️ 커넥션 상한선: max_size가 DB 인스턴스의 max_connections 허용치를 초과하지 않는가?
✔️ 필수 3종 옵션: autocommit, row_factory, prepare_threshold가 모두 등록되었는가?
✔️ 초기화 멱등성: checkpointer.setup()이 서버 기동 시 싱글톤으로 1회만 호출되는가?
✔️ 보안 프로토콜: LANGGRAPH_STRICT_MSGPACK=true 환경 변수가 적용되었는가?

이 블로그 검색

giantturtle