Airflow Human-in-the-Loop Operators

DAG の実行を一時停止し、ユーザーが Airflow UI または REST API 経由で応答するまで待機します。HITL operators は遅延可能 (deferrable) — 待機中はワーカースロットを解放します。

Airflow 3.1+ が必須 (af config version)。

UI の場所: Browse → Required Actions。タスクインスタンスページの Required Actions タブから応答します。

クロスリファレンス: airflow-ai は AI/LLM タスクデコレータ用、airflow は以下で使用するレジストリと API ディスカバリーコマンド用。

---

Step 1 — 必要な機能を選ぶ

機能	クラス (Step 2 で検証)
承認または却下。却下時は下流をスキップ	ApprovalOperator
N 個のオプションを表示し、選択されたものを返す	HITLOperator
選択肢に基づいて 1 つ以上の下流タスクに分岐	HITLBranchOperator
フォーム (承認/選択ステップなし) を収集	HITLEntryOperator
HITL trigger を直接使用 (高度な/カスタムオペレータ)	HITLTrigger

クラス名がハードコードされているのはこの場所だけです。プロバイダはリリース間でパラメータを追加、名前変更、削除します — メモリからパラメータリストをコピーしないでください。コードを書く前に現在のシグネチャを取得してください。

---

Step 2 — Airflow Registry から現在のシグネチャを検出

HITL コードを書く前に、以下を実行して、ライブロスターとコンストラクタパラメータを確認してください (airflow スキルの完全な af registry リファレンスを参照):

# 標準プロバイダ内のすべての HITL 関連モジュール
af registry modules standard \
  | jq '.modules[] | select(.import_path | test("\\.hitl\\.")) | {name, type, import_path, short_description, docs_url}'

# コンストラクタシグネチャ: name, type, default, required, description
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>

レジストリがこのスキルで説明していないパラメータを表示している場合は、レジストリを優先してください。レジストリが Step 1 にないクラスを表示している場合は、それを追加として扱ってください — 上記の決定テーブルが古い可能性があります。

---

Step 3 — 標準的な例 (承認ゲート)

あらゆる HITL タスクの開始点。Step 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()

Step 1 の他のクラスについては、形状は同じ (task_id、subject プラス クラス固有のパラメータ)。各コンストラクタを Step 2 で検証してください — 例えば、HITLBranchOperator は、すべてのオプションが下流タスク ID と直接一致するか、Step 2 でレジストリに表示されるマッピングパラメータ経由で解決される必要があります。

---

Step 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 コールバック kwargs (on_success_callback、on_failure_callback など) を受け入れます。

Notifiers

HITL オペレータは notifiers リストを受け入れます。notifier の notify(context) メソッド内で、HITLOperator.generate_link_to_ui_from_context(context, base_url=...) を使用して保留中のタスクへのリンクを構築します。

応答者を限定

パラメータ名と受け入れられる識別子形式はアクティブな auth manager によって異なります。ハードコードしないでください — どれがアクティブかを確認し、現在のプロバイダがどの kwarg を公開しているかを確認してください:

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

その後、Step 2 で現在の kwarg を検索してください (執筆時点では assigned_users、アクティブな auth manager が使用する識別子形式を受け入れます — Astro は Astro ユーザー ID を使用、FabAuthManager はメール、SimpleAuthManager はユーザー名)。

---

Step 5 — 外部統合から応答

Slack bボット、カスタムアプリ、またはスクリプト用。パスをハードコードするのではなく、ライブエンドポイントを検出してください:

af api ls --filter hitl           # ライブエンドポイントリスト
af api spec \
  | jq '.paths | to_entries[] | select(.key | test("hitl"))'   # リクエスト/レスポンススキーマ

応答への PATCH パターンは安定しています。正確なパスは検出されます。典型的な形状:

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"}},
)

---

Step 6 — 安全チェック

• Airflow バージョン ≥ 3.1 (af config version)。
• コンストラクタ kwargs が Step 2 の現在のレジストリ出力と一致 — respondents vs assigned_users 形式のドリフトなし。
• 分岐の場合: すべてのオプションが下流タスク ID に解決される (直接またはStep 2 のマッピング kwarg 経由)。
• defaults のすべての値も options に含まれている。
• execution_timeout が設定されている、タイムアウト時に成功すべき場合は defaults が設定されている。
• 外部応答者がフローの一部である場合、API トークンが設定されている。

---

References

上流の docs URL はレジストリによってモジュールごとに表示されます — ハードコードしないでください:

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

関連スキル

• airflow — af registry、af api、af config コマンドリファレンス。
• airflow-ai — AI/LLM タスクデコレータと GenAI パターン。
• authoring-dags — 一般的な DAG 作成のベストプラクティス。
• testing-dags — 反復的なテスト → デバッグ → 修正サイクル。