지속 실행 백엔드 만들기

지속 실행 백엔드 만들기 (Building a durable execution backend)

Pydantic AI의 지속 실행 빌더는 모델 요청, 도구 발견, 도구 검증, 도구 호출, 이벤트 처리, 메시지 압축, 장식된 캐퍼빌리티 연산을 하나의 엔진 백엔드로 라우팅하게 해 줘요. 아직 지원되지 않는 지속 실행 시스템을 통합할 때 사용합니다.

출처: 문서

본문

Temporal, DBOS, Prefect의 완전한 구현이 유용한 참고가 됩니다. 외부 Restate, AWS Lambda, Absurd 통합들은 JSON 저널과 함께 같은 공개 빌더를 보여줍니다.

백엔드 계층 선택 (Choose a backend tier)

엔진 SDK가 지속 단위를 호출할 때마다 콜백을 받아들인다면 CallableOperationBackend를 서브클래스하세요. execute를 구현해서 타입 있는 연산 식별자, 이름, 콜백, 캐시 식별자, 해석된 설정을 SDK에 전달하세요. 연산 식별자는 엔진이 영속된 이름에 의존하지 않고 연산 종류에 동작을 적용하게 해 줍니다.

엔진이 워커 시작 전에 핸들러 등록을 요구한다면 RegisteredOperationBackend를 쓰세요. register를 구현해서 바인딩된 콜러를 만들고 그 등록 핸들을 반환합니다. registrations() 메서드는 수집된 SDK 등록 핸들을 바인딩 순서대로 반환해요. 에이전트 조립 중에 베이스가 워커 시작 전에 네 가지 모델 연산을 바인딩하므로, 모델 요청을 먼저 실행하지 않아도 이 등록들이 존재합니다. 워커를 만들 때 완전한 registrations() 결과를 엔진 SDK에 전달하세요.

두 계층 모두 DurableOperationBackend를 구현합니다. 결과 인코딩, 캐시 투영, 설정 해석, 엔진 특정 프리미티브 주변의 이름 붙이기를 소유합니다.

최소 callable 백엔드 (Minimal callable backend)

이 완전한 인프로세스 예제는 각 연산을 즉시 실행해요. 실제 통합은 ImmediateBackend.execute를 SDK의 activity·step·task 호출로 바꾸고 in_durable_context를 엔진 런타임을 질의하도록 바꿉니다.

from collections.abc import Awaitable, Callable

from pydantic_ai import Agent
from pydantic_ai.durable_exec import (
    JSON_CODEC,
    BaseDurabilityCapability,
    DurabilityEngineSpec,
    DurableOperationId,
    JournalCallableOperationBackend,
    RoleBasedOperationConfig,
)
from pydantic_ai.models.test import TestModel


class TerminalError(Exception):
    pass


class ImmediateBackend(JournalCallableOperationBackend[None]):
    def __init__(self, agent_name: str, default_model_id: str | None) -> None:
        super().__init__(
            agent_name=agent_name,
            default_model_id=default_model_id,
            config=RoleBasedOperationConfig(model=None, event=None, capability=None, tool=None),
        )

    async def execute(
        self,
        *,
        operation_id: DurableOperationId,
        name: str,
        body: Callable[[], Awaitable[object]],
        cache_key: tuple[object, ...],
        config: None,
    ) -> object:
        return await body()


class ImmediateDurability(BaseDurabilityCapability[None]):
    engine_spec = DurabilityEngineSpec(
        engine_name='Immediate',
        durable_unit_noun='operation',
        durable_container_noun='run',
        codec=JSON_CODEC,
        serialization_failure=lambda exc: TerminalError(str(exc)),
    )

    @property
    def in_durable_context(self) -> bool:
        return True

    def get_durable_operation_backend(self) -> ImmediateBackend:
        return ImmediateBackend(self.name, self.default_model_id)


agent = Agent(TestModel(), name='example', capabilities=[ImmediateDurability()])
result = agent.run_sync('hello')
assert result.output == 'success (no tool calls)'

공개 DurabilityEngineSpec은 선언적 엔진 표면을 하나의 불변 객체로 묶어요. 필수 필드는 엔진, 지속 단위, 지속 컨테이너를 이름 짓습니다. 옵션 필드는 코덱, 감싸는 툴셋 종류, 라이프사이클 정책, 업그레이드 호환성, 지속 발견, 순차 도구 실행, 미지원 런타임 툴셋, 툴별 설정 키를 선택합니다. 기본 라이프사이클 정책은 매 런마다 함수·MCP 툴셋을 입력하고 동적 툴셋은 절대 입력하지 않아요.

엔진 동작이 기본값과 다른 모든 필드를 정의하세요. 스펙은 비어 있지 않은 명사를 검증하고, 생성 시 감싸는 모든 툴셋 종류에 라이프사이클을 요구하므로 무효한 엔진 선언은 그 클래스가 정의되는 동안 실패합니다. 라이프사이클 선택은 성공·오류·취소 시 툴셋 리소스가 닫히도록 보장해야 해요.

직렬화와 설정 (Serialization and configuration)

엔진 SDK가 파이썬 객체 직렬화를 소유하면 IDENTITY_CODEC을 쓰세요. 통합이 직접 JSON 호환 저널 페이로드를 쓴다면 JSON_CODEC을 써요. 둘 다 DurabilityCodec를 구현합니다. 인수, 결과, 도구 제어 흐름 신호, 장식된 캐퍼빌리티 연산이 모두 선택된 코덱 경계를 통과합니다.

JSON 저널 엔진은 DurabilityEngineSpec.serialization_failure를 설정해서 결정적 코덱 실패를 엔진의 종료 또는 비-재시도 가능 예외 타입으로 변환해야 해요. 그러한 값은 재시도 시 직렬화 가능해질 수 없습니다.

RoleBasedOperationConfig는 연산 역할당 하나의 설정을 공급하고, 툴별 오버라이드용 선택적 resolve_tool 콜백을 받아요. 콜백은 완전한 타입 있는 연산 ID, 도구 객체, 도구 이름을 받습니다.

JournalCallableOperationBackend는 callable 백엔드와 JournalOperationNamer를 결합합니다. 바인딩된 캐퍼빌리티의 default_model_id를 전달해서 에이전트의 기본 문자열 모델이 표준 비-접미사 영속 이름을 유지하게 하세요.

백엔드 설정 객체는 백엔드 설정 프로토콜을 구현합니다. 공개 base와 for_tool 메서드는 OperationConfigRole과 DurableOperationId를 받아요. 설정이 모델, 툴셋, 연산에 따라 다르면 구체적인 ID 변형을 일치시키세요. 역할은 대략적인 설정 버킷입니다: 'model', 'event', 'tool', 'capability'. 연산 ID는 세밀한 식별자를 담습니다. 캐퍼빌리티 연산 ID는 @durable_operation(name='...')의 명시적 이름을 포함합니다. 그 이름은 영속된 호환성 데이터가 되므로 파이썬 메서드 이름이 바뀌어도 안정적이어야 해서 필수예요. ID 유니온은 설치된 Pydantic AI 버전에 존재하는 ID를 나타냅니다. 툴별 설정은 함수·동적 도구를 지속 단위에서 제외하도록 False를 반환할 수 있어요. MCP 도구는 I/O를 수행하고 항상 지속 단위에서 실행되므로, 하나에 False를 반환하면 UserError가 발생합니다.

내장 ID는 ModelRequestId, ModelCompactMessagesId, ModelCancelSuspendedResponseId, EventStreamHandlerId, ToolsetGetToolsId, ToolsetGetInstructionsId, ToolsetValidateToolArgumentsId, ToolsetCallToolId, CapabilityOperationId예요. 파이썬 클래스 이름은 영속된 연산 이름을 결정하지 않습니다.

API 진화 (API evolution)

DurableOperationId는 Pydantic AI가 지속 단위를 추가함에 따라 마이너 릴리스에서 커져요. 샌드박스 연산이 계획된 예 중 하나입니다. 따라서 엔진 설정은 ID를 일치시킬 때 기본 분기를 포함해야 해요. 그 분기로 안전한 기본 설정을 적용하거나 실행 가능한 미지원-연산 오류를 발생시키세요. 현재 유니온이 절대 다른 팔을 얻지 않을 것이라고 가정하는 완전 일치에 의존하지 마세요.

영속 이름과 회복 (Persisted names and recovery)

연산 이름은 영속된 호환성 데이터예요. 파이썬 클래스 이름에 의존하지 않습니다. JournalOperationNamer가 표준 시퀀스 기반 이름 짓기 체계를 제공하고, 엔진에 다른 요구사항이 있으면 DurableOperationNamer를 구현하세요. 에이전트 이름, 모델 ID, 툴셋 ID, 캐퍼빌리티 ID, 연산 이름을 리팩터링하기 전에 테스트에서 생성된 모든 이름을 고정하세요. 마이그레이션 없는 이름 변경은 진행 중인 실행이 기록된 작업을 찾지 못하게 합니다.

네이머는 타입 있는 연산 ID와 선택적 keyword-only label을 받아요. 호출별 접미사가 필요한 연산은 선언 시 타입 있는 invocation_label callable을 제공합니다. 내장된 도구 호출·인수 검증 연산은 도구 이름을 사용합니다. 네이머는 연산의 파라미터 객체를 검사하면 안 됩니다.

지속 단위는 사이드 이펙트 후 체크포인트 커밋 전에 워커가 실패하면 두 번 이상 실행될 수 있어요. 따라서 엔진이 적절한 at-most-once 모드를 제공하지 않는 한 도구와 장식된 캐퍼빌리티 연산은 멱등(idempotent)이어야 합니다. 통합을 배포하기 전에 재생·회복, 리소스 정리, 제어 흐름 예외, 영속 출력 업그레이드, 지속 런타임 밖의 일반 실행을 테스트하세요.

캐퍼빌리티 작성자가 같은 백엔드를 통해 도착하는 장식 메서드를 기여하는 방법은 durable capability operations을 보세요.

지속 실행 백엔드 만들기