커스텀 체크포인터 사용하기

커스텀 체크포인터 사용하기

에이전트 배포에서 내장 Postgres 체크포인터를 커스텀 BaseCheckpointSaver 구현으로 교체해요.

LangSmith에 에이전트를 배포할 때 서버는 그래프 실행 간 상태 영속성을 처리하는 내장 Postgres 기반 체크포인터를 제공해요. 이를 다른 스토리지 백엔드를 사용하는 자체 BaseCheckpointSaver 구현으로 교체할 수 있어요.

BaseCheckpointSaver 인스턴스를 생성(yield)하는 비동기 컨텍스트 매니저에 대한 경로를 제공하면 서버가 그 수명 주기를 자동으로 관리해요.

커스텀 체크포인터는 알파 단계예요. 이 기능은 마이너 버전 업데이트에서 호환되지 않는 변경이 발생할 수 있어요.

체크포인트 저장에 PostgreSQL 대신 MongoDB를 사용하려면 체크포인터 백엔드 구성을 참고하세요. 이 페이지는 완전히 커스텀 스토리지 백엔드를 구현하기 위한 것이에요.

출처: 문서

본문

체크포인터 정의하기

기존 LangSmith 애플리케이션에서 시작해, 커스텀 체크포인터를 생성하는 비동기 컨텍스트 매니저를 정의하는 파일을 만드세요. 새 프로젝트를 시작한다면 CLI를 사용해 템플릿에서 앱을 만들 수 있어요.

langgraph new --template=new-langgraph-project-python my_new_project

비동기 컨텍스트 매니저 패턴을 통해 서버는 애플리케이션 수명 주기의 적절한 지점에서 데이터베이스 연결을 열고 닫을 수 있어요:

# ./src/agent/checkpointer.py
import contextlib

class MyCheckpointer(BaseCheckpointSaver):
    def __init__(self):
        super().__init__()
        # Initialize your custom checkpointer here
    ...

    @contextlib.asynccontextmanager
    async def aget(self, config: RunnableConfig):
        # Your custom logic to create a connection pool and initialize your checkpointer here.
        yield


@contextlib.asynccontextmanager
async def generate_checkpointer():
    """Yield a BaseCheckpointSaver, open for the duration of the server."""
    async with AsyncSqliteSaver.from_conn_string("./checkpoints.db") as saver:
        await saver.setup()
        yield saver

컨포먼스 스위트로 테스트하기

대부분의 오픈 소스 체크포인터 구현은 아직 Agent Server가 요구하는 모든 작업을 구현하지 않아요. 체크포인터를 구성하기 전에 컨포먼스 테스트 스위트로 검증해 호환성을 확인하세요.

패키지를 설치하세요:

pip install langgraph-checkpoint-conformance

체크포인터를 등록하고 검증을 실행하세요:

import asyncio

from langgraph.checkpoint.conformance import checkpointer_test, validate


@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    async with MyCheckpointer(...) as saver:
        yield saver


async def main():
    report = await validate(my_checkpointer)
    report.print_report()
    assert report.passed_all_base()


asyncio.run(main())

스위트는 체크포인터가 구현한 확장 기능을 자동 감지하고 적절한 테스트를 실행해요. pytest 테스트로도 실행할 수 있어요:

import pytest

from langgraph.checkpoint.conformance import checkpointer_test, validate


@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    async with MyCheckpointer(...) as saver:
        yield saver


@pytest.mark.asyncio
async def test_conformance():
    report = await validate(my_checkpointer)
    report.print_report()
    assert report.passed_all_base()

스위트가 검증하는 기본 및 확장 작업의 전체 목록은 기능 섹션을 참고하세요.

langgraph.json 구성하기

langgraph.json 구성 파일에 checkpointer 키를 추가하세요. path는 앞서 정의한 비동기 컨텍스트 매니저를 가리켜요.

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "env": ".env",
  "checkpointer": {
    "path": "./src/agent/checkpointer.py:generate_checkpointer"
  }
}

서버 시작

서버를 로컬에서 테스트하세요:

langgraph dev --no-browser

서버 로그에서 커스텀 체크포인터가 활성화되었는지 확인할 수 있어요.

기능

서버는 시작 시 체크포인터의 기본(필수) 및 확장(선택) 기능을 확인해요. 확장 기능이 없으면 서버는 폴백을 사용하거나 해당 기능을 비활성화해요.

기본 기능 (필수)

확장 기능 (선택)

배포

이 앱을 그대로 LangSmith나 셀프 호스팅 플랫폼에 배포할 수 있어요.

다음 단계

• 델타 채널 지원을 포함한 커스텀 체크포인터 구축.
• 내장 장기 메모리 스토어를 교체하려면 커스텀 스토어 사용.
• LangGraph의 영속성 및 메모리에 대해 알아보기.

더 알아보기 (Learn more)

• 체크포인터 백엔드 구성
• 커스텀 스토어
• LangGraph 영속성