用 Airflow 编排 AI 流水线

启动 AI 原型很容易。一个笔记本、一个 API 密钥，再加上几行 Python 代码，也许就足以完成文本分类、文档摘要或推荐生成。

然而，要让这个原型每天都稳定运行，则是完全不同的挑战。

当模型 API 达到速率限制时会怎样？如果某个请求超时、输出不符合预期格式，或下游数据库暂时不可用，又该如何处理？

当一条 AI 流水线变成周期性工作流时，可靠性、监控与恢复会迅速变得与模型本身同等重要。

在本文中，我将向您展示如何使用 Apache Airflow（一个用于开发、调度与监控工作流的开源平台）构建一条完整的 AI 流水线。

我们的流水线将每天运行，获取客户评价、使用大语言模型（LLM）对情感进行分类，并存储结果——全部在一个可靠的工作流中完成！

为何选择 Airflow 作为 AI 流水线的编排工具？

Jupyter 笔记本非常适合试验。我们都喜欢用它来测试想法、写点快速代码，并边走边可视化结果。

一个cron 任务往往足以让一个简单脚本在每晚运行。虽然，实话说，很多脚本会被手动运行数周，直到有人终于给它们安排一个正式的计划。

问题从脚本膨胀为多个脚本并且开始相互依赖时出现。此时，我们不再是一个独立程序：而是一个工作流。

无论是笔记本还是 cron 任务，都不是为可靠地管理复杂工作流而设计的。

依赖关系常常隐藏在代码中，故障需要手动排查，重新运行流水线可能意味着重复已经成功完成的步骤。

传统的抽取-转换-加载（ETL）流水线中，我们已经会遇到部分这类复杂性。ETL 工作流可能需要获取数据、进行转换、验证，然后加载到其他系统中。

即便输入与转换是可预测的，每多一步都会新增一个潜在的失败点。如今，AI 流水线让这一挑战更为突出。

如果您专门对 ETL 流水线感兴趣，推荐这门关于Airflow 中的 ETL 流水线课程。

AI 流水线

考虑一个典型的 AI 工作流：收集数据、清洗数据、调用模型 API、验证响应、存储结果，并在出现问题时通知相关人员。

每一步都依赖前一步，而且每一步都可能以不同方式失败：API 可能返回速率限制错误；模型请求可能超时；生成的响应也许是有效的 JSON，却仍无法通过质量检查。

输出也可能不完整、不一致，或者与下一步所期望的格式不同。

Airflow 并不能消除这些失败，但能让它们更易于管理。

借助 Airflow，调用 LLM 的任务在遇到临时速率限制时可以重试，而无需重跑数据摄取步骤。验证任务能在结果进入生产前拦截格式错误或质量不佳的输出。只有在用尽配置的重试后，告警才会通知团队。

Airflow 之所以能做到这一点，是因为它将工作流表示为有向无环图（DAG）。

与其将整条流水线藏在一段长脚本中，不如用 DAG 将每个操作定义为独立任务，并明确依赖关系、重试策略、日志与执行状态。

这正是 Airflow 非常适合 AI 流水线的主要原因：它把失败视为工作流中的「预期组成部分」，而非意外的惊喜。

当一个实验成长为周期性的生产流程时，Airflow 提供所需的编排层，让其可观测、可恢复且更易维护。

在开始构建我们的流水线之前，先来完成环境搭建。

环境搭建

我们需要三个简短步骤来完成环境设置。

安装 Airflow 3 与 Airflow AI SDK

在本教程中，我们将在本地运行 Airflow。

这种设置非常适合学习与开发，因为只需一条命令就能初始化元数据数据库，并启动使用 Airflow 所需的服务。

首先，让我们创建并激活一个虚拟环境，以便将 Airflow 及其依赖与您其他 Python 安装隔离：

python3 -m venv ~/airflow-venv
source ~/airflow-venv/bin/activate

接下来，我们需要使用版本钉定与官方约束文件来安装 Airflow 3：

AIRFLOW_VERSION=3.3.0
PYTHON_VERSION="$(python -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')"

CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

pip install "apache-airflow==${AIRFLOW_VERSION}" \
 --constraint "${CONSTRAINT_URL}"

请注意，Airflow 依赖许多 Python 包，而这些包也有各自的依赖。若不使用约束文件，pip 可能会选择更新或不兼容的包版本。

安装表面上也许正常，但 Airflow 之后在导入 provider、启动服务或运行任务时可能失败。Airflow 的官方安装指南建议使用约束文件来安装一组已知兼容的依赖。

由于我们的工作流会调用 LLM，我们还需要安装 Airflow 的 Common AI provider，以及 Pydantic AI 的 OpenAI 依赖：

pip install \
 "apache-airflow-providers-common-ai" \
 "pydantic-ai-slim[openai]"

接下来，配置 OpenAI 连接。Airflow 使用连接来将凭据与连接设置存储在 DAG 代码之外。这样既能保持 Python 文件更简洁，又能避免在流水线中硬编码密钥。

在验证安装后，就该设置 OpenAI 密钥来允许与模型进行实际通信了。这方面有多种方式可选。

我偏好的方式是：

1. 在当前终端会话中导出我们的 OpenAI API 密钥：

export OPENAI_API_KEY="your-openai-api-key"

2. 创建一个名为 pydanticai_default 的 Airflow 连接：

airflow connections add pydanticai_default \
 --conn-type pydanticai \
 --conn-password "$OPENAI_API_KEY" \
 --conn-extra '{"model":"openai:gpt-4o-mini"}'

Common AI provider 使用 pydanticai 连接类型，通过Pydantic AI 与 LLM 提供商通信。

该命令会将连接存储到 Airflow 的元数据数据库，因此之后您无需保留 .env 文件或在每次启动 Airflow 时再次传递密钥。

3. 最后，我们可以用以下命令启动本地 Airflow 环境：

airflow standalone

在 Airflow 运行期间，请保持该终端窗口打开。Standalone 模式非常适合本地开发与教程，但并非面向生产环境的部署模式。

要访问 Airflow 界面，现在可以在浏览器中打开 http://localhost:8080。首先看到的是登录页面。但该使用什么用户名与密码呢？

Apache Airflow 的登录视图。

当 Airflow 以 standalone 模式启动时，会自动生成登录凭据。用户名与密码会打印在终端输出中，同时也会存储在 $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated。

默认情况下，AIRFLOW_HOME 为 ~/airflow，因此我们可以通过以下命令获取凭据：

cat ~/airflow/simple_auth_manager_passwords.json.generated

最后，使用这些凭据即可登录 Airflow 界面。

浏览界面

登录后，Airflow 会打开首页。此视图可让您快速了解本地 Airflow 环境的概况。

首次登录后的 Airflow UI 视图。

顶部的统计卡（Stats）汇总了 DAG 的当前状态，包括失败、运行中与活跃的工作流。

由于我们尚未创建流水线，这些数值可能仍为零。

健康状况（Health）部分显示主要的 Airflow 服务是否在运行。

下方的 Pool Slots 条显示可用的执行容量。本文不会配置自定义池，但在需要控制同时运行的任务数量时，这一区域会很有用。

最后，历史（History）部分汇总了最近的 DAG 运行。一旦我们触发评价情感流水线，这里就会开始显示排队、运行、成功与失败的运行情况。

项目结构

现在 Airflow 已在运行，让我们为流水线创建所需文件：

mkdir -p ~/airflow/dags
touch ~/airflow/dags/review_sentiment_pipeline.py
touch ~/airflow/requirements.txt

在本地 standalone 安装中，Airflow 默认使用 ~/airflow 作为 AIRFLOW_HOME。我们会将 DAG 文件放入 dags 目录，并保留一个 requirements.txt 文件来记录项目依赖。

因此，项目会具有如下结构：

最简 Airflow 项目结构。

文件 eview_sentiment_pipeline.py 将包含工作流定义，包括获取评价、进行情感分类与存储结果的任务。

Airflow 会监控配置的 dags 目录，并在 Python 文件中查找有效的 DAG 定义。一旦我们的文件包含一个有效的 DAG，该工作流就会出现在 Airflow UI 的Dags 页面上。

如果您在上手过程中遇到一些问题，建议查看教程Getting Started with Apache Airflow。

构建您的第一个 Airflow 流水线

为了开始动手实践 Airflow，我们将构建一个每日作业来处理客户评价并对其情感进行分类。

设想我们刚搭建了一个电商平台。客户已经开始留下评价，我们希望每天自动分析它们。

与其手动运行脚本，不如使用 Airflow 来编排工作流。

步骤 1：获取评价

我们来设想原始评价存储在一个 .jsonl 文件中。

每一行都包含一个 JSON 对象表示的一条评价，新的评价会在白天不断追加到文件末尾。在生产系统中，这些数据可能来自 API、数据库、对象存储或消息队列。在本教程中，JSONL 文件为我们提供了一个简单且贴近真实的原始评价数据源。

在这种场景下，流水线的第一步是将这些评价加载到 Airflow。我们从一个常规 Python 函数开始，就像在笔记本或脚本中那样：

REVIEWS_FILE = Path.home() / "airflow" / "data" / "reviews.jsonl"

def fetch_reviews() -> list[dict]:
   """Load raw customer reviews from the JSONL landing file."""
   reviews = []

   with REVIEWS_FILE.open("r", encoding="utf-8") as file:
       for line_number, line in enumerate(file, start=1):
           if not line.strip():
               continue

           try:
               reviews.append(json.loads(line))
           except json.JSONDecodeError as error:
               raise ValueError(
                   f"Invalid JSON on line {line_number}"
               ) from error

   print(f"Fetched {len(reviews)} reviews")
   return reviews

该函数打开 JSONL 文件，逐行读取，跳过空行，并将每个 JSON 对象转换为 Python 字典。

如果某一行包含无效 JSON，任务会以清晰的错误信息（包含行号）失败。

到目前为止，这只是纯 Python。下一步是将其转换为 Airflow 任务并放入 DAG 中。

将任务封装进 DAG

要将函数纳入 Airflow 工作流，我们需要三个部分：

1. 首先，用 @task 装饰该函数。这样 Airflow 就知道该函数是一个任务。
2. 其次，使用 @dag 定义一个 DAG。DAG 为工作流提供名称、调度、开始日期与其他配置。
3. 第三，在 DAG 函数中调用该任务，并在文件末尾实例化该 DAG。

以下是最小版本：

from airflow.sdk import dag

@dag(    fetch_reviews()
   dag_id="fetch_reviews_pipeline",
)

def fetch_reviews_pipeline():
   fetch_reviews()

fetch_reviews_pipeline()

请注意，这个 DAG 目前还不会自动执行。

按当前写法，我们需要在 Airflow UI 中手动触发它。要配置工作流的运行时机，我们可以在 DAG 定义中加入调度参数。一个可选方案是：

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)

我们来逐项解析！

• dag_id 为 DAG 在 Airflow 中提供唯一名称。

• schedule="@daily" 让 DAG 每天运行一次。
• start_date 告诉 Airflow 何时可以开始调度。
• catchup=False 防止 Airflow 为过去的日期创建运行。
• tags 有助于在 Airflow UI 中组织与筛选该 DAG。

一旦定义了 DAG，我们就可以切换到 UI，看看实际效果：

Airflow UI 中的流水线 review_sentiment_pipeline 视图。注意两种运行类型（计划与手动）以及右上角的手动 Trigger 按钮。

在 UI 中，我们可以看到 Airflow 已自动发现名为 review_sentiment_pipeline 的 DAG。

我们还可以看到它按照计划运行，并且我们通过界面右上角的Trigger 按钮手动触发了一次额外的运行。

一个很棒的功能是日历视图，Airflow 会将流水线运行记录在日历上，并显示每次运行的状态：

我们的 Airflow 流水线的日历视图，显示计划与手动运行。

不妨先暂停阅读，自己探索一下 Airflow UI！

步骤 2：LLM 分类

在加载评价后，下一步就是对它们的情感进行分类。为此，我们将使用 @task.llm，这是一个将 LLM 调用转换为原生 Airflow 任务的装饰器。

我们可以在常规的 @task 中手动调用 OpenAI API，但 @task.llm 能提供更简洁的集成。

它会处理与 LLM 相关的细节，如模型连接、系统提示、结构化输出与结果序列化。同时，该任务仍能享受 Airflow 的标准特性，如重试、日志、监控与依赖管理。

这第二步的组件比前一步多一些，我们循序渐进地构建：

1. 首先，定义我们期望从模型得到的输出结构：

from typing import Literal
from pydantic import BaseModel

class ReviewClassification(BaseModel):
   """Structured result returned by the LLM."""

   review_id: str
   sentiment: Literal["positive", "neutral", "negative"]

这告诉 LLM 任务：每个响应应包含评价 ID 以及三种情感标签之一：positive、neutral 或 negative。

2. 接着，创建一个为单条评价构建提示的函数：

@task.llm(
   llm_conn_id="pydanticai_default",
   system_prompt=(
       "Classify the customer review as positive, neutral, or negative. "
       "Use both the written review and the star rating as context. "
       "Return the review ID and sentiment."
   ),
   output_type=ReviewClassification,
   serialize_output=True,
   retries=3,
)
def classify_review(review: dict) -> str:
   """Build the prompt sent to the LLM."""
   return (
       f"Review ID: {review['review_id']}\n"
       f"Product: {review['product']}\n"
       f"Rating: {review['rating']} out of 5 stars\n"
       f"Review: {review['review_text']}"
   )

函数体返回一个字符串提示，而 @task.llm 装饰器负责使用我们之前配置的 pydanticai_default 连接将该提示发送给模型。

其中 system_prompt 向模型传达任务说明，而 output_type=ReviewClassification 要求返回结构化响应。我们还设置了 serialize_output=True，以便 Airflow 将结果存为普通字典，便于传递给下游任务。

对需要 LLM 推理的任务感兴趣？可以考虑这门更专业的 LangChain 课程！

可视化模型分类结果

分类任务返回结构化结果，但在开发阶段，提前查看模型的输出会很有帮助，而不必等到最终存储步骤。

实际上，一个流水线步骤可以由多个 Airflow 任务组成。

在这里，我们将添加一个小任务，将每条分类结果打印到 Airflow 日志中：

@task
def show_classification(classification: dict) -> dict:
   """Print one model classification to the Airflow task logs."""
   print(
       f"Review {classification['review_id']}: "
       f"{classification['sentiment']}"
   )

   return classification

该任务不会更改结果。它只是记录模型输出并返回相同的字典，以便分类结果仍可传递到流水线的下一步。

 3. 最后，我们需要在 DAG 函数中连接这些新任务：

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)
def review_sentiment_pipeline():
   reviews = fetch_reviews()

   classifications = classify_review.expand(
       review=reviews
   )

   displayed_classifications = show_classification.expand(
       classification=classifications
   )

review_sentiment_pipeline()

注意：在这条流水线中，我们使用了带有 .expand(review=reviews) 的动态任务映射。这告诉 Airflow 为每条评价创建一个 classify_review 任务实例，从而每条评价都以独立请求发送给 LLM。

这通常优于将完整评价列表发送给一个 LLM 任务。像 classify_reviews(reviews) 这样的调用会将所有评价一起处理，可能让故障更难调试、重试更难精确定位，且大输入更容易触发模型上下文或 API 限制。

在完成流水线第二步前，我们可以回到 Airflow UI，看看工作流如何演进。

此时，DAG 包含三个任务：fetch_reviews, classify_review 和 show_classification：

Airflow UI 中我们 review_sentiment_pipeline DAG 的三个任务视图。

故障排查：处理临时 LLM 失败

由于我们使用了动态任务映射，Airflow 会对每条评价发送一次 LLM 请求。这很有用，因为每条评价都能被独立跟踪：

Airflow UI 显示各条评价被独立分类。有些任务已完成或失败，另一些仍在运行。

LLM 请求可能因 API 速率限制、网络问题或模型超时等原因而失败。

我们可以让 Airflow 自动重试受影响的任务，而不是立刻使整条流水线失败。

让我们加上这一特性！

from datetime import timedelta

@task.llm(
   llm_conn_id="pydanticai_default",
   system_prompt=(
       "Classify the customer review as positive, neutral, or negative. "
       "Use both the written review and the star rating as context. "
       "Return the review ID and sentiment."
   ),
   output_type=ReviewClassification,
   serialize_output=True,
   retries=3,
   retry_delay=timedelta(seconds=30),
   retry_exponential_backoff=True,
   max_retry_delay=timedelta(minutes=5),
)

这里，retries=3 允许在首次失败后最多再尝试三次。

第一次重试会等待 30 秒，随后指数退避会增加后续尝试之间的等待时间，最长不超过 5 分钟。

更好的一点在于 Airflow 只会重试失败的映射任务。

如果有 49 条评价成功分类，只有 1 条失败，那么仅会为失败的那条安排重试。流水线的其余部分无需重复已经成功的工作。

一段时间后，失败的评价分类任务将再次等待重试：

Airflow UI 显示所有失败的评价分类任务正在等待重试。

一旦评价分类成功，每个映射的 show_classification() 任务实例都会在其日志中显示对应评价的模型响应。

这意味着显示步骤也是独立的：无论其他评价的状态如何，您都可以打开、检查或调试其中一条评价的输出。

Airflow UI 显示对评价 47 的 LLM 情感分类结果。

步骤 3：存储结果

在开发 DAG 时，打印每条分类结果很有用，因为我们可以在 Airflow 日志中快速检查模型响应。

不过，日志并不是存储流水线结果的可靠场所。日志更难查询、复用，也难以与下游系统集成。

因此，最后一步就是汇总分类结果并写入结构化输出。

在本文中，我们将移除对 show_classification() 的调用，并将结果存储到本地 JSON 文件中。

在生产流水线中，这个任务通常会写入数据库、数据仓库或对象存储。

这个新 @task 的一种可能实现如下：

OUTPUT_FILE = (
   Path.home()
   / "airflow"
   / "output"
   / "classified_reviews.json"
)

@task
def store_results(results: list[dict]) -> str:
   """Write all classified reviews to a local JSON file."""
   OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)

   classifications = list(results)

   with OUTPUT_FILE.open("w", encoding="utf-8") as file:
       json.dump(
           classifications,
           file,
           indent=2,
           ensure_ascii=False,
       )

   print(
       f"Stored {len(classifications)} classifications "
       f"in {OUTPUT_FILE}"
   )

   return str(OUTPUT_FILE)

该任务是我们的加载步骤。

它接收由 LLM 生成的分类结果，必要时创建输出文件夹，并将所有内容写入本地 JSON 文件。

小小的 list(results) 转换很有用，因为分类结果来自映射任务。

Airflow 会帮我们收集这些映射输出，该任务则在保存前将其转为常规列表。现在我们只需在 DAG 中调用 store_results(classifications) 来连接该任务。

有了这行代码，Airflow 就知道 store_results() 依赖分类步骤。换言之，只有当所有评价的分类都成功结束后，输出文件才会被写入。

运行流水线

三步全部连通后，完整的 DAG 会先获取评价，为每条评价创建一个 LLM 任务，并存储合并后的结果。

在 DAG 函数内，流水线如下所示：

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)
def review_sentiment_pipeline():
   reviews = fetch_reviews()
   classifications = classify_review.expand(
       review=reviews
   )
   store_results(classifications)

review_sentiment_pipeline()

保存文件后，Airflow 需要几秒钟来解析并发现完整的 DAG。一旦 review_sentiment_pipeline 出现在 DAG 列表中，使用开关启用它，并点击 Trigger 以启动一次手动运行。

这样我们就能立即测试整条流水线，无需等待下一次计划的每日运行。一旦运行成功，最终的 JSON 文件会在本地的 ~/airflow/output/classified_reviews.json 可用。

该输出 JSON 文件是工作流的持久化产物，可以在 DAG 外复用，或稍后传递给下游系统。

相比之下，Airflow UI 为我们提供流水线的运维视图：哪些任务运行了、哪些成功或失败，以及到哪里查看日志。

最终的流水线如下所示：

Airflow UI 中我们流水线的三个任务视图。右下面板显示每个任务的属性，包括名称、触发规则、算子或装饰器，以及重试次数。

最后，在接下来的几天中，Airflow UI 会逐渐填充自动的每日运行，为我们提供清晰的运行历史与每次执行是否成功：

Airflow UI 中流水线运行及其状态的视图。完成步骤 3 后，流水线不再使用 show_classification，而是改为运行 store_results。

更进一步：可继续探索的 Airflow 3 AI 特性

我们的流水线现在能够按日程获取评价、用 LLM 进行分类，并存储结果。

Airflow 3 还支持更高级的模式，可让 AI 工作流更具响应性、更易于监督。

事件驱动的调度

与其每天在固定时间运行流水线，Airflow 可以在外部事件发生时启动它。

例如，当新的评价文件上传到对象存储，或队列中有消息到达时，就可运行评价流水线。

Airflow 3 通过资产（assets）和兼容事件的触发器支持事件驱动的调度。这免去了反复轮询新数据的需求，并让流水线在检测到事件后立即响应。

人参与环（HITLOperator）

并非每个模型决策都应被自动接受。

有些评价可能含糊、敏感，或需要人工判断。Airflow 的人参与环算子可以暂停正在运行的工作流，并通过 Airflow 界面请求输入。

审核者可以批准或拒绝结果、提供信息，或选择应继续的工作流分支。

Common AI provider 还支持代理型工作流的交互式审核模式，审核者可以对 LLM 生成的结果进行批准、拒绝或要求修改。

事件驱动调度与人工审核结合在一起，使我们能够构建既能快速响应、又不丢失关键环节人工把关的 AI 流水线。

要进一步了解Airflow 3.0 的特性，建议查看我们的独立指南。

结语

在本文中，我们使用 Apache Airflow 构建了一条完整的 AI 流水线。

该流水线会获取客户评价，使用 LLM 对每条评价进行分类，并将最终结果存储到结构化 JSON 文件中。

在此过程中，我们看到 Airflow 如何将 Python 函数转变为可调度、可观测的任务。这对 AI 工作流尤其有用，因为 API 限制、模型超时与意外输出都可能发生。

借助任务依赖、动态任务映射、重试、日志与 Airflow UI，我们可以让 AI 流水线更易监控、调试与恢复。

本文示例使用本地文件以保持简洁，但相同结构可以扩展到 API、数据库、事件驱动工作流以及人参与环审核。

如果您准备继续探索，建议先学习我们的Introduction to Apache Airflow in Python 课程，然后继续学习 Building Data Pipelines with Airflow，进行更多实操练习。

若希望获得更广泛的学习路径， Data Engineer in Python 职业路径涵盖构建与管理生产级数据流水线所需的更广泛技能。