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 を使用してタイムアウト時の選択肢を設定します。