Airflow

HITL이 해결하는 문제

Human-in-the-Loop(HITL) 기능은 Airflow 3.1에서 도입되었으며, 사람의 의사결정을 워크플로우에 직접 포함시킬 수 있게 해준다. 이 강력한 기능은 워크플로우가 일시 정지되어 사람의 입력을 기다릴 수 있게 해주므로, 승인 프로세스, 수동 품질 검사, 그 밖에 사람의 판단이 필수적인 시나리오에 안성맞춤이다.

HITL 기능은 LLM(대규모 언어 모델) 워크플로우에서 특히 가치가 있다 — 사람이 제공하는 가이드가 더 나은 결과를 얻는 데 핵심적인 경우가 많기 때문이다. 또한 엔터프라이즈 데이터 파이프라인에서도 매우 유용한데, 사람의 검증이 자동화된 프로세스를 보완하고 강화할 수 있기 때문이다.

3.3의 변경점: awaiting_input 상태

Airflow 3.3부터, 입력을 기다리는 HITL 태스크는 트리거러(triggerer)로 지연(defer)되는 대신 스케줄러가 직접 관리하는 전용 ``awaiting_input`` 태스크 상태를 사용한다. 대기하는 동안 이 태스크는 워커 슬롯도, 트리거러도 점유하지 않는다 — 그래서 HITL 태스크가 응답을 기다리는 중이어도 트리거러를 0개로 스케일 다운할 수 있다. 태스크는 사람의 응답이 오거나, 스케줄러의 응답 타임아웃 스윕(response-timeout sweep)에 걸릴 때 재개된다. Airflow 3.1과 3.2에서는 HITL 태스크가 예전 방식인 트리거 기반 지연(trigger-based deferral)을 사용했다.

대기 중인 ``awaiting_input`` 태스크는 풀(pool) 슬롯도 차지하지 않는다. 이는 예전의 지연 방식과 다른 점인데, 예전 방식에서는 ``include_deferred``가 활성화된 풀이 있다면 지연된 HITL 태스크가 그 풀의 슬롯을 소모했었다.

네 가지 입력 패턴

HITL은 네 가지 형태의 사람 입력을 지원한다.

**입력 제공(Input Provision)**: 사용자가 params 형태로 입력을 제공하면 이후 태스크에서 사용된다. LLM 워크플로우 안에서 사람의 가이드가 필요한 경우에 유용하다.

**옵션 선택(Option Selection)**: 사용자가 제시된 옵션 중 하나를 선택해 워크플로우 방향을 정할 수 있다. 여러 옵션을 동시에 선택하는 것도 허용된다.

**승인 또는 거부(Approval or Rejection)**: 옵션 선택의 특수한 형태로, 'Approval'과 'Rejection'만 옵션으로 제공된다. ``assigned_users``를 설정해 응답 가능한 사용자를 제한할 수 있으며, 이는 사용자 id와 이름을 모두 담은 딕셔너리 리스트(예: ``[{"id": "1", "name": "user1"}]``)로 지정한다 — 이 목록에 있는 사용자만 응답할 수 있다.

**브랜치 선택(Branch Selection)**: 사용자가 Dag 안에서 어떤 브랜치를 따를지 선택할 수 있다. 콘텐츠 검수처럼 사람의 판단이 때때로 필요한 시나리오에서 흔히 쓰인다. 옵션 선택과 비슷하지만, 옵션 자체가 태스크여야 하고 그 관계를 워크플로우에 명시해야 한다는 차이가 있다.

알림과 UI 링크

Notifier(알림자)는 태스크가 입력을 기다리거나, 성공하거나, 실패하는 것 같은 HITL 이벤트를 처리하는 콜백 메커니즘이다. 예제에서는 시연 목적으로 메시지를 로그로 남기는 ``LocalLogNotifier``를 사용한다.

``HITLOperator.generate_link_to_ui_from_context`` 메서드는 사용자가 응답해야 할 UI 페이지로 바로 이동하는 링크를 생성하는 데 쓸 수 있다. 이 메서드는 네 개의 인자를 받는다 — 알림자에 의해 ``notify``로 자동 전달되는 ``context``, (선택) UI의 base URL(``base_url``, 지정하지 않으면 설정의 ``api.base_url``을 사용), (선택) UI 페이지에 미리 선택되어 있을 옵션(``options``), (선택) UI 페이지에 미리 채워질 입력값(``params_inputs``)이다. 이를 이용하면 알림이나 로그에 실행 가능한(actionable) 링크를 손쉽게 포함시킬 수 있다. 사용자는 원한다면 직접 커스텀 Notifier를 구현할 수도 있다.

HITL 오퍼레이터에는 ``notifiers`` 인자로 여러 개의 Notifier 리스트를 전달할 수 있다. 오퍼레이터가 사람의 응답을 기다리는 HITL 요청을 생성하면, 각 Notifier의 ``notify`` 메서드가 ``context`` 하나를 인자로 받아 호출된다.

로컬에서 HITL Dag 테스트하기: REST API로 응답 제출

``airflow dags test``(그리고 그 기반이 되는 ``dag.test()``)는 HITL 태스크를 지원한다. ``awaiting_input`` 상태에 도달한 태스크는 그대로 멈춰(parked) 있고 — 테스트 실행 자체가 그것을 스스로 해결해주지는 않는다 — 실행은 외부에서 응답이 기록될 때까지 대기하면서, 어떤 태스크가 입력을 기다리는지 로그로 남긴다. 이 응답은 실제 배포 환경과 동일한 경로를 거친다 — Required Actions 페이지나, 메타데이터 데이터베이스를 공유하는 api-server(예: ``airflow standalone``이나 별도로 실행한 ``airflow api-server``)의 HITL REST API(``PATCH .../hitlDetails``)를 통해서다. 응답이 기록되면 테스트 실행은 해당 태스크를 재개하고 다운스트림 태스크로 이어간다.

이 덕분에 AI 에이전트도 HITL 파이프라인을 로컬에서 엔드투엔드로 구동할 수 있다 — ``airflow dags test``를 실행하고, 대기 로그 라인을 지켜보다가, 사람에게 물어보고, HITL REST API로 그 답을 제출하면 된다. 관련된 두 호출은 다음과 같다 (``~``는 ``dag_id``와 ``dag_run_id``에 대한 와일드카드로 동작한다).

# 응답 대기 중인 요청 조회 (subject, options, params, run/task 식별자)
GET /api/v2/dags/~/dagRuns/~/hitlDetails?response_received=false

# 응답 제출 — 테스트 실행은 다음 폴링 시점에 태스크를 재개한다
# map_index는 매핑되지 않은(non-mapped) 태스크의 경우 -1이다
PATCH /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/hitlDetails
{"chosen_options": ["Approve"], "params_input": {}}

유의할 점은, ``response_timeout``이나 그 밖의 타임아웃 기본값은 스케줄러가 강제하는데, ``airflow dags test``에서는 스케줄러가 실행되지 않는다는 것이다. 따라서 멈춰 있는 태스크는 응답이 올 때까지 무기한 대기하므로, UI나 REST API를 통해 응답을 제공해야만 실행이 끝난다.