kaxil opened a new pull request, #73586:
URL: https://github.com/apache/airflow/pull/73586
Stacked on #73578 (templated toolset connections). Only the last commit is
new here.
An agent that can look things up freely should not refund an order, send an
email or write to a production table without a person seeing the call first.
Mark those tools with pydantic-ai's own approval API, and `AgentOperator` /
`@task.agent` pause the task in front of the call and ask on the **Required
Actions** page, with the tool name and its arguments:
```python
shop = FunctionToolset(tools=[lookup_order, refund_order]).approval_required(
lambda ctx, tool_def, args: tool_def.name == "refund_order"
)
AgentOperator(
task_id="handle_ticket",
llm_conn_id="pydanticai_default",
prompt="The customer says order 7 never arrived. Resolve it.",
toolsets=[shop],
tool_approval_timeout=timedelta(hours=4),
)
```
**Approve** runs the call and the agent carries on. **Reject** does not fail
the task: the agent gets the reviewer's reason as the tool result and carries
on without the call. While it waits, the task is in `awaiting_input` and holds
no worker slot. Needs Airflow 3.3+.
## Design rationale
**Why not `require_approval` or `enable_hitl_review`?** Both review an
*output*: `require_approval` gates an `LLMOperator` result after the run, and
`enable_hitl_review` reviews an agent's final answer while polling in the
worker. Neither can stop a side effect before it happens, which is what a tool
call needs.
**Marking is pydantic-ai's; Airflow supplies the backend.**
`toolset.approval_required()` and `Tool(..., requires_approval=True)` already
exist and make pydantic-ai end the run with `DeferredToolRequests`. The
operator adds `DeferredToolRequests` to the output type only while building the
agent. pydantic-ai strips it from the schema the model sees, and
`self.output_type` (part of the serialized Dag) is unchanged. The pause is the
same `TaskAwaitingInput` path `LLMApprovalMixin` uses on 3.3+, and the resume
continues the run from its transcript with `DeferredToolResults`.
**The transcript goes to the task state store, not the continuation
kwargs.** It holds tool results such as query rows, which do not belong on the
task instance row. The continuation carries the transcript's hash, the pending
call ids, the usage so far, and the rendered toolset ids. The resume fails if
the transcript changed, or if a templated connection id now renders to a
different connection than the reviewer saw.
**One approval per task instance per Dag run.** Core keeps a single approval
request per task instance, across retries and clears, and a repeat request
keeps the first one's subject and body (`execution_api/routes/hitl.py`). A
second request would show the reviewer the earlier call while Approve applied
to the new one. So the operator records a marker in the task state store, and a
second request fails with a non-retryable `ToolApprovalAlreadyRequestedError`.
Allowing several approvals needs core to overwrite the request content on a
repeat upsert; that is a separate change.
**No approve-on-timeout.** `on_tool_approval_timeout` is `"fail"` (default)
or `"deny"`, and `"deny"` tells the agent that nobody answered, not that a
person refused. `usage_limits` covers both sides of the pause, so a
`cost_limit` is not reset by it.
## Screenshots
On a real scheduler and API server with a `test` model: two runs waiting on
the same tool, answered through the UI.
Approve with the reason left empty: the refund runs and the agent finishes.
The resumed run's `run_id` carries the `-resumed` suffix.
Reject with a reason: the agent receives it as the tool result, and the task
still succeeds.
## Gotchas
- Not available together with `durable=True`, `enable_hitl_review=True`,
`code_mode=True` or a `SandboxToolset`, each of which assumes the run finishes
in one go. There, a marked tool fails the task as it does today, and adding
`DeferredToolRequests` to `output_type` by hand raises
`UnsupportedToolDeferralError`.
- Tools that defer to external execution (pydantic-ai's `CallDeferred`) are
not supported and fail without retrying.
- A task whose approved call ran and that then failed cannot ask again in
the same Dag run; the retry fails at the gate. The docs say to keep gated tools
idempotent and trigger a new run.
- pydantic-ai refuses a `run_id` already present in the message history, so
the resumed run uses `<task-instance id>-resumed`. The `run_id` XCom holds that
id; `usage` covers both sides of the pause.
- `LoggingToolset` now logs a call waiting for approval at INFO rather than
as an ERROR "Tool ... failed".
- `LLMOperator`'s assigned-users validation moved into a helper shared with
the new `tool_approval_assigned_users`; its behaviour and messages are
unchanged.
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
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]
