Apache Airflow Common AI Provider：基于 pydantic-ai 的 LLM Hooks、Operators 安装与依赖全景解析

Apache Airflow Common AI Provider基于 pydantic-ai 的 LLM Hooks、Operators 安装与依赖全景解析【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本文基于 Apache Airflow 仓库中common.aiprovider 的官方包说明 README.rst 展开完整解析apache-airflow-providers-common-ai0.9.0 的安装方式、硬性依赖与全部可选依赖extras体系并结合仓库源码说明其 Hook 的模型解析机制、Connection 配置方式与快速上手路径帮助你在 Airflow 3 环境中正确安装、配置并验证这一 AI/LLM 能力包。包定位面向 Airflow 流水线的 AI/LLM 扩展apache-airflow-providers-common-ai是 Airflow 的 provider 包官方描述为「AI/LLM hooks and operators for Airflow pipelines using pydantic-ai」——即基于 pydantic-ai 框架该外链见原文档此处仅作概念说明为 Airflow 工作流提供 LLM 接入的 Hook 与 Operator。所有 Python 类都位于airflow.providers.common.ai包内当前发布版本为0.9.0。从源码结构看该 provider 的能力面由 provider.yaml 完整声明涵盖9 个 Operator 模块operators/ 目录agent、llm、llm_file_analysis、llm_branch、llm_sql、llm_schema_compare、document_loader、llamaindex_embedding、llamaindex_retrieval7 个task装饰器decorators/ 目录与 Operator 一一对应agent、llm、llm_file_analysis、llm_branch、llm_sql、llm_schema_compare9 个 toolset 模块toolsets/ 目录hook、sql、datafusion、logging、mcp、sandbox、skills、langchain_bridge、managed_agent用于给 Agent 提供「基于 Airflow Hook、SQL 数据库或 MCP Server 构建的工具」4 个 HookPydantic AI、MCP Server、LangChain、LlamaIndexhooks/ 目录一个HITL人工审批UI 插件hitl_review。provider.yaml 还声明了其声明式集成integration覆盖Common AI、Pydantic AI、MCP Server、LangChain、LlamaIndex、Docker Sandboxes 等标签统一为ai。安装与运行环境要求在已安装的 Airflow 之上叠加安装该包即可pip install apache-airflow-providers-common-ai该包支持的 Python 版本为3.10、3.11、3.12、3.13、3.14。硬性依赖RequirementsREADME 的 Requirements 表与 pyproject.toml 第 69–75 行的dependencies完全一致PIP 包版本要求apache-airflow3.0.0apache-airflow-providers-common-compat1.15.0apache-airflow-providers-standard1.12.1pydantic-ai-slim2.0.0其中pydantic-ai-slim2.0.0在 pyproject.toml 中有注释说明该包依赖pydantic-ai 2.x 的 agent/instrumentation API对应上游 issue #69122。版本下限不只是元数据层面的声明而是在包初始化时做了运行时硬校验。init.py 中if packaging.version.parse(packaging.version.parse(airflow_version).base_version) packaging.version.parse( 3.0.0 ): raise RuntimeError( fThe package apache-airflow-providers-common-ai:{__version__} needs Apache Airflow 3.0.0 )也就是说若运行环境是 Airflow 2.x导入该包会直接抛出RuntimeError。适用前提必须使用 Airflow 3.0 及以上。可选依赖extras体系全解README 将可选依赖分为两类「跨 provider 包依赖Optional cross provider package dependencies」和「可选依赖Optional dependencies」。下面完整继承原文档的两张表并结合 pyproject.toml 中的注释补充各 extra 的实际作用。跨 provider 包依赖这类依赖用于启用涉及其他 provider 的全部功能安装时通过 extras 一并装齐例如pip install apache-airflow-providers-common-ai[common.sql]依赖包Extraapache-airflow-providers-common-sqlcommon.sqlapache-airflow-providers-gitgit完整可选依赖表以下即 README「Optional dependencies」表的完整内容与 pyproject.toml 的[project.optional-dependencies]逐项对应Extra依赖anthropicpydantic-ai-slim[anthropic]2.0.0bedrockpydantic-ai-slim[bedrock]2.0.0googlepydantic-ai-slim[google]2.0.0openaipydantic-ai-slim[openai]2.0.0mcppydantic-ai-slim[mcp]2.0.0code-modepydantic-ai-harness[codemode]0.3.0shieldspydantic-ai-shields0.3.4skillsapache-airflow-providers-git0.4.0、pydantic-ai-skills1.2.0avrofastavro1.10.0Python 3.14/fastavro1.12.1Python 3.14parquetpyarrow18.0.0Python 3.14/pyarrow22.0.0Python 3.14sqlapache-airflow-providers-common-sql1.33.0、sqlglot30.0.0common.sqlapache-airflow-providers-common-sql1.33.0langchainlangchain1.0.0llamaindexdataclasses-json0.6.7、llama-index-core0.13.0、llama-index-embeddings-openai0.6.0、llama-index-llms-openai0.6.0pdfpypdf4.0.0docxpython-docx1.0.0gitapache-airflow-providers-gitpyproject.toml 中的注释揭示了几个 extra 的设计意图值得留意code-mode将工具调用折叠为单个run_code工具由模型通过编写 Python 来驱动代码在 Monty 沙箱pydantic-monty中执行启用AgentOperator(code_modeTrue)。注释明确指出 Monty 处于 pre-1.0 阶段因此作为 opt-in extra 单独钉住版本「其变动不会破坏基础 provider 安装」。shields提供 InputGuard、OutputGuard、ToolGuard、CostTracking 等防护/治理能力。skillsAgent Skillsagentskills.io支持——pydantic-ai-skills提供 toolset而apache-airflow-providers-git提供 GitHook GitPython用于凭gitconnection 的凭据克隆 Git 仓库中的技能定义。avro/parquet对 fastavro、pyarrow 采用按 Python 版本分叉的最低版本要求3.14 起要求更高版本这与 README 中声明 Python 3.14 支持是一致的。安装命令应把 extra 拼在包名后例如pip install apache-airflow-providers-common-ai[openai]。官方 quickstart.rst 给出的推荐安装方式是先确定要用的模型 SDK再带上对应 extrapip install apache-airflow-providers-common-ai[extra]快速上手Connection 配置与第一个 LLM 任务理解依赖关系后最短可用路径是配置连接并跑一个task.llm任务。所有 LLM 调用都经由 Pydantic AI 类型的 Connectionconn_type为pydanticai默认连接 id 为pydanticai_default。模型以provider:model格式填写在 extra 的model字段API Key 放在 password 字段。最快的配置方式是通过环境变量把模型和 key 换成你实际可用的值export AIRFLOW_CONN_PYDANTICAI_DEFAULT{conn_type: pydanticai, password: sk-..., extra: {model: openai:gpt-5.6-sol}}也可以经由 Airflow UIAdmin Connections或 CLIairflow connections add创建。各字段的具体含义与各类 provider 的完整 JSON 示例OpenAI、Anthropic、本地 Ollama、AWS Bedrock、Google Vertex AI / Gemini API见 connections/pydantic_ai.rst。第一个 Dag 使用task.llm装饰器——它把一个「返回 prompt 字符串」的函数变成「把该 prompt 发给 LLM 并返回其响应」的任务。这正是仓库示例 example_quickstart.py 中的写法from airflow.sdk import dag, task dag(scheduleNone, tags[example]) def quickstart_llm(): task.llm(llm_conn_idpydanticai_default, system_promptYou are a helpful assistant. Be concise.) def summarize(text: str): return fSummarize this article: {text} summarize( Apache Airflow is a platform for programmatically authoring, scheduling, and monitoring workflows. ) quickstart_llm()以airflow dags test quickstart_llm运行即可summarize任务会把 LLM 的响应推送到 XCom。需要类型化数据而非字符串时为任务设置 PydanticBaseModel类型的output_type模型实例会被原样推入 XCom。源码纵深PydanticAIHook 的模型解析与校验机制PydanticAIHook 是整个 provider 的枢纽。其类属性第 63–66 行声明了连接契约conn_name_attr llm_conn_id default_conn_name pydanticai_default conn_type pydanticai hook_name Pydantic AI该 Hook 覆盖所有使用「标准api_key 可选base_url」模式的 providerOpenAI、Anthropic、Groq、Mistral、DeepSeek、Ollama、vLLM 等对于认证方式非标准的云厂商则由PydanticAIAzureHook、PydanticAIBedrockHook、PydanticAIVertexHook三个子类处理——对应 provider.yaml 中声明的pydanticai_azure、pydanticai_bedrock、pydanticai_vertex三种专用连接类型带 region、IAM 密钥、project/location、API version 等专属 UI 字段。模型解析顺序get_conn()pydantic_ai.py 第 130–185 行显式凭据路径_get_provider_kwargs把连接的password/host/extra映射为 provider 构造参数非空时通过infer_provider_class(pname)(**kwargs)实例化 provider 并包装为provider_factory传给 pydantic-ai 的infer_model。若 provider 构造函数拒绝这些 kwargsTypeError会降级为环境变量认证并打 warning。默认解析路径没有任何显式凭据时直接交给 pydantic-ai 的infer_model由其读取标准环境变量OPENAI_API_KEY、AWS_PROFILE等。模型名的取值优先级为model_id构造参数 连接 extra 中的model字段两者都缺省则抛出ValueErrorNo model specified...。解析结果在 Hook 实例生命周期内缓存。test_connection的克制设计同样值得注意第 289–302 行它只解析模型、验证「模型字符串合法且 provider 可用给定凭据实例化」刻意不发起真实的 LLM API 调用——文档注释明确解释这样做是因为真实调用「昂贵且会因配额、计费、限流等与连通性无关的原因失败」。create_agent()第 228–287 行则展示了 Hook 与可观测性的集成点当配置项[common.ai] otel_export_enabled开启且 worker 配好了 OpenTelemetry exporter 时创建的 Agent 会自动注入 instrumentation通过 Airflow 的 tracing 管线输出 GenAI spansagent 运行、模型调用、工具调用、token 用量。调用方若显式传入instrument参数仍然优先于 provider 的自动注入。声明式注册与 [common.ai] 配置项pyproject.toml 声明了两个关键的 entry point第 184–188 行它们解释了 Airflow 如何「发现」这个 provider[project.entry-points.apache_airflow_provider] provider_info airflow.providers.common.ai.get_provider_info:get_provider_info [project.entry-points.airflow.plugins] hitl_review airflow.providers.common.ai.plugins.hitl_review:HITLReviewPlugin前者让 Airflow 读取 provider 元信息即 provider.yaml 中的 operators、hooks、connection-types 等清单后者把 HITL 审批插件注册为 Airflow UI 插件。provider.yaml 还声明了[common.ai]配置段下的 3 个配置项是使用该包时最容易被忽略的调优入口配置项类型/默认值作用durable_cache_pathstring默认在 Airflow 3.3上以durableTrue运行AgentOperator/task.agent时持久化每步缓存的 ObjectStorage URIfile://、s3://、gs://等。每次任务执行写一个 JSON 文件缓存的模型响应与工具结果重试时回放已完成步骤而非重复发起 LLM 调用任务成功完成后删除文件。Airflow 3.3 时缓存存入 AIP-103 task state store此项被忽略otel_export_enabledboolean默认False为本 provider 创建的 Agent 附加 pydantic-ai OpenTelemetry instrumentation输出 GenAI spans。spans 经由 Airflow 既有 OTel exporter 发出[traces]/OTEL_EXPORTER_OTLP_*并嵌套在 task span 之下若 worker 未开启核心 tracing[traces] otel_on则不发出任何 span。默认关闭保证安装本包不会「悄悄」开始上报 spanscapture_contentboolean默认False在输出的 GenAI spans 上捕获 prompt/completion/工具调用内容gen_ai.input.messages/gen_ai.output.messages。默认关闭时只记录 token 数、模型 id、延迟、工具名与结束原因绝不记录消息文本。开启会把模型输入输出不做脱敏地导出到 tracing 后端——Airflow 的 secret 脱敏作用于日志和模板渲染字段不作用于 span 属性。文档明确建议「仅在可信环境中为调试开启」且otel_export_enabled为True时本项才有效Operator / 装饰器选型参考README 指向的文档体系以 docs/operators/index.rst 的选型表为核心。该 provider 的 Operator 与装饰器选型对照继承自原文档的 how-to 索引需求Operator装饰器单轮 prompt → 文本或结构化输出LLMOperatortask.llm用一次 prompt 分析文件、前缀、图片或 PDFLLMFileAnalysisOperatortask.llm_file_analysis由 LLM 决定下游执行哪个任务LLMBranchOperatortask.llm_branch自然语言 → SQL 生成不执行LLMSQLQueryOperatortask.llm_sql跨数据源比较 schema、检测漂移LLMSchemaCompareOperatortask.llm_schema_compare多轮推理 工具调用查库、调 API 等AgentOperatortask.agent把文件PDF、DOCX、CSV 等解析为供嵌入的文档 dictDocumentLoaderOperator无装饰器文档分块并生成嵌入向量LlamaIndexEmbeddingOperator无装饰器从向量索引检索相关分块LlamaIndexRetrievalOperator无装饰器其中LLMOperator/task.llm是无状态单轮调用适合分类、摘要、抽取AgentOperator/task.agent是多轮工具调用循环工具由toolsets配置注入。文档特别提示Agent 即使不配 toolsets 也能运行但如果没有工具需求LLMOperator更简单、意图更明确。小结与延伸阅读回到 README 本身它作为自动生成文件模板位于dev/breeze/src/airflow_breeze/templates的PROVIDER_README_TEMPLATE.rst.jinja2承担的是「包级事实清单」职责包名、发布版本、Python 版本范围、硬性依赖、跨 provider 依赖与全部 optional extras。本文在其基础上补充了三条可验证的源码线索pyproject.toml中逐字对应的 extras 定义与设计注释、__init__.py中对 Airflow 3.0 的运行时硬校验、以及PydanticAIHook的模型解析/校验实现。进一步深入可按以下仓库路径继续快速上手与连接配置docs/quickstart.rst、docs/connections/pydantic_ai.rst以及 bedrock、azure、vertex、mcp、langchain、llamaindex 各连接文档Operator 详解docs/operators/ 下各篇agent、llm、llm_sql、llm_branch、llm_schema_compare、llm_file_analysis、document_loader、llamaindex_*工具、可观测性与安全docs/toolsets.rst、docs/observability.rst、docs/security.rst版本变更记录docs/changelog.rst【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考