Робочі процеси з участю людини (human-in-the-loop)

Створення конвеєрів даних з Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Випадки використання HITL

Погодження відповіді, згенерованої ШІ
Маршрутизація звернення підтримки до відповідної команди
Перевірка звіту про якість даних перед публікацією

Візуалізація 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 подальшого завдання
Необрані гілки пропускаються

Required Actions в інтерфейсі

Завдання HITL створюють Required Action
Рецензенти відповідають у перегляді екземпляра завдання

Усі оператори HITL відкладені
Вони вивільняють слот воркера під час очікування
Triggerer обробляє асинхронне очікування

HITL в інтерфейсі Airflow

Давайте потренуємось!

Створення конвеєрів даних з Airflow