Airflow 3.1+ DAG 需要人工审批…

当 Airflow 3.1+ DAG 需要在运行中插入人工审批/拒绝、表单填写、选项选择、人工驱动分支或外部响应时，使用 HITL 算子与触发器。

适用场景

解决思路

1. 用 `af config version` 确认 Airflow ≥ 3.1；HITL 算子在 Airflow UI 的 Browse → Required Actions 中响应，也可在任务实例页面的 Required Actions tab 响应。
2. 按能力选择类：审批或拒绝且下游在拒绝后跳过用 `ApprovalOperator`；展示 N 个选项并返回哪些被选用 `HITLOperator`；基于选择分支到一个或多个下游任务用 `HITLBranchOperator`；只收集表单、无审批/选择步骤用 `HITLEntryOperator`；直接使用 HITL trigger（高级/自定义算子）用 `HITLTrigger`。
3. 从 Airflow Registry 发现当前签名，不要凭记忆硬编码类名和参数。运行： `af registry modules standard | jq '.modules[] | select(.import_path | test("\\.hitl\\.")) | {name, type, import_path, short_description, docs_url}'` `af registry parameters standard | jq '.classes | to_entries[] | select(.key | test("\\.hitl\\.")) | {fqn: .key, parameters: .value.parameters}'` `af config providers | jq '.providers[] | select(.package_name == "apache-airflow-providers-standard") | .version'` 然后运行 `af registry parameters standard --version <VERSION>` 固定到已安装的 provider 版本。
4. 如果 registry 显示本技能未提到的参数，优先采用 registry；如果 registry 显示 Step 1 中没有的类，将其视为新增，Step 1 的决策表可能已过时。
5. 以审批门为规范示例开始： ```python from airflow.providers.standard.operators.hitl import ApprovalOperator from airflow.sdk import dag, task, chain, Param from pendulum import datetime @dag(start_date=datetime(2025, 1, 1), schedule="@daily") def approval_example(): @task def prepare(): return "Review quarterly report" approval = ApprovalOperator( task_id="approve_report", subject="Report Approval", body="{{ ti.xcom_pull(task_ids='prepare') }}", defaults="Approve", params={"comments": Param("", type="string")}, ) @task def after_approval(result): print(f"Decision: {result['chosen_options']}") chain(prepare(), approval) after_approval(approval.output) approval_example() ```
6. 其他类形状相同：都使用 `task_id`、`subject`，再加类特定参数；通过 Step 2 验证每个构造函数。例如 `HITLBranchOperator` 要求每个选项要么直接匹配下游 task id，要么通过 registry 中暴露的 mapping 参数解析。
7. 处理超时：设置 `defaults` 时，超时后任务成功并选中默认选项；不设置 `defaults` 时，超时后任务失败。
8. 在 `body` 中使用 Markdown 和 Jinja 模板，并直接渲染 XCom 上下文，例如： ```python body = """**Total Budget:** {{ ti.xcom_pull(task_ids='get_budget') }} | Category | Amount | |----------|--------| | Marketing | $1M | """ ```
9. 所有 HITL 算子都接受标准 Airflow callback kwargs，如 `on_success_callback`、`on_failure_callback` 等。
10. HITL 算子接受 `notifiers` 列表；在 notifier 的 `notify(context)` 方法内，用 `HITLOperator.generate_link_to_ui_from_context(context, base_url=...)` 构建指向待处理任务的链接。
11. 限制谁可响应时，不要硬编码参数名和标识符格式：先运行 `af config show | jq '.auth_manager // .core.auth_manager'` 查看当前 auth manager，再在 Step 2 中查当前 provider 暴露的 kwarg；撰写时是 `assigned_users`，接受的标识符格式由当前 auth manager 决定：Astro 用 Astro user ID，FabAuthManager 用 email，SimpleAuthManager 用 username。
12. 从外部集成（Slack bot、自定义应用或脚本）响应时，先发现实时端点，不要硬编码路径： `af api ls --filter hitl` `af api spec | jq '.paths | to_entries[] | select(.key | test("hitl"))'` PATCH-to-respond 模式稳定；确切路径需发现。典型形式： ```python import os, requests HOST = os.environ["AIRFLOW_HOST"] TOKEN = os.environ["AIRFLOW_API_TOKEN"] HEADERS = {"Authorization": f"Bearer {TOKEN}"} # 列出待处理项 — 使用 `af api ls --filter hitl` 得到的路径 requests.get(f"{HOST}/<path>", headers=HEADERS, params={"state": "pending"}) # 响应 — 同一发现路径族，使用 PATCH requests.patch( f"{HOST}/<path>/{dag_id}/{run_id}/{task_id}", headers=HEADERS, json={"chosen_options": ["Approve"], "params_input": {"comments": "ok"}}, ) ```
13. 安全检查： - 用 `af config version` 确认 Airflow 版本 ≥ 3.1。 - 构造函数 kwargs 与 Step 2 当前 registry 输出一致，避免出现 `respondents` 与 `assigned_users` 这类漂移。 - 做分支时：每个选项都能解析到下游 task id（直接解析或通过 Step 2 的 mapping kwarg 解析）。 - `defaults` 中的每个值也都存在于 `options` 中。 - 设置 `execution_timeout`；如果希望超时也成功而不是失败，则配置 `defaults`。 - 如果流程包含外部响应者，则配置 API token。
14. 不要硬编码上游文档 URL；通过 registry 按模块获取： `af registry modules standard | jq '.modules[] | select(.import_path | test("\\.hitl\\.")) | {name, docs_url}'`

适用边界 · 注意事项

• Airflow 版本低于 3.1，或没有使用/启用包含 HITL 算子的 provider（如 `apache-airflow-providers-standard`）时，不适用。
• 需要 AI/LLM 任务调用或 GenAI 模式时，不适用；应改用 `migrating-ai-sdk-to-common-ai`。
• 没有人工响应者、无法访问 Airflow UI/REST API，或外部响应但未配置 API token 时，不应依赖本流程。