Human-in-the-loop 워크플로

Airflow로 데이터 파이프라인 구축하기

Volker Janz

Senior Developer Advocate at Astronomer

HITL 활용 사례

AI 생성 응답 승인
지원 티켓을 담당 팀으로 라우팅
데이터 품질 보고서 검토 후 게시

Human in the loop 시각화

HITL 오퍼레이터 종류

오퍼레이터	활용 사례
`ApprovalOperator`	이진 승인/거부
`HITLBranchOperator`	다음 태스크 선택
`HITLOperator`	커스텀 옵션 + 폼
`HITLEntryOperator`	순수 폼 입력, 옵션 없음

모든 HITL 오퍼레이터는 지연 가능 오퍼레이터(deferrable operators)로 구현됩니다

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의 Required Actions

HITL 태스크는 Required Action을 생성합니다
검토자는 태스크 인스턴스 뷰에서 응답합니다

모든 HITL 오퍼레이터는 지연 가능(deferrable)합니다
대기 중에는 워커 슬롯을 반환합니다
Triggerer가 비동기 대기를 처리합니다

Airflow UI의 HITL

연습해 봅시다!

Airflow로 데이터 파이프라인 구축하기