airflow-hitl

Airflow 人工介入操作符

使用 HITL 操作符，在 Airflow DAG 中实现人工审批关卡、表单输入和人工驱动的分支。这些可延迟操作符会暂停工作流执行，直到有人通过 Airflow UI 或 REST API 作出响应。

实施清单

按顺序执行各步骤。优先使用可延迟的 HITL 操作符，而非自定义传感器或轮询循环。

关键提示：需要 Airflow 3.1+。Airflow 2.x 不支持。

所有 HITL 操作符都可延迟，并会在等待输入时释放工作槽位。

UI 位置：Airflow UI 中的 Browse -> Required Actions。可通过任务实例的 Required Actions 选项卡或 REST API 作出响应。

步骤 1：选择操作符

操作符	人工操作	结果
ApprovalOperator	批准或拒绝	拒绝会导致下游任务被跳过（审批任务本身会成功）
HITLOperator	选择选项和填写表单	返回所选内容
HITLBranchOperator	选择下游任务	运行所选任务，跳过其他任务
HITLEntryOperator	提交表单	返回表单数据

步骤 2：实现操作符

ApprovalOperator

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()

HITLOperator

必需参数：subject 和 options。

from airflow.providers.standard.operators.hitl import HITLOperator
from airflow.sdk import dag, task, chain, Param
from datetime import timedelta
from pendulum import datetime

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def hitl_example():
    hitl = HITLOperator(
        task_id="select_option",
        subject="Select Payment Method",
        body="Choose how to process payment",
        options=["ACH", "Wire", "Check"],
        defaults=["ACH"],
        multiple=False,
        execution_timeout=timedelta(hours=4),
        params={"amount": Param(1000, type="number")},
    )

    @task
    def process(result):
        print(f"Selected: {result['chosen_options']}")
        print(f"Amount: {result['params_input']['amount']}")

    process(hitl.output)

hitl_example()

HITLBranchOperator

选项可以直接匹配下游任务 ID，也可以使用 options_mapping。

from airflow.providers.standard.operators.hitl import HITLBranchOperator
from airflow.sdk import dag, task, chain
from pendulum import datetime

DEPTS = ["marketing", "engineering", "sales"]

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def branch_example():
    branch = HITLBranchOperator(
        task_id="select_dept",
        subject="Select Departments",
        options=[f"Fund {d}" for d in DEPTS],
        options_mapping={f"Fund {d}": d for d in DEPTS},
        multiple=True,
    )

    for dept in DEPTS:
        @task(task_id=dept)
        def handle(dept_name: str = dept):
            print(f"Processing {dept_name}")
        chain(branch, handle())

branch_example()

HITLEntryOperator

from airflow.providers.standard.operators.hitl import HITLEntryOperator
from airflow.sdk import dag, task, chain, Param
from pendulum import datetime

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def entry_example():
    entry = HITLEntryOperator(
        task_id="get_input",
        subject="Enter Details",
        body="Provide response",
        params={
            "response": Param("", type="string"),
            "priority": Param("p3", type="string"),
        },
    )

    @task
    def process(result):
        print(f"Response: {result['params_input']['response']}")

    process(entry.output)

entry_example()

步骤 3：可选功能

通知器

from airflow.sdk import BaseNotifier, Context
from airflow.providers.standard.operators.hitl import HITLOperator

class MyNotifier(BaseNotifier):
    template_fields = ("message",)
    def __init__(self, message=""):
        self.message = message
    def notify(self, context: Context):
        if context["ti"].state == "running":
            url = HITLOperator.generate_link_to_ui_from_context(context, base_url="https://airflow.example.com")
            self.log.info(f"Action needed: {url}")

hitl = HITLOperator(..., notifiers=[MyNotifier("{{ task.subject }}")])

限制响应者

格式取决于你的身份验证管理器：

身份验证管理器	格式	示例
SimpleAuthManager	用户名	["admin", "manager"]
FabAuthManager	邮箱	["[email protected]"]
Astro	Astro ID	["cl1a2b3cd456789ef1gh2ijkl3"]

hitl = HITLOperator(..., respondents=["[email protected]"])

超时行为

使用 execution_timeout 自动处理，并使用 defaults 设置超时时的选项。