Apache Airflow 3.3.1 та AI-агенти в DAG-ах: що змінилося для data engineering у 2026

Коротко: Apache Airflow 3.3.1 — актуальний patch-реліз лінійки 3.3, у якій ключовою для agentic workflows стала поява Task State Store. Самі LLM-оператори та AI-агенти доступні через окремий apache-airflow-providers-common-ai, а в Airflow 3.3+ AgentOperator може використовувати Task State Store для durable execution: завершені LLM- і tool-кроки кешуються та не виконуються повторно після retry. Для data engineering це означає більш контрольовану, відновлювану й спостережувану інтеграцію LLM у DAG-и без необхідності ховати всю agentic-логіку всередині PythonOperator.

Вступ

Автоматизація data pipelines у 2026 виходить за межі класичного ETL. Команди дедалі частіше хочуть, щоб пайплайни не лише переміщували й трансформували дані, а й класифікували неструктурований контент, пояснювали аномалії, генерували звіти або взаємодіяли з базами даних та API через LLM.

Раніше такі сценарії в Airflow часто реалізовували через PythonOperator із прямими викликами LLM API або через зовнішні agent frameworks. Це працює, але вимагає самостійно продумувати tool calling, retries, ліміти використання, логування, передачу результатів і поведінку після часткового збою.

У 2026 Apache Airflow отримав офіційний Common AI Provider з LLMOperator, AgentOperator, @task.llm і @task.agent. Водночас Airflow 3.3 додав Task & Asset State Store, а версія 3.3.1 стабілізує цю гілку та виправляє низку проблем, зокрема навколо state store, deferred tasks, XCom і сумісності з pandas 3.

У статті — що саме означає Airflow 3.3.1 для AI/LLM-пайплайнів, як правильно використовувати AgentOperator, де потрібен LLMOperator замість агента, які use cases мають практичний сенс і яких помилок варто уникати.

Чому Airflow 3.3.1 — це не просто черговий patch-реліз?

Airflow як хребет сучасного data engineering

З 2014 року Airflow пройшов шлях від внутрішнього інструменту Airbnb до одного з найпоширеніших оркестраторів у data engineering. Airflow 2.0 приніс новий scheduler і TaskFlow API, серія 2.x розвинула deferrable operators і dynamic task mapping, а Airflow 3.x продовжив відокремлення runtime через Task SDK та розширив data-aware orchestration.

Важливо розділяти дві речі. AI-агенти не були додані саме в Airflow 3.3.1: AgentOperator з’явився в apache-airflow-providers-common-ai, який працює з Airflow 3.0+. Натомість гілка Airflow 3.3 дає цим агентам особливо корисний механізм — first-class Task State Store, на якому Common AI Provider будує durable execution для Airflow 3.3+.

Що змінилося: від одноразового LLM-виклику до відновлюваного agent workflow

Airflow завжди керував порядком виконання задач, retries, залежностями та observability. Тепер Common AI Provider додає до цієї моделі LLM-оператори та agent loop: модель може отримати prompt, викликати дозволені tools, проаналізувати їх результати й продовжувати роботу до фінальної відповіді.

Для Airflow 3.3.1 ключова різниця — durable execution. Якщо agent task має кілька LLM- і tool-кроків та падає посередині, повторний запуск може відтворити вже завершені кроки зі state store і продовжити з місця збою. Це зменшує повторні API-виклики, витрати й ризик дублювання роботи.

Сам 3.3.1 є patch-релізом: він не вводить новий AgentOperator, але додає важливі виправлення для production-експлуатації 3.3, зокрема коректну роботу DataFrame XCom із pandas 3, виправлення task state store для ключів зі слешами та коректні retries для частини deferred task failures.

Що таке AI-агенти в DAG-ах і як це працює в Apache Airflow 3.3.1?

Нова AI-абстракція: LLMOperator, AgentOperator і @task.agent

Common AI Provider розділяє прості LLM-виклики та агентні сценарії. LLMOperator / @task.llm — це single-turn виклик: prompt → відповідь. Його варто використовувати для класифікації, summarization, extraction і structured output. AgentOperator / @task.agent — це multi-turn loop, у якому модель може самостійно вирішувати, які tools викликати й коли завершити роботу.

• Стандартизований lifecycle: Airflow керує task lifecycle, retries, залежностями та XCom, а Common AI Provider — запуском LLM/agent logic.
• Tool calling: AgentOperator отримує явно дозволені toolsets — наприклад SQLToolset або HookToolset.
• Durable execution: durable=True кешує завершені model/tool steps; на Airflow 3.3+ кеш зберігається через Task State Store.
• Usage limits: можна обмежувати кількість model requests, токени та tool calls через Pydantic AI UsageLimits.
• Observability: agent/tool execution потрапляє у стандартні task logs Airflow; нові версії Common AI Provider також розвивають tracing та HITL-сценарії.

Приклад робочого agent DAG для аналітичного SQL-сценарію:

from datetime import timedelta

from airflow.sdk import dag
from airflow.providers.common.ai.operators.agent import AgentOperator
from airflow.providers.common.ai.toolsets.sql import SQLToolset


@dag(
    schedule="@daily",
    catchup=False,
    default_args={"retries": 3, "retry_delay": timedelta(seconds=30)},
    tags=["ai", "analytics"],
)
def ai_analytics_pipeline():
    AgentOperator(
        task_id="durable_analyst",
        prompt="What are the top 5 customers by order count?",
        llm_conn_id="pydanticai_default",
        system_prompt=(
            "You are a SQL analyst. Use only the available tools "
            "and answer with data from the database."
        ),
        durable=True,
        toolsets=[
            SQLToolset(
                db_conn_id="postgres_default",
                allowed_tables=["customers", "orders"],
                max_rows=20,
            )
        ],
    )


ai_analytics_pipeline()

from datetime import timedelta

from airflow.sdk import dag
from airflow.providers.common.ai.operators.agent import AgentOperator
from airflow.providers.common.ai.toolsets.sql import SQLToolset


@dag(
    schedule="@daily",
    catchup=False,
    default_args={"retries": 3, "retry_delay": timedelta(seconds=30)},
    tags=["ai", "analytics"],
)
def ai_analytics_pipeline():
    AgentOperator(
        task_id="durable_analyst",
        prompt="What are the top 5 customers by order count?",
        llm_conn_id="pydanticai_default",
        system_prompt=(
            "You are a SQL analyst. Use only the available tools "
            "and answer with data from the database."
        ),
        durable=True,
        toolsets=[
            SQLToolset(
                db_conn_id="postgres_default",
                allowed_tables=["customers", "orders"],
                max_rows=20,
            )
        ],
    )


ai_analytics_pipeline()

У production connection postgres_default має використовувати least-privilege роль. allowed_tables — корисний application-level guardrail, але не повноцінна security boundary.

Як Airflow передає контекст агенту і отримує результат

AgentOperator отримує prompt, system_prompt, llm_conn_id та набір дозволених toolsets. Дані з попередніх задач можна передавати через XComArg, Jinja templates або аргументи декорованої @task.agent-функції. Для @task.agent функція повертає саме prompt, який буде переданий агенту.

Фінальний результат AgentOperator повертається через XCom, як і результат звичайної Airflow task. Для structured output можна вказати Pydantic output_type. Великі payloads краще зберігати в object storage або іншому зовнішньому сховищі, а через XCom передавати URI, ключ чи компактні metadata.

Durable execution у 3.3.1: навіщо тут Task State Store

LLM-агент може зробити кілька кроків: отримати schema, виконати SQL, викликати API, проаналізувати відповідь і сформувати звіт. Якщо task падає на останньому кроці, звичайний retry починає agent run заново.

З durable=True кожна завершена LLM-відповідь і результат tool call кешуються. Якщо Airflow повторює task після transient failure, Common AI Provider перевіряє fingerprint запиту та відтворює валідні завершені кроки з кешу. Виконуються лише ті кроки, яких бракує або контекст яких змінився.

Саме тут Airflow 3.3.x має пряме значення: на Airflow >= 3.3 durable cache використовує AIP-103 Task State Store без окремого durable_cache_path. Для великих model/tool payloads варто налаштувати [workers] state_store_backend і винести значення в зовнішнє сховище.

Підтримувані моделі та інтеграції

Common AI Provider побудований на Pydantic AI й дає єдину Airflow-абстракцію поверх багатьох model providers. Це не означає, що для кожного framework існує окремий AgentOperator-провайдер.

Порівняння підходів до запуску LLM у пайплайні:

Де реально застосовувати AI-агентів у data pipeline автоматизації?

Use case 1: автоматична класифікація та збагачення даних

Типовий сценарій — ETL-пайплайн, який обробляє неструктуровані записи: клієнтські відгуки, тікети служби підтримки, описи документів. Для простої класифікації тут не потрібен AgentOperator: офіційно для classification, summarization та extraction краще використовувати LLMOperator або @task.llm.

Наприклад, @task.llm може отримати батч записів, повернути structured output із категоріями, а наступна задача збереже результат у data warehouse. Якщо ж класифікація вимагає додатково звертатися до БД, CRM або API, тоді сценарій уже може виправдовувати AgentOperator.

from typing import Literal
from airflow.sdk import task

@task.llm(
    llm_conn_id="pydanticai_default",
    system_prompt="Classify the support ticket into one allowed category.",
    output_type=Literal[
        "COMPLAINT",
        "REFUND_REQUEST",
        "POSITIVE_FEEDBACK",
        "OTHER",
    ],
)
def classify_ticket(ticket_text: str):
    return ticket_text

from typing import Literal
from airflow.sdk import task

@task.llm(
    llm_conn_id="pydanticai_default",
    system_prompt="Classify the support ticket into one allowed category.",
    output_type=Literal[
        "COMPLAINT",
        "REFUND_REQUEST",
        "POSITIVE_FEEDBACK",
        "OTHER",
    ],
)
def classify_ticket(ticket_text: str):
    return ticket_text

Для контролю витрат batching потрібно організовувати на рівні upstream task, Dynamic Task Mapping або application code. batch_size не є універсальним параметром AgentOperator.

Use case 2: агент як валідатор якості даних з поясненнями

Стандартні Data Quality checks зазвичай повертають True/False, лічильник або технічне повідомлення. LLM може отримати вже знайдені аномалії та сформувати зрозуміле пояснення: наприклад, що значна частина записів певної дати втратила region_id після зміни джерела.

Важливо, щоб deterministic checks залишалися джерелом істини, а LLM пояснював результат, а не сам вирішував, чи пройшла перевірка там, де правило можна формалізувати.

Use case 3: динамічна генерація SQL і трансформацій

У мультитенантних або дослідницьких системах agent може аналізувати schema metadata, формувати read-only SQL і виконувати його через SQLToolset. Це корисно для ad hoc аналітики та автоматизованого дослідження даних.

Але LLM-generated SQL не можна запускати з надмірними правами. Production-патерн: read-only або least-privilege DB role, allowlist таблиць, max_rows, query timeout, аудит запитів і окремі права для DDL/DML. allowed_tables допомагає обмежити поведінку агента, але реальним security boundary залишаються права БД.

Use case 4: агент-репортер — від сирих даних до бізнес-інсайту

Після завершення тижневого пайплайну LLMOperator або AgentOperator може зібрати агреговані показники — revenue, churn, conversion, кількість аномалій — і сформувати короткий executive summary для Slack, email або внутрішнього dashboard.

Якщо всі показники вже підготовлені, достатньо LLMOperator. Якщо моделі потрібно самостійно звертатися до кількох джерел або виконувати додаткові перевірки через tools, тоді використання AgentOperator є виправданим.

Які обмеження та типові помилки при роботі з Airflow AI agents?

Помилка 1: агент як заміна детермінованої логіки

Агент потрібен там, де є семантична неоднозначність або необхідність працювати з tools. Якщо правило детерміноване — наприклад, «якщо значення > 1000, позначити як велике замовлення» — LLM тут зайвий. Він додає затримку, вартість і недетермінованість там, де достатньо звичайного if/else або SQL.

Помилка 2: ігнорування вартості та rate limits LLM

Кожен LLM-виклик коштує токени, а agent loop може зробити кілька model requests і tool calls у межах однієї task. Для контролю витрат використовуйте batching на рівні pipeline design, Airflow Pools, model-specific rate limits і UsageLimits — наприклад request_limit, total_tokens_limit або tool_calls_limit.

Помилка 3: відсутність retry- та fallback-стратегії

LLM/API-виклики можуть падати через timeout, rate limit або мережеві збої. Airflow дає task retries, Pydantic AI дозволяє налаштовувати agent/model behavior, а durable=True допомагає не повторювати завершені agent steps після retry. Але fallback-гілку — наприклад, перехід на deterministic logic або alert — потрібно проєктувати явно на рівні DAG.

Для tools із side effects потрібна ідемпотентність. Durable execution кешує результат завершеного tool call, але якщо tool частково виконав side effect і впав до кешування, retry може повторити дію.

Помилка 4: зберігання великих LLM-відповідей у XCom

Default XCom backend не варто використовувати як сховище великих документів, transcript histories або великих батчів. Зберігайте payload у S3, GCS, Azure Blob, базі даних чи іншому object storage, а в XCom передавайте компактний результат або посилання. Для великих durable cache values в Airflow 3.3+ аналогічно варто розглянути worker-side state_store_backend.

Реальні обмеження поточної версії 3.3.1

• Common AI Provider все ще розвивається окремо від core Airflow; його API та можливості можуть змінюватися швидше, ніж API самого Airflow.
• Agent workflows залишаються недетермінованими: однаковий prompt не гарантує буквально однаковий шлях tool calls або текстову відповідь.
• By default кожен agent run є cold single-turn conversation. Для міжзапускової пам’яті потрібно явно організувати message_history та її зберігання.
• Не всі способи підключення tools однаково покриваються durable caching; built-in toolsets мають найпрозоріший сценарій replay.
• Task State Store за замовчуванням використовує metadata DB; великі значення краще offload-ити через [workers] state_store_backend.

Загальна позиція: AI-агенти в Airflow — корисний інструмент для multi-step задач із tools і семантичною складністю, але не універсальна заміна звичайних операторів, SQL або Python-коду.

Погляд практика: чи варто вже переходити на Airflow 3.3.1 для LLM-пайплайнів?

Коли переходити вже зараз

3.3.1 — стабілізуючий patch-реліз гілки 3.3. Якщо команда вже планує Airflow 3.x і використовує або тестує Common AI Provider, саме 3.3.1 виглядає логічнішим вибором, ніж 3.3.0, оскільки містить додаткові bug fixes і важливу сумісність із pandas 3 для DataFrame XCom.

• Нові проєкти й greenfield pipelines: можна одразу будувати на Task SDK, state store та сучасних provider APIs.
• Команди з multi-step LLM/tool workloads: durable=True на Airflow 3.3+ прибирає необхідність окремо налаштовувати durable_cache_path.
• Пайплайни з реальною семантичною складністю: SQL agent, API exploration, enrichment через tools, пояснення та автоматичні звіти.

Коли краще почекати

• Критичні production-системи з великим legacy DAG-кодом: міграція з 2.x/ранніх 3.x потребує regression testing, provider compatibility checks і плану rollback.
• Команди без досвіду з LLM integration: AgentOperator спрощує orchestration, але не замінює розуміння token costs, prompt injection, model limits і tool security.
• Сценарії з жорсткою детермінованістю або SLA: якщо результат можна надійно отримати SQL/Python, агент часто лише збільшує операційний ризик.

Напрямок розвитку Airflow зрозумілий: core відповідає за orchestration, state, retries і execution model, а provider ecosystem додає AI/LLM primitives. Для production adoption це здоровіша модель, ніж заховати всю agentic-логіку в одному PythonOperator.

FAQ: питання про Apache Airflow 3.3.1 та AI-агентів

Питання: Що таке AI-агенти в Apache Airflow 3.3.1?

Відповідь: AI-агенти запускаються через AgentOperator або @task.agent з пакета apache-airflow-providers-common-ai. Це не нова фіча саме 3.3.1: provider підтримує Airflow 3.0+. Перевага Airflow 3.3+ у тому, що durable agent execution може використовувати вбудований Task State Store для кешування завершених LLM- і tool-кроків між retries.

Питання: Як налаштувати AI-агента в DAG-у на Airflow 3.3.1?

Відповідь: Встановіть Apache Airflow 3.3.1 із відповідними constraints, додайте apache-airflow-providers-common-ai та налаштуйте Connection для моделі. У AgentOperator задайте prompt, llm_conn_id, за потреби system_prompt і toolsets. Для multi-step agent workflows із retries можна ввімкнути durable=True. Якщо потрібен лише single-turn prompt — classification, extraction або summarization — краще використати LLMOperator / @task.llm.

Питання: Apache Airflow vs Prefect — що краще для AI-агентів у пайплайнах?

Відповідь: Однозначного переможця немає. Airflow має зрілу ecosystem orchestration, provider model, DAG-level observability та Common AI Provider. Prefect має іншу developer experience і також може оркеструвати LLM/agent workloads. Для організацій із наявним Airflow-стеком перехід на Common AI Provider зазвичай природніший; для нового проєкту рішення варто приймати за вимогами до orchestration, deployment, observability та досвідом команди.

Питання: Чи є безкоштовна версія Apache Airflow 3.3.1?

Відповідь: Так. Apache Airflow — open-source проєкт під Apache License 2.0, тому сам Airflow 3.3.1 можна безкоштовно розгортати на власній інфраструктурі. Managed-сервіси на кшталт Google Cloud Composer, Amazon MWAA або Astronomer оплачуються окремо за інфраструктуру та managed service.

Питання: Які помилки роблять при інтеграції AI-агентів в Airflow DAG-и?

Відповідь: Найтиповіші помилки — використовувати AgentOperator для задач, де достатньо LLMOperator або звичайного Python/SQL; не обмежувати token/tool usage; давати agent tools надмірні права; ігнорувати ідемпотентність; не проєктувати retries/fallback; зберігати великі payloads у default XCom; і вважати application-level allowlists повноцінною security boundary.

Підсумок і що далі: оркестрація даних входить в епоху AI-агентів

Ключові висновки

Apache Airflow 3.3.1 не є релізом, який сам по собі «додав AI-агентів». Агентні та LLM-абстракції приходять із apache-airflow-providers-common-ai. Але гілка Airflow 3.3 суттєво покращує їх production story завдяки Task State Store: AgentOperator з durable=True може кешувати завершені model/tool steps у state store й відновлювати виконання після retry без повторення всієї дорогої послідовності.

Для single-turn задач — classification, summarization, extraction — використовуйте LLMOperator. Для multi-step reasoning із SQL/API/tools — AgentOperator. А Airflow залишайте відповідальним за те, у чому він сильний: orchestration, scheduling, retries, dependencies, state та operational visibility.

Серед практичних нововведень саме 3.3.1 варто врахувати сумісність DataFrame XCom із pandas 3 та набір виправлень для стабільності state store, deferred tasks, UI/API й оновлень з попередніх версій.

Що далі: куди рухається Airflow і data engineering у 2026

Тренд очевидний: data orchestration поступово охоплює не лише детерміновані ETL/ELT tasks, а й LLM-driven workloads. Водночас правильна архітектура не означає «додати агента всюди». Найкращий підхід — залишати deterministic logic детермінованою, використовувати LLMOperator для single-turn AI-задач і вмикати AgentOperator лише там, де справді потрібні multi-step reasoning та tools.

Практичний крок — спробувати Common AI Provider у тестовому середовищі на Airflow 3.3.1: спочатку простий LLMOperator, потім невеликий read-only SQL agent із retries та durable=True. Це швидко покаже, де agentic orchestration дає цінність, а де достатньо звичайної задачі Airflow.

А якщо хочете освоїти Airflow, dbt та побудову пайплайнів, то спеціалізація Analytics & Data Engineer від Data Lab дає системне розуміння сучасного data stack.