This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 64c207d56f8 Document how common.ai agent tasks recover from provider
and worker failures (#74708)
64c207d56f8 is described below
commit 64c207d56f833225381c0df63ef42f0b06be2427
Author: Kaxil Naik <[email protected]>
AuthorDate: Wed Oct 14 12:20:27 2026 +0100
Document how common.ai agent tasks recover from provider and worker
failures (#74708)
Add a "Recovering from failures" page that states, per failure, which
setting handles it and whether completed steps are kept, replayed or run
again, with a runnable example combining fallback_conn_ids and
durable=True. Correct the timeout cost of a fallback chain in the
provider fallback and retry policy pages: the OpenAI and Anthropic SDKs
retry a timed-out request twice, and a rejected 400 is not retried.
---
providers/common/ai/docs/operations.rst | 3 +
providers/common/ai/docs/provider_fallback.rst | 22 +--
providers/common/ai/docs/recovery.rst | 164 +++++++++++++++++++++
providers/common/ai/docs/retry_policies.rst | 5 +-
.../ai/example_dags/example_agent_recovery.py | 70 +++++++++
5 files changed, 253 insertions(+), 11 deletions(-)
diff --git a/providers/common/ai/docs/operations.rst
b/providers/common/ai/docs/operations.rst
index 9a0ddb28025..0eae34cd5e6 100644
--- a/providers/common/ai/docs/operations.rst
+++ b/providers/common/ai/docs/operations.rst
@@ -22,6 +22,8 @@ Reliability and operations
An AI task fails in more ways than a SQL task, and it costs money each time it
runs.
+- :doc:`recovery` keeps an agent run going through a provider outage, and
replays the
+ steps it completed when every provider fails or the worker dies.
- :doc:`durable_execution` replays the model and tool calls an agent already
completed, so a
retry pays only for the steps that did not finish.
- :doc:`retry_policies` lets a model classify a failure and decide whether a
retry is worth
@@ -38,6 +40,7 @@ An AI task fails in more ways than a SQL task, and it costs
money each time it r
:hidden:
:titlesonly:
+ Recovering from failures <recovery>
Durable execution <durable_execution>
Retry policies <retry_policies>
Observability <observability>
diff --git a/providers/common/ai/docs/provider_fallback.rst
b/providers/common/ai/docs/provider_fallback.rst
index 5e5a3965242..a8a74574b5c 100644
--- a/providers/common/ai/docs/provider_fallback.rst
+++ b/providers/common/ai/docs/provider_fallback.rst
@@ -21,7 +21,9 @@ Provider fallback
A single ``llm_conn_id`` gives a task one provider. When that provider is
down, the task
fails and retries into the same outage. ``fallback_conn_ids`` gives the
connection an
ordered list of other connections to try, so a provider outage moves to the
next vendor
-inside the same task attempt.
+inside the same task attempt. The next vendor receives the run's message
history and
+continues an agent run from where it stopped; :doc:`recovery` covers that, and
what
+happens when every connection fails.
Configure it on the connection
------------------------------
@@ -146,12 +148,13 @@ Costs to know before configuring a long chain
---------------------------------------------
**The timeout multiplies.** pydantic-ai applies a ``ModelSettings`` timeout to
each model
-in the chain, not to the chain as a whole. A 30-second timeout across three
connections is
-a 90-second worst case for one call.
+in the chain, not to the chain as a whole, and the OpenAI and Anthropic SDKs
retry a
+timed-out request twice before giving up on that model. A 30-second timeout
across three
+such connections is therefore up to 270 seconds, plus backoff, for one call.
**There is no circuit breaker.** Every call tries the primary first. During an
outage each
-task instance pays the primary's timeout again before failing over, so 500
mapped tasks pay
-it 500 times. Keeping the primary's timeout short bounds both of these.
+task instance pays the primary's timeouts and retries again before failing
over, so 500
+mapped tasks pay them 500 times. Keeping the primary's timeout short bounds
both of these.
**Chains are not resolved recursively.** If a connection listed as a fallback
declares its
own ``fallback_conn_ids``, resolution fails with an error rather than
following it. List
@@ -162,10 +165,11 @@ connection.
``ModelAPIError`` family, which includes ``ModelHTTPError`` -- raised for any
4xx as well as
5xx. A malformed prompt is the one error every connection in the chain shares:
the same
request body goes to each of them, so all reject it alike before the task
finally sees the
-failure -- N requests, N timeouts, and N billable calls for a request that was
never going to
-succeed. An expired key does not cost the same way -- it is per-connection, so
the next
-connection in the chain presents its own credentials and, if they are still
valid, answers
-normally; that is the chain doing its job, not a repeated failure. A
misspelled model name is
+failure -- N requests for a request that was never going to succeed. Each
rejection comes
+back at once, and the OpenAI and Anthropic SDKs do not retry a 400, so the
cost is the
+requests themselves rather than N timeouts. An expired key does not cost the
same way -- it
+is per-connection, so the next connection in the chain presents its own
credentials and, if
+they are still valid, answers normally; that is the chain doing its job, not a
repeated failure. A misspelled model name is
shared across the chain only in the narrower case where the name is bare and
every fallback it
reaches configures no ``model`` of its own: a bare name that itself embeds a
``:`` (a vendor's
native id) only forwards to a fallback on the *same* platform, and a name that
already pins a
diff --git a/providers/common/ai/docs/recovery.rst
b/providers/common/ai/docs/recovery.rst
new file mode 100644
index 00000000000..5ccea3db064
--- /dev/null
+++ b/providers/common/ai/docs/recovery.rst
@@ -0,0 +1,164 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/recovery:
+
+Recovering from failures
+========================
+
+An agent task can fail partway through a run: the model provider goes down or
rate
+limits the task, or the worker running it dies. Two settings keep the work the
agent
+already did: ``fallback_conn_ids`` moves a failed model request to another
connection,
+and ``durable=True`` lets an Airflow retry replay the steps an earlier attempt
+completed. To decide whether a failure is worth retrying at all, see
+:doc:`retry_policies`.
+
+What each failure costs
+-----------------------
+
+.. list-table::
+ :header-rows: 1
+ :widths: 30 25 45
+
+ * - Failure
+ - Handled by
+ - What happens to the completed steps
+ * - The provider returns any HTTP 4xx or 5xx error, times out, or cannot be
reached
+ - ``fallback_conn_ids``, inside the attempt
+ - Kept. The next connection receives the run's message history, including
every
+ tool result, and continues the run.
+ * - Every connection in the chain fails, and the task raises
+ ``pydantic_ai.exceptions.FallbackExceptionGroup``
+ - Airflow ``retries`` with ``durable=True``
+ - Replayed. The retry replays the steps an earlier attempt recorded,
whichever
+ connection answered them, and continues on the first connection that
answers now.
+ * - The worker dies, or the task hits its ``execution_timeout``
+ - Airflow ``retries`` with ``durable=True``
+ - Replayed, as above. A step that was still running runs again.
+ * - Either of the last two, without ``durable=True``
+ - Airflow ``retries``
+ - Run again from the start: every model call is paid for again and every
tool runs
+ again.
+
+A response the provider does return but pydantic-ai rejects, such as a
content-filter
+refusal (``ContentFilterError``), does not fail over: the chain moves on only
for
+``ModelAPIError``.
+
+The OpenAI and Anthropic SDKs retry a connection error, a timeout, and a 408,
409, 429 or
+5xx response twice by default before the chain sees the failure. A refused
connection
+therefore costs a few seconds before failover, and a request that hangs costs
up to three
+times the ``timeout`` in ``model_settings``, plus backoff. Without a
``timeout``, each
+attempt waits for the SDK's default of 600 seconds. Both SDKs also wait out a
429's
+``Retry-After`` header before retrying: the OpenAI SDK for up to two minutes,
the
+Anthropic SDK for as long as the header asks.
+
+On Airflow before 3.3 the journal is a file under ``[common.ai]
durable_cache_path``; a
+retry that runs on another worker replays only if that path is shared storage.
See
+:doc:`durable_execution`.
+
+Keep a run going through an outage
+----------------------------------
+
+This task declares both settings: a standby connection for a failed request,
and a
+journal for a failed attempt:
+
+.. exampleinclude::
/../../ai/src/airflow/providers/common/ai/example_dags/example_agent_recovery.py
+ :language: python
+ :start-after: [START howto_agent_recovery]
+ :end-before: [END howto_agent_recovery]
+
+See :doc:`provider_fallback` for the chain itself, including how a bare model
name is
+forwarded to connections that set no ``model`` of their own. See
+:doc:`durable_execution` for the journal, including the ``id`` it needs on
every
+``FunctionToolset``, ``MCPToolset`` and ``DynamicToolset``.
+
+What the next connection receives
+---------------------------------
+
+Failover happens per model request. Each request carries the run's whole
message
+history: the prompt, every model response so far and every tool result. The
next
+connection in the chain gets that same history and continues the run; a tool
that
+already ran is not called again. The next connection can be another vendor's
model,
+for example an Anthropic model taking over a run an OpenAI model started.
+
+A streamed request is the exception. When the run streams, for example with an
+``event_stream_handler`` in ``agent_params``, the chain moves on only if the
stream fails
+to open; an error after the stream started propagates and fails the attempt.
+
+To drill the example, point ``llm_primary`` at a port nothing listens on. The
task
+succeeds, and its log shows the chain, the OpenAI SDK's two retries against
the primary,
+and the model that finished the run:
+
+.. code-block:: text
+
+ Resolving LLM fallback chain: llm_primary -> llm_standby
+ Retrying request to /responses in 0.459459 seconds
+ Retrying request to /responses in 0.913650 seconds
+ ::group::Tool call: list_tables
+ Retrying request to /responses in 0.485229 seconds
+ Retrying request to /responses in 0.849597 seconds
+ ::group::Tool call: get_schema
+ ...
+ Durable: replayed 0 steps (0 model, 0 tool, 0 other), recorded 7 new steps
(4 model, 3 tool, 0 other)
+ ::group::LLM run complete: model=claude-sonnet-5, requests=4,
tool_calls=3, input_tokens=5135, output_tokens=334, total_tokens=5469
+ Tool call sequence: list_tables -> get_schema -> query
+
+Each of the run's four model requests tried the primary first and paid its
retries
+before moving on. The chain keeps no memory of an outage between requests, so
keep the
+primary's ``timeout`` short.
+
+Retries after the whole chain fails
+-----------------------------------
+
+With ``durable=True``, each step is recorded in the task instance's journal
whichever
+connection answered it. When every connection fails, the task raises
+``FallbackExceptionGroup`` and Airflow retries it. The retry replays the
recorded steps,
+then sends its first live request through the chain from the top. The run
continues on
+the first connection that answers, whether or not that is the one that
answered before.
+
+A recorded model step replays only if the retry's request has the same
fingerprint, and
+for a chain the model in the fingerprint is the whole chain, whichever
connection
+answered. Adding, removing or reordering a connection in
``fallback_conn_ids``, or
+changing the ``model`` of a connection in it, between attempts therefore makes
the retry
+run every step live, tools included. Changing a connection's credentials or
host does
+not. "How a retry replays" in :doc:`durable_execution` gives the full matching
rules.
+
+What is not covered
+-------------------
+
+- A tool that sent an email and then raised sends it again on retry, and so
does one
+ whose worker died before its result was recorded. Make such tools
idempotent; see
+ "Side effects and idempotency" in :doc:`durable_execution`.
+- A request every provider rejects, such as a malformed one that each returns
a 400
+ for, walks the whole chain on every attempt. A ``RetryRule`` cannot stop
those retries
+ without stopping outage retries too: once a chain is configured, both raise
+ ``FallbackExceptionGroup``, and a rule compares only that top-level type.
Retry
+ policies need Airflow 3.3 or later; see "When the connection also carries a
fallback
+ chain" in :doc:`retry_policies`.
+- No connection type builds the ``AnthropicVertex`` client that Claude on
Vertex AI
+ needs, so a chain cannot include Claude on Vertex AI. A Vertex AI connection
reaches
+ Gemini.
+- Failover and replay cover pydantic-ai agents only: ``fallback_conn_ids`` on
the
+ pydantic-ai hooks and the operators built on them, and replay through
``durable=True``
+ or the ``AirflowDurability`` capability in a plain ``@task``. A Strands
+ Agents agent run through ``invoke_durably`` resumes from its own checkpoint
on
+ Airflow 3.3 and later (:ref:`howto/frameworks:strands-durable`), and a
retried
+ ``LLMBatchOperator`` re-attaches to its running batch
(:ref:`llm-batch-reattach`).
+- A ``ManagedAgentToolset`` runs again on every attempt unless it is built with
+ ``replayable=True``. Durable execution cannot be combined with
``enable_hitl_review``,
+ a ``CodeMode`` capability, a ``SandboxToolset`` or tool approval; see the
limitations
+ in :doc:`durable_execution`.
diff --git a/providers/common/ai/docs/retry_policies.rst
b/providers/common/ai/docs/retry_policies.rst
index 5b19a1f2966..1ef857944f6 100644
--- a/providers/common/ai/docs/retry_policies.rst
+++ b/providers/common/ai/docs/retry_policies.rst
@@ -368,8 +368,9 @@ retry behaviour with no change to the Dag. Two things
follow:
* ``timeout`` stops bounding the whole classification call. pydantic-ai
applies a
``ModelSettings`` timeout to each model in the chain, not to the chain as a
whole, so a
- 30-second ``timeout`` across a three-connection chain is a 90-second worst
case before the
- policy falls back to ``fallback_rules``.
+ 30-second ``timeout`` across a three-connection chain can take up to 90
seconds before the
+ policy falls back to ``fallback_rules``, or 270 when the connections use the
OpenAI or
+ Anthropic SDK, which retry a timed-out request twice.
* If every connection in the chain fails, the classification call raises
``pydantic_ai.exceptions.FallbackExceptionGroup``. ``evaluate()`` still
degrades to
``fallback_rules`` correctly -- it catches the broad ``Exception``, and an
exception group is
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_agent_recovery.py
b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_agent_recovery.py
new file mode 100644
index 00000000000..53b2d3210d1
--- /dev/null
+++
b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_agent_recovery.py
@@ -0,0 +1,70 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example Dag: an agent task that keeps its progress through a provider outage.
+
+``fallback_conn_ids`` moves a failed model request to the next connection
inside the
+attempt, and that connection continues the run from the same message history.
+``durable=True`` keeps what each attempt completed, so when every connection
fails, or
+the worker running the task dies, the retry replays those steps and continues
the run
+on whichever connection answers.
+
+Prerequisites:
+ - Connection ``llm_primary`` with ``conn_type='pydanticai'``,
``password=<API key>``
+ and ``extra='{"model": "openai:gpt-5-mini"}'``
+ - Connection ``llm_standby`` with ``conn_type='pydanticai'``,
``password=<API key>``
+ and ``extra='{"model": "anthropic:claude-sonnet-5"}'``
+ - Connection ``warehouse`` to a database with an ``orders`` table
+ - The provider installed with the ``anthropic``, ``openai`` and ``sql``
extras
+ - On Airflow before 3.3, ``[common.ai] durable_cache_path`` set to shared
storage
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+
+from airflow.providers.common.ai.operators.agent import AgentOperator
+from airflow.providers.common.compat.sdk import dag
+
+try:
+ from airflow.providers.common.ai.toolsets.sql import SQLToolset
+except Exception:
+ SQLToolset = None # type: ignore[assignment,misc]
+
+
+# [START howto_agent_recovery]
+if SQLToolset is not None:
+
+ @dag(
+ default_args={"retries": 2, "retry_delay": timedelta(minutes=1)},
+ catchup=False,
+ tags=["example", "fallback", "durable"],
+ )
+ def example_agent_recovery():
+ AgentOperator(
+ task_id="investigate_orders",
+ prompt="Which region placed the most orders, and what share of all
orders is that?",
+ llm_conn_id="llm_primary",
+ fallback_conn_ids=["llm_standby"],
+ durable=True,
+ system_prompt="You are a SQL analyst. Use the tools to answer with
data.",
+ toolsets=[SQLToolset(db_conn_id="warehouse",
allowed_tables=["orders"], max_rows=20)],
+ )
+
+ # [END howto_agent_recovery]
+
+ example_agent_recovery()
