airflow-hitl

Airflow 人机协同操作符

暂停 DAG，直到有人通过 Airflow UI 或 REST API 响应。HITL 操作符支持延迟执行 — 等待期间会释放工作槽位。

需要 Airflow 3.1+（af config version）。

UI 位置：浏览 → 必需操作。在任务实例页面的“必需操作”选项卡中响应。

交叉引用：migrating-ai-sdk-to-common-ai 用于 AI/LLM 任务装饰器；airflow 用于下文使用的注册表和 API 发现命令。

---

步骤 1 — 选择所需的功能

功能	类（在步骤 2 中验证）
批准或拒绝；拒绝时下游任务会跳过	ApprovalOperator
显示 N 个选项并返回所选项	HITLOperator
根据选择分支到一个或多个下游任务	HITLBranchOperator
收集表单（无需批准/选择步骤）	HITLEntryOperator
直接使用 HITL 触发器（高级/自定义操作符）	HITLTrigger

这里只硬编码类名。Provider 会在不同版本中添加、重命名和移除参数 — 请勿凭记忆复制参数列表。编写代码前，请获取当前签名。

---

步骤 2 — 从 Airflow Registry 发现当前签名

编写 HITL 代码前，运行以下命令，查看当前类列表和构造函数参数（完整的 af registry 参考请参阅 airflow 技能）：

# Every HITL-related module in the standard provider
af registry modules standard \
  | jq '.modules[] | select(.import_path | test("\\.hitl\\.")) | {name, type, import_path, short_description, docs_url}'

# Constructor signatures: name, type, default, required, description
af registry parameters standard \
  | jq '.classes | to_entries[] | select(.key | test("\\.hitl\\.")) | {fqn: .key, parameters: .value.parameters}'

# Pin to the exact installed provider version
af config providers \
  | jq '.providers[] | select(.package_name == "apache-airflow-providers-standard") | .version'
# then: af registry parameters standard --version <VERSION>

如果注册表显示了此技能未提及的参数，请优先采用注册表中的信息。如果注册表显示了步骤 1 中未列出的类，请将其视为新增项 — 上方的决策表可能已过时。

---

步骤 3 — 标准示例（审批关卡）

任何 HITL 任务都可从此示例开始。根据步骤 2 替换类名和参数进行调整。

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",              # Auto-selected on timeout
        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()

步骤 1 中的其他类使用相同结构（task_id、subject，以及类专属参数）。请通过步骤 2 验证每个构造函数 — 例如，HITLBranchOperator 要求每个选项要么直接匹配某个下游任务 ID，要么通过注册表中列出的映射参数进行解析。

---

步骤 4 — 行为约定（跨版本稳定）

超时

• 设置了 defaults：任务在超时时成功，并选择默认选项。
• 未设置 defaults：任务在超时时失败。

body 中的 Markdown 和 Jinja

body 支持 Markdown 和 Jinja 模板。直接渲染 XCom 上下文：

body = """**Total Budget:** {{ ti.xcom_pull(task_ids='get_budget') }}

| Category | Amount |
|----------|--------|
| Marketing | $1M |
"""

回调

所有 HITL 操作符都接受标准 Airflow 回调参数（on_success_callback、on_failure_callback 等）。

通知器

HITL 操作符接受 notifiers 列表。在通知器的 notify(context) 方法中，使用 HITLOperator.generate_link_to_ui_from_context(context, base_url=...) 构建指向待处理任务的链接。

限制响应者

参数名称和可接受的标识符格式取决于当前启用的身份验证管理器。请勿硬编码 — 检查当前启用的是哪一个，以及当前 Provider 提供了哪个关键字参数：

af config show | jq '.auth_manager // .core.auth_manager'

然后在步骤 2 中查找当前的关键字参数（撰写本文时为 assigned_users，接受当前身份验证管理器使用的任意格式标识符 — Astro 使用 Astro 用户 ID，FabAuthManager 使用电子邮件，SimpleAuthManager 使用用户名）。

---

步骤 5 — 通过外部集成响应

适用于 Slack 机器人、自定义应用或脚本。请发现当前端点，而不是硬编码路径：

af api ls --filter hitl           # live endpoint list
af api spec \
  | jq '.paths | to_entries[] | select(.key | test("hitl"))'   # request/response schemas

通过 PATCH 响应的模式稳定；具体路径需通过发现确定。典型结构如下：

import os, requests

HOST = os.environ["AIRFLOW_HOST"]
TOKEN = os.environ["AIRFLOW_API_TOKEN"]
HEADERS = {"Authorization": f"Bearer {TOKEN}"}

# List pending — use the path from `af api ls --filter hitl`
requests.get(f"{HOST}/<path>", headers=HEADERS, params={"state": "pending"})

# Respond — same discovered path family, PATCH
requests.patch(
    f"{HOST}/<path>/{dag_id}/{run_id}/{task_id}",
    headers=HEADERS,
    json={"chosen_options": ["Approve"], "params_input": {"comments": "ok"}},
)

---

步骤 6 — 安全检查

• Airflow 版本 ≥ 3.1（af config version）。
• 构造函数 kwargs 与步骤 2 中当前注册表的输出一致——不存在 respondents 与 assigned_users 这类差异。
• 对于分支：每个选项都能解析为下游任务 ID（直接解析，或通过步骤 2 中的映射 kwargs 解析）。
• defaults 中的每个值也都包含在 options 中。
• 已设置 execution_timeout；如果超时后应成功而非失败，则已配置 defaults。
• 如果流程中包含外部响应者，则已配置 API 令牌。

---

参考资料

注册表会按模块显示上游文档 URL——请勿硬编码：

af registry modules standard \
  | jq '.modules[] | select(.import_path | test("\\.hitl\\.")) | {name, docs_url}'

相关技能

• airflow — af registry、af api、af config 命令参考。
• migrating-ai-sdk-to-common-ai — AI/LLM 任务装饰器和 GenAI 模式（common-ai 提供方）。
• authoring-dags — 编写 DAG 的通用最佳实践。
• testing-dags — 迭代式测试 → 调试 → 修复流程。