pankajkoti commented on code in PR #75358:
URL: https://github.com/apache/airflow/pull/75358#discussion_r4352252218
##########
providers/standard/src/airflow/providers/standard/version_compat.py:
##########
@@ -37,6 +37,7 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
AIRFLOW_V_3_1_3_PLUS: bool = get_base_airflow_version_tuple() >= (3, 1, 3)
AIRFLOW_V_3_2_PLUS: bool = get_base_airflow_version_tuple() >= (3, 2, 0)
AIRFLOW_V_3_3_PLUS: bool = get_base_airflow_version_tuple() >= (3, 3, 0)
+AIRFLOW_V_3_4_PLUS: bool = get_base_airflow_version_tuple() >= (3, 4, 0)
Review Comment:
Small question on the gate. `get_base_airflow_version_tuple()` drops the
pre-release part, so 3.4.0b1 counts as 3.4+. From the tags I checked, 3.4.0b1
has the upsert rewrite but not `request_created_at` (b2 has both), so the docs
line about a replaced request being refused would not hold there. Is a beta a
concern for you, or fine as is? Totally your call.
##########
providers/standard/src/airflow/providers/standard/operators/hitl.py:
##########
@@ -50,11 +66,165 @@
from airflow.sdk.exceptions import TaskAwaitingInput
if TYPE_CHECKING:
+ from datetime import datetime
+
from airflow.providers.common.compat.sdk import Context
from airflow.sdk.execution_time.hitl import HITLUser
from airflow.sdk.types import RuntimeTaskInstanceProtocol
+def _request_input_and_wait(
+ task: BaseOperator,
+ context: Context,
+ *,
+ options: list[str],
+ subject: str,
+ body: str | None,
+ defaults: list[str] | None,
+ multiple: bool,
+ params: dict[str, Any],
+ assigned_users: list[HITLUser] | None,
+ notifiers: Sequence[BaseNotifier],
+ timeout: timedelta | None,
+ timeout_datetime: datetime | None,
+ method_name: str,
+ kwargs: dict[str, Any] | None = None,
+) -> NoReturn:
+ """Write the Human-in-the-loop request, notify, and stop the task until a
response arrives."""
+ ti_id = context["task_instance"].id
+ upsert_hitl_detail(
+ ti_id=ti_id,
+ options=options,
+ subject=subject,
+ body=body,
+ defaults=defaults,
+ multiple=multiple,
+ params=params,
+ assigned_users=assigned_users,
+ )
+
+ task.log.info("Waiting for response")
+ for notifier in notifiers:
+ notifier(context)
+
+ if AIRFLOW_V_3_3_PLUS:
+ # Airflow 3.3+: park the task in AWAITING_INPUT -- no trigger, no
triggerer. The task
+ # is resumed by the Core API response handler or the scheduler timeout
sweep, so the
+ # triggerer no longer needs to run for Human-in-the-loop tasks to make
progress.
+ raise TaskAwaitingInput(method_name=method_name, kwargs=kwargs,
timeout=timeout)
+
+ # Fallback for Airflow versions < 3.3: defer the response check to
HITLTrigger on the triggerer.
+ task.defer(
+ trigger=HITLTrigger(
+ ti_id=ti_id,
+ options=options,
+ defaults=defaults,
+ params=params,
+ multiple=multiple,
+ timeout_datetime=timeout_datetime,
+ ),
+ method_name=method_name,
+ kwargs=kwargs,
+ )
+
+
+def request_input(
+ *,
+ subject: str,
+ options: list[str],
+ body: str | None = None,
+ defaults: str | list[str] | None = None,
+ multiple: bool = False,
+ params: ParamsDict | dict[str, Any] | None = None,
+ notifiers: Sequence[BaseNotifier] | BaseNotifier | None = None,
+ assigned_users: HITLUser | list[HITLUser] | None = None,
+ timeout: timedelta | None = None,
+ key: str | None = None,
+) -> InputResponse:
+ """
+ Ask a person for input from inside a running Python task, and return their
answer.
+
+ The first time a question is asked, the request appears on the Required
Actions page, the same
+ as one from :class:`HITLOperator`, and the task stops there: it waits in
the ``awaiting_input``
+ state without holding a worker slot. Once the request is answered, the
task runs again from the
+ start of its callable, and this call returns the answer instead of asking
again. Save anything
+ the second run should not redo (for example an agent framework's run
state) to
+ ``task_state_store`` before calling this.
+
+ It works in ``@task`` and ``PythonOperator`` callables (sync or async),
including ``@task.branch``
+ and ``@task.short_circuit``. It can't pause tasks that run their callable
in another process, such
+ as ``@task.virtualenv``, ``@task.external_python``, ``@task.docker`` or
``@task.kubernetes``.
+
+ Needs Airflow 3.4+. Earlier versions keep the first request's subject,
body and options when a
+ task asks again, after a resume or on a retry, so the reviewer would
answer a question they were
+ not shown.
+
+ :param subject: Headline shown to the reviewer.
+ :param options: Options the reviewer chooses from.
+ :param body: Details the reviewer needs to decide (Markdown).
+ :param defaults: Options applied if ``timeout`` passes without an answer.
Without defaults, the
+ task fails with
:class:`~airflow.providers.standard.exceptions.HITLTimeoutError` at the timeout.
+ :param multiple: Whether the reviewer can choose more than one option.
+ :param params: Form fields for the reviewer to fill in, defined like Dag
params.
+ :param notifiers: Notifiers called with the task context once the request
is written.
+ :param assigned_users: Users allowed to answer. Anyone with access can
answer if not set.
+ :param timeout: How long to wait for an answer. Waits indefinitely if not
set.
+ :param key: Which question this is, so a resumed task gets each answer
back at the right call.
+ Defaults to ``subject``. Each key is asked at most once per try. Set
it when the subject can
+ change between runs, or when the task can ask the same subject more
than once, for example
+ one approval per agent tool call (use the call's id).
+ """
+ if not AIRFLOW_V_3_4_PLUS:
+ raise AirflowOptionalProviderFeatureException("request_input() needs
Airflow 3.4+.")
+ context = get_current_context()
+ task = context["task"]
+ if not isinstance(task, PythonOperator) or isinstance(task,
_BasePythonVirtualenvOperator):
+ raise TypeError(
+ f"request_input() was called from {type(task).__name__}, which
cannot resume it. It works "
+ "only in tasks that run their Python callable in the task's own
process, such as @task "
+ "and PythonOperator."
+ )
+ defaults = [defaults] if isinstance(defaults, str) else defaults
+ params = params if isinstance(params, ParamsDict) else ParamsDict(params
or {})
+ validate_options(options)
+ validate_params(params)
+ validate_defaults(defaults, options, multiple)
+ key = subject if key is None else key
+ answer_log = answer_log_var.get()
+ if answer_log is not None and (answer := answer_log.take(key)) is not None:
+ # The code may have changed since the reviewer answered; never hand
back an option it no
+ # longer offers.
+ validate_chosen_options(answer["chosen_options"], options)
+ return answer
+ serialized_params = {name: params.get_param(name).serialize() for name in
params}
Review Comment:
A required field with no default, for example `params={"x":
Param(schema={"type": "string"})}`, serializes with `value: NOTSET`. The task
parks fine, but on resume `serde.deserialize` raises `TypeError` and the
fallback raises `KeyError`, so the task fails after the reviewer has already
answered. I reproduced this at the serde layer only, so I might be misreading
how the task runner calls it. `HITLOperator` doesn't hit it because it passes
`kwargs=None`. Would rebuilding the params from the original call or the
HITLDetail row work here? If you'd rather leave it for a follow-up, a clear
error at call time for a param with no default would also help.
##########
providers/standard/src/airflow/providers/standard/operators/hitl.py:
##########
@@ -50,11 +66,165 @@
from airflow.sdk.exceptions import TaskAwaitingInput
if TYPE_CHECKING:
+ from datetime import datetime
+
from airflow.providers.common.compat.sdk import Context
from airflow.sdk.execution_time.hitl import HITLUser
from airflow.sdk.types import RuntimeTaskInstanceProtocol
+def _request_input_and_wait(
+ task: BaseOperator,
+ context: Context,
+ *,
+ options: list[str],
+ subject: str,
+ body: str | None,
+ defaults: list[str] | None,
+ multiple: bool,
+ params: dict[str, Any],
+ assigned_users: list[HITLUser] | None,
+ notifiers: Sequence[BaseNotifier],
+ timeout: timedelta | None,
+ timeout_datetime: datetime | None,
+ method_name: str,
+ kwargs: dict[str, Any] | None = None,
+) -> NoReturn:
+ """Write the Human-in-the-loop request, notify, and stop the task until a
response arrives."""
+ ti_id = context["task_instance"].id
+ upsert_hitl_detail(
+ ti_id=ti_id,
+ options=options,
+ subject=subject,
+ body=body,
+ defaults=defaults,
+ multiple=multiple,
+ params=params,
+ assigned_users=assigned_users,
+ )
+
+ task.log.info("Waiting for response")
+ for notifier in notifiers:
+ notifier(context)
+
+ if AIRFLOW_V_3_3_PLUS:
+ # Airflow 3.3+: park the task in AWAITING_INPUT -- no trigger, no
triggerer. The task
+ # is resumed by the Core API response handler or the scheduler timeout
sweep, so the
+ # triggerer no longer needs to run for Human-in-the-loop tasks to make
progress.
+ raise TaskAwaitingInput(method_name=method_name, kwargs=kwargs,
timeout=timeout)
+
+ # Fallback for Airflow versions < 3.3: defer the response check to
HITLTrigger on the triggerer.
+ task.defer(
+ trigger=HITLTrigger(
+ ti_id=ti_id,
+ options=options,
+ defaults=defaults,
+ params=params,
+ multiple=multiple,
+ timeout_datetime=timeout_datetime,
+ ),
+ method_name=method_name,
+ kwargs=kwargs,
+ )
+
+
+def request_input(
+ *,
+ subject: str,
+ options: list[str],
+ body: str | None = None,
+ defaults: str | list[str] | None = None,
+ multiple: bool = False,
+ params: ParamsDict | dict[str, Any] | None = None,
+ notifiers: Sequence[BaseNotifier] | BaseNotifier | None = None,
+ assigned_users: HITLUser | list[HITLUser] | None = None,
+ timeout: timedelta | None = None,
+ key: str | None = None,
+) -> InputResponse:
+ """
+ Ask a person for input from inside a running Python task, and return their
answer.
+
+ The first time a question is asked, the request appears on the Required
Actions page, the same
+ as one from :class:`HITLOperator`, and the task stops there: it waits in
the ``awaiting_input``
+ state without holding a worker slot. Once the request is answered, the
task runs again from the
+ start of its callable, and this call returns the answer instead of asking
again. Save anything
+ the second run should not redo (for example an agent framework's run
state) to
+ ``task_state_store`` before calling this.
+
+ It works in ``@task`` and ``PythonOperator`` callables (sync or async),
including ``@task.branch``
+ and ``@task.short_circuit``. It can't pause tasks that run their callable
in another process, such
+ as ``@task.virtualenv``, ``@task.external_python``, ``@task.docker`` or
``@task.kubernetes``.
+
+ Needs Airflow 3.4+. Earlier versions keep the first request's subject,
body and options when a
+ task asks again, after a resume or on a retry, so the reviewer would
answer a question they were
+ not shown.
+
+ :param subject: Headline shown to the reviewer.
+ :param options: Options the reviewer chooses from.
+ :param body: Details the reviewer needs to decide (Markdown).
+ :param defaults: Options applied if ``timeout`` passes without an answer.
Without defaults, the
+ task fails with
:class:`~airflow.providers.standard.exceptions.HITLTimeoutError` at the timeout.
+ :param multiple: Whether the reviewer can choose more than one option.
+ :param params: Form fields for the reviewer to fill in, defined like Dag
params.
+ :param notifiers: Notifiers called with the task context once the request
is written.
+ :param assigned_users: Users allowed to answer. Anyone with access can
answer if not set.
+ :param timeout: How long to wait for an answer. Waits indefinitely if not
set.
+ :param key: Which question this is, so a resumed task gets each answer
back at the right call.
+ Defaults to ``subject``. Each key is asked at most once per try. Set
it when the subject can
+ change between runs, or when the task can ask the same subject more
than once, for example
+ one approval per agent tool call (use the call's id).
+ """
+ if not AIRFLOW_V_3_4_PLUS:
+ raise AirflowOptionalProviderFeatureException("request_input() needs
Airflow 3.4+.")
+ context = get_current_context()
+ task = context["task"]
+ if not isinstance(task, PythonOperator) or isinstance(task,
_BasePythonVirtualenvOperator):
+ raise TypeError(
+ f"request_input() was called from {type(task).__name__}, which
cannot resume it. It works "
+ "only in tasks that run their Python callable in the task's own
process, such as @task "
+ "and PythonOperator."
+ )
+ defaults = [defaults] if isinstance(defaults, str) else defaults
+ params = params if isinstance(params, ParamsDict) else ParamsDict(params
or {})
+ validate_options(options)
+ validate_params(params)
+ validate_defaults(defaults, options, multiple)
+ key = subject if key is None else key
+ answer_log = answer_log_var.get()
Review Comment:
I tried `ThreadPoolExecutor.submit`, parallel `asyncio.gather`, and
`asyncio.TaskGroup` around `request_input`, outside the real task runner. With
`submit` the thread has an empty context, so a resumed call looks like a first
ask and parks again. With `gather` the single HITLDetail row ends up with the
last question's subject while the parked key is the first. With `TaskGroup` the
park arrives as a `BaseExceptionGroup`, so the task fails. The docs say "sync
or async", so would a short caveat make sense? Also, the key-mismatch guards
raise `ValueError` and `RuntimeError`, so a wrapper that turns exceptions into
tool-error text can swallow them. Is "the task fails" in the docs meant to
cover that?
##########
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:
Minor. `HITLTimeoutError` and `HITLTriggerEventError` are unused in this
module now, and I found no in-repo importers. `HITLRejectException` is still
used. Would you prefer to drop the `noqa` re-export, or keep it for third-party
code with a short note?
##########
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:
Minor wording. The GET field is `created_at` and `request_created_at` is the
name on the answering call. "The request's `created_at`" may read clearer.
##########
providers/standard/src/airflow/providers/standard/operators/python.py:
##########
@@ -236,6 +245,43 @@ def __prepare_execution() ->
tuple[ExecutionCallableRunner, OutletEventAccessors
return return_value
+ def resume_with_input(
+ self,
+ context: Context,
+ event: dict[str, Any],
+ key: str,
+ options: list[str],
+ params: dict[str, dict[str, Any]],
+ answered: list[list[Any]],
+ ) -> Any:
+ """
+ Run the task again once the input it asked for with
``request_input()`` arrives.
+
+ The callable runs from the top. Each ``request_input()`` call that was
answered, in this or
+ an earlier pause, returns its answer instead of asking again, and the
run fails if it never
+ reads the answer it resumed with. A request that timed out without
defaults raises
+ :class:`~airflow.providers.standard.exceptions.HITLTimeoutError`.
+ """
+ if "error" in event:
+ raise_for_error_event(event)
+ validate_chosen_options(event["chosen_options"], options)
+ validate_params_input(event["params_input"],
params_from_serialized(params))
Review Comment:
Minor. `execute_complete` uses `event["params_input"] or {}` and this path
indexes it directly. Core sends `{}` today, so I don't think it fails now, but
mirroring `or {}` would keep the two consistent.
--
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]
