airflow-hitl

Airflow Human-in-the-Loop Operators

Airflow UI または REST API を介して人間から応答があるまで、DAG を一時停止します。HITL オペレーターは遅延可能なため、待機中にワーカースロットを解放します。

Airflow 3.1+ が必要です（af config version）。

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

相互参照：AI/LLM タスクデコレーターについては migrating-ai-sdk-to-common-ai、以下で使用するレジストリおよび API の確認コマンドについては airflow を参照してください。

---

ステップ 1 — 必要な機能を選ぶ

機能	クラス（ステップ 2 で確認）
承認または却下し、却下時は後続タスクをスキップする	ApprovalOperator
N 個の選択肢を提示し、どれが選ばれたかを返す	HITLOperator
選択に基づいて、1 つ以上の後続タスクに分岐する	HITLBranchOperator
フォームを収集する（承認／選択ステップなし）	HITLEntryOperator
HITL トリガーを直接使用する（高度な用途／カスタムオペレーター）	HITLTrigger

クラス名をハードコードするのはここだけです。プロバイダーはリリースごとにパラメーターを追加、改名、削除するため、記憶を頼りにパラメーター一覧をコピーしないでください。コードを書く前に、現在のシグネチャを取得してください。

---

ステップ 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 など）を受け付けます。

Notifier

HITL オペレーターは notifiers リストを受け付けます。Notifier の notify(context) メソッド内で、HITLOperator.generate_link_to_ui_from_context(context, base_url=...) を使って保留中のタスクへのリンクを作成してください。

応答できるユーザーの制限

パラメーター名と受け付ける識別子の形式は、アクティブな認証マネージャーによって異なります。ハードコードしないでください。どの認証マネージャーがアクティブか、また現在のプロバイダーがどのキーワード引数を公開しているかを確認してください。

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 — テスト → デバッグ → 修正を繰り返すサイクル。