从数周到数分钟：用 Data Agent Kit 构建智能数据管道与 MLOps 闭环

2026-09-01 50 预计阅读时间: 1 分钟
来源: cloud.google.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：12 分钟

数据管道一直是企业数据平台的骨架，但 Airflow DAG、计算资源、模型训练、批量推理和生产部署之间存在较高的工程门槛。Google Cloud 推出的 Orchestration Pipelines 框架与开源 Data Agent Kit，试图把这段复杂流程直接带进开发者熟悉的 IDE 或 CLI，让数据分析师、数据工程师和 ML 工程师能够用自然语言描述目标，再通过声明式 YAML 管理生产级流程。

这并不意味着可以完全跳过代码审查和生产治理。更准确的理解是：把重复的编排样板交给工具生成，把人的精力集中到数据契约、模型指标、权限和运行边界上。

Data Agent Kit 改变了什么

Data Agent Kit 可以集成到 VS Code、Claude Code、Codex 等开发环境中，并通过两种方式接入 Orchestration Pipelines：

  • 提供专门的数据工程面板，用于创建、查看、部署和监控管道。
  • 提供 gcp-pipelines-orchestration agent skill，让代理理解管道语法、变量替换、密钥管理以及 Airflow 运行故障诊断。

核心变化在于，编排逻辑与计算执行被拆开了。开发者可以在 YAML 中声明 SQL、PySpark、dbt、AI 推理和跨管道触发关系，而具体计算由 BigQuery、Managed Service for Apache Spark、Gemini Enterprise Agent Platform 或 Airflow 执行。

这种方式尤其适合 MLOps 场景：训练通常需要重型计算，日常推理更偏向轻量批处理，模型质量评估则需要根据数据漂移决定是否重新训练。三者可以分别维护，又能组成一个完整闭环。

一个供应链预测闭环

可以用 bigquery-public-data.thelook_ecommerce 公开数据集构造示例。目标是根据仓库位置、客户位置和订单属性，预测订单运输天数，并在预计违反 SLA 时提前通知客户或升级配送服务。

一个实用的拆分方式是：

  1. 训练管道：从 BigQuery 提取历史已完成订单，在无服务器 Spark 中计算地理距离并训练模型，然后将模型上传到模型注册表。
  2. 推理管道：每天查询仍在运输中的订单，调用模型进行批量预测，并将结果写回 BigQuery。
  3. 评估管道：通过 dbt 将预测结果与真实送达时间关联，计算误差和 SLA 违约情况；当误差超过阈值时，触发训练管道。

训练管道可以这样表达。下面的配置是可改造的示例，运行前需要替换项目、区域、Bucket 和模型路径，并确认对应服务账号拥有 BigQuery、Spark 和模型平台权限。

modelVersion: "1.0"
pipelineId: "training-pipeline"
runner: airflow
owner: "mlops"
defaults:
  projectId: "your-project-id"
  location: "us-central1"
  executionConfig:
    retries: 1

actions:
  - sql:
      name: "extract_training_data"
      engine:
        bigquery:
          location: "US"
          destinationTable: "your-project-id.mlops.training_dataset"
          query:
            path: "blogpostdemo/training_query.sql"

  - pyspark:
      name: "train_model_dataproc"
      dependsOn:
        - "extract_training_data"
      engine:
        dataprocServerless:
          location: "us-central1"
          resourceProfile:
            inline:
              runtimeConfig:
                version: "2.3"
              properties:
                "spark.dataproc.driverEnv.PYTHONPATH": "./libs/lib/python3.11/site-packages"
                "spark.executorEnv.PYTHONPATH": "./libs/lib/python3.11/site-packages"
          mainFilePath: "blogpostdemo/train_model.py"
          environment:
            requirements:
              inline:
                list:
                  - "tensorflow==2.14.1"
                  - "numpy<2.0.0"
                  - "protobuf<5.0.0dev"
                  - "google-cloud-storage"

  - ai:
      name: "upload_model"
      dependsOn:
        - "train_model_dataproc"
      agentPlatform:
        projectId: "your-project-id"
        location: "us-central1"
        modelUpload:
          modelName: "transit_days_predictor"
          modelArtifactUri: "gs://your-bucket-name/models/tf_transit_days_model"
          servingContainerImageUri: "us-docker.pkg.dev/vertex-ai/prediction/tf2-cpu.2-14:latest"

这段配置的关键不是 YAML 本身，而是依赖关系。只有训练数据提取成功，Spark 训练才会开始；只有训练完成，模型上传动作才会执行。对于生产环境，建议把项目 ID、Bucket、数据集名称和模型版本放入环境变量或密钥管理系统，不要把真实凭据写入仓库。

从自然语言到可部署管道

在支持 Data Agent Kit 的 IDE 中，可以用自然语言描述目标，例如：

Create three production-ready orchestration pipelines for transit-time prediction.
Use BigQuery for historical and in-transit order extraction, serverless Spark for
model training, dbt for evaluation, and batch inference through the model platform.
When the daily error metric exceeds 20%, trigger training-pipeline asynchronously.
Add dependencies, retry policies, parameterized project and location values, and
create validation and CI/CD configuration for deployment to Managed Airflow.

代理可能一次生成 YAML、PySpark、SQL、dbt 配置和 CI 工作流，也可能遗漏某个参数。响应会受到模型版本、工作区上下文和提示长度影响，因此应把生成过程当作协作式开发：检查文件，运行验证，然后用短提示补齐缺失项。例如：

The inference action is missing its BigQuery destination table and the evaluation
pipeline does not pass the drift threshold to the condition check. Update the
configuration and add a validation step for these two parameters.

日常推理的核心结构可以简化为：

pipelineId: "inference-pipeline"
runner: airflow
actions:
  - sql:
      name: "extract_inference_data"
      engine:
        bigquery:
          location: "US"
          destinationTable: "your-project-id.mlops.inference_dataset"
          query:
            path: "blogpostdemo/inference_query.sql"

  - ai:
      name: "run_batch_prediction"
      dependsOn:
        - "extract_inference_data"
      agentPlatform:
        projectId: "your-project-id"
        location: "us-central1"
        batchInference:
          jobDisplayName: "daily_transit_prediction"
          modelName: "projects/your-project-id/locations/us-central1/models/your-model-id"
          bigquerySource: "bq://your-project-id.mlops.inference_dataset"
          bigqueryDestinationPrefix: "bq://your-project-id.mlops"

评估管道则把 dbt 和条件分支连接起来：

pipelineId: "evaluation-pipeline"
runner: airflow
actions:
  - pipeline:
      name: "run_dbt_models"
      framework:
        dbt:
          airflowWorker:
            projectDirectoryPath: "blogpostdemo/dbt_project"

  - python:
      name: "check_retraining_condition"
      dependsOn:
        - "run_dbt_models"
      mainFilePath: "blogpostdemo/evaluate_drift.py"
      pythonCallable: "check_drift"
      engine:
        local: {}

  - orchestrationPipeline:
      name: "trigger_retraining_pipeline"
      dependsOn:
        - "check_retraining_condition"
      pipelineId: "training-pipeline"
      bundleId: "mlops-bundle"
      waitForCompletion: false

示例中的 evaluate_drift.py 可以承担阈值判断。具体输入输出契约需要按照实际框架和团队规范调整，下面只展示一种清晰的决策逻辑：

from dataclasses import dataclass


@dataclass
class EvaluationResult:
    mean_absolute_error: float
    sla_breach_rate: float


def check_drift(result: EvaluationResult, max_mae: float = 2.0) -> bool:
    """Return True when the model should be retrained."""
    return result.mean_absolute_error > max_mae


if __name__ == "__main__":
    current = EvaluationResult(mean_absolute_error=2.4, sla_breach_rate=0.08)
    print({"retrain": check_drift(current)})

部署和故障处理也进入同一工作流

编排逻辑写完只是开始。Data Agent Kit 可以生成适用于工作区的 CI/CD 配置，将管道打包后部署到 Managed Airflow。团队可以把 YAML、SQL、PySpark、dbt 和测试一起放进代码仓库，通过代码评审与 GitHub Actions 等标准流程发布。

一个最小化的验证步骤可以这样运行：

# 认证并选择目标项目，命令名称按本地 Google Cloud 环境调整
gcloud auth application-default login
gcloud config set project your-project-id

# 在项目根目录执行静态检查和测试
python -m compileall blogpostdemo
pytest -q

# 提交变更后由 CI 工作流打包并部署管道 bundle

生产部署前至少应验证以下内容：

  • BigQuery 源表和目标表是否存在，区域是否匹配。
  • Spark 运行时、Python 依赖和模型服务容器版本是否兼容。
  • 服务账号是否遵循最小权限原则。
  • 重试是否会造成重复写入、重复推理或重复触发训练。
  • 模型版本、数据快照和评估指标是否可以追溯。

运行阶段的价值在于减少上下文切换。开发者可以直接在 IDE 中查看 Managed Airflow 运行状态和失败上下文。当 Spark 因数据量季节性增长发生内存不足，或 BigQuery 触发配额限制时，Data Agent Kit 的 agentic troubleshooting 能够根据日志和运行信息归纳根因，并给出调整计算模板或修复代码的建议。

这类建议仍然需要人工确认。配额问题、权限问题、数据质量问题和代码缺陷的处理方式不同，不能把代理给出的修复直接视为已验证的变更。

落地时的取舍

Orchestration Pipelines 的主要收益是缩短从需求到可运行管道的时间，并降低编写 Airflow Operator 样板代码的门槛。但它并没有消除系统复杂度，复杂度只是从 DAG Python 代码转移到了数据契约、资源配置、依赖版本、权限和运维策略上。

建议采用渐进式路径：

  1. 先用公开或脱敏数据集验证 SQL、训练和推理链路。
  2. 为每个动作定义输入、输出、幂等性和失败重试语义。
  3. 将生成的 YAML 和脚本纳入代码评审、静态检查与自动化测试。
  4. 再接入真实数据、模型注册表和生产 Airflow 环境。
  5. 用误差、漂移、SLA 违约率和资源成本共同决定是否自动重训。

当这些边界被明确后，代理生成的管道才真正具备可维护性。最理想的结果不是“用一句话替代所有工程工作”，而是让数据团队用更少的样板代码建立更快、更透明、可审计的 MLOps 闭环。

# AI # Docker # GCP # Go # LLM # Python # TensorFlow # 云原生 # 云平台 # 全栈 # 可观测性 # 后端 # 安全 # 工程效能 # 数据库 # 架构设计

相关推荐