에이전트 애플리케이션은 실행 도중 실패할 수 있습니다. 네트워크 호출이 끊기거나, 서버 프로세스가 충돌하거나, LLM 요청이 시간 초과될 수 있기 때문입니다. 내구성이 확보되지 않으면 이러한 실패는 진행 상황의 손실을 의미하며, 프로세스를 처음부터 다시 시작해야 할 때 반복적인 (낭비되는) LLM 호출을 유발합니다.

Pydantic AI는 내구성 있는 실행 엔진인 Temporal과 네이티브하게 통합됩니다. Temporal은 워크플로우(workflow) 실행의 각 단계를 이벤트 히스토리(event history)로 기록하므로, 프로세스가 충돌해도 처음부터 다시 시작하는 대신 중단했던 지점부터 재개할 수 있습니다.

Pydantic AI는 TemporalAgent 클래스를 통해 이를 노출하며, 이 클래스는 기존 에이전트를 래핑하여 그 도구 호출(tool calls), MCP 호출, 모델 요청을 Temporal Activity로 변환합니다.

from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import TemporalAgent

...

temporal_agent에 대해 .run(...)을 호출하는 방식은 기본 에이전트에 호출하는 방식과 동일합니다. 차이점은 내부적으로 작동한다는 것입니다: 각 도구 호출, MCP 호출 또는 모델 요청이 이제 Temporal Activity로 실행됩니다.

세 가지 핵심 개념

Activity — 작업 단위(unit of work)를 수행하는 함수입니다. LLM 호출, API 요청, 파일 쓰기 등이 될 수 있습니다. 만약 실패하면, Temporal은 구성된 재시도 정책(retry policy)에 따라 이를 재시도할 수 있습니다.

from temporalio import activity

@activity.defn
...

Workflow — Activity를 오케스트레이션하며, 무엇이 어떤 순서로 실행될지 정의합니다. Workflow 코드는 상태를 재구성하기 위해 Temporal에 의해 재생(replay)되므로 결정론적(deterministic)이어야 합니다 (직접적인 네트워크 호출 금지, 무작위 값 금지, 시스템 시계 읽기 금지).

from temporalio import workflow
from datetime import timedelta

...

Worker — 예약된 작업을 작업 큐(task queue)에서 폴링하고 실행합니다. Workflow나 Activity는 실행되기 전에 Worker에 등록되어야 합니다.

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
...

작업 큐 이름이 워크플로우를 시작하는 코드와 이를 실행하는 Worker 사이의 유일한 연결 고리입니다. 이 둘은 직접적으로 호출하지 않습니다.

Workers는 Activities와 Workflows를 다르게 실행합니다: Activities는 스케줄링될 때 함수 본문을 실행하며, 재시도 정책에 따라 실패 시 재시도될 수 있습니다. 반면, Workflows는 진행할 때마다 이벤트 기록(event history)을 기반으로 재생됩니다. 이것이 워크플로우가 충돌 후 새로운 Worker에서 올바르게 재개될 수 있게 하는 원리입니다.

Signal 및 Query

Workflow가 실행 중일 때, Signal과 Query를 사용하여 외부에서 상호작용할 수 있습니다.

Signal은 반환 값이 없이 비동기적으로 실행 중인 Workflow에 데이터를 전송합니다:

@workflow.signal
async def approve_plan(self) -> None:
    self._plan_approved = True
...

Signals는 Workflow가 여전히 실행 중일 때만 작동하며, 완료되거나 실패하거나 종료되면 더 이상 Signal을 받을 수 없습니다.

Query는 부수 효과 없이 동기적으로 Workflow의 현재 상태를 읽어옵니다:

@workflow.query
def state(self) -> WorkflowState:
    return WorkflowState(
...

Queries는 또한 네임스페이스의 보존 기간 내에 완료된 Workflow에 대해서도 실행될 수 있습니다. 하지만 종료된 Workflows에는 적용되지 않으며, 이는 안전하게 쿼리할 수 없습니다.

Human In The Loop (사람 개입)

Temporal은 전체 이벤트 기록을 유지하기 때문에, Workflow는 상태를 잃지 않고 일시 중지했다가 재개될 수 있으며, 이는 사람의 승인 단계에 적합한 패턴입니다.

from temporalio import workflow

@workflow.defn
...

workflow.wait_condition은 주어진 조건이 참일 때까지 실행을 일시 중지합니다. 여기서는 Workflow가 Signal(위의 approve_plan과 같은)이 self._plan_approved를 설정할 때까지 기다립니다. 이 Workflow를 실행하는 프로세스가 일시 중지된 상태에서 충돌하더라도, 새로운 Worker가 이를 가져와서 동일한 지점에서 대기하는 것을 재개할 수 있습니다.

이 예제는 타임아웃이 없으므로 무기한 대기합니다. 실제 운영 환경에서는 일반적으로 wait_condition에 timeout을 전달하고 결정을 내리지 못했을 경우를 처리해야 합니다.

Resources

Resources

• 이 시리즈의 파트 4를 시청하세요: https://youtu.be/J0_GeI8Srzc
• 이 예제의 전체 코드는 다음과 같습니다: https://github.com/bjoxiah/pydantic-ai-series/tree/agent-workflow

이는 제 채널에서 진행되는 Pydantic AI 마스터하기 시리즈의 일부입니다.