kaxil commented on code in PR #75358:
URL: https://github.com/apache/airflow/pull/75358#discussion_r4352909076
##########
providers/standard/src/airflow/providers/standard/operators/hitl.py:
##########
@@ -25,19 +25,35 @@
AIRFLOW_V_3_1_3_PLUS,
AIRFLOW_V_3_1_PLUS,
AIRFLOW_V_3_3_PLUS,
+ AIRFLOW_V_3_4_PLUS,
)
if not AIRFLOW_V_3_1_PLUS:
raise AirflowOptionalProviderFeatureException("Human in the loop
functionality needs Airflow 3.1+.")
from collections.abc import Collection, Mapping, Sequence
from datetime import timedelta
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, NoReturn
from urllib.parse import ParseResult, urlencode, urlparse, urlunparse
-from airflow.providers.common.compat.sdk import BranchMixIn, SkipMixin, conf
-from airflow.providers.standard.exceptions import HITLRejectException,
HITLTimeoutError, HITLTriggerEventError
+from airflow.providers.common.compat.sdk import BranchMixIn, SkipMixin, conf,
get_current_context
+from airflow.providers.standard.exceptions import ( # noqa: F401 - kept
importable from this module
Review Comment:
I'd keep them: they were importable from this module before, so third-party
code may rely on that. The noqa comment says so.
##########
providers/standard/docs/operators/hitl.rst:
##########
@@ -42,3 +42,69 @@ For an end-to-end guide on using HITL operators, including
examples and
recommended patterns, see the HITL tutorial:
* `HITL tutorial
<https://airflow.apache.org/docs/apache-airflow/stable/tutorial/hitl.html>`_
+
+.. _howto/operator:request_input:
+
+Asking for input from inside a Python task
+------------------------------------------
+
+Sometimes the question only comes up halfway through a task. An agent built
with the OpenAI
+Agents SDK, Strands or Google ADK, for example, stops in front of a tool call
and needs a person
+to approve it.
:func:`~airflow.providers.standard.operators.hitl.request_input` asks the
question
+and returns the answer:
+
+.. exampleinclude::
/../../standard/src/airflow/providers/standard/example_dags/example_hitl_request_input.py
+ :language: python
+ :start-after: [START howto_hitl_request_input]
+ :end-before: [END howto_hitl_request_input]
+
+The first time the task reaches ``request_input()``, the question appears on
the Required Actions
+page and the task stops there. It waits in the ``awaiting_input`` state
without holding a worker
+slot. When the reviewer answers, the task runs again from the start of its
callable, and this
+time the same call returns the answer: a dict with ``chosen_options``,
``params_input``,
+``responded_at`` (a ``datetime``), ``responded_by_user`` and ``timedout``.
+
+``request_input()`` takes the same request fields as
+:class:`~airflow.providers.standard.operators.hitl.HITLOperator` (``subject``,
``options``, ``body``,
+``defaults``, ``multiple``, ``params``, ``notifiers``, ``assigned_users``), a
``timeout``, and a
+``key``. A task can ask several questions; each run returns every answer it
already has and stops
+at the first question that has none.
+
+Things to know:
+
+* Code before ``request_input()`` runs again on every run. Keep anything
expensive, such as an
+ agent framework's saved run state, in ``task_state_store`` so the next run
restores it instead of
+ redoing it.
+* Each answer goes back to the call with the same ``key``, which defaults to
``subject``, and each
+ key is asked at most once per try. Pass ``key`` when the subject can change
between runs (for
+ example when it includes a value computed fresh each run), or when the task
can ask the same
+ subject more than once, such as one approval per agent tool call: key on the
call's id.
+* The task fails instead of guessing when the keys don't line up: when it asks
a key whose answer
+ it has already read, and when a run that resumed with an answer asks a new
question, or finishes,
+ before reading that answer.
+* Every pause carries the answers the task has received so far in this try,
including their
+ ``params_input``, so keep the number of questions per try small (a handful,
not hundreds) and the
+ form fields short.
+* A side effect between two questions, such as issuing a refund before asking
whether to email the
+ customer, runs again when the task resumes after the second answer. Record
it in
+ ``task_state_store`` and skip it when the record is there:
+
+ .. code-block:: python
+
+ if task_state_store.get("refund_id") is None:
+ task_state_store.set("refund_id", issue_refund(order_id))
+
+* A retry asks every question again, but ``task_state_store`` keeps its values
across retries, so
+ the retry still sees what the failed try saved. Delete those values when you
no longer need them.
+* Without ``defaults``, the task fails with
+ :class:`~airflow.providers.standard.exceptions.HITLTimeoutError` when
``timeout`` passes. With
+ ``defaults``, it resumes with ``timedout`` set to ``True`` and
``responded_by_user`` set to
+ ``None``.
+* It works in ``@task``, ``@task.branch``, ``@task.short_circuit`` and
``PythonOperator`` callables,
+ sync or async. It can't pause tasks that run their callable in another
process
+ (``@task.virtualenv``, ``@task.external_python``, ``@task.docker``,
``@task.kubernetes``).
+* It needs Airflow 3.4+.
+* The Required Actions page refuses an answer to a request that the task has
since replaced. A
+ client answering through the REST API gets the same protection only if it
sends the
Review Comment:
Done, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
