企业官网建设流程全解析

Apache Airflow 3.3 引入awaiting_input任务状态：Human-in-the-loop 从 Triggerer 卸载到 Scheduler 的架构演进

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow

Human-in-the-loop（HITL，人在环路）任务在 Airflow 3.3 起改用由 Scheduler 管理的一等任务状态awaiting_input等待人工响应，不再 defer 到 Triggerer，使 Triggerer 可以在任务等待人工回复期间缩容到零。本文基于当前仓库（Apache Airflow 主分支，对应 3.3+ 行为）剖析这一机制：awaiting_input状态如何定义、Scheduler 如何执行超时扫描、响应如何直接恢复任务，以及在 3.1/3.2 上如何回退到旧的 trigger-based 路径。

背景：HITL 任务为什么需要独立于 Triggerer 的等待机制

Airflow 3.1 引入的 Human-in-the-loop 功能让工作流可以在审批、人工质检、内容审核等场景暂停并等待人工决策（参见 HITL 官方教程）。在 3.1 与 3.2 中，等待人工响应的 HITL 任务通过defer()进入DEFERRED状态，由 Triggerer 上的HITLTrigger轮询数据库中的响应。这意味着只要存在一个等待人工响应的任务，Triggerer 就必须保持运行——尽管它几乎无事可做。

3.3 起，HITL 等待被重构为 Scheduler 管理的专用任务状态awaiting_input：任务既不占用 worker 槽位，也不占用 Triggerer，Triggerer 可以在 HITL 任务等待响应期间缩容到零。等待中的任务要么在收到人工响应时直接恢复，要么在 Scheduler 的响应超时扫描（response-timeout sweep）中被解决。

awaiting_input状态的定义与语义

awaiting_input是任务实例的中间状态（IntermediateTIState），定义于 airflow-core/src/airflow/utils/state.py，并纳入 TaskInstanceState 枚举：

class IntermediateTIState(str, Enum): """States that a Task Instance can be in that indicate it is not yet in a terminal or running state.""" SCHEDULED = "scheduled" QUEUED = "queued" RESTARTING = "restarting" UP_FOR_RETRY = "up_for_retry" UP_FOR_RESCHEDULE = "up_for_reschedule" DEFERRED = "deferred" AWAITING_INPUT = "awaiting_input"

从状态机的编排看，awaiting_input与deferred一样属于"尚未终结也尚未运行"的中间态，但语义不同：deferred意味着任务正等待某个 Trigger 触发事件（通常由 Triggerer 承载），而awaiting_input表示任务正在等待来自外部（UI 或 REST API）的人工输入，与 Triggerer 无关。

从 Scheduler 的资源核算角度看，awaiting_input与DEFERRED一并被排除在 worker 槽位/max_active_tasks统计之外（见 scheduler_job_runner.py），并且在 Scheduler 的指标统计中与DEFERRED并列计入等待类状态（scheduler_job_runner.py）。

值得注意的是池（pool）语义的变化：旧版 defer 路径中，若池开启了include_deferred，等待中的 HITL 任务会计入池槽位；而在新机制下，awaiting_input任务不再占用池槽位（官方文档明确说明）。

从 defer 到awaiting_input：operator 层的双路径实现

标准 Provider 的 HITL 操作符（HITLOperator、HITLEntryOperator、HITLBranchOperator、ApprovalOperator等）位于 providers/standard/src/airflow/providers/standard/operators/hitl.py，其execute方法用AIRFLOW_V_3_3_PLUS版本开关区分两条路径（hitl.py#L234-L254）：

if AIRFLOW_V_3_3_PLUS: # New core (3.3+): park the task in AWAITING_INPUT -- no trigger, no triggerer. raise TaskAwaitingInput( method_name="execute_complete", timeout=self.response_timeout, ) # Fallback for cores < 3.3: defer the response check to HITLTrigger on the triggerer. self.defer( trigger=HITLTrigger( ti_id=ti_id, options=self.options, defaults=self.defaults, params=self.serialized_params, multiple=self.multiple, timeout_datetime=timeout_datetime, ), method_name="execute_complete", )

• Airflow 3.3+：抛出TaskAwaitingInput（从 airflow-core/src/airflow/sdk/exceptions.py 引入），将任务停驻在awaiting_input状态，不创建任何 Trigger。
• Airflow 3.1/3.2：回退到defer()+HITLTrigger的旧路径，行为与之前版本完全一致。

无论哪条路径，任务最终都在execute_complete中恢复（hitl.py#L262），因此上层业务代码不受影响。execute_complete会校验选项合法性、校验params_input，并把人工输入通过 XCom 暴露给下游任务。

响应如何到达：hitl_detail表与恢复事件

人工请求及其响应持久化在hitl_detail表，模型定义于 airflow-core/src/airflow/models/hitl.py（3.1 引入，迁移脚本见 0076_3_1_0_add_human_in_the_loop_response.py；3.2 起每次尝试的历史记录存于HITLDetailHistory，见 hitl_history.py）。

关键方法HITLDetail.as_resume_event_payload()（hitl.py#L181-L197）把响应列映射为execute_complete消费的事件字典，从而让人工响应（通过 Core API 提交）或 Scheduler 的超时扫描都能不经 Trigger直接恢复awaiting_input任务，并完整复刻 Provider 侧HITLTriggerEventSuccessPayload的契约：

def as_resume_event_payload(self, *, timedout: bool = False) -> dict[str, Any]: return { "chosen_options": list(self.chosen_options or []), "params_input": self.params_input or {}, "responded_at": self.responded_at, "responded_by_user": self.responded_by_user, "timedout": timedout, }

responded_by_user在超时默认值场景下为None，因为此时没有真实的人工响应者。

人工响应的提交入口：Core API 的 PATCH 路由

人工响应通过 Core API 提交，路由实现在 airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py（PATCH .../taskInstances/{task_id}/{map_index}/hitlDetails）。该处理器会：

1. 以固定顺序加行锁（先TaskInstance后HITLDetail），避免与 worker 的 park 转换、清除（clear）操作死锁；
2. 校验响应幂等性：已响应（response_received为真）的任务返回 409；
3. 校验assigned_users授权：若请求指定了响应人白名单，非白名单用户返回 403（is_authorized_hitl_task）；
4. 写侧校验：选项合法性（非法选项返回 400）、单选任务提交多选项返回 400——这避免了响应在恢复时才失败；
5. 写入responded_by/responded_at/chosen_options/params_input；
6. 事件驱动恢复：若任务正处于AWAITING_INPUT或DEFERRED状态，直接调用handle_event_submit把as_resume_event_payload()打包进next_kwargs["event"]，将状态置为SCHEDULED并设置scheduled_dttm，随后 Scheduler 重新入队执行execute_complete——全程无需 Triggerer。

列表查询路由（GET .../hitlDetails）支持按response_received、ti_state、subject/body模式、responded_by_user_id/name、created_at范围等多种过滤（hitl.py#L301-L399），并接入 DAG 级权限过滤与 HITL_DETAIL 访问实体权限。

Scheduler 侧的响应超时扫描（response-timeout sweep）

Scheduler 是awaiting_input任务"活性保证"（liveness guarantee）的承载者。check_awaiting_input_timeouts是 Scheduler 心跳循环中的一项（注册于 scheduler_job_runner.py#L1826），其实现位于 scheduler_job_runner.py#L3543-L3602，与 Triggerer 完全独立：

query = ( select(TI) .where( TI.state == TaskInstanceState.AWAITING_INPUT, TI.trigger_timeout < now, ) .options(joinedload(TI.hitl_detail)) .limit(100) ) query = with_row_locks(query, of=TI, session=session, skip_locked=True)

对每个超时任务，扫描按以下优先级解决：

1. 截止前刚到达的响应：若hitl_detail.responded_at非空，则以真实响应恢复（timedout=False）；
2. 配置了 defaults：把默认选项写为响应并恢复为成功（timedout=True，responded_by=None）；
3. 两者皆无：任务失败（镜像check_trigger_timeouts的语义）。

实现细节体现了工程上的健壮性考虑：

• 每批最多处理 100 个超时任务，避免单个 tick 锁住/处理无界积压而阻塞并发的响应与清除；
• 只对TI行加锁（of=TI）+skip_locked=True，使 HA 多 Scheduler 不会重复解决同一任务，且避免对hitl_detail外连接可空侧误加FOR UPDATE。

response_timeout与execution_timeout语义分离：response_timeout控制开始等待之后的人工响应等待上限，由 Scheduler 的awaiting_input超时扫描（3.3+）或 Triggerer（旧版本）强制；execution_timeout只控制任务执行（等待之前）阶段，不应再用于控制等待时长（operator 已对旧用法发出弃用警告并自动迁移，见 hitl.py#L94-L106）。response_timeout未设置时不施加等待超时。

HITL 操作符使用指南（附完整示例）

以下内容取自标准 Provider 的示例 DAG example_hitl_operator.py，完整覆盖 3.3 新机制下的各种等待形态。

1. 通用参数

HITLOperator（所有 HITL 操作符的基类，hitl.py#L58-L74）核心参数：

2. 输入收集（HITLEntryOperator）

用户通过params提供结构化输入，供下游任务使用——这对 LLM 工作流中的人工引导尤为实用：

wait_for_input = HITLEntryOperator( task_id="wait_for_input", subject="Please provide required information: ", params={"information": Param("", type="string")}, notifiers=[hitl_request_callback], on_success_callback=hitl_success_callback, on_failure_callback=hitl_failure_callback, )

3. 单选 / 多选

# 单选 wait_for_option = HITLOperator( task_id="wait_for_option", subject="Please choose one option to proceed: ", options=["option 1", "option 2", "option 3"], notifiers=[hitl_request_callback], ) # 多选 wait_for_multiple_options = HITLOperator( task_id="wait_for_multiple_options", subject="Please choose option to proceed: ", options=["option 4", "option 5", "option 6"], multiple=True, notifiers=[hitl_request_callback], )

4. 超时与默认值

设置response_timeout与defaults：超时后 Scheduler 扫描会把默认选项写入响应并成功恢复任务：

wait_for_default_option = HITLOperator( task_id="wait_for_default_option", subject="Please choose option to proceed: ", options=["option 7", "option 8", "option 9"], defaults=["option 7"], response_timeout=datetime.timedelta(seconds=1), notifiers=[hitl_request_callback], )

5. 审批 / 拒绝（ApprovalOperator）

审批是选项选择的特化形态，仅包含 'Approval' 与 'Rejection'。通过assigned_users限制可响应者（id 与 name 必须同时提供），未在名单内的用户会被 API 层以 403 拒绝：

valid_input_and_options = ApprovalOperator( task_id="valid_input_and_options", subject="Are the following input and options valid?", body=""" Input: {{ ti.xcom_pull(task_ids='wait_for_input')["params_input"]["information"] }} Option: {{ ti.xcom_pull(task_ids='wait_for_option')["chosen_options"] }} """, defaults="Reject", response_timeout=datetime.timedelta(minutes=5), notifiers=[hitl_request_callback], assigned_users=[{"id": "1", "name": "airflow"}, {"id": "admin", "name": "admin"}], )

如示例所示，用户输入通过 XCom 获取：chosen_options（所选选项）与params_input（输入参数）。

6. 分支选择（HITLBranchOperator）

用户选择的选项是任务，用于内容审核等需要人工判断决定流程走向的场景。选项需对应 DAG 中的任务，且必须在工作流中声明依赖关系：

choose_a_branch_to_run = HITLBranchOperator( task_id="choose_a_branch_to_run", subject="You're now allowed to proceeded. Please choose one task to run: ", options=["task_1", "task_2", "task_3"], notifiers=[hitl_request_callback], ) @task def task_1(): ... @task def task_2(): ... @task def task_3(): ... ( [wait_for_input, wait_for_option, wait_for_default_option, wait_for_multiple_options] >> valid_input_and_options >> choose_a_branch_to_run >> [task_1(), task_2(), task_3()] )

7. Notifier 通知机制

Notifier 是 HITL 事件的回调机制（任务等待输入、成功、失败时触发）。示例使用LocalLogNotifier演示，其中HITLOperator.generate_link_to_ui_from_context可生成直达 UI 响应页面的链接，接受四个参数：context（notifier 自动传入）、base_url（可选，默认取配置api.base_url）、options（可选，预选选项）、params_inputs（可选，预填输入）：

class LocalLogNotifier(BaseNotifier): template_fields = ("message",) def __init__(self, message: str) -> None: self.message = message def notify(self, context: Context) -> None: url = HITLOperator.generate_link_to_ui_from_context( context=context, base_url="http://localhost:28080", ) self.log.info(self.message) self.log.info("Url to respond %s", url)

将 notifier 列表传给 HITL 操作符的notifiers参数，操作符创建等待人工响应的 HITL 请求时即以单个参数context调用notify。Notifier 自定义方式见 创建 Notifier 指南 与 Notifications 扩展文档。

在 UI 中响应

Airflow UI 提供专门的 HITL 交互入口：任务进入等待后，点击任务可在详情面板的Required Actions标签页中看到待响应请求（对应 UI 组件 HITLTaskInstances.tsx、HITLReviewDrawer.tsx），可直接在页面上选择选项、填写输入或提交审批。

UI 侧通过 Core API 的PATCH .../hitlDetails提交响应，与命令行/脚本方式走同一条路由。

本地测试与 AI Agent 驱动

airflow dags test（及其底层dag.test()）支持 HITL 任务：到达awaiting_input的任务保持停驻——测试运行本身不会自行解决它——并持续记录等待日志，直到外部记录响应。响应通道与真实部署一致：Required Actions 页面，或与元数据库共享的 api-server 的 HITL REST API（如airflow standalone或单独启动的airflow api-server）。响应落地后测试运行恢复任务并继续下游任务。

这也让 AI Agent 可以端到端驱动 HITL 流水线：运行airflow dags test，监听等待日志行，询问人类并把答案通过 HITL REST API 提交。两个关键调用（~可作为dag_id与dag_run_id的通配符）：

# 发现待处理请求（subject、options、params、run/task 标识） GET /api/v2/dags/~/dagRuns/~/hitlDetails?response_received=false # 提交响应；测试运行会在下一次轮询时恢复任务。 # 非映射任务 map_index 为 -1。 PATCH /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/hitlDetails {"chosen_options": ["Approve"], "params_input": {}}

注意：response_timeout及超时默认值由 Scheduler 强制，而airflow dags test不运行 Scheduler。因此停驻任务会无限期等待响应；请通过 UI 或 REST API 提供响应以结束测试运行（官方文档说明）。

状态变更、API 与测试覆盖

• 状态枚举与调度：awaiting_input已纳入TaskInstanceState与IntermediateTIState（utils/state.py），Scheduler 在任务状态指标与槽位核算中将其与DEFERRED并列处理。
• REST API 与 UI：OpenAPI 定义见 v2-rest-api-generated.yaml 与 execution_api v2026_06_30.py；UI 的 HITL 状态工具函数见 ui/src/utils/hitl.ts 与 ui/src/utils/stateUtils.ts。
• 测试覆盖：单元测试覆盖 Scheduler 的awaiting_input超时扫描（tests/unit/jobs/test_scheduler_job.py）、HITL 公共 API 路由（tests/unit/api_fastapi/core_api/routes/public/test_hitl.py）以及 UI 侧的 HITL 交互（ui/src/pages/HITLTaskInstances/HITLTaskInstances.test.tsx）。

典型应用场景

• LLM 工作流：大语言模型任务链中的人工引导，人工提供的方向性输入常能显著改善结果；
• 企业数据管道：人工验证补充自动化流程（审批、质量门禁）；
• 内容审核 / 风险控制：HITLBranchOperator按人工判断路由到不同处理分支；
• 人机协作的自动化运维：变更审批、发布确认等需要人工把关的场景。

小结

Airflow 3.3 将 Human-in-the-loop 等待从 Triggerer 卸载到 Scheduler：新增一等状态awaiting_input，通过hitl_detail表持久化请求与响应，PATCH .../hitlDetails在收到人工响应时事件驱动地直接恢复任务，Scheduler 的check_awaiting_input_timeouts独立于 Triggerer 保证超时活性（响应优先、其次默认值、否则失败）。由此，即使存在大量等待人工响应的任务，Triggerer 也可以缩容到零；而在 Airflow 3.1/3.2 上，operator 通过版本开关自动回退到旧的 trigger-based defer 路径，确保行为兼容。对使用 HITL 的用户而言，操作符 API 与响应通道完全不变，升级 3.3 即可自动获得这一资源效率改进。

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow