Workflows com humano no loop

Criando Pipelines de Dados com Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Casos de uso de HITL

Aprovar uma resposta gerada por IA
Roteamento de um ticket de suporte para o time certo
Revisar um relatório de qualidade de dados antes de publicar

Visual de humano no loop

A família de operadores HITL

Operador	Caso de uso
`ApprovalOperator`	Binário aprovar/rejeitar
`HITLBranchOperator`	Escolher quais tarefas rodam depois
`HITLOperator`	Opções personalizadas + formulário
`HITLEntryOperator`	Entrada via formulário apenas, sem opções

Todos os operadores HITL são implementados como operadores adiáveis

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

O pipeline pausa até os campos serem preenchidos

Acessando a resposta

@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)

A entrada do formulário está em params_input
As opções (se houver) estão em 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"],
)

Humano seleciona uma opção, mapeada para um ID de tarefa downstream
Caminhos não selecionados são pulados

Ações necessárias na UI

Tarefas HITL criam uma Ação necessária
Revisores respondem na visualização da instância da tarefa

Todos os operadores HITL são adiáveis
Eles liberam o slot do worker enquanto aguardam
O Triggerer gerencia a espera assíncrona

HITL na UI do Airflow

Vamos praticar!

Criando Pipelines de Dados com Airflow