Apache Airflow 發佈 AI Provider 1.0，為 DAG 加入智能體任務、沙盒執行及人工審批

Apache Airflow just shipped AI Provider 1.0, bringing agent orchestration primitives to a workflow engine already running in thousands of production data pipelines. This is not a new agent framework. It is a set of DAG tasks that let you run LLM agents inside existing Airflow workflows, with sandboxing, human approval gates, and retry logic that understands when a model fails versus when a tool call fails.

The provider installs on Airflow 2.11 and later, so teams still on Airflow 2 can start using agents without upgrading the entire platform. The core primitives are agent tasks, sandboxed code execution, and human-in-the-loop approval gates that integrate with Airflow's XCom system for passing context between tasks.

What the Provider Actually Does

The Common AI provider (apache-airflow-providers-common-ai) adds three main capabilities to Airflow DAGs:

• Agent tasks that run LLM loops with tool calls inside a DAG step
• Sandboxed execution that isolates agent-generated code from the Airflow process
• Human approval gates (Airflow 3.3+) that pause agent execution until a person approves a tool call

An agent task is a model working in a loop. It calls tools (functions the DAG author allows), reads results, and picks the next step until it has an answer. The provider wraps this loop in Airflow's TaskFlow API, so it behaves like any other DAG task with retries, logging, and XCom data passing.

The provider supports OpenAI, Anthropic, Google, and Bedrock models through a unified connection type called pydanticai. You configure the model and API key in an Airflow connection, and the agent task reads it at runtime.

Architecture: How Agents Fit Into DAG Topology

Airflow DAGs are directed acyclic graphs where each node is a task. The AI Provider adds agent tasks that fit into this topology like any other task. The difference is what happens inside the task.

A traditional Airflow task runs a function once. An agent task runs a loop:

1. Model receives the prompt and available tools
2. Model calls a tool (SQL query, API request, code execution)
3. Tool result flows back to the model
4. Model decides to call another tool or return an answer
5. Loop continues until the model returns a final result or hits a retry condition

The agent task returns its final answer through XCom, so downstream tasks can use it. If the agent fails (model timeout, invalid tool call, sandbox error), Airflow's retry logic kicks in. The provider lets you configure whether retries happen automatically or only after a model decides the failure is transient.

Sandboxing: Isolation Boundaries

When an agent writes code, the provider runs it in a sandbox instead of the Airflow worker process. This prevents agent-generated code from:

• Accessing Airflow credentials or connections
• Modifying DAG state or task metadata
• Reading files outside the sandbox directory
• Making network calls to internal services (unless explicitly allowed)

The sandbox is a separate process with limited filesystem access and no access to Airflow's internal state. The agent task passes code to the sandbox, waits for the result, and returns it to the model. If the sandbox crashes or times out, the agent task fails and Airflow retries it.

The sandbox does not use containers or VMs. It is a Python subprocess with restricted imports and filesystem access. This keeps overhead low but means you need to trust the sandbox implementation to enforce boundaries. For higher isolation, you would wrap the agent task in a KubernetesPodOperator or similar.

Human Approval Gates: Pausing Agent Execution

On Airflow 3.3 and later, agent tasks can pause and wait for human approval before executing a tool call. This is useful when the agent wants to post to Slack, send an email, or run a write query.

The approval gate works through Airflow's existing approval mechanism. When the agent calls a tool marked as requiring approval, the task pauses and creates an approval request. A person reviews the request in the Airflow UI, approves or rejects it, and the agent task resumes.

The approval request includes:

• The tool name and arguments
• The model's reasoning for calling the tool
• Context from previous tool calls in the loop
• The current state of the agent's plan

If the person rejects the tool call, the agent task fails and Airflow retries it (or not, depending on retry configuration). If approved, the tool executes and the result flows back to the model.

This integrates with Airflow's XCom system. The approval request is an XCom value that downstream tasks can read. The approval decision is another XCom value that the agent task reads to continue or fail.

Retry Logic: When Models Fail vs. When Tools Fail

Airflow retries failed tasks automatically, but agent tasks need smarter retry logic. A model timeout is transient. An invalid SQL query is not.

The provider adds two retry modes:

Automatic retries treat every failure as transient. If the model times out, the tool call fails, or the sandbox crashes, Airflow retries the entire agent task. This works for flaky APIs or rate limits but wastes tokens on persistent errors.

Model-decided retries let the model decide whether to retry. When a tool call fails, the error message flows back to the model. The model can:

• Retry the tool call with different arguments
• Call a different tool
• Return an error to the DAG author
• Request human approval before retrying

This mode requires the model to understand error messages and make retry decisions. It saves tokens on persistent errors but adds latency because the model has to process the error and decide.

You configure retry mode per agent task. The default is automatic retries with a max retry count.

Observability: Tracking Token Usage and Latency

The provider exposes metrics for each agent task:

• Total tokens used (prompt + completion)
• Number of tool calls
• Latency per tool call
• Sandbox execution time
• Approval wait time (if applicable)

These metrics flow into Airflow's existing observability stack. You can query them through Airflow's REST API, export them to Prometheus, or view them in the task logs.

The provider also logs every tool call with arguments and results. This is useful for debugging why an agent made a particular decision or why a tool call failed. The logs include:

• Model reasoning (if the model exposes it)
• Tool call arguments
• Tool call result or error
• Time spent in each step of the loop

For token usage tracking, the provider reads token counts from the model API response. If the model does not expose token counts, the provider estimates them using a tokenizer.

Code Example: Agent Task with SQL Tools

Here is an agent task that queries a warehouse to explain a revenue drop:

from airflow.decorators import dag, task
from datetime import datetime
from airflow.providers.common.ai.decorators import agent_task

@dag(start_date=datetime(2026, 10, 1), schedule=None)
def revenue_analysis():

    @agent_task(
        conn_id="pydanticai_default",
        tools=["sql_query"],
        sandbox=True,
        require_approval=False
    )
    def explain_revenue_drop(context):
        """
        Query the warehouse to explain why revenue dropped 15% last week.
        You have read-only access to the sales and customer tables.
        """
        return {
            "prompt": "Revenue dropped 15% last week. Query the warehouse to find the cause.",
            "max_iterations": 10
        }

    @task
    def post_to_slack(explanation: str):
        # Post the explanation to Slack
        pass

    result = explain_revenue_drop()
    post_to_slack(result)

revenue_analysis()

The agent_task decorator wraps the function in an agent loop. The tools parameter lists available tools (in this case, a SQL query tool). The sandbox=True parameter runs agent-generated code in a sandbox. The require_approval=False parameter allows the agent to execute tool calls without human approval.

The agent task returns a dictionary with the prompt and max iterations. The provider runs the agent loop, calls tools as needed, and returns the final answer through XCom. The downstream post_to_slack task receives the explanation and posts it.

Trade-offs: When to Use Airflow for Agent Orchestration

Airflow adds scheduling overhead. If your agent workflow is latency-sensitive (sub-second responses), Airflow is not the right tool. If your agent workflow runs on a schedule or as part of a larger data pipeline, Airflow makes sense.

The sandbox isolation is weaker than containers or VMs. If you need strong isolation (untrusted code, compliance requirements), wrap the agent task in a KubernetesPodOperator or run it in a separate cluster.

Human approval gates are useful for high-risk actions (posting to Slack, running write queries) but add latency. If your agent workflow is fully automated, disable approval gates.

Failure Modes and Mitigations

Model timeouts: The provider sets a default timeout of 60 seconds per model call. If the model does not respond, the task fails and Airflow retries it. You can increase the timeout in the agent task configuration.

Invalid tool calls: If the model calls a tool with invalid arguments, the tool returns an error. In automatic retry mode, Airflow retries the entire task. In model-decided retry mode, the error flows back to the model, which can retry with different arguments.

Sandbox crashes: If the sandbox process crashes, the agent task fails. Airflow retries it. The sandbox does not persist state between retries, so the agent starts from scratch.

Approval timeouts: If a human does not approve a tool call within the timeout (default 1 hour), the agent task fails. You can configure the timeout per task.

Token limit exhaustion: If the agent loop uses too many tokens, the model API returns an error. The provider does not track cumulative token usage across retries, so you need to monitor this separately.

Airflow 2.11 Compatibility

The provider installs on Airflow 2.11, but some features require Airflow 3.3:

• Human approval gates require Airflow 3.3+
• Model-decided retries work on Airflow 2.11+
• Sandboxed execution works on Airflow 2.11+

On Airflow 2.11, change imports from airflow.decorators to airflow.operators.python. The agent_task decorator is the same, but the import path is different.

Technical Verdict

Use Airflow AI Provider when:

• You already run Airflow and want to add agent tasks to existing DAGs
• Your agent workflow runs on a schedule or as part of a data pipeline
• You need human approval gates for high-risk actions
• You want to reuse Airflow's retry logic, observability, and credential management

Avoid it when:

• Your agent workflow is latency-sensitive (sub-second responses)
• You need strong isolation (containers, VMs) for agent-generated code
• You do not already run Airflow and do not want to operate it
• Your agent workflow is standalone and does not integrate with data pipelines

The provider is useful for teams that already run Airflow and want to add agent tasks without deploying a separate agent framework. It is not a replacement for standalone agent frameworks like LangGraph or CrewAI, but it fills a gap for teams that want to run agents inside existing workflow orchestration.

Source Links

• Airflow AI Provider 1.0 announcement
• Lobsters discussion