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 c65451f5397 Walk through the reviewer's side of Common AI approval
gates (#74737)
c65451f5397 is described below
commit c65451f5397fd0c37248147f4708eddd205aac2e
Author: Kaxil Naik <[email protected]>
AuthorDate: Thu Oct 15 15:15:38 2026 +0100
Walk through the reviewer's side of Common AI approval gates (#74737)
The approval-gates guide explained every parameter but never showed the
reviewer's side. It now starts from a complete Dag (an LLMOperator with
require_approval and allow_modifications, its input in the snippet, and a
downstream task that uses the approved text), then shows where the review
appears in the UI, what the form contains, and what approve, edit, reject
and each timeout setting do to the task state, the task log and the
downstream task. The screenshots, states and log lines come from a run on
Airflow main.
Running it showed that from Airflow 3.4 a retried or cleared task rewrites
the open review (subject, body, output field and assigned users), while 3.1
to 3.3 keep the first attempt's. The guide, the parameter list and the
operator and mixin docstrings stated the older behaviour unconditionally;
they now say which version does what.
---
providers/common/ai/docs/approval_gates.rst | 211 +++++++++++++++++----
.../ai/docs/img/approval_required_actions.png | Bin 0 -> 78595 bytes
.../common/ai/docs/img/approval_review_form.png | Bin 0 -> 95532 bytes
providers/common/ai/docs/operators/llm.rst | 7 +-
.../common/ai/example_dags/example_llm_approval.py | 54 ++++++
.../airflow/providers/common/ai/mixins/approval.py | 12 +-
.../airflow/providers/common/ai/operators/llm.py | 18 +-
7 files changed, 254 insertions(+), 48 deletions(-)
diff --git a/providers/common/ai/docs/approval_gates.rst
b/providers/common/ai/docs/approval_gates.rst
index e45d2735d51..a9907e1909b 100644
--- a/providers/common/ai/docs/approval_gates.rst
+++ b/providers/common/ai/docs/approval_gates.rst
@@ -21,15 +21,144 @@ Approval gates for LLM operators
================================
.. seealso::
- This page covers the one-shot approve, edit or reject gate on
``LLMOperator`` and its
- subclasses. For a multi-round review loop on ``AgentOperator`` with a chat
UI and REST
- API, see :doc:`hitl_review`.
+ For a multi-round review loop on ``AgentOperator`` with a chat UI and REST
API, see
+ :doc:`hitl_review`.
-Set ``require_approval=True`` to pause the task after the LLM generates its
-output and wait for a human reviewer to approve or reject it via the Airflow
-HITL interface. Optionally allow the reviewer to edit the output before
-approving with ``allow_modifications=True``, and set a deadline with
-``approval_timeout``.
+Set ``require_approval=True`` and the task stops after the model answers. A
person
+reads the output in the Airflow UI and approves it, edits it, or rejects it,
and only an
+approved answer reaches the tasks downstream. To review each tool call an
agent makes
+rather than its final answer, see :doc:`tool_approval`.
+
+Approval gates need Airflow 3.1 or later. On an older Airflow version the
operator
+raises ``AirflowOptionalProviderFeatureException: require_approval=True needs
Airflow 3.1+.``
+when it is constructed, so the Dag file fails to import, and with it every Dag
+defined in that file. A dynamically mapped
+task (``.expand()``) is only constructed when it runs, so there the same error
+surfaces as a task failure, before the model is called.
+
+A Dag with an approval gate
+---------------------------
+
+``draft_release_note`` asks the model for a release note and waits for a
reviewer.
+``publish`` runs only after the note is approved, and receives the approved
text,
+including any edit the reviewer made:
+
+.. exampleinclude::
/../../ai/src/airflow/providers/common/ai/example_dags/example_llm_approval.py
+ :language: python
+ :start-after: [START howto_operator_llm_approval_walkthrough]
+ :end-before: [END howto_operator_llm_approval_walkthrough]
+
+The Dag runs as-is once the ``pydanticai_default`` connection points at a
model; see
+:doc:`connections/pydantic_ai`.
+
+Where the review appears
+------------------------
+
+Once the model has answered, ``draft_release_note`` writes its review request
and pauses.
+On Airflow 3.3 and later the task is in the ``awaiting_input`` state and holds
no worker
+slot. The last line of its log, inside the **Post Execute** group that the log
view
+shows collapsed, is:
+
+.. code-block:: text
+
+ Pausing task as AWAITING_INPUT.
+
+On Airflow 3.1 and 3.2 the task is ``deferred`` instead, and the triggerer
watches for the
+answer, so a triggerer must be running.
+
+Every review is listed under **Browse > Required Actions**. An open one
carries the
+**Approval Required** badge; once answered, the row stays and the badge
changes to
+**Approval Received** or **Rejection Received**:
+
+.. image:: img/approval_required_actions.png
+ :alt: The Required Actions page listing three open reviews for
draft_release_note, each with an Approval Required badge
+
+The subject links to the review. The task instance page opens it too: the task
shows the
+**Awaiting Input** badge, and its **Required Action (1)** tab holds the form.
+
+What the reviewer sees
+----------------------
+
+.. image:: img/approval_review_form.png
+ :alt: The Required Action tab of draft_release_note, showing the prompt
and the model's answer, an editable output field, and Approve and Reject buttons
+
+The form shows:
+
+* The subject, :literal:`Review output for task \`draft_release_note\`` by
default.
+* The body: the rendered prompt, then the model's output. The system prompt is
not shown.
+* An ``output`` field holding the model's output, which the reviewer can
change before
+ approving. It is there because the Dag sets ``allow_modifications=True``;
without it the
+ form has only the two buttons.
+* **Approve** and **Reject**.
+
+After the reviewer clicks a button, the form is disabled and shows
+``Response received at <time> from <user>``, and the task resumes on a worker
to apply
+the answer.
+
+What each answer does
+---------------------
+
+Each row comes from a run of the Dag above, answered by the user ``admin``,
against a
+stand-in model that always returns the same sentence; your model's wording
will differ.
+The timeout rows set ``approval_timeout=timedelta(minutes=1)`` and the
+``on_approval_timeout`` value shown. An exception appears in the log under
+``Task failed with exception``.
+
+.. list-table::
+ :header-rows: 1
+
+ * - Answer
+ - Result
+ - ``draft_release_note`` log after it resumes
+ * - Approve
+ - ``draft_release_note`` succeeds; ``publish`` runs with the model's
output.
+ - ``Output approved by admin.``
+ * - Edit the ``output`` field, then Approve
+ - ``draft_release_note`` succeeds; ``publish`` runs with the edited text.
+ - ``Output approved by admin.``, then ``output=<edited text> modified by
the
+ reviewer=admin``
+ * - Reject
+ - ``draft_release_note`` fails; ``publish`` is upstream_failed.
+ - ``HITLRejectException: Output was rejected by the reviewer admin.``
+ * - No answer before ``approval_timeout``, with
``on_approval_timeout="fail"`` (the
+ default)
+ - ``draft_release_note`` fails; ``publish`` is upstream_failed.
+ - ``HITLTimeoutError: Approval timed out: The Human-in-the-loop response
timeout has
+ passed without a response.``
+ * - No answer, with ``on_approval_timeout="approve"``
+ - ``draft_release_note`` succeeds; ``publish`` runs with the model's
output.
+ - ``Output approved by the approval timeout default.``
+ * - No answer, with ``on_approval_timeout="reject"``
+ - ``draft_release_note`` fails; ``publish`` is upstream_failed.
+ - ``HITLRejectException: Output was rejected automatically:
approval_timeout expired
+ with on_approval_timeout='reject'.``
+
+After an approval, ``publish`` logs the text it received. For the plain
approval:
+
+.. code-block:: text
+
+ Publishing release note: Failed webhook deliveries are now retried three
times with backoff instead of dropping the event.
+
+For the edited one:
+
+.. code-block:: text
+
+ Publishing release note: Webhook deliveries that fail with a 5xx response
are now retried three times with exponential backoff.
+
+The approved text is the task's ``return_value`` XCom, which is what
``draft.output``
+passes to ``publish``.
+
+A rejection, or a timeout with ``"fail"`` or ``"reject"``, fails the task like
any other
+exception, so ``retries`` apply: with a retry left, the task goes to
``up_for_retry``, and
+the retry calls the model again and opens the review again, unanswered. On
Airflow 3.4
+and later the reopened review shows the new subject, body and output. On
Airflow 3.1 to
+3.3 it keeps the first attempt's subject, body and ``output`` field, so the
form shows the
+earlier output: with ``allow_modifications=True``, Approve returns that
earlier text
+unless the reviewer edits it; without it, Approve returns the regenerated
output, which
+the reviewer did not see. With ``retries=0``, the default, a rejection is
final.
+
+Editing structured output
+-------------------------
On ``LLMOperator``, ``LLMFileAnalysisOperator`` and
``LLMSchemaCompareOperator``, the edited text is
converted back into ``output_type`` when the task resumes. Write it as JSON
for the
@@ -54,56 +183,72 @@ text. A model with a serializer that cannot be reversed,
or a required field ex
the dump, shows text that does not validate as shown, so write an edit in the
form the model
accepts.
-Human-in-the-loop review needs Airflow 3.1+, whether ``require_approval`` or a
-``decision_policy`` with ``on_uncertain="review"`` opens it. On an older
Airflow version the
-operator raises ``AirflowOptionalProviderFeatureException`` when it is
constructed, so the Dag file
-fails to import, and with it every Dag defined in that file. A dynamically
mapped
-task (``.expand()``) is only constructed when it runs, so there the same error
-surfaces as a task failure -- still before the model is called.
+Timeouts
+--------
+
+``approval_timeout`` bounds the wait; the default ``None`` waits indefinitely.
On Airflow
+3.3 and later the scheduler resolves expired reviews, checking every
+``[scheduler] trigger_timeout_check_interval`` seconds (15 by default), so a
review can
+stay open that long past its deadline. On Airflow 3.1 and 3.2 the triggerer
enforces the
+deadline. ``on_approval_timeout="approve"`` returns the generated output when
the
+timeout expires. ``"reject"`` still fails this operator; only
+:class:`~airflow.providers.common.ai.operators.llm_branch.LLMBranchOperator`
turns a
+rejection into a downstream skip. The chosen option is also pre-highlighted as
the
+default in the review form, so ``"reject"`` makes Reject the primary button.
-When ``approval_timeout`` expires without a review, the task fails by default.
-Set ``on_approval_timeout="approve"`` to return the generated output instead,
so
-an unattended pipeline keeps moving. ``"reject"`` answers the review with a
-rejection, which still fails this operator; only
-:class:`~airflow.providers.common.ai.operators.llm_branch.LLMBranchOperator`
-turns a rejection into a downstream skip. The chosen option is also
-pre-highlighted as the default in the review form, so ``"reject"`` makes
-Reject the primary button:
+This example approves the output if nobody answers within 24 hours, and logs
each new
+review through a notifier:
.. exampleinclude::
/../../ai/src/airflow/providers/common/ai/example_dags/example_llm.py
:language: python
:start-after: [START howto_operator_llm_approval]
:end-before: [END howto_operator_llm_approval]
-A pending review is not surfaced as a notification. Pass
+It runs as-is with the same ``pydanticai_default`` connection.
+
+Notifying reviewers
+-------------------
+
+Airflow sends no notification when a review opens. Pass
``approval_notifiers`` to tell the reviewers about it through any Airflow
notifier (Slack, email, ...), the way
:class:`~airflow.providers.standard.operators.hitl.HITLOperator` does with
``notifiers``. The notifiers run once the review is open and can reference
the review ``{{ task.subject }}`` and ``{{ task.body }}`` in their templates,
-as the example above does. The ``@task.llm`` decorator and the operator
-subclasses accept the same parameter. A notifier whose delivery fails is
-logged and the task still waits for the review; a template error fails the
-task. A retry re-runs the LLM and re-notifies with the regenerated output,
-while the open review keeps the original subject and body.
+as the example under `Timeouts`_ does: its notifier writes the subject and
body to the
+task log, just before ``Pausing task as AWAITING_INPUT.`` The ``@task.llm``
decorator
+and the ``LLMOperator`` subclasses (``LLMBranchOperator``,
``LLMSQLQueryOperator``,
+``LLMSchemaCompareOperator``, ``LLMFileAnalysisOperator``) accept the same
+parameter. A notifier that raises is logged as
+``Approval notifier <notifier> failed; the review stays open`` and the task
still
+waits for the review. A Jinja ``TemplateError`` in a notifier's template
fails the
+task. A retry re-runs the LLM and re-notifies with the regenerated output.
What the
+reopened review shows depends on the Airflow version, as described under
+`What each answer does`_.
The default ``body`` contains the rendered prompt and the output. Where either
is sensitive, template only ``{{ task.subject }}`` and a link to the review
into channels outside Airflow's auth boundary.
-By default any user with the permission can answer the review. Pass
+Restricting who can review
+--------------------------
+
+By default any user allowed to edit the Dag's Human-in-the-loop details can
answer the
+review; under the FAB auth manager that is the ``can edit on HITL Detail``
permission,
+which the User role and every role above it have. Pass
``approval_assigned_users=[{"id": "<auth-manager-user-id>", "name":
"<user-name>"}]``
to restrict it to named reviewers, the way
:class:`~airflow.providers.standard.operators.hitl.HITLOperator` does with
``assigned_users``. ``id`` is the user id reported by the auth manager: with
the default ``SimpleAuthManager`` it is the username from
``simple_auth_manager_users``; under the FAB auth manager it is the numeric
-user row id as a string, not the username. This needs Airflow 3.1+. On
Airflow 3.1.0 through 3.1.5 both
+user row id as a string, not the username. On Airflow 3.1.0 through 3.1.5 both
``id`` and ``name`` must match what the auth manager reports, so a wrong
``name`` blocks the assigned reviewer as well as everyone else; from 3.1.6 only
-``id`` is compared. The list is stored when the review is first created:
-clearing the task re-runs it against the existing review row, so a changed
-list does not take effect.
+``id`` is compared. On Airflow 3.4 and later a cleared or retried task
replaces the
+list along with the rest of the review. On Airflow 3.1 to 3.3 the list is
stored when the
+review is first created, so if you edit ``approval_assigned_users`` and clear
the task,
+the old list still applies.
.. _approval-gates:uncertain:
diff --git a/providers/common/ai/docs/img/approval_required_actions.png
b/providers/common/ai/docs/img/approval_required_actions.png
new file mode 100644
index 00000000000..64ca24f77a0
Binary files /dev/null and
b/providers/common/ai/docs/img/approval_required_actions.png differ
diff --git a/providers/common/ai/docs/img/approval_review_form.png
b/providers/common/ai/docs/img/approval_review_form.png
new file mode 100644
index 00000000000..b5bafe89d41
Binary files /dev/null and
b/providers/common/ai/docs/img/approval_review_form.png differ
diff --git a/providers/common/ai/docs/operators/llm.rst
b/providers/common/ai/docs/operators/llm.rst
index 38c4bceacaa..aaecd73e7b9 100644
--- a/providers/common/ai/docs/operators/llm.rst
+++ b/providers/common/ai/docs/operators/llm.rst
@@ -288,8 +288,9 @@ Parameters
``dict`` of the same fields (templated via Jinja, then coerced per field
type).
Fails the task when token / request / tool-call budgets are exceeded, or
when a
templated dict value cannot be coerced. Default ``None``.
-- ``require_approval``: If ``True``, the task defers after generating output
and waits
- for human review. Default ``False``. Needs Airflow 3.1+.
+- ``require_approval``: If ``True``, the task pauses after generating output
and waits
+ for human review (``awaiting_input`` on Airflow 3.3+, deferred on 3.1 and
3.2).
+ Default ``False``. Needs Airflow 3.1+.
- ``approval_timeout``: Maximum time to wait for a review (``timedelta``).
``None``
means wait indefinitely. Default ``None``.
- ``on_approval_timeout``: Outcome when ``approval_timeout`` expires without a
@@ -305,7 +306,7 @@ Parameters
- ``approval_assigned_users``: Users allowed to answer the review, as
``{"id": ..., "name": ...}`` dicts where ``id`` is the auth manager's user
id.
``None`` (default) lets any user with the permission respond. Fixed at first
- run. Needs Airflow 3.1+.
+ run on Airflow 3.1 to 3.3; replaced on clear or retry from 3.4. Needs
Airflow 3.1+.
Logging
-------
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm_approval.py
b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm_approval.py
new file mode 100644
index 00000000000..ef0a20fdfc4
--- /dev/null
+++
b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_llm_approval.py
@@ -0,0 +1,54 @@
+# 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: LLM output that a person approves, edits or rejects before a
downstream task uses it."""
+
+from __future__ import annotations
+
+# [START howto_operator_llm_approval_walkthrough]
+from datetime import timedelta
+
+from airflow.providers.common.ai.operators.llm import LLMOperator
+from airflow.sdk import dag, task
+
+CHANGE = """
+Webhook deliveries that fail with a 5xx response are now retried three times
+with exponential backoff before the delivery is logged as failed.
+Previously a single failed delivery dropped the event.
+"""
+
+
+@dag(schedule=None, tags=["example"])
+def example_llm_approval():
+ draft = LLMOperator(
+ task_id="draft_release_note",
+ prompt=f"Write a one-sentence release note for this change:\n{CHANGE}",
+ llm_conn_id="pydanticai_default",
+ system_prompt="You write release notes for the users of a developer
platform. Be plain and specific.",
+ require_approval=True,
+ allow_modifications=True,
+ approval_timeout=timedelta(hours=24),
+ )
+
+ @task
+ def publish(note: str) -> None:
+ print(f"Publishing release note: {note}")
+
+ publish(draft.output)
+
+
+example_llm_approval()
+# [END howto_operator_llm_approval_walkthrough]
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py
b/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py
index a5854a855b9..af970dd636c 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/mixins/approval.py
@@ -147,15 +147,17 @@ class LLMApprovalMixin:
``{{ task.subject }}`` and ``{{ task.body }}`` in notifier templates. A
notifier whose delivery raises is logged without failing the task, while a
template error fails the task. A retry re-runs the LLM and re-notifies
- with the regenerated output, while the open review keeps the original
- subject and body.
+ with the regenerated output. On Airflow 3.4+ the reopened review shows the
+ regenerated output too; on Airflow 3.1 to 3.3 it keeps the original
subject,
+ body and editable output, so approving it unedited returns the first
+ attempt's output.
``approval_assigned_users`` restricts the review to the named users, the
way
:class:`~airflow.providers.standard.operators.hitl.HITLOperator` does with
``assigned_users``. Leaving it empty lets any user with the permission
- respond. The list is stored when the review is first created; clearing
- the task re-runs it against the existing review row, so a changed list
- does not take effect.
+ respond. On Airflow 3.1 to 3.3 the list is stored when the review is first
+ created, so clearing the task after changing it does not take effect; on
+ 3.4+ a cleared or retried task replaces it.
Operators that use this mixin must set the following attributes:
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py
b/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py
index 1033bc6c414..4913dcfad46 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/operators/llm.py
@@ -148,9 +148,10 @@ class LLMOperator(CancellableAgentRunMixin, BaseOperator,
LLMApprovalMixin, Usag
On Airflow >= 3.3, a set ``usage_limits`` bounds the whole task
instance:
usage from earlier attempts counts against it. See
:ref:`howto/operator:llm` for the full set of caveats.
- :param require_approval: If ``True``, the task defers after generating
- output and waits for a human reviewer to approve or reject via the
- HITL interface. Default ``False``. Needs Airflow 3.1+.
+ :param require_approval: If ``True``, the task pauses after generating
+ output (``awaiting_input`` on Airflow 3.3+, deferred on 3.1 and 3.2)
+ and waits for a human reviewer to approve or reject via the HITL
+ interface. Default ``False``. Needs Airflow 3.1+.
:param approval_timeout: Maximum time to wait for a review. When
exceeded, ``on_approval_timeout`` decides the outcome.
:param on_approval_timeout: What to do when ``approval_timeout`` expires
@@ -168,13 +169,16 @@ class LLMOperator(CancellableAgentRunMixin, BaseOperator,
LLMApprovalMixin, Usag
Default ``False``.
:param approval_notifiers: Notifiers called once the review is open, so a
reviewer is told about it. Only takes effect when a review is
- opened. A retry re-notifies with the regenerated
- output while the open review keeps the original subject and body.
- Default ``None``.
+ opened. A retry re-notifies with the regenerated output. On
+ Airflow 3.4+ the reopened review shows it too; on Airflow 3.1 to 3.3
+ the review keeps the original subject, body and editable output, so
+ approving it unedited returns the first attempt's output. Default
+ ``None``.
:param approval_assigned_users: Users allowed to answer the review, as
``{"id": ..., "name": ...}`` dicts where ``id`` is the auth manager's
user id. ``None`` (default) lets any user with the permission respond.
- The list is fixed when the review is first created. Needs Airflow
3.1+.
+ On Airflow 3.1 to 3.3 the list is fixed when the review is first
+ created; on 3.4+ a cleared or retried task replaces it. Needs Airflow
3.1+.
:param decision_policy: Experimental. A
:class:`~airflow.providers.common.ai.policies.decision.DecisionPolicy`
saying how confident the model has to be for the operator to return
its answer by
