Airflow 工作流接入智能体与审批机制

Apache Airflow 刚刚发布了 AI Provider 1.0，为这个已在数千条生产数据流水线中运行的工作流引擎带来了 Agent 编排基础能力。这不是一个新的 Agent 框架，而是一组 DAG 任务，让你可以在现有 Airflow 工作流中运行 LLM Agent，并提供沙箱隔离、人工审批关卡，以及能够区分模型失败和工具调用失败的重试逻辑。

这个 Provider 可以安装在 Airflow 2.11 及更高版本上，因此仍在使用 Airflow 2 的团队，无须升级整个平台，就能开始使用 Agent。它的核心能力包括 Agent 任务、沙箱代码执行，以及人在回路中的人工审批关卡。这些能力与 Airflow 的 XCom 系统集成，可以在任务之间传递上下文。

这个 Provider 到底能做什么

Common AI Provider（apache-airflow-providers-common-ai）为 Airflow DAG 增加了三项主要能力：

• Agent 任务：在一个 DAG 步骤内运行包含工具调用的 LLM 循环。
• 沙箱执行：将 Agent 生成的代码与 Airflow 进程隔离。
• 人工审批关卡（Airflow 3.3+）：暂停 Agent 执行，直到有人批准工具调用。

Agent 任务的本质，是让模型在循环中工作。它调用工具，也就是 DAG 作者允许使用的函数，读取结果，再选择下一步，直到得到答案。这个 Provider 用 Airflow 的 TaskFlow API 封装了这个循环，因此它与其他 DAG 任务一样，具备重试、日志记录和通过 XCom 传递数据的能力。

这个 Provider 通过名为 pydanticai 的统一连接类型，支持 OpenAI、Anthropic、Google 和 Bedrock 模型。你在 Airflow 连接中配置模型和 API key，Agent 任务会在运行时读取它们。

架构：Agent 如何融入 DAG 拓扑

Airflow DAG 是有向无环图，每个节点都是一个任务。AI Provider 新增的 Agent 任务与其他任务一样，可以融入这个拓扑结构。区别在于任务内部发生的事情。

传统 Airflow 任务只运行一次函数。Agent 任务则运行一个循环：

1. 模型接收 prompt 和可用工具。
2. 模型调用一个工具，例如 SQL 查询、API 请求或代码执行。
3. 工具结果返回给模型。
4. 模型决定继续调用其他工具，还是返回答案。
5. 循环持续进行，直到模型返回最终结果，或触发重试条件。

Agent 任务通过 XCom 返回最终答案，供下游任务使用。如果 Agent 失败，例如模型超时、无效的工具调用或沙箱错误，Airflow 的重试逻辑就会介入。这个 Provider 允许你配置：是自动重试，还是仅在模型判断故障属于暂时性问题后才重试。

沙箱：隔离边界

当 Agent 编写代码时，这个 Provider 会在沙箱中运行代码，而不是在 Airflow worker 进程中运行。这可以防止 Agent 生成的代码执行以下操作：

• 访问 Airflow 凭据或连接。
• 修改 DAG 状态或任务元数据。
• 读取沙箱目录之外的文件。
• 向内部服务发起网络调用，除非明确允许。

沙箱是一个独立进程，文件系统访问权限受限，也无法访问 Airflow 的内部状态。Agent 任务将代码传给沙箱，等待执行结果，再将结果返回给模型。如果沙箱崩溃或超时，Agent 任务就会失败，随后由 Airflow 重试。

这个沙箱不使用容器或 VM，而是一个限制了导入和文件系统访问权限的 Python 子进程。这使它的开销较低，但也意味着你需要信任沙箱实现能够守住隔离边界。如果需要更高的隔离级别，可以用 KubernetesPodOperator 或类似机制来封装 Agent 任务。

人工审批关卡：暂停 Agent 执行

在 Airflow 3.3 及更高版本中，Agent 任务可以在执行工具调用前暂停，等待人工批准。当 Agent 想要向 Slack 发消息、发送邮件或执行写入查询时，这个能力就很有用。

审批关卡通过 Airflow 现有的审批机制运行。当 Agent 调用被标记为需要审批的工具时，任务会暂停，并创建一条审批请求。审批人在 Airflow UI 中查看请求，批准或拒绝后，Agent 任务便会恢复执行。

审批请求包含：

• 工具名称和参数。
• 模型调用该工具的推理依据。
• 本轮循环中此前工具调用的上下文。
• Agent 当前的计划状态。

如果审批人拒绝这次工具调用，Agent 任务就会失败，Airflow 随后会根据重试配置决定是否重试。如果获得批准，工具就会执行，结果再返回给模型。

这套机制与 Airflow 的 XCom 系统集成。审批请求是一个 XCom 值，下游任务可以读取它。审批决定则是另一个 XCom 值，Agent 任务读取后，决定继续执行还是失败退出。

重试逻辑：模型失败与工具失败的区别

Airflow 会自动重试失败的任务，但 Agent 任务需要更智能的重试逻辑。模型超时是暂时性故障，无效的 SQL 查询则不是。

这个 Provider 增加了两种重试模式：

• 自动重试：把所有失败都视为暂时性故障。无论是模型超时、工具调用失败，还是沙箱崩溃，Airflow 都会重试整个 Agent 任务。这适合应对不稳定的 API 或速率限制，但会在持续性错误上浪费 token。
• 由模型决定重试：让模型决定是否重试。工具调用失败时，错误信息会返回给模型。模型可以：
  • 使用不同参数重试工具调用。
  • 调用其他工具。
  • 向 DAG 作者返回错误。
  • 在重试前请求人工批准。

后一种模式要求模型能够理解错误信息，并作出重试决策。它可以减少持续性错误造成的 token 浪费，但会增加延迟，因为模型需要先处理错误，再作决定。

你可以为每个 Agent 任务配置重试模式。默认采用自动重试，并设置最大重试次数。

可观测性：跟踪 token 用量和延迟

这个 Provider 为每个 Agent 任务提供以下指标：

• token 总用量，包括 prompt 和 completion。
• 每次工具调用的延迟。
• 沙箱执行时间。
• 审批等待时间，如果涉及审批。

这些指标会进入 Airflow 现有的可观测性体系。你可以通过 Airflow 的 REST API 查询，将它们导出到 Prometheus，或在任务日志中查看。

这个 Provider 还会记录每次工具调用的参数和结果。这有助于排查 Agent 为什么作出某个决定，或某次工具调用为什么失败。日志包括：

• 模型的推理过程，如果模型提供。
• 工具调用的结果或错误。
• 循环中每个步骤的耗时。

在跟踪 token 用量时，这个 Provider 会从模型 API 响应中读取 token 数量。如果模型不提供 token 数量，则使用 tokenizer 进行估算。

代码示例：使用 SQL 工具的 Agent 任务

下面这个 Agent 任务会查询数据仓库，解释收入下降的原因：

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

agent_task 装饰器将函数封装进一个 Agent 循环。tools 参数列出可用工具，这里是一个 SQL 查询工具。sandbox=True 参数让 Agent 生成的代码在沙箱中运行。require_approval=False 参数允许 Agent 无须人工审批就执行工具调用。

Agent 任务返回一个包含 prompt 和最大迭代次数的字典。这个 Provider 运行 Agent 循环，按需调用工具，并通过 XCom 返回最终答案。下游的 post_to_slack 任务接收解释结果，并将其发布出去。

取舍：什么时候用 Airflow 编排 Agent

Airflow 会带来调度开销。如果你的 Agent 工作流对延迟敏感，需要亚秒级响应，Airflow 就不合适。如果 Agent 工作流按计划运行，或属于更大数据流水线的一部分，使用 Airflow 就很合理。

沙箱的隔离能力弱于容器或 VM。如果你需要强隔离，例如运行不可信代码或满足合规要求，可以用 KubernetesPodOperator 封装 Agent 任务，或将其放在独立集群中运行。

人工审批关卡适合发布 Slack 消息、执行写入查询等高风险操作，但会增加延迟。如果你的 Agent 工作流完全自动化，就关闭审批关卡。

故障类型与应对方式

• 模型超时：这个 Provider 默认将每次模型调用的超时时间设为 60 秒。如果模型没有响应，任务就会失败，随后由 Airflow 重试。你可以在 Agent 任务配置中延长超时时间。
• 无效的工具调用：如果模型使用无效参数调用工具，工具会返回错误。在自动重试模式下，Airflow 会重试整个任务。在由模型决定重试的模式下，错误会返回给模型，模型可以使用不同参数再次尝试。
• 沙箱崩溃：如果沙箱进程崩溃，Agent 任务就会失败，随后由 Airflow 重试。沙箱不会在重试之间保留状态，因此 Agent 会从头开始。
• 审批超时：如果审批人没有在超时时间内批准工具调用，Agent 任务就会失败，默认超时时间为 1 小时。你可以为每个任务单独配置超时时间。
• token 限额耗尽：如果 Agent 循环消耗了过多 token，模型 API 会返回错误。这个 Provider 不会跨重试跟踪累计 token 用量，因此你需要单独监控。

Airflow 2.11 兼容性

这个 Provider 可以安装在 Airflow 2.11 上，但部分功能需要 Airflow 3.3：

• 人工审批关卡需要 Airflow 3.3+。
• 由模型决定重试支持 Airflow 2.11+。
• 沙箱执行支持 Airflow 2.11+。

在 Airflow 2.11 上，需要将导入从 airflow.decorators 改为 airflow.operators.python。agent_task 装饰器相同，但导入路径不同。

适合使用 Airflow AI Provider 的情况：

• 你已经在运行 Airflow，希望向现有 DAG 添加 Agent 任务。
• 你的 Agent 工作流按计划运行，或属于数据流水线的一部分。
• 你需要为高风险操作设置人工审批关卡。
• 你希望复用 Airflow 的重试逻辑、可观测性和凭据管理。
• 你的 Agent 工作流对延迟敏感，需要亚秒级响应。
• 你需要为 Agent 生成的代码提供容器或 VM 级别的强隔离。
• 你目前没有运行 Airflow，也不想运维它。
• 你的 Agent 工作流独立运行，不与数据流水线集成。

对于已经运行 Airflow、希望添加 Agent 任务，又不想部署独立 Agent 框架的团队，这个 Provider 很有用。它不能替代 LangGraph 或 CrewAI 这样的独立 Agent 框架，但填补了一类需求：在现有工作流编排体系内运行 Agent。

Airflow AI Provider 1.0 发布公告

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。