This is an automated email from the ASF dual-hosted git repository. kaxil pushed a commit to branch commonai-approval-walkthrough in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 4b851b4350405bcdbcd3568702c9776e55cb11e5 Author: Kaxil Naik <[email protected]> AuthorDate: Wed Oct 14 14:30:06 2026 +0100 Walk through the reviewer's side of Common AI approval gates 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 | 208 +++++++++++++++++---- .../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, 251 insertions(+), 48 deletions(-) diff --git a/providers/common/ai/docs/approval_gates.rst b/providers/common/ai/docs/approval_gates.rst index a42ccf2786a..141c6ae74e7 100644 --- a/providers/common/ai/docs/approval_gates.rst +++ b/providers/common/ai/docs/approval_gates.rst @@ -21,66 +21,208 @@ 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`. - -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``. - -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 + For a multi-round review loop on ``AgentOperator`` with a chat UI and REST API, see + :doc:`hitl_review`. + +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. Use it where a wrong answer is costly to +publish or act on. 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 -- still before the model is called. +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**. -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: +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. + +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"`` keeps an +unattended pipeline moving with output nobody reviewed. ``"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. + +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 b789fa0472a..301e6ba69e9 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 @@ -303,7 +304,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 dc33aef7b73..2442270c815 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 @@ -113,15 +113,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 ac894d4903d..1a3fa805a69 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 @@ -166,13 +167,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
