Short answer: - Within a single flow run, a failed PrefectAgent tool call is retried in isolation (just that tool task runs again). The agent’s prior reasoning and earlier tool calls are not re-executed. - If you manually retry the entire flow run (e.g., from the UI or with

prefect flow-run retry <id>

), Prefect 3.x will restart the flow but use durable/idempotent execution to skip already-completed AI/model/tool tasks. Only the work that failed will be re-executed. Below is a minimal, reproducible example you can run to see both behaviors. What it demonstrates - Tool-level retry: a flaky tool fails once, then succeeds on retry during the same flow run. - Flow-level retry: the flow intentionally fails after the agent completes. When you retry the flow run, the PrefectAgent reuses cached AI/model/tool work and does not re-run them. Code ``` import os from typing import Any from pydantic import BaseModel from pydantic_ai import Agent, RunContext from pydantic_ai.durable_exec.prefect import PrefectAgent, TaskConfig from prefect import flow from prefect import runtime # A flaky tool that fails the first time and succeeds on retry def flaky_tool(ctx: RunContext[None], value: int) -> int: attempt = runtime.task_run.run_count() print(f"[flaky_tool] attempt={attempt}, value={value}") if attempt == 0: # fail once to trigger tool-task retry raise RuntimeError("Boom! Simulated tool failure on first attempt.") return value * 2 # Structured output from the agent (keeps the demo simple/predictable) class Result(BaseModel): doubled: int def build_agent() -> PrefectAgent[None, Result]: # Base pydantic-ai agent: ask it to use the flaky tool to compute the result agent = Agent( "openai:gpt-4o-mini", # any supported model name="prefectagent-retry-demo", # name is required when wrapping with PrefectAgent tools=[flaky_tool], output_type=Result, system_prompt=( "You must call the tool 'flaky_tool' exactly once to double the given value, " "and then return Result(doubled=<the result>). No other output." ), ) # Wrap in PrefectAgent for durable execution and retries return PrefectAgent( agent, # Retries for the LLM call itself (if model API fails/transient errors) model_task_config=TaskConfig( retries=2, retry_delay_seconds=[1.0, 2.0], timeout_seconds=60.0, ), # Retries for any tool calls (this is what will catch our flaky_tool failure) tool_task_config=TaskConfig( retries=2, retry_delay_seconds=[0.5, 1.0], # capture prints from the tool for easy debugging log_prints=True, ), ) @flow(name="prefectagent-retry-semantics", log_prints=True) async def run_demo(initial_value: int = 7) -> Any: agent = build_agent() # Ask the agent to double the value using the tool print("[flow] Asking agent to double the value via flaky_tool...") resp = await agent.run( f"Double the number {initial_value} using the 'flaky_tool' tool and return only the structured result.", ) print(f"[flow] Agent result: {resp.output}") # Intentionally fail the first flow run so you can test a manual flow-run retry. # On retry (run_count > 0), we do not fail; you can then observe that the agent/tool # work is skipped thanks to durable execution and caching. flow_attempt = runtime.flow_run.run_count() print(f"[flow] flow_run.run_count={flow_attempt}") if flow_attempt == 0: raise RuntimeError("Intentional post-agent failure to demo 'prefect flow-run retry' behavior.") print("[flow] No crash on retry; flow completed successfully.") return resp.output if name == "__main__": # Option A: run the flow locally once to observe tool-level retry # - You'll see flaky_tool fail on attempt 0, then succeed on retry attempt 1.

# import anyio; anyio.run(run_demo, 7) # Option B (recommended): serve as a deployment to test both tool-level and flow-level retries # 1) Run this script: it will register and serve the deployment locally. # 2) Trigger a run from the UI or CLI. The first run will fail after the agent completes. # 3) Retry the same flow run from the UI or CLI to see the agent/tool work get skipped. run_demo.serve(name="prefectagent-retry-demo")

How to run and what to observe
- Install deps:

pip install "pydantic-ai[prefect]" prefect export OPENAI_API_KEY=... # if using an OpenAI model

- Start the script. It will register and serve a deployment named prefectagent-retry-demo.
- Trigger a flow run (UI or CLI). The first run will:
  - Call the tool and fail on the first attempt,
  - Retry the tool once (succeeds),
  - Then the flow intentionally crashes after the agent so you can test a flow-run retry.
- Retry the same flow run (manual retry):

prefect flow-run retry <FLOW_RUN_ID> ``` On the retried run, PrefectAgent will reuse cached/model/tool steps: - The LLM call and

flaky_tool

will not execute again (they’re cached due to durable execution). - The flow will skip straight past those steps and complete. Answers to your specific questions - If a tool call fails during a run: only that tool task is retried (per

tool_task_config

). The earlier steps of the agent are not re-run, and the flow does not restart. - If you retry the entire flow run (from UI/CLI): the flow restarts, but Prefect 3.x durable execution and caching skip already-completed AI/model/tool tasks. Only the failed work is re-executed. References - pydantic‑ai + Prefect durable execution - AI data analyst with pydantic‑ai (example) - Retries in Prefect 3.x - Transactions and idempotency in Prefect 3.x

Happy to! Can you paste the test script (or point me to the file/path) so I can patch it directly? In the meantime, here are two small ways to log cache hits in Prefect 3.x. Both rely on the fact that a cached task run ends in a state named "Cached". Option A — Check the state at the call site

from prefect import task, flow
from prefect.cache_policies import INPUTS

@task(persist_result=True, cache_policy=INPUTS)
def compute(x):
    return x * 2

@flow
def test_cache():
    # first call (likely executes)
    s1 = compute(5, return_state=True)
    print("first:", "CACHE HIT" if s1.name == "Cached" else "executed", s1.result())

    # second call with same input (cache hit)
    s2 = compute(5, return_state=True)
    print("second:", "CACHE HIT" if s2.name == "Cached" else "executed", s2.result())

if __name__ == "__main__":
    test_cache()

Option B — Add an on_completion hook to the task

import logging
from prefect import task, flow
from prefect.cache_policies import INPUTS

logger = logging.getLogger(__name__)

def log_cache_hit(task, task_run, state):
    if state.name == "Cached":
        <http://logger.info|logger.info>(f"Cache hit for task={task.name} run_id={task_run.id}")
    else:
        <http://logger.info|logger.info>(f"Executed task={task.name} run_id={task_run.id}")

@task(persist_result=True, cache_policy=INPUTS, on_completion=[log_cache_hit])
def compute(x):
    return x * 2

@flow
def test_cache():
    compute(5)
    compute(5)  # same input → should log a cache hit

if __name__ == "__main__":
    logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
    test_cache()

Notes: -

state.name == "Cached"

is the reliable indicator of a cache hit. -

persist_result=True

and a suitable

cache_policy

(e.g.,

INPUTS

) ensure the cache is used for identical inputs. - If you want to force a miss for testing, call the task with

refresh_cache=True

. Share your script and I’ll apply this directly.

Great goal. In Prefect 3.x, “agents” have been replaced by workers and work pools. Here’s a concise, durable reference architecture you can use, plus a minimal setup you can copy/paste. Assumptions: Prefect 3.x and you want HA, fault tolerance, and easy scale. If you’re on 2.x, I recommend upgrading or using workers in 2.x until you can. High-level reference architecture - Control plane: Prefect Cloud or Prefect Server (API/UI). This tracks state, schedules, retries, etc. - Execution plane: One or more work pools with multiple workers (HA), each submitting flow runs to your compute (Kubernetes, Docker, ECS, etc.). - Code + deployments: Deploy flows from source (git) with versioned images. Avoid embedding secrets in code. - Durability features: Persist results, make tasks idempotent, set retries/backoff/timeouts, and apply concurrency limits. - Event-driven orchestration: Use schedules and automations/triggers for SLAs, dependencies, and alerts. - Observability and recovery: Prefect manages leases/heartbeats—if a worker crashes, leases expire and capacity is freed automatically. Minimal implementation (durable by default) 1) Write resilient flows and tasks - Persist results so runs are resumable/auditable. - Use retries/backoff and timeouts to survive transient failures.

from datetime import timedelta
from prefect import flow, task

@task(
    retries=5,
    retry_delay_seconds=30,
    timeout_seconds=300,
    persist_result=True,
)
def fetch_data(source: str) -> dict:
    # call external API, db, etc.
    ...

@task(persist_result=True)
def transform(d: dict) -> dict:
    ...

@task(persist_result=True)
def load(d: dict) -> None:
    ...

@flow(name="etl", persist_result=True)
def etl(source: str = "default"):
    d = fetch_data(source)
    d = transform(d)
    load(d)

Tip: For heavy durability, enable default persistence in your environment:

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

. See Results &amp; persistence. 2) Deploy from source (durable, versioned) - This packages the flow from your repo and registers a deployment that a worker can pick up. Workers will pull your image and code at run time.

from prefect.flows import Flow

Flow.from_source(
    "<https://github.com/acme/data-pipelines>",
    "flows/etl.py:etl",
).deploy(
    name="etl-prod",
    work_pool_name="k8s-prod",
    image="<http://ghcr.io/acme/etl:{{|ghcr.io/acme/etl:{{> git_sha }}",  # reproducible image
    tags=["prod"],
    cron="0 * * * *",                       # also supports interval/rrule or event triggers
    concurrency_limit=5,                    # deployment-level limit
    parameters={"source": "primary"},
)

Links: - Deployments: deploy from source and images -

Flow.from_source

,

Flow.deploy

reference: Python deploy API 3) Run highly-available workers - Create a work pool (e.g., type “kubernetes”, “docker”, “ecs”) and run at least two worker replicas for HA. - On Kubernetes, run a Deployment with 2+ replicas of the Prefect worker that points at your pool; on VMs, run multiple

prefect worker start

processes under a supervisor (systemd, PM2, etc.). - Workers maintain leases and heartbeats—on crash or node loss, leases expire and capacity is automatically released. Docs: - Work pools &amp; queues - Workers overview 4) Add SLAs, automations, and event-driven triggers - Use automations for failure notifications, late-runs, and SLAs (e.g., “alert if not completed within 15m”, “start B when A succeeds”). - Combine cron/interval schedules with event triggers if you need reactive orchestration (e.g., run when a file lands, when an upstream flow completes, etc.). Docs: - Automations and triggers 5) Set concurrency limits where it matters

- Limit per deployment, per tag, or globally to protect shared resources (APIs, DBs). - Combine with work queues to prioritize important jobs. Docs: - Concurrency limits 6) Centralize secrets and config - Use Blocks for secrets and credentials (S3/GCS/DB) and Variables for runtime knobs. - Keep credentials out of images and code; bind them via worker/job environment.

from prefect.blocks.system import Secret
from prefect.variables import Variable

# One-time setup (via UI or code)
# Secret(value="...").save("my-api-key")
# Variable.set(name="batch_size", value="100")

api_key = Secret.load("my-api-key").get()
batch_size = int(Variable.get("batch_size", default=50))

Docs: - Blocks (secrets, storage, connections) - Variables 7) Storage for durability - Use remote result storage (S3/GCS/Azure) for persisted results and artifacts. - This enables resume, caching, and audits even if local disk is lost. Docs: - Results &amp; storage Notes on “agents” vs 3.x workers - Prefect 3.x removed the old agent model. Use work pools + workers instead. - Also removed in 3.x:

Deployment.build_from_flow()

and

prefect deployment build

. Use

Flow.from_source(...).deploy(...)

or

prefect deploy

. What you get from this setup - Survives worker crashes and node reboots (leases/heartbeats free capacity; new workers pick up remaining work). - Durable run history and results (remote result storage). - Idempotent, retrying tasks with timeouts and backoff. - Controlled concurrency and back-pressure at the deployment/tag/global levels. - Event-driven orchestration and SLAs with automated notifications. A few questions to tailor this for you - Where will you run your workers? (Kubernetes, VMs, Docker, ECS, other) - Do you prefer Prefect Cloud or self-hosted Prefect Server? - Target throughput and SLA/latency goals? - Any specific failure modes you need to harden against (e.g., spot/preemptible, frequent network partitions)? - Preferred result store (S3/GCS/Azure/local) and secret backend? If you share those, I’ll give you a concrete HA blueprint (including worker manifests/config) tuned to your stack. More docs to dig deeper - Deployments overview: Deployments - Workers and pools: Workers and Work pools - Results and persistence: Results - Automations/triggers: Automations - Concurrency limits: Concurrency