14.Python API 自动化

在命令行之外，DVC 还提供了一套完整的 Python API。这套 API 让我们能在代码中直接操作 DVC 项目，读取数据、查询实验、获取参数和指标，甚至访问模型制品。对于需要构建自动化流程、开发内部工具，或者想在 Jupyter Notebook 里直接分析实验结果的场景，Python API 显得特别实用。

DVCFileSystem 文件操作

DVCFileSystem 是 DVC Python API 中最基础也最强大的组件之一。它提供了一个符合 fsspec 标准的文件系统接口，让我们能用 Pythonic 的方式访问 DVC 仓库中的文件，无论是 Git 追踪的代码，还是 DVC 追踪的数据和模型。

基本用法

DVCFileSystem 的核心是一个只读文件系统，它不支持写入操作，但提供了完整的读取和遍历能力。我们可以用它来访问本地仓库，也可以直接操作远程 Git 仓库，而无需事先克隆到本地。

from dvc.api import DVCFileSystem

# 访问本地仓库
fs = DVCFileSystem("/path/to/local/repository")

# 访问远程仓库
repo = "https://github.com/iterative/example-get-started.git"
fs = DVCFileSystem(repo, rev="main")

初始化 DVCFileSystem 时，主要参数有两个：repo 指定仓库位置，可以是 URL 或本地路径；rev 指定 Git 版本，可以是分支名、标签或提交哈希。如果不指定 repo，默认使用当前工作目录所在的 DVC 项目。

读取文件内容

DVCFileSystem 提供了多种读取文件的方式，最常用的是 open() 方法，它返回一个文件对象，可以像普通文件一样读取。

import pickle

# 打开模型文件
with fs.open("model.pkl") as f:
    model = pickle.load(f)

需要注意的是，与 Python 内置的 open() 不同，DVCFileSystem 的 open() 默认使用二进制模式（"rb"）。如果要读取文本文件，需要显式指定模式和编码。

# 读取文本文件
text = fs.read_text("get-started/data.xml", encoding="utf-8")

# 读取二进制内容
contents = fs.read_bytes("get-started/data.xml")

read_text() 和 read_bytes() 是便捷方法，分别用于读取文本和二进制内容，它们内部调用了 open() 并处理了编码问题。

列出仓库文件

DVCFileSystem 的 find() 方法可以递归列出仓库中的所有文件，这在探索远程仓库结构时非常有用。

# 列出所有 DVC 追踪的文件
dvc_files = fs.find("/", detail=False, dvc_only=True)
print(dvc_files)

输出结果类似这样：

[
    '/data/data.xml',
    '/data/features/test.pkl',
    '/data/features/train.pkl',
    '/model.pkl'
]

dvc_only=True 参数确保只返回 DVC 追踪的文件，不包括 Git 追踪的代码文件。如果想查看所有文件，包括 Git 追踪的，可以去掉这个参数。

# 列出所有文件（包括 Git 追踪的）
all_files = fs.find("/", detail=False)

find() 方法返回的是相对于仓库根路径的列表，根路径用 "/" 表示。我们也可以指定子目录来缩小搜索范围。

# 只列出 data 目录下的文件
data_files = fs.find("/data", detail=False, dvc_only=True)

对于非递归列出，可以使用 ls() 方法，它类似于命令行的 ls 命令。

下载文件或目录

除了读取文件内容，DVCFileSystem 还支持将文件或目录下载到本地。

import os

# 下载单个文件
fs.get_file("data/data.xml", "local_data.xml")

# 下载整个目录
fs.get("data", "local_data", recursive=True)

get_file() 用于下载单个文件，第一个参数是远程路径，第二个是本地路径。get() 方法配合 recursive=True 可以下载整个目录树。DVC 会优先从本地缓存读取，如果缓存中没有，则从远程存储流式传输。

处理子仓库

在 monorepo 结构中，DVC 可能在子目录中初始化。这时需要设置 subrepos=True 参数才能访问这些子项目。

repo = "https://github.com/iterative/monorepo-example.git"

# 默认情况下，子目录中的 DVC 项目会被忽略
fs = DVCFileSystem(repo, rev="develop")
print(fs.find("nlp", detail=False, dvc_only=True))  # 返回空列表

# 使用 subrepos=True 来访问子项目
fs = DVCFileSystem(repo, subrepos=True, rev="develop")
print(fs.find("nlp", detail=False, dvc_only=True))
# 返回：['nlp/data/data.xml', 'nlp/model.pkl', ...]

与 fsspec 生态集成

DVCFileSystem 基于 fsspec 构建，这意味着它能无缝集成到 Python 数据生态中。当安装了 DVC 和其他 fsspec 兼容库（如 Hugging Face Datasets）时，可以直接使用 dvc:// URL 方案。

import fsspec

# 使用 dvc:// URL 访问 DVC 追踪的文件
f = fsspec.open(
    "dvc://workshop/satellite-data/jan_train.csv",
    repo="https://github.com/iterative/dataset-registry.git",
)

需要注意的是，dvc:// URL 只包含文件路径，不包含 Git 仓库地址。仓库地址需要通过 repo 参数单独提供。对于需要 storage_options 的方法，可以将 DVCFileSystem 的参数传递进去。

实验数据查询

在机器学习项目中，我们经常需要程序化地获取实验数据进行分析。dvc.api.exp_show() 提供了这个功能，它返回一个实验列表，每个实验包含参数、指标和其他元数据。

基本用法

最简单的调用方式不需要任何参数，它会返回当前项目 HEAD 的所有实验。

import dvc.api

exps = dvc.api.exp_show()

返回的数据结构是一个字典列表，每个字典代表一个实验：

[
  {
    "Experiment": "paled-acre",
    "rev": "883442c",
    "Created": "Apr 19, 2023",
    "dice_multi": 0.8590125166103912,
    "train.arch": "squeezenet1_1"
  },
  {
    "Experiment": "vocal-suer",
    "rev": "231e504",
    "Created": "Apr 19, 2023",
    "dice_multi": 0.8997336177828745,
    "train.arch": "resnet34"
  }
]

每个实验字典包含了实验名称、Git 修订版本、创建时间以及所有参数和指标的值。字段名直接对应 params.yaml 中的参数路径和指标文件名。

与 Pandas 集成

返回的列表格式可以直接转换为 Pandas DataFrame，方便进行数据分析和可视化。

import dvc.api
import pandas as pd

df = pd.DataFrame(dvc.api.exp_show())

转换后，我们可以利用 Pandas 强大的数据处理能力来清洗和分析实验数据。

# 只选择需要的列
columns = ["Experiment", "dice_multi", "train.arch"]
df = pd.DataFrame(dvc.api.exp_show(), columns=columns)

# 清理数据
df.dropna(inplace=True)
df.reset_index(drop=True, inplace=True)

# 输出为 Markdown 表格
print(df.to_markdown())

输出结果：

|     | Experiment | train.arch    | dice_multi |
| --: | :--------- | :------------ | ---------: |
|   0 | paled-acre | squeezenet1_1 |   0.859013 |
|   1 | vocal-suer | resnet34      |   0.891619 |
|   2 | banal-hogs | alexnet       |   0.875823 |

筛选特定实验

exp_show() 提供了多个参数来控制返回的实验范围。repo 参数可以指定远程仓库，revs 参数可以指定特定的 Git 版本。

# 获取特定仓库的实验
exps = dvc.api.exp_show(
    repo="https://github.com/iterative/example-get-started"
)

# 获取特定版本的实验
exps = dvc.api.exp_show(revs="HEAD~1")

revs 参数可以接受单个版本字符串，也可以接受列表。结合 dvc.api.scm 模块，我们可以方便地获取所有标签或分支对应的实验。

import dvc.api

# 获取所有标签对应的实验
all_tags = dvc.api.scm.all_tags()
exps = dvc.api.exp_show(revs=all_tags)

控制输出细节

param_deps 参数可以限制只返回作为阶段依赖的参数，这在只想查看影响流水线执行的参数时很有用。

# 只返回阶段依赖的参数
exps = dvc.api.exp_show(param_deps=True)

config 参数允许传递配置字典，可以用来提供远程存储的认证信息，类似于其他 API 函数中的 remote_config。

参数读取 params_show

dvc.api.params_show() 专门用于读取 DVC 项目中追踪的参数。与 exp_show() 不同，它专注于参数，提供了更精细的控制。

基本用法

不带参数调用时，params_show() 返回项目中所有的参数。

import dvc.api

params = dvc.api.params_show()
print(params)

返回的字典结构直接对应 params.yaml 的内容：

{
  "split": 0.2,
  "seed": 20170428,
  "train": {
    "epochs": 10,
    "lr": 0.001
  }
}

按阶段筛选

在复杂的流水线中，不同阶段可能使用不同的参数。stages 参数可以限制只返回特定阶段的参数。

# 获取单个阶段的参数
params = dvc.api.params_show(stages="train")
print(params)

输出：

{
  "n_est": 50,
  "min_split": 0.01
}

stages 也可以接受阶段名称列表，返回多个阶段的参数合并结果。

# 获取多个阶段的参数
params = dvc.api.params_show(stages=["featurize", "train"])

返回的字典可以直接在代码中使用，例如初始化模型时传入超参数。

from sklearn.ensemble import RandomForestClassifier

params = dvc.api.params_show(stages="train")
clf = RandomForestClassifier(
    n_estimators=params["n_est"],
    min_samples_split=params["min_split"]
)

读取特定参数文件

如果项目使用了多个参数文件，可以通过 targets 参数指定要读取的文件路径。

# 读取特定参数文件
params = dvc.api.params_show("configs/params_dev.yaml")

# 读取多个参数文件
params = dvc.api.params_show(
    "params.yaml", 
    "configs/model.yaml"
)

当指定多个目标时，返回的字典会按文件名组织，避免参数名冲突。

{
  "params.yaml": {
    "split": 0.2,
    "seed": 20170428
  },
  "configs/model.yaml": {
    "arch": "resnet50",
    "pretrained": True
  }
}

读取依赖参数

deps 参数控制是否只返回被声明为阶段依赖的参数。当设置为 True 时，只有那些在 dvc.yaml 中 params 字段下列出的参数会被返回。

# 只返回阶段依赖的参数
params = dvc.api.params_show(deps=True)

这在只想查看影响流水线执行的参数时非常有用，可以过滤掉一些临时或调试用的参数。

指定版本和仓库

与其他 API 一样，params_show() 支持 repo 和 rev 参数，可以读取远程仓库或特定版本的参数。

# 从远程仓库读取参数
params = dvc.api.params_show(
    repo="https://github.com/iterative/example-get-started"
)

# 读取特定版本的参数
params = dvc.api.params_show(rev="v1.0.0")

指标读取 metrics_show

dvc.api.metrics_show() 用于读取 DVC 项目中追踪的指标文件。指标通常是模型评估的结果，如准确率、损失值等。

基本用法

不带参数调用时，metrics_show() 返回所有追踪的指标。

import dvc.api

metrics = dvc.api.metrics_show()
print(metrics)

返回的字典结构以指标文件名为键，指标值为嵌套字典：

{
  "avg_prec": { "train": 0.97, "val": 0.92 },
  "roc_auc": { "train": 0.98, "val": 0.94 }
}

如果只有一个指标文件，返回的字典会直接包含指标值。

筛选指标文件

targets 参数可以指定要读取的指标文件路径，支持一个或多个文件。

# 读取单个指标文件
metrics = dvc.api.metrics_show("eval/train_val_metrics.json")

# 读取多个指标文件
metrics = dvc.api.metrics_show(
    "eval/train_val_metrics.json", 
    "eval/test_metrics.json"
)

当指定多个目标时，返回的字典会按文件名组织：

{
  "eval/train_val_metrics.json": {
    "avg_prec": { "train": 0.97, "val": 0.92 },
    "roc_auc": { "train": 0.98, "val": 0.94 }
  },
  "eval/test_metrics.json": {
    "avg_prec": { "test": 0.72 },
    "roc_auc": { "test": 0.77 }
  }
}

指定版本和仓库

metrics_show() 同样支持 repo 和 rev 参数，可以读取远程仓库或特定版本的指标。

# 从远程仓库读取指标
metrics = dvc.api.metrics_show(
    repo="https://github.com/iterative/example-get-started"
)

# 读取特定版本的指标
metrics = dvc.api.metrics_show(rev="HEAD~1")

这在比较不同版本的模型性能时非常有用，可以程序化地获取历史实验的结果。

Artifact 访问

Artifact 是 DVC 中用于管理模型和数据集的重要概念。dvc.api.artifacts_show() 提供了程序化访问 artifact 的能力，可以获取 artifact 的路径和 Git 版本信息。

基本用法

artifacts_show() 需要至少指定 artifact 的名称。

import dvc.api

artifact = dvc.api.artifacts_show(
    'text-classification',
    repo='https://github.com/iterative/example-get-started.git',
)
print(artifact)

返回的字典包含两个关键字段：

{
    'path': 'model.pkl',
    'rev': 'c7c6ae0',
}

path 是 artifact 在 DVC 仓库中的相对路径，rev 是对应的 Git 修订版本。当不提供 version 或 stage 参数时，默认返回最新版本的 artifact。

按版本或阶段获取

Artifact 可以通过语义化版本或生命周期阶段来访问，这两个参数是互斥的。

# 按版本获取
artifact = dvc.api.artifacts_show(
    'text-classification',
    version='v1.0.0',
    repo='https://github.com/iterative/example-get-started.git',
)

# 按阶段获取
artifact = dvc.api.artifacts_show(
    'text-classification',
    stage='prod',
    repo='https://github.com/iterative/example-get-started.git',
)

这在模型部署场景中特别有用，可以确保获取到特定版本或已部署到生产环境的模型。

读取 Artifact 内容

获取到 artifact 的路径和版本后，可以结合 dvc.api.read() 来读取内容。

import pickle
import dvc.api

# 获取 artifact 信息
artifact = dvc.api.artifacts_show(
    'text-classification',
    version='v1.0.0',
    repo='https://github.com/iterative/example-get-started.git',
)

# 读取 artifact 内容
data = dvc.api.read(
    artifact['path'],
    rev=artifact['rev'],
    repo='https://github.com/iterative/example-get-started.git',
    mode='rb',
)

# 反序列化模型
model = pickle.loads(data)

这种方式避免了硬编码模型路径，通过 artifact 名称来访问，使代码更具可维护性。

下载 Artifact

如果需要将 artifact 下载到本地，可以结合 DVCFileSystem 使用。

import os
import dvc.api

# 获取 artifact 信息
artifact = dvc.api.artifacts_show(
    'text-classification',
    stage='prod',
    repo='https://github.com/iterative/example-get-started.git',
)

# 创建文件系统对象
fs = dvc.api.DVCFileSystem(
    'https://github.com/iterative/example-get-started.git',
    rev=artifact['rev'],
)

# 下载 artifact
fs.get_file(artifact['path'], os.path.basename(artifact['path']))

这段代码将生产环境的模型下载到当前工作目录，文件名保持原样。这在部署脚本中很常见，可以自动化获取最新模型。

Artifact 名称解析

默认情况下，DVC 会在仓库根目录的 dvc.yaml 文件中查找 artifact 定义。如果 artifact 定义在其他位置的 dvc.yaml 中，需要使用特定格式指定名称。

# 指定路径到 dvc.yaml
artifact = dvc.api.artifacts_show(
    'path/to/dvc.yaml:artifact_name',
    repo='https://github.com/iterative/example-get-started.git',
)

# 或省略 dvc.yaml 文件名
artifact = dvc.api.artifacts_show(
    'path/to:artifact_name',
    repo='https://github.com/iterative/example-get-started.git',
)

这种灵活的命名方式支持复杂的项目结构，可以在不同子目录中管理多个 artifact。

总结

DVC 的 Python API 为自动化和集成提供了强大工具。DVCFileSystem 让我们能以统一接口访问 Git 和 DVC 追踪的文件，支持本地和远程仓库。实验数据查询 API 使程序化分析实验结果变得简单，特别是与 Pandas 结合时。参数和指标读取 API 为超参数优化和模型评估提供了基础设施。Artifact 访问 API 则简化了模型管理和部署流程。

这些 API 设计遵循 Python 惯例，返回标准数据结构，易于集成到现有工作流中。无论是构建自动化训练流水线、开发实验分析工具，还是实现模型服务，DVC Python API 都能提供必要的支持。

下一章将探讨大规模数据管理，包括大数据集优化、云存储版本控制和缓存迁移等高级主题，这些内容与 Python API 结合使用，能构建出更强大的 MLOps 系统。