Human-in-the-loop ワークフロー

Airflow によるデータパイプラインの構築

Volker Janz

Senior Developer Advocate at Astronomer

HITL のユースケース

AI が生成した回答の承認
サポートチケットの適切なチームへのルーティング
データ品質レポートの公開前レビュー

Human-in-the-loop の図解

HITL オペレーターファミリー

オペレーター	ユースケース
`ApprovalOperator`	承認／拒否の二択
`HITLBranchOperator`	次に実行するタスクの選択
`HITLOperator`	カスタムオプション＋フォーム
`HITLEntryOperator`	フォーム入力のみ、オプションなし

すべての HITL オペレーターは遅延実行オペレーターとして実装されています

HITLEntryOperator

from airflow.providers.standard.operators.hitl import HITLEntryOperator
from airflow.sdk import Param


review = HITLEntryOperator(
    task_id="human_review",
    subject="Review AI draft response",
    body="{{ ti.xcom_pull(task_ids='generate_draft') }}",
    params={
        "feedback": Param("", type="string"),
        "urgency": Param("p3", type="string"),
    },
)

フィールドが入力されるまでパイプラインは一時停止します

レスポンスの取得

@task
def send_response(hitl_output):
    feedback = hitl_output["params_input"]["feedback"]
    urgency = hitl_output["params_input"]["urgency"]
    print(f"Sending response (urgency: {urgency}): {feedback}")


draft = generate_draft()
chain(draft, review)
send_response(review.output)

フォーム入力は params_input に格納されます
オプション（選択した場合）は chosen_options に格納されます

HITLBranchOperator

from airflow.providers.standard.operators.hitl import HITLBranchOperator


route = HITLBranchOperator(
    task_id="route_complaint",
    subject="Route customer complaint",
    body="A customer reported an issue. Choose the responsible team.",
    options=["Billing Issue", "General Inquiry", "Technical Issue"],
    options_mapping={
        "Billing Issue": "handle_billing",
        "Technical Issue": "handle_technical",
        "General Inquiry": "handle_general",
    },
    defaults=["General Inquiry"],
)

担当者がオプションを選択すると、下流のタスク ID にマッピングされます
未選択のパスはスキップされます

UI での必要なアクション

HITL タスクは必要なアクションを生成します
レビュー担当者はタスクインスタンスビューで回答します

すべての HITL オペレーターは遅延実行に対応しています
待機中はワーカースロットを解放します
非同期の待機は Triggerer が処理します

Airflow UI 上の HITL

練習しましょう！

Airflow によるデータパイプラインの構築