환경 구성

사용된 라이브러리 버전:

• Python: 3.12.9
• langgraph: 0.5.4
• langgraph-checkpoint-postgres: 2.0.23
• psycopg: 3.2.10

체크포인트 초기화 방식

방법 1: 직접 연결 객체로 초기화

PostgresSaver 클래스를 직접 사용하여 데이터베이스 연결을 설정할 수 있습니다.

from psycopg import Connection
from langgraph.checkpoint.postgres import PostgresSaver

db_uri = "postgres://<user>:<password>@<host>:<port>/<database>"
conn = Connection.connect(db_uri)
conn.autocommit = True  # 필수: 트랜잭션 내에서 CREATE INDEX CONCURRENTLY 실행 불가 → 자동 커밋 필요

checkpointer = PostgresSaver(conn)
checkpointer.setup()

방법 2: 연결 문자열로 간편 초기화

from_conn_string 메서드를 사용하면 자동으로 컨텍스트 관리를 지원합니다.

from langgraph.checkpoint.postgres import PostgresSaver

db_uri = "postgres://<user>:<password>@<host>:<port>/<database>"
with PostgresSaver.from_conn_string(db_uri) as checkpointer:
    checkpointer.setup()

스키마 지정 옵션

기본적으로 테이블은 public 스키마에 생성됩니다. 특정 스키마에 저장하려면 연결 문자열에 options 파라미터를 추가하세요.

db_uri = "postgres://<user>:<password>@<host>:<port>/<database>?options=-csearch_path%3D<your_schema>,public"

이 경우, your_schema에 테이블이 생성되며, public도 검색 경로에 포함됩니다.

비동기 처리 설정

작업 흐름의 비동기 출력을 사용하려면 AsyncPostgresSaver를 사용해야 합니다. 동기 호출로 접근하면 오류가 발생합니다.

from langgraph.graph import StateGraph

workflow = StateGraph(MainState)
workflow.add_node("node1", execute_node_func)

# ... 노드 및 상태 정의 생략 ...

async def generate_async_response(*args, **kwargs):
    async with AsyncPostgresSaver.from_conn_string(db_uri) as async_checkpointer:
        graph = workflow.compile(checkpointer=async_checkpointer)

        inputs = {
            # 입력 데이터
        }
        config = {
            "configurable": {
                "thread_id": "123456",
            }
        }

        async for output in graph.astream(inputs, config, subgraphs=True, stream_mode="updates"):
            # 결과 처리 로직
            pass

비동기 모드에서의 상태 조회 변경

비동기 체크포인트 사용 시 동기 메서드(get_state)는 사용 불가능합니다. 대신 aget_state를 사용해야 합니다.

# 기존 동기 방식 (비동기 체크포인트에서는 작동하지 않음)
def is_thread_interrupted(graph, thread_id: str):
    state = graph.get_state({"configurable": {"thread_id": thread_id}})
    return state.next not in [(), ("__end__",)]

# 수정된 비동기 방식
async def is_thread_interrupted(graph, thread_id: str):
    state = await graph.aget_state({"configurable": {"thread_id": thread_id}})
    return state.next not in [(), ("__end__",)]

비동기 함수 내에서 상태를 확인하고 조건부 재개 로직을 적용할 수 있습니다.

async def generate_async_response(*args, **kwargs):
    async with AsyncPostgresSaver.from_conn_string(db_uri) as async_checkpointer:
        graph = workflow.compile(checkpointer=async_checkpointer)

        inputs = { ... }
        config = { "configurable": { "thread_id": "123456" } }

        if await is_thread_interrupted(graph, thread_id="123456"):
            inputs = Command(resume=inputs)

        async for output in graph.astream(inputs, config, subgraphs=True, stream_mode="updates"):
            # 스트림 출력 처리
            pass