用途:把 Airflow 项目从 airflow-ai-sdk 迁移到官方 AI provider apache-airflow-providers-common-ai(目标 0.4.0+,基于 PydanticAI),也适用于从 common-ai 0.1.x 升级(多模态提示、toolsets、embedding operator、结构化输出 XCom 行为有变化)。
前提与限制:新 provider 要求 Airflow 3.0+,0.4.0 需 pydantic-ai-slim>=1.71.0。LLM 配置从代码里的模型字符串/对象改为 Airflow connection(conn_type=pydanticai,默认 ID pydanticai_default)。新 provider 没有 @task.embed,需改用 LlamaIndex 集成或普通 @task。
主要步骤:
- 用 grep 盘点 airflow_ai_sdk、@task.llm/agent/llm_branch/embed 及模型配置。
- 改 requirements.txt:移除 airflow-ai-sdk[...],加入 apache-airflow-providers-common-ai[openai]>=0.4.0;按需加 [llamaindex] 等 extras。
- 建 pydanticai 连接,模型用 provider:model,如 openai:gpt-5;自定义端点用 host。
- 迁移装饰器:model= 改为 llm_conn_id=;@task.agent 不再手建 Agent,改由 system_prompt、toolsets、agent_params 等参数配置;支持 model_id、require_approval、serialize_output 等。
- 替换 @task.embed:A 用 LlamaIndexEmbeddingOperator,B 用 LlamaIndexHook,C 用 sentence-transformers 的普通 @task。
- 更新导入:BaseModel 改自 pydantic;删除 airflow_ai_sdk 与多余 pydantic_ai.models 导入。
- 更新 connections.yaml 与清理冗余环境变量,保留 AIRFLOW_CONN_*。
- 校验残留引用并端到端运行 DAG;HITL 任务需人工或在 /hitl-review 接口处理。
注意:切换嵌入模型会改变向量维度,需重算存量向量并调整表结构;结构化输出在旧 core 为 dict、新 core 为实例,可用 serialize_output=True 固定;output_type 类必须在模块级定义。