Prefect와 함께하는 지속 실행
Prefect와 함께하는 지속 실행 (Durable Execution with Prefect)
Prefect는 Python에서 탄력적인 데이터 파이프라인을 만들기 위한 워크플로 오케스트레이션 프레임워크로, Pydantic AI에 네이티브로 통합되어 있어요. 이 문서에서는 PrefectDurability 캐퍼빌리티로 에이전트를 지속으로 만드는 법을 정리할게요.
출처: 문서
본문
지속 실행 (Durable Execution)
Prefect 3.0은 Python 워크플로에 트랜잭션 시맨틱을 가져와, 태스크를 원자적 단위로 묶고 실패 모드를 정의할 수 있게 해 줘요. 트랜잭션의 어떤 부분이 실패해도 전체 트랜잭션을 깨끗한 상태로 롤백할 수 있습니다.
- 플로우 (Flows) 는 워크플로의 최상위 진입점이에요. 태스크와 다른 플로우를 담을 수 있습니다.
- 태스크 (Tasks) 는 독립적으로 재시도·캐시·모니터링될 수 있는 개별 작업 단위입니다.
Prefect 3.0의 트랜잭션 오케스트레이션 접근은 워크플로를 자동으로 멱등(idempotent) 하게 만듭니다: 어떤 환경에서도 중복이나 불일치 없이 재실행 가능해요. 모든 태스크는 태스크 결과 레코드가 언제 어디에 영속되는지를 규율하는 트랜잭션 안에서 실행됩니다. 태스크가 동일한 컨텍스트에서 다시 실행되면 재실행하지 않고 이전 결과를 로드합니다.
아래 다이어그램은 Prefect로 만든 에이전트 애플리케이션의 전체 아키텍처를 보여줍니다. Prefect는 기본적으로 클라이언트 측 태스크 오케스트레이션을 사용하며, 스케줄링·모니터링 같은 고급 기능에는 선택적 서버 연결이 있어요.
+---------------------+
| Prefect Server | (Monitoring,
| or Cloud | scheduling, UI,
+---------------------+ orchestration)
^
|
Flow state, | Schedule flows,
metadata, | track execution
logs |
|
+------------------------------------------------------+
| Application Process |
| +----------------------------------------------+ |
| | Flow (Agent.run) | |
| +----------------------------------------------+ |
| | | | |
| v v v |
| +-----------+ +------------+ +-------------+ |
| | Task | | Task | | Task | |
| | (Tool) | | (MCP Tool) | | (Model API) | |
| +-----------+ +------------+ +-------------+ |
| | | | |
| Cache & Cache & Cache & |
| persist persist persist |
| to to to |
| v v v |
| +----------------------------------------------+ |
| | Result Storage (Local FS, S3, etc.) | |
| +----------------------------------------------+ |
+------------------------------------------------------+
| | |
v v v
[External APIs, services, databases, etc.]
자세한 내용은 Prefect 문서를 보세요.
지속 에이전트 (Durable Agent)
PrefectDurability 캐퍼빌리티를 붙여 어떤 Agent든 지속 실행을 추가할 수 있어요. 에이전트가 Prefect 플로우 안에서 실행되면, 이 캐퍼빌리티는 모델 요청, 도구 호출, MCP 통신을 Prefect 태스크로 라우팅합니다. 런을 지속으로 만들려면 @flow 안에서 agent.run()을 호출하세요.
에이전트는 어디서나 일반 Agent로 유지됩니다 — Prefect 플로우 밖에서는 캐퍼빌리티가 투명하고, 원래 에이전트·모델·MCP 서버를 평소처럼 쓸 수 있어요.
태스크와 플로우 코드 안의 이벤트 처리는 Streaming을 보세요.
에이전트에 지속 실행을 붙이는 간단하지만 완전한 예제입니다. Prefect와 함께 Pydantic AI를 설치하기만 하면 됩니다:
pip install pydantic-ai[prefect]
uv add pydantic-ai[prefect]
slim 패키지를 쓰고 있다면 prefect 옵션 그룹으로 설치할 수 있어요:
pip install pydantic-ai-slim[prefect]
uv add pydantic-ai-slim[prefect]
from prefect import flow
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectDurability
agent = Agent(
'openai:gpt-5.6-sol',
instructions="You're an expert in geography.",
name='geography', # (1)
capabilities=[PrefectDurability()], # (2)
)
@flow # (3)
async def answer(question: str) -> str:
result = await agent.run(question)
return result.output
async def main():
answer_text = await answer('What is the capital of Mexico?')
print(answer_text)
#> Mexico City (Ciudad de México, CDMX)
- (1) 에이전트의
name은 그 플로우와 태스크를 고유하게 식별하는 데 쓰입니다. - (2)
capabilities=[...]로 지속성을 붙입니다. 에이전트가 플로우 안에서 실행될 때 이 캐퍼빌리티가 모델 요청, 도구 호출, MCP 통신을 Prefect 태스크로 라우팅합니다. - (3)
agent.run()을 자신의@flow로 감싸 런을 지속으로 만듭니다.
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요; 다른 변경은 필요 없어요.)
같은 에이전트가 Prefect 플로우 안팎에서 동작하므로, PrefectDurability는 각각 Prefect 특정 래퍼 변형 없이도 다른 모든 캐퍼빌리티와 조립됩니다.
파이썬 애플리케이션에서 Prefect를 쓰는 법은 Python documentation을 보세요.
래퍼 에이전트 경로 (Wrapper-agent path, 더 이상 사용되지 않음)
더 이상 사용되지 않음 (Deprecated)
PrefectAgent는 Prefect 통합의 원래 래퍼 에이전트 경로이고 v3에서 제거됩니다. 새 코드는 위의PrefectDurability캐퍼빌리티를 사용하세요.
마이그레이션할 때는 런을 스스로 플로우로 감싸야 합니다. PrefectAgent는 run/run_sync를 Prefect 플로우로 자동으로 감쌌지만, PrefectDurability는 의도적으로 그러지 않아요 — 런은 agent.run()을 자신의 @flow 안에서 호출할 때만 지속됩니다. 생성자 인수를 옮겨 놓고 agent.run()을 직접 호출하면 동작하지만 지속되지 않는 런이 만들어져요.
진행 중인 플로우 런은 마이그레이션을 가로질러 캐시에서 재개되지 않습니다. PrefectAgent 아래에 기록된 태스크 결과는 태스크의 소스 코드에 키가 매겨지므로, 마이그레이션 배포 후 재시도되는 플로우 런은 기록된 결과를 재생하는 대신 모델 요청·도구 호출을 실시간으로 재실행합니다. 재실행이 중요하다면 전환 전에 진행 중인 플로우 런을 끝내세요.
어떤 에이전트든 PrefectAgent로 감싸서 모델 요청, 도구 호출, MCP 통신을 Prefect 태스크로 라우팅하는 지속 에이전트 변형을 얻을 수 있습니다:
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectAgent
agent = Agent('openai:gpt-5.6-sol', name='geography')
prefect_agent = PrefectAgent(agent) # Use `prefect_agent` in place of `agent`.
캐퍼빌리티로 마이그레이션한다는 것은 PrefectDurability를 붙이고 PrefectAgent가 예전에 대신 적용해 주던 플로우 데코레이터를 추가하는 것입니다:
-prefect_agent = PrefectAgent(agent)
-result = await prefect_agent.run(prompt)
+agent = Agent(..., capabilities=[PrefectDurability()])
+
+@flow
+async def answer(prompt: str) -> str:
+ result = await agent.run(prompt)
+ return result.output
Prefect 통합 고려사항 (Prefect Integration Considerations)
Prefect를 Pydantic AI 에이전트와 함께 쓸 때 워크플로가 올바르게 동작하도록 몇 가지 중요 고려사항이 있어요.
에이전트 요구사항 (Agent Requirements)
각 에이전트 인스턴스는 Prefect가 그 플로우와 태스크를 올바르게 식별·추적하도록 고유한 name을 가져야 해요.
자체 도구 나열·호출을 구현하는 툴셋(FunctionToolset, MCPToolset, DynamicToolset)은 고유한 id가 설정되어야 해요. 플로우 안에서 그 태스크를 식별하는 데 쓰이니까요.
런타임 캐퍼빌리티 (Capabilities at Runtime)
Temporal과 DBOS와 달리, Prefect는 지속 단위를 미리 등록하지 않고 호출당 태스크를 만들므로, 플로우 안에서 agent.run(capabilities=[...])에 전달된 캐퍼빌리티는 허용됩니다. 실행 중인 툴셋을 기여하는 캐퍼빌리티는 여전히 거부되는데, run(toolsets=...)을 거부하는 것과 같은 가드에 의해서예요. 툴셋이 에이전트의 툴셋이 감싸진 후 도착하기 때문입니다. 그것들은 에이전트 생성 시 붙이세요.
런타임 모델 선택 (Model Selection at Runtime)
Agent.run(model=...)은 모델 문자열('openai:gpt-5.6-sol' 같은)과 모델 인스턴스를 모두 지원해요. 모델 인스턴스는 태스크 경계를 넘어 직렬화할 수 없고, model_id 문자열에서 재구성하면 다른 모델을 만들 것입니다 — 워커 환경이 암시하는 어떤 공급자의 같은 모델 이름이므로, 요청이 다른 자격 증명으로 다른 엔드포인트에 갈 거예요. 그래서 미리 등록되지 않은 인스턴스는 UserError로 거부됩니다. 특정 인스턴스를 쓰는 방법은 두 가지예요: PrefectDurability에 models dict를 전달해 사전 등록하고 키로 참조(또는 등록된 인스턴스 전달)하거나, 모델 이름 문자열을 전달하고 ResolveModelId 캐퍼빌리티로 태스크 안에서 인스턴스를 만드세요 — 모델이 런의 deps에 의존할 때(예: 사용자별 자격 증명) 올바른 선택입니다. 모델 이름 문자열은 등록이 절대 필요 없어요. 생성 시 설정된 에이전트 자신의 모델은 항상 기본값으로 사용 가능합니다.
모델 문자열이 만들어지는 방식을 커스터마이즈 — 커스텀 공급자, 또는 런의 deps에 실린 사용자별 자격 증명 — 하려면 PrefectDurability 앞에 ResolveModelId 캐퍼빌리티를 추가하세요. 그것이 모든 문자열에 첫 기회를 얻고, 해석기는 런의 실제 deps로 태스크 안에서 다시 실행되므로 주어진 (model_id, deps)에 대해 결정적이어야 하며 외부 I/O를 수행하면 안 됩니다.
도구 감싸기 (Tool Wrapping)
에이전트 도구는 자동으로 Prefect 태스크로 감싸져 이점을 얻습니다:
- 재시도 로직: 실패한 도구 호출을 자동으로 재시도
- 캐싱: 도구 결과가 입력에 기반해 캐시됨
- 관측성: 도구 실행이 Prefect UI에서 추적됨
DynamicCapability이 기여한 것 포함 DynamicToolset의 경우, 도구 발견과 각 도구 호출이 Prefect 태스크로 실행되고, 플로우 재시도가 기록된 발견·도구 결과를 재생합니다. 팩토리가 per_run_step=False로 만들어지면, 런은 툴셋을 한 번 해석하고 각 태스크가 재사용하며, 첫 태스크가 필요할 때 입력하고 런의 나머지 동안 입력된 채로 둡니다. 툴셋을 만드는 것은 연결하는 것과 같지 않아요 — MCPToolset은 입력될 때까지 아무것도 열지 않습니다 — 그래서 연결은 태스크 안에서 일어나고 다른 태스크 실패처럼 재시도됩니다. 팩토리 자체는 플로우 코드에서 실행되고 그것이 재생될 때마다 다시 실행되므로, 아래 캐퍼빌리티 팩토리처럼 런의 deps에 대해 결정적이어야 합니다: 팩토리에서 툴셋을 만들고 I/O는 태스크에 맡기세요. 이것은 비-지속 런이 툴셋에 주는 라이프사이클이므로, 팩토리가 반환하는 MCPToolset의 cache_tools 같은 툴셋 자신의 캐싱은 태스크 사이에 버려지는 대신 플로우 밖에서처럼 동작합니다. per_run_step=True 팩토리는 요청한 대로 각 태스크 안에서 재해석됩니다.
MCPToolset의 경우, 도구 발견과 각 도구 호출이 Prefect 태스크로 실행되고, 서버는 플로우 코드가 아니라 그것이 필요한 첫 태스크 안에서 연결되므로, 실패한 연결은 그 태스크의 재시도 정책으로 커버됩니다. 플로우는 런이 끝날 때까지 세션을 보유하므로 서버는 런당 한 번 연결되고, 그 자신의 cache_tools가 이후의 발견 태스크에 답합니다. 한 프로세스의 동시 런은 플로우 밖처럼 서버 세션을 공유하며, 마지막 런이 끝나면 닫힙니다.
args_validator가 있는 도구는 Validate Tool Args: {name} 태스크를 받아 검증자의 I/O가 도구 호출처럼 체크포인트됩니다. 없는 도구는 추가 태스크를 받지 않고, metadata={'prefect': False}가 있는 도구는 그 호출과 함께 플로우 코드에서 검증됩니다. 검증은 승인과 지연 전에 실행되므로 거부된 인수는 승인자에게 닿지 않습니다. 검증자는 태스크 안에서 지연시킬 수도 있는데, 승인으로 재개하면 tool_call_approved가 설정된 채 검증이 다시 실행됩니다.
모든 도구에 대한 기본 TaskConfig를 tool_task_config로 PrefectDurability 생성자에 전달할 수 있어요. 툴별 설정은 도구의 metadata 필드에 살아있는데 — PrefectDurability는 'prefect' 키를 찾습니다. 도구 정의에 직접 메타데이터를 설정하거나, SetToolMetadata 캐퍼빌리티로 도구 선택에 걸쳐 적용할 수 있어요. 전체 선택기 어휘는 capabilities documentation을 보세요.
from pydantic_ai import Agent
from pydantic_ai.capabilities import SetToolMetadata
from pydantic_ai.durable_exec.prefect import PrefectDurability, TaskConfig
from pydantic_ai.toolsets import FunctionToolset
toolset = FunctionToolset(id='research')
@toolset.tool(metadata={'prefect': TaskConfig(timeout_seconds=10.0)}) # (1)
def fetch_data(url: str) -> str: ...
@toolset.tool(metadata={'prefect': False}) # (2)
def simple_tool() -> str: ...
agent = Agent(
'openai:gpt-5.6-sol',
name='research',
toolsets=[toolset],
capabilities=[
SetToolMetadata( # (3)
tools=['fetch_data', 'fetch_dataset'],
prefect=TaskConfig(timeout_seconds=10.0),
),
PrefectDurability(tool_task_config=TaskConfig(retries=3)), # (4)
],
)
- (1) 인라인: 도구 정의와 함께 태스크 설정을 선언. 툴별 설정은 기본
tool_task_config위에 병합됩니다. - (2) 그 도구에 대해 태스크 감싸기를 완전히 건너뛰려면
'prefect': False를 설정. - (3) 선택기 기반:
SetToolMetadata가 도구 선택('all', 이름 목록, dict, 또는 callable)에 걸쳐 같은 메타데이터를 적용. - (4)
tool_task_config가 모든 도구의 기본 설정을 정합니다.
이 옵트아웃은 함수·동적 도구에만 적용됩니다. MCP 도구는 I/O를 수행하고 항상 그 Prefect 태스크에서 실행되므로, MCP 도구의 metadata={'prefect': False}는 UserError를 냅니다.
스트리밍 (Streaming)
Agent.run_stream(), Agent.run_stream_events(), Agent.iter()는 Prefect 플로우 안에서 동작하지만, 그 이벤트는 실시간으로 전달되지 않고 버퍼링됩니다. 모델 스트림은 지속 태스크 안에서 실행되고, 그 이벤트는 태스크가 완료된 후 플로우에 재생됩니다.
I/O 사이드 이펙트가 있는 핸들러는 event_stream_handler=를 PrefectDurability에 전달하세요. 모델 이벤트는 각 모델 요청 태스크 안에서 실시간 전달되고, 각 도구 이벤트는 자체 이벤트 핸들러 태스크에서 전달됩니다. 그 이벤트별 태스크는 event_stream_handler_task_config=로 구성하세요. 다른 Prefect 태스크처럼, 태스크가 재시도되면 핸들러가 두 번 이상 실행될 수 있으니 사이드 이펙트를 멱등으로 유지하세요.
대안으로 ProcessEventStream을 등록하세요. 그 핸들러는 플로우 코드에서 실행되고, 플로우 재생 때 다시 실행되므로 결정적이어야 해요. 도구와 최종 출력 이벤트는 실시간으로 도착하고, 실제 캡처된 모델 이벤트는 각 모델 요청이 완료된 후 재생됩니다. 예제는 streaming docs를 보세요.
지속성 event_stream_handler=와 별도로 등록된 ProcessEventStream은 두 개의 별개 핸들러이며 각각 한 번 발동합니다. 지속 핸들러는 지속 태스크 안에서 실시간 이벤트를 받고, ProcessEventStream은 플로우 코드의 버퍼링된 재생을 봅니다.
Agent.run(event_stream_handler=...)에 전달된 런별 핸들러도 재생된 모델 이벤트에 대해 플로우 측에서 실행됩니다.
지속 태스크 안에서 — 캐퍼빌리티 자신의 도구가 발행한 캐퍼빌리티 이벤트 포함 — ctx.emit()으로 발행된 이벤트는 태스크가 실제로 실행될 때 전달되며, 플로우 재시도나 캐시 히트에서 기록된 결과가 재생될 때 재발행되지 않아요. 태스크 안에 쓴 로그 줄처럼, 발행된 이벤트는 기록된 결과의 일부가 아니라 태스크를 실행하는 사이드 이펙트입니다. 리스너가 매 시도에서 이벤트를 봐야 한다면 캐퍼빌리티 훅 같은 플로우 수준 코드에서 emit 하세요. @on_event로 등록된 캐퍼빌리티 리스너는 태스크가 아니라 플로우 코드에서 실행되므로 플로우 재시도 시 다시 실행되고 결정적이어야 해요. I/O는 자신의 태스크에서 실행되는 지속성 event_stream_handler=에 두세요. ctx.enqueue()는 태스크 안에서 버리는 것이 모델이 보는 것을 바꾸므로 거부되지만, 놓친 이벤트는 관찰자가 통지받지 않았다는 뜻일 뿐이에요. 지속 단위의 발행된 이벤트를 기록된 출력에 실어 재생이 재현하게 하는 것은 pydantic-ai#7971에서 추적합니다.
모델 스트림이 태스크 안에서 소비되므로, 플로우 측에서 취소하는 것(AgentStream.cancel() 같은)은 지속 경계를 넘어 사용할 수 없어요.
CancellationToken과 RunContext.cancel()은 같은 프로세스 취소 핸들이며 Prefect 지속 경계를 넘을 수 없어요. 대신 Prefect 플로우를 취소하세요.
Agent.run_stream_sync()는 플로우 코드용이 아니에요. 실행 중인 이벤트 루프가 필요 없고 run_stream()을 감쌉니다. PrefectDurability 아래에서는 위 버퍼링 async 스트리밍 API나 이벤트 스트림 핸들러가 있는 Agent.run()을 쓰세요. 플로우 밖에서는 PrefectDurability가 있는 에이전트가 일반 에이전트처럼 동작하므로 run_stream_sync()이 평소처럼 동작합니다. (래퍼 PrefectAgent는 플로우 안에서 run_stream을 금지합니다 — 거기서는 run + 이벤트 스트림 핸들러를 쓰세요.)
일시 정지 턴과 백그라운드 모드 (Suspended Turns and Background Mode)
공급자가 모델 턴을 도중에 일시 정지하거나(Anthropic pause_turn) 준비될 때까지 폴링되는 서버 측 작업으로 실행하면(OpenAI background mode), 각 세그먼트는 별도의 모델 요청 태스크에서 실행됩니다. 일시 정지된 ModelResponse와 백그라운드 작업 ID는 세그먼트 사이에 체크포인트되고, 최종 응답은 병합되며 사용량은 한 번 기록됩니다. 일시 정지된 응답으로 끝나는 message_history가 첫 태스크에 전달됩니다. Task Configuration의 timeout_seconds는 공급자 왕복 한 번으로 잡으세요. 오류가 일시 정지된 작업을 버리면, 그 공급자 정리는 전용 취소 태스크에서 실행됩니다.
런타임 툴셋 (Toolsets at Runtime)
지속 감싸기가 필요한 실행 중인 모든 툴셋을 에이전트 생성자에 전달해서 플로우 실행 전에 태스크가 등록되게 하세요. DynamicToolset 포함: 명시적 id를 주고 Agent(toolsets=[...])로 전달합니다. @agent.toolset 데코레이터는 엔진의 지속 단위가 만들어진 후 등록하므로, PrefectDurability 아래 플로우 안에서 쓰면 UserError가 납니다. 더 이상 사용되지 않는 PrefectAgent는 이 검사를 실행하지 않아요: 플로우 안에서 감싸는 시점에 동결된 태스크-감싸진 툴셋 목록을 실행하므로, 그렇게 늦게 등록된 툴셋은 조용히 런에서 빠집니다.
추가 툴셋은 agent.run(toolsets=...)으로 런별 전달할 수 있지만, 지속 감싸기가 필요 없는 툴셋만 지원됩니다: 에이전트 런 밖에서 실행되는 도구를 가진 ExternalToolset 같은 비-실행 툴셋과, 모든 도구가 metadata={'prefect': False}로 태스크 감싸기를 옵트아웃한 FunctionToolset이요. 다른 실행 툴셋(FunctionToolset, MCPToolset)과 런타임에 전달된 동적 툴셋은 UserError를 냅니다.
플로우 안에서 agent.override(toolsets=...)로 바꿔 넣은 툴셋도 같은 규칙에 묶입니다. 그것들도 에이전트의 태스크가 등록된 후 도착하기 때문이에요. 런타임에 추가된 툴셋은 에이전트를 생성할 때 만든 것의 id를 재사용할 수도 없는데, id가 도구 호출이 어떤 등록된 툴셋의 태스크로 디스패치되는지 식별하기 때문입니다.
태스크 설정 (Task Configuration)
TaskConfig 객체를 PrefectDurability 생성자에 전달해 재시도·타임아웃 같은 Prefect 태스크 동작을 커스터마이즈할 수 있어요:
mcp_task_config: MCP 서버 통신 태스크 설정model_task_config: 모델 요청 태스크 설정event_stream_handler_task_config: 이벤트 스트림 핸들러 태스크 설정tool_task_config: 모든 도구 호출의 기본 설정 (툴별 오버라이드는 도구의'prefect'메타데이터에 — 위 Tool Wrapping 참고)
사용 가능한 TaskConfig 옵션:
retries: 태스크의 최대 재시도 횟수 (기본:0)retry_delay_seconds: 재시도 사이 지연(초). 단일 값 또는 지수 백오프용 목록일 수 있음 (기본:1.0)timeout_seconds: 태스크 완료 최대 시간(초)cache_policy: 태스크의 커스텀 Prefect 캐시 정책persist_result: 태스크 결과를 영속화할지 여부result_storage: 태스크의 Prefect 결과 저장소 (예:'s3-bucket/my-storage'또는WritableFileSystem블록)log_prints: 태스크의 print 문을 로그할지 여부 (기본:False)
예제:
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectDurability, TaskConfig
agent = Agent(
'openai:gpt-5.6-sol',
instructions="You're an expert in geography.",
name='geography',
capabilities=[
PrefectDurability(
model_task_config=TaskConfig(
retries=3,
retry_delay_seconds=[1.0, 2.0, 4.0], # Exponential backoff
timeout_seconds=30.0,
),
),
],
)
async def main():
result = await agent.run('What is the capital of France?')
print(result.output)
#> Paris
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요; 다른 변경은 필요 없어요.)
재시도 고려사항 (Retry Considerations)
Pydantic AI와 공급자 API 클라이언트는 자체 재시도 로직이 있어요. Prefect를 쓸 때는 다음과 같이 하면 좋습니다:
- Pydantic AI에서 트랜스포트 재시도 비활성화
- 공급자 API 클라이언트의 재시도 로직 끄기 (예: 커스텀 OpenAI 클라이언트의
max_retries=0) - 일관성을 위해 Prefect의 태스크 수준 재시도 설정에 의존
이렇게 하면 요청이 서로 다른 계층에서 여러 번 재시도되는 것을 방지합니다. 계층은 곱해져요: 산술은 Retry multiplication을 보세요.
캐싱과 멱등성 (Caching and Idempotency)
Prefect 3.0은 내장 캐싱과 트랜잭션 시맨틱을 제공해요. 동일한 입력의 태스크는 결과가 이미 캐시되어 있으면 재실행되지 않으므로, 워크플로가 자연스럽게 멱등이고 실패에 탄력적입니다.
동적 도구 캐시 키가 바뀌었어요.
동적 도구 태스크 키는 이제 준비된 도구 정의를 포함합니다. 동적 도구의 기존 캐시 결과는 업그레이드 후 한 번 미스하고 재계산됩니다. 수동 캐시 삭제는 필요 없고, 이후 호출은 새 값 주소 키를 재사용합니다.
- 태스크 입력 (Task inputs): 모델 요청의 메시지·설정·파라미터; 도구 호출의 이름·인수·정의·
tool_call_id(그래서 같은 도구에 같은 인수의 두 병렬 호출이 각각 실행됩니다); 그리고 태스크의 작업이 의존할 수 있는 런 상태: 의존성,metadata,validation_context, 프롬프트, 메시지 히스토리.
run_id나 conversation_id 같은 런별 식별자와 메시지 타임스탬프는 의도적으로 빠져 있어, 그 외에는 동일한 런이 재실행 대신 기록된 결과를 재생합니다.
참고: 사용자 의존성이 캐시 키에 포함되려면 직렬화 가능해야 해요 (예: Pydantic 모델 또는 기본 파이썬 타입). 비-직렬화 값은 캐시 계산에서 자동으로 제외됩니다.
Prefect와 Logfire로 관측성 (Observability with Prefect and Logfire)
Prefect는 플로우 런, 태스크 실행, 실패를 모니터링하는 내장 UI를 제공해요. 다음을 할 수 있습니다:
- 실시간 플로우 런 상태 보기
- 전체 스택 트레이스로 실패 디버깅
- 알림·통지 설정
Prefect UI에 접근하려면 다음 중 하나를 쓸 수 있어요:
- Prefect Cloud (관리형 서비스) 사용
prefect server start로 로컬 Prefect 서버 실행
세부 관측성에는 Pydantic Logfire를 쓸 수도 있어요. Prefect와 Logfire를 모두 쓰면 상호 보완적인 보기를 얻습니다:
- Prefect: 워크플로 수준 오케스트레이션, 태스크 상태, 재시도 이력
- Logfire: 에이전트 런, 모델 요청, 도구 호출의 세밀한 트레이싱
Logfire를 Prefect와 함께 쓸 때 분산 트레이싱을 활성화하면 에이전트 런·모델 요청·도구 호출과 함께 Prefect 런의 스팬을 볼 수 있어요.
Prefect 모니터링에 대한 자세한 내용은 Prefect 문서를 보세요.
배포와 스케줄링 (Deployments and Scheduling)
Prefect 지속 에이전트를 배포·스케줄링하려면 Prefect 플로우로 감싸고 플로우의 serve() 또는 deploy() 메서드를 쓰세요:
from prefect import flow
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectDurability
@flow
async def daily_report_flow(user_prompt: str):
"""Generate a daily report using the agent."""
agent = Agent( # (1)
'openai:gpt-5.6-sol',
name='daily_report_agent',
instructions='Generate a daily summary report.',
capabilities=[PrefectDurability()],
)
result = await agent.run(user_prompt)
return result.output
# Serve the flow with a daily schedule
if __name__ == '__main__':
daily_report_flow.serve(
name='daily-report-deployment',
cron='0 9 * * *', # Run daily at 9am
parameters={'user_prompt': "Generate today's report"},
tags=['production', 'reports'],
)
- (1) 각 플로우 런은 격리된 프로세스에서 실행되며 모든 입력·의존성이 직렬화 가능해야 해요.
Agent인스턴스는 직렬화할 수 없으므로, 모듈 수준이 아니라 플로우 안에서 에이전트를 생성하세요.
serve() 메서드는 스케줄링 옵션을 받습니다:
cron: Cron 스케줄 문자열 (예:'0 9 * * *'— 매일 9시)interval: 초 또는 timedelta로 된 스케줄 간격rrule: iCalendar RRule 스케줄 문자열
Docker, Kubernetes 등의 프로덕션 배포에는 플로우의 deploy() 메서드를 쓰세요. 자세한 내용은 Prefect deployment documentation을 보세요.