kaxil opened a new pull request, #75358:
URL: https://github.com/apache/airflow/pull/75358
An agent built with the OpenAI Agents SDK, Strands or Google ADK usually
runs inside a plain `@task`. All three can stop before a tool call and wait for
a person, but Airflow had no way to connect that pause to Required Actions:
- **Polling doesn't work.** The review form only accepts an answer while the
task instance is `deferred` or `awaiting_input`, so a task that waits while
running can never be answered.
- **`TaskAwaitingInput` doesn't work either.** It resumes into a method
named on the operator, and a TaskFlow task has no method that accepts the
answer.
This adds two functions to `airflow.providers.standard.operators.hitl`:
- `request_input(...)` writes a Required Actions request with the same
fields as `HITLOperator` (`subject`, `options`, `body`, `defaults`, `multiple`,
`params`, `notifiers`, `assigned_users`) plus a `timeout`. It then stops the
task in `awaiting_input` without holding a worker slot.
- `get_input_response()` returns the answer once the task resumes, and
`None` on the first run.
```python
@task
def refund(order_id: str, task_state_store=None):
answer = get_input_response()
if answer is None:
task_state_store.set("amount", 42.5)
request_input(subject=f"Refund 42.5 for {order_id}?",
options=["Approve", "Reject"])
if answer["chosen_options"] == ["Approve"]:
return f"Refunded {task_state_store.get('amount')} for {order_id}"
return "Rejected"
```
When the request is answered, the callable runs again from the top. An agent
framework fits this naturally: save its run state (for example the OpenAI
Agents SDK's `RunState.to_string()`) to `task_state_store` before asking, then
rebuild and continue from it on the second run.
## Design rationale
**Why in the standard provider and not in core.** Everything this touches
outside core already lives in the standard provider: the TaskFlow operator,
`PythonOperator.execute`, and `HITLOperator`'s write-then-park path. The core
pieces it relies on are already public:
- `TaskAwaitingInput`;
- the task runner resuming into `next_method`;
- the response handler keeping the parked kwargs and adding the answer as
`event`.
A core change would ship in 3.5 at the earliest, since 3.4 is already in
beta. As a provider change it reaches users on 3.4.
**The resume method is on `PythonOperator`, not `BaseOperator`.** Classic
operators that pause already define their own resume methods
(`HITLOperator.execute_complete`, for example), so a hook on `BaseOperator`
would have no second caller. The implementations that need it are the ones that
run a Python callable in the task's own process: `PythonOperator`, `@task`,
`@task.branch` and `@task.short_circuit`.
`resume_with_input()` re-runs `self.execute`, so a branch decision or
short-circuit condition made after the answer is applied as usual. In tasks
that run their callable in another process (`@task.virtualenv`,
`@task.external_python`, `@task.docker`, `@task.kubernetes`), the pause could
never reach the task runner, so `request_input()` raises before it writes a
request. No task can end up waiting on a request nothing can resume.
**One copy of the request logic.** `HITLOperator` and `request_input()`
share:
- one helper that writes the request, calls the notifiers and parks;
- one set of request checks (options, defaults, the reserved `_options`
param);
- one set of answer checks (chosen options, and `params_input` against the
request's params and schemas).
`HITLOperator`'s `validate_*` methods keep their names and signatures and
now delegate to these, so subclasses that override them behave as before. Its
existing tests pass unchanged.
The resume path checks `params_input` itself because the PATCH endpoint
validates only which options were chosen and how many. To do that,
`request_input()` parks with the request's options and serialized params, and
the answer is checked against what the reviewer was actually shown.
**The answer is a typed accessor, not a context key.**
`get_input_response()` returns a `TypedDict` (`InputResponse`) backed by a
`ContextVar`, which the resume method sets and resets around `execute`.
A provider-defined context key isn't one of the `Context` keys the decorator
knows about. A call-time argument with the same name would silently replace the
answer, and nothing would type-check it.
**Why 3.4+ only.** Before 3.4, writing a second request for the same task
instance clears the earlier answer but keeps the earlier subject, body and
options. That happens when a task asks again after resuming, and on a retry,
because the request row follows the task instance to its new try. The reviewer
would approve a question they were never shown. On 3.4 the request is rewritten
every time, and the review form refuses an answer to a request that has since
been replaced.
**It doesn't hold a worker slot.** Checked against a running Airflow with
LocalExecutor: while the task ran, the default pool showed 1 slot occupied and
a busy executor worker. Once it parked, the pool showed 0 occupied and all
workers idle. The same run covered:
- approve in the UI and reject over the REST API;
- a timeout without defaults (`HITLTimeoutError`);
- two approvals in one task instance;
- two mapped indexes each restoring their own saved state;
- a retry that sees the previous try's `task_state_store` values.
The agent was a real OpenAI Agents SDK `Agent` and `RunState` with a
scripted model.
## Gotchas
- **Code before `request_input()` runs again on the second run.** Branch on
`get_input_response()`. A callable that can ask more than one question should
record which one it asked, because the answer doesn't say.
- **`task_state_store` values survive retries.** A retry starts with
`get_input_response()` returning `None` but sees what the failed try saved.
- **API callers should send `request_created_at`.** A client answering
through the REST API is protected against answering a replaced request only if
it sends the `request_created_at` it read from the GET. The field is optional,
and without it the answer goes to whatever request is waiting.
--
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]
