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/airflowHuman-in-the-loopHITL人在环路任务在 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 槽位也不占用 TriggererTriggerer 可以在 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_inputoperator 层的双路径实现标准 Provider 的 HITL 操作符HITLOperator、HITLEntryOperator、HITLBranchOperator、ApprovalOperator等位于 providers/standard/src/airflow/providers/standard/operators/hitl.py其execute方法用AIRFLOW_V_3_3_PLUS版本开关区分两条路径hitl.py#L234-L254if AIRFLOW_V_3_3_PLUS: # New core (3.3): park the task in AWAITING_INPUT -- no trigger, no triggerer. raise TaskAwaitingInput( method_nameexecute_complete, timeoutself.response_timeout, ) # Fallback for cores 3.3: defer the response check to HITLTrigger on the triggerer. self.defer( triggerHITLTrigger( ti_idti_id, optionsself.options, defaultsself.defaults, paramsself.serialized_params, multipleself.multiple, timeout_datetimetimeout_datetime, ), method_nameexecute_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.py3.1 引入迁移脚本见 0076_3_1_0_add_human_in_the_loop_response.py3.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.pyPATCH .../taskInstances/{task_id}/{map_index}/hitlDetails。该处理器会以固定顺序加行锁先TaskInstance后HITLDetail避免与 worker 的 park 转换、清除clear操作死锁校验响应幂等性已响应response_received为真的任务返回 409校验assigned_users授权若请求指定了响应人白名单非白名单用户返回 403is_authorized_hitl_task写侧校验选项合法性非法选项返回 400、单选任务提交多选项返回 400——这避免了响应在恢复时才失败写入responded_by/responded_at/chosen_options/params_input事件驱动恢复若任务正处于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 sweepScheduler 是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, ofTI, sessionsession, skip_lockedTrue)对每个超时任务扫描按以下优先级解决截止前刚到达的响应若hitl_detail.responded_at非空则以真实响应恢复timedoutFalse配置了 defaults把默认选项写为响应并恢复为成功timedoutTrueresponded_byNone两者皆无任务失败镜像check_trigger_timeouts的语义。实现细节体现了工程上的健壮性考虑每批最多处理 100 个超时任务避免单个 tick 锁住/处理无界积压而阻塞并发的响应与清除只对TI行加锁ofTIskip_lockedTrue使 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核心参数参数类型说明subjectstr必填展示给用户的标题/主题optionslist[str]用户可选择的选项列表bodystr | None描述性正文支持 Markdowndefaultsstr | list[str]默认选项超时未响应时作为响应使用multiplebool是否允许多选默认Falseparamsdict参数定义格式同 DAG params可渲染表单校验后写入任务结果 XComresponse_timeouttimedelta | None等待人工响应的最长时间3.3 由 Scheduler 超时扫描强制assigned_userslist[dict]允许响应的用户白名单id 与 name 都必须提供仅名单内用户可响应notifiersBaseNotifier | listHITL 事件等待/成功/失败回调2. 输入收集HITLEntryOperator用户通过params提供结构化输入供下游任务使用——这对 LLM 工作流中的人工引导尤为实用wait_for_input HITLEntryOperator( task_idwait_for_input, subjectPlease provide required information: , params{information: Param(, typestring)}, notifiers[hitl_request_callback], on_success_callbackhitl_success_callback, on_failure_callbackhitl_failure_callback, )3. 单选 / 多选# 单选 wait_for_option HITLOperator( task_idwait_for_option, subjectPlease choose one option to proceed: , options[option 1, option 2, option 3], notifiers[hitl_request_callback], ) # 多选 wait_for_multiple_options HITLOperator( task_idwait_for_multiple_options, subjectPlease choose option to proceed: , options[option 4, option 5, option 6], multipleTrue, notifiers[hitl_request_callback], )4. 超时与默认值设置response_timeout与defaults超时后 Scheduler 扫描会把默认选项写入响应并成功恢复任务wait_for_default_option HITLOperator( task_idwait_for_default_option, subjectPlease choose option to proceed: , options[option 7, option 8, option 9], defaults[option 7], response_timeoutdatetime.timedelta(seconds1), notifiers[hitl_request_callback], )5. 审批 / 拒绝ApprovalOperator审批是选项选择的特化形态仅包含 Approval 与 Rejection。通过assigned_users限制可响应者id 与 name 必须同时提供未在名单内的用户会被 API 层以 403 拒绝valid_input_and_options ApprovalOperator( task_idvalid_input_and_options, subjectAre the following input and options valid?, body Input: {{ ti.xcom_pull(task_idswait_for_input)[params_input][information] }} Option: {{ ti.xcom_pull(task_idswait_for_option)[chosen_options] }} , defaultsReject, response_timeoutdatetime.timedelta(minutes5), 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_idchoose_a_branch_to_run, subjectYoure 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 响应页面的链接接受四个参数contextnotifier 自动传入、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( contextcontext, base_urlhttp://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_receivedfalse # 提交响应测试运行会在下一次轮询时恢复任务。 # 非映射任务 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与IntermediateTIStateutils/state.pyScheduler 在任务状态指标与槽位核算中将其与DEFERRED并列处理。REST API 与 UIOpenAPI 定义见 v2-rest-api-generated.yaml 与 execution_api v2026_06_30.pyUI 的 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Airflow 3.3 引入 `awaiting_input` 任务状态：Human-in-the-loop 从 Triggerer 卸载到 Scheduler 的架构演进
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/airflowHuman-in-the-loopHITL人在环路任务在 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 槽位也不占用 TriggererTriggerer 可以在 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_inputoperator 层的双路径实现标准 Provider 的 HITL 操作符HITLOperator、HITLEntryOperator、HITLBranchOperator、ApprovalOperator等位于 providers/standard/src/airflow/providers/standard/operators/hitl.py其execute方法用AIRFLOW_V_3_3_PLUS版本开关区分两条路径hitl.py#L234-L254if AIRFLOW_V_3_3_PLUS: # New core (3.3): park the task in AWAITING_INPUT -- no trigger, no triggerer. raise TaskAwaitingInput( method_nameexecute_complete, timeoutself.response_timeout, ) # Fallback for cores 3.3: defer the response check to HITLTrigger on the triggerer. self.defer( triggerHITLTrigger( ti_idti_id, optionsself.options, defaultsself.defaults, paramsself.serialized_params, multipleself.multiple, timeout_datetimetimeout_datetime, ), method_nameexecute_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.py3.1 引入迁移脚本见 0076_3_1_0_add_human_in_the_loop_response.py3.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.pyPATCH .../taskInstances/{task_id}/{map_index}/hitlDetails。该处理器会以固定顺序加行锁先TaskInstance后HITLDetail避免与 worker 的 park 转换、清除clear操作死锁校验响应幂等性已响应response_received为真的任务返回 409校验assigned_users授权若请求指定了响应人白名单非白名单用户返回 403is_authorized_hitl_task写侧校验选项合法性非法选项返回 400、单选任务提交多选项返回 400——这避免了响应在恢复时才失败写入responded_by/responded_at/chosen_options/params_input事件驱动恢复若任务正处于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 sweepScheduler 是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, ofTI, sessionsession, skip_lockedTrue)对每个超时任务扫描按以下优先级解决截止前刚到达的响应若hitl_detail.responded_at非空则以真实响应恢复timedoutFalse配置了 defaults把默认选项写为响应并恢复为成功timedoutTrueresponded_byNone两者皆无任务失败镜像check_trigger_timeouts的语义。实现细节体现了工程上的健壮性考虑每批最多处理 100 个超时任务避免单个 tick 锁住/处理无界积压而阻塞并发的响应与清除只对TI行加锁ofTIskip_lockedTrue使 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核心参数参数类型说明subjectstr必填展示给用户的标题/主题optionslist[str]用户可选择的选项列表bodystr | None描述性正文支持 Markdowndefaultsstr | list[str]默认选项超时未响应时作为响应使用multiplebool是否允许多选默认Falseparamsdict参数定义格式同 DAG params可渲染表单校验后写入任务结果 XComresponse_timeouttimedelta | None等待人工响应的最长时间3.3 由 Scheduler 超时扫描强制assigned_userslist[dict]允许响应的用户白名单id 与 name 都必须提供仅名单内用户可响应notifiersBaseNotifier | listHITL 事件等待/成功/失败回调2. 输入收集HITLEntryOperator用户通过params提供结构化输入供下游任务使用——这对 LLM 工作流中的人工引导尤为实用wait_for_input HITLEntryOperator( task_idwait_for_input, subjectPlease provide required information: , params{information: Param(, typestring)}, notifiers[hitl_request_callback], on_success_callbackhitl_success_callback, on_failure_callbackhitl_failure_callback, )3. 单选 / 多选# 单选 wait_for_option HITLOperator( task_idwait_for_option, subjectPlease choose one option to proceed: , options[option 1, option 2, option 3], notifiers[hitl_request_callback], ) # 多选 wait_for_multiple_options HITLOperator( task_idwait_for_multiple_options, subjectPlease choose option to proceed: , options[option 4, option 5, option 6], multipleTrue, notifiers[hitl_request_callback], )4. 超时与默认值设置response_timeout与defaults超时后 Scheduler 扫描会把默认选项写入响应并成功恢复任务wait_for_default_option HITLOperator( task_idwait_for_default_option, subjectPlease choose option to proceed: , options[option 7, option 8, option 9], defaults[option 7], response_timeoutdatetime.timedelta(seconds1), notifiers[hitl_request_callback], )5. 审批 / 拒绝ApprovalOperator审批是选项选择的特化形态仅包含 Approval 与 Rejection。通过assigned_users限制可响应者id 与 name 必须同时提供未在名单内的用户会被 API 层以 403 拒绝valid_input_and_options ApprovalOperator( task_idvalid_input_and_options, subjectAre the following input and options valid?, body Input: {{ ti.xcom_pull(task_idswait_for_input)[params_input][information] }} Option: {{ ti.xcom_pull(task_idswait_for_option)[chosen_options] }} , defaultsReject, response_timeoutdatetime.timedelta(minutes5), 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_idchoose_a_branch_to_run, subjectYoure 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 响应页面的链接接受四个参数contextnotifier 自动传入、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( contextcontext, base_urlhttp://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_receivedfalse # 提交响应测试运行会在下一次轮询时恢复任务。 # 非映射任务 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与IntermediateTIStateutils/state.pyScheduler 在任务状态指标与槽位核算中将其与DEFERRED并列处理。REST API 与 UIOpenAPI 定义见 v2-rest-api-generated.yaml 与 execution_api v2026_06_30.pyUI 的 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考