# 工件

## 简介

**artifact**（工件）是附加到项目上的、带有版本和名称的文件包——例如训练好的模型、数据集或一组评估输出。将文件作为工件记录（而不是将其保留在磁盘上）可以为您提供：

- **版本控制** — 每次以相同名称记录时，Trackio 都会记录一个新版本（`v0`、`v1` 等）。相同的内容会被去重，因此重新记录未更改的文件会复用现有版本。
- **别名** — 移动指针，例如 `latest`（自动分配）或您自定义的（`prod`、`best`），它们始终解析为单个版本。
- **血缘关系** — 工件与生成和消费它们的运行相关联，因此您可以追溯哪个运行创建了模型，以及哪些运行使用了该模型。

Artifacts 可完全离线运行，使用本地存储，并在配置了 Hugging Face Space（或自托管服务器）时与之同步，其机制与 metrics 完全相同。

## 记录工件

最快的方式是直接记录文件或目录。这会创建一个新工件，添加路径，并将其记录为当前运行的输出：

```python
import trackio

trackio.init(project="my_project")

# ... training ...

trackio.log_artifact("checkpoints/model.safetensors", type="model")
```

当传入路径时，工件的 `name` 默认为 `run-<run_id>-<basename>`，`type` 默认为 `"unspecified"`。传入 `name` 和 `type` 可显式设置它们：

```python
trackio.log_artifact("checkpoints/", name="my-model", type="model")
```

### 工件类型

`type` 是自由格式的类别字符串。Trackio 不强制使用固定的类型列表，但常见的约定包括：

| 类型 | 用于 |
| --- | --- |
| `"model"` | 模型检查点、适配器权重、分词器、模型配置 |
| `"dataset"` | 训练、验证、测试或生成的数据集 |
| `"evaluation"` | 评估输出、预测结果、基准测试结果、分数文件 |
| `"report"` | 图表、绘图、表格、笔记本或人类可读的分析包 |
| `"unspecified"` | 未提供类型时的默认值 |

在项目中使用稳定的类型名称，以便一致地筛选工件，并使用 `trackio.use_artifact(..., type="...")` 来断言该运行正在消费预期类型的工件。

### 显式构建工件

如需更精细的控制——例如多个文件、自定义逻辑路径、描述或元数据——请构造一个 [Artifact](/docs/trackio/v0.45.0/zh/api#trackio.Artifact)，向其添加文件，然后记录它：

```python
artifact = trackio.Artifact(
    name="my-model",
    type="model",
    description="Fine-tuned on the v2 dataset",
    metadata={"base_model": "bert-base", "epochs": 3},
)
artifact.add_file("checkpoints/model.safetensors")
artifact.add_file("checkpoints/config.json")
artifact.add_dir("tokenizer/")

logged = trackio.log_artifact(artifact)
```

记录完成后，工件将被冻结，其 `version`、`aliases`、`size` 和 `manifest` 将被填充：

```python
print(logged.version)   # "v0"
print(logged.aliases)   # ("latest",)
```

### 版本和别名

每个在现有名称下记录的日志都会创建下一个整数版本，并将 `latest` 别名指向该版本。使用 `aliases=` 可以分配自定义别名：

```python
trackio.log_artifact(artifact, aliases=["prod", "best"])
```

别名是移动指针：使用相同别名记录新版本时，该别名会重新指向新版本。版本说明符（`v0`、`v1`、...）是保留的，不能用作别名。

如果您记录的内容与现有版本完全相同，Trackio 会复用该版本，而不是创建新版本。您显式传入的任何 `aliases` 仍会轮换到该版本上，但 `latest` 会保持在其当前位置——重新记录相同（或更旧）的内容永远不会使其回退。

## 使用工件

[use_artifact()](/docs/trackio/v0.45.0/zh/api#trackio.use_artifact) 会获取一个工件，并将其记录为当前运行的 **输入**，这正是用于在消费该工件的运行与工件之间建立血缘关系的方式：

```python
trackio.init(project="my_project")

artifact = trackio.use_artifact("my-model")          # resolves to :latest
artifact = trackio.use_artifact("my-model:v2")       # pin a version
artifact = trackio.use_artifact("my-model:prod")     # resolve an alias
```

传入 `type` 以断言工件的类型，如果不匹配则引发异常：

```python
artifact = trackio.use_artifact("my-model", type="model")
```

### 下载文件

`use_artifact` 返回一个 [Artifact](/docs/trackio/v0.45.0/zh/api#trackio.Artifact)，你可以使用 `download()` 来实例化其中的文件：

```python
artifact = trackio.use_artifact("my-model:latest")
path = artifact.download()
# files are now under ./.trackio/artifact-downloads/my_project/my-model_v2/
```

默认情况下，文件会写入 `./.trackio/artifact-downloads/<project>/<name>_v<version>/`，并按项目作为键，以确保来自不同项目的同名 artifact 不会发生冲突；传入 `root` 可指定其他目录。`download()` 是幂等的——已存在的文件会被跳过——并且当 run 由 Space 提供支持时，本地缺失的文件会从远程获取。

## 引用外部数据

某些数据体积过大无法复制，或者已经存在于您不希望重复的持久化存储中。`add_reference` 通过 **URI** 在清单中记录此类数据，*无需将任何字节暂存* 到 Trackio 的存储中，因此您仍然可以对从未移动的数据进行版本控制、去重、设置别名并追踪其血缘关系：

```python
artifact = trackio.Artifact(name="training-set", type="dataset")
artifact.add_reference("file:///mnt/nvme/datasets/corpus.parquet")
artifact.add_reference("file:///mnt/nvme/datasets/shards/")            # a directory — expands to one entry per file
artifact.add_reference("https://example.com/data/eval.json")
artifact.add_reference("hf://datasets/username/my-dataset/train.parquet")
trackio.log_artifact(artifact)
```

内置方案为 `file://`、`http(s)://` 和 `hf://` —— 均由 Trackio 的现有依赖提供支持。其他任何方案（`s3://`、`gs://`、Azure blob URL 等）都可以通过注册自定义处理器来支持；完整的 S3、GCS 和 Azure 示例请参见下文中的 [引用其他存储](#referencing-other-stores-s3-gcs-azure)。使用 `checksum=True`（默认值）时，Trackio 会探测源以填充 `size` 和校验和：

- `file://` 会通过流式哈希计算得到 sha256 值，其大小从磁盘读取。
- `hf://` 使用 LFS sha256（或 git blob id）；`http(s)://` 使用服务器的 `ETag`。

每个校验和都存储在一个单独的 `digest` 字段中——对于本地/LFS 引用，它是 sha256；对于 HTTP 或对象存储引用，它是提供方的 ETag。

表示目录（`file://`）或仓库路径（`hf://`）的 URI 会被**展开**为每个对象一个引用条目，每个条目以其相对于前缀的路径命名（如果提供了 `name=`，则嵌套在其下）。`max_objects`（默认值为 10,000）限制展开数量，超出时抛出异常。

当无法获取校验和时——例如服务器没有 `ETag`——引用仍会被记录，使用 **URI 本身作为 `digest`**，并会发出警告。不存在的本地 `file://` 路径是例外：它会引发错误而不是被记录，因为缺失的本地文件几乎总是错误——请使用 `add_file` 来暂存本地文件。传入 `checksum=False` 可跳过针对每个对象的校验和——此时单个对象仅记录其 URI——并使用 `name=` 来设置逻辑路径。

引用的版本标识由其 URI、大小和 `digest` 决定，因此重复记录相同的引用会去重为一个版本，而 URI、大小或校验和的任何变更都会创建新版本——这与文件内容的处理方式完全一致。当 URI 被用作 `digest` 时，版本标识因此跟随 URI 而非内容。

URI 会按原样记录，并且清单可以同步到 Hugging Face Dataset。如果 URI 中嵌入了凭证——例如预签名的 `https://` URL 或 Azure SAS 令牌——那么任何能够读取该 Dataset 的人都将看到该凭证，因此当 URI 的查询字符串看起来像是已签名时，`add_reference` 会发出警告；建议优先使用对象的规范无签名 URI（例如 `s3://bucket/key`）。

Reference 字节永远不会上传到 Space 或存储桶，但可按需获取。`download()` 会将每个引用的对象下载到目录中（通过 `httpx2`、`huggingface_hub` 或您注册的处理器），这可能会传输大量数据。在获取之前，会重新探测每个对象的来源，并将其当前校验和与记录的校验和进行比较——如果不匹配，会发出警告提示来源可能已更改，但下载仍会继续进行。每个对象都是原子写入的，如果磁盘上已存在该对象，则会跳过而不重新检查来源，因此重复调用成本低且是幂等的：

```python
artifact = trackio.use_artifact("training-set:latest")
path = artifact.download()   # downloads staged files AND referenced objects
```

要处理指针且*无需*下载，请直接从工件中读取它们：

```python
print(artifact.references)                  # [{'path': ..., 'ref': 's3://...', 'size': ..., 'digest': ...}, ...]
print(artifact.get_entry_uri("eval.json"))  # "s3://my-bucket/eval.json"
```

当 `download()` 遇到无法获取的引用（缺失的本地文件、无法访问的 URL，或未注册处理程序的协议）时，会引发异常。

## 引用其他存储（S3、GCS、Azure）

Trackio 刻意不附带任何云厂商的 SDK。若要引用其未原生支持的存储中的对象，请子类化 `trackio.ReferenceHandler`，并使用 `trackio.register_reference_handler` 注册一个实例。该处理器需实现四个方法：

- `matches(scheme, uri)` — 判断此处理器是否拥有该 URI。注册的处理器会在内置处理器*之前*被查询，因此处理器也可以声明特定的 `https://` 主机（如下面的 Azure 示例所示），优先于通用的 HTTP 处理器。
- `resolve(uri, checksum, max_objects)` — 探测源并返回一个 `trackio.ResolvedReference` 条目列表：对于单个对象，返回一个 `relkey=None` 的条目；当 URI 表示一个前缀时，为每个对象返回一个条目（其中 `relkey` 设置为该对象相对于前缀的路径）。如果源可以被探测，则设置 `size` 和 `digest`（提供方的 ETag 可以很好地用作摘要——Trackio 将引用摘要视为不透明的令牌）；如果返回 `None`，则记录该引用而不进行校验和，在这种情况下 Trackio 会回退到使用 URI 作为摘要并发出警告。
- `fetch(uri, dest)` — 将 `uri` 处的单个对象下载到本地路径 `dest`（由 `Artifact.download()` 调用）。
- `hint()` — 当 `resolve` 未生成校验和时，在警告中显示的一行可操作提示消息。

在合成每个条目的 `uri` 时，前缀扩展过程中发现的键应进行百分号编码（并在 `fetch` 中解码），因为清单 URI 中可能不包含空格。

### S3 （`s3://`，通过 `boto3`）

```python
from urllib.parse import quote, unquote, urlsplit

import boto3

import trackio

class S3Handler(trackio.ReferenceHandler):
    def matches(self, scheme, uri):
        return scheme == "s3"

    def resolve(self, uri, checksum, max_objects):
        parts = urlsplit(uri)
        bucket, key = parts.netloc, unquote(parts.path.lstrip("/"))
        client = boto3.client("s3")
        if key and not key.endswith("/"):
            head = client.head_object(Bucket=bucket, Key=key)
            return [
                trackio.ResolvedReference(
                    relkey=None,
                    uri=uri,
                    size=int(head["ContentLength"]),
                    digest=head["ETag"].strip('"') if checksum else None,
                )
            ]
        base = key[: key.rfind("/") + 1]
        entries = []
        pages = client.get_paginator("list_objects_v2").paginate(
            Bucket=bucket, Prefix=key
        )
        for page in pages:
            for obj in page.get("Contents", []):
                if obj["Key"].endswith("/"):
                    continue
                if len(entries) >= max_objects:
                    raise ValueError(
                        f"{uri} expands to more than {max_objects} objects; "
                        "pass max_objects= to raise the limit."
                    )
                entries.append(
                    trackio.ResolvedReference(
                        relkey=obj["Key"][len(base):],
                        uri=f"s3://{bucket}/{quote(obj['Key'], safe='/')}",
                        size=int(obj["Size"]),
                        digest=obj["ETag"].strip('"') if checksum else None,
                    )
                )
        return entries

    def fetch(self, uri, dest):
        parts = urlsplit(uri)
        with open(dest, "wb") as f:
            boto3.client("s3").download_fileobj(
                parts.netloc, unquote(parts.path.lstrip("/")), f
            )

    def hint(self):
        return "Configure AWS credentials so the object can be probed for an ETag."

trackio.register_reference_handler(S3Handler())

artifact = trackio.Artifact(name="training-set", type="dataset")
artifact.add_reference("s3://my-bucket/shards/")   # expands to one entry per object
trackio.log_artifact(artifact)
```

### Google Cloud Storage（`gs://`，通过 `google-cloud-storage`）

```python
from urllib.parse import quote, unquote, urlsplit

from google.cloud import storage

import trackio

class GcsHandler(trackio.ReferenceHandler):
    def matches(self, scheme, uri):
        return scheme == "gs"

    def resolve(self, uri, checksum, max_objects):
        parts = urlsplit(uri)
        bucket, key = parts.netloc, unquote(parts.path.lstrip("/"))
        client = storage.Client()
        if key and not key.endswith("/"):
            blob = client.bucket(bucket).get_blob(key)
            if blob is None:
                raise ValueError(f"No such GCS object: {uri}")
            return [
                trackio.ResolvedReference(
                    relkey=None,
                    uri=uri,
                    size=blob.size,
                    digest=blob.etag.strip('"') if checksum else None,
                )
            ]
        base = key[: key.rfind("/") + 1]
        entries = []
        for blob in client.list_blobs(bucket, prefix=key):
            if blob.name.endswith("/"):
                continue
            if len(entries) >= max_objects:
                raise ValueError(
                    f"{uri} expands to more than {max_objects} objects; "
                    "pass max_objects= to raise the limit."
                )
            entries.append(
                trackio.ResolvedReference(
                    relkey=blob.name[len(base):],
                    uri=f"gs://{bucket}/{quote(blob.name, safe='/')}",
                    size=blob.size,
                    digest=blob.etag.strip('"') if checksum else None,
                )
            )
        return entries

    def fetch(self, uri, dest):
        parts = urlsplit(uri)
        key = unquote(parts.path.lstrip("/"))
        storage.Client().bucket(parts.netloc).blob(key).download_to_filename(
            str(dest)
        )

    def hint(self):
        return "Configure GCP credentials so the object can be probed for an ETag."

trackio.register_reference_handler(GcsHandler())
```

### Azure Blob Storage（通过 `azure-storage-blob` + `azure-identity`）

Azure 存储块通过普通的 `https://` URL（`https://<account>.blob.core.windows.net/<container>/<blob>`）进行寻址，因此该处理器通过主机名后缀来声明它们——注册后，它将优先于 Trackio 的通用 HTTP 处理器：

```python
from urllib.parse import quote, unquote, urlsplit

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

import trackio

class AzureHandler(trackio.ReferenceHandler):
    def matches(self, scheme, uri):
        return scheme in ("http", "https") and urlsplit(
            uri
        ).netloc.lower().endswith(".blob.core.windows.net")

    def _split(self, uri):
        parts = urlsplit(uri)
        container, _, blob_name = parts.path.lstrip("/").partition("/")
        return f"{parts.scheme}://{parts.netloc}", container, unquote(blob_name)

    def _service(self, account_url):
        return BlobServiceClient(account_url, credential=DefaultAzureCredential())

    def resolve(self, uri, checksum, max_objects):
        account_url, container, blob_name = self._split(uri)
        service = self._service(account_url)
        if blob_name and not blob_name.endswith("/"):
            props = service.get_blob_client(container, blob_name).get_blob_properties()
            return [
                trackio.ResolvedReference(
                    relkey=None,
                    uri=uri,
                    size=props.size,
                    digest=props.etag.strip('"') if checksum else None,
                )
            ]
        base = blob_name[: blob_name.rfind("/") + 1]
        entries = []
        container_client = service.get_container_client(container)
        for blob in container_client.list_blobs(name_starts_with=blob_name):
            if blob.name.endswith("/"):
                continue
            if len(entries) >= max_objects:
                raise ValueError(
                    f"{uri} expands to more than {max_objects} objects; "
                    "pass max_objects= to raise the limit."
                )
            entries.append(
                trackio.ResolvedReference(
                    relkey=blob.name[len(base):],
                    uri=f"{account_url}/{container}/{quote(blob.name, safe='/')}",
                    size=blob.size,
                    digest=blob.etag.strip('"') if checksum else None,
                )
            )
        return entries

    def fetch(self, uri, dest):
        account_url, container, blob_name = self._split(uri)
        downloader = (
            self._service(account_url)
            .get_blob_client(container, blob_name)
            .download_blob()
        )
        with open(dest, "wb") as f:
            downloader.readinto(f)

    def hint(self):
        return "Configure Azure credentials so the blob can be probed for an ETag."

trackio.register_reference_handler(AzureHandler())
```

这些处理器在无法探测源时会抛出异常；如果你希望在这种情况下记录一个未校验和的引用（例如，在记录日志时无法访问需要身份验证的数据目录时），请在 `resolve` 内部捕获 provider 错误并返回 `[trackio.ResolvedReference(relkey=None, uri=uri)]` — Trackio 将回退到使用 URI 作为摘要并发出警告，同时使用你的 `hint()`。

在添加*或下载*此类引用的每个进程中注册处理程序：`Artifact.download()` 需要处理程序来获取字节，而方案未注册处理程序的 URI 会引发异常。

## 在仪表板中浏览工件

仪表板的 **Artifacts** 选项卡（在项目拥有工件后显示）按类型列出工件，并显示每个版本的别名、大小和文件数。选择某个版本会显示其文件清单——点击文件即可下载——以及其元数据和血缘关系：生成该工件的运行和使用该工件的运行，每个运行都链接到其详情页。

每个版本还包含一个 **Lineage** 部分，展示该版本谱系的交互式图形：制品版本和运行作为节点，箭头从运行指向其记录的制品，从制品指向使用它的运行。选中的版本会被高亮显示。为了保持大型图形的可读性，五个或更多具有相同连接的相似节点（例如，单个运行记录的许多检查点）会被分组为一个显示数量的 **cluster** 节点——点击该集群可搜索其成员，并将单个成员提取回图形中。拖动以平移，使用 Ctrl/Cmd + 滚轮（或 +/− 按钮）进行缩放，点击任意节点可打开摘要，其中包含指向该运行或制品版本的链接。

运行详情页面从另一个方向展示相同的血缘关系，其中 **输出工件** 和 **输入工件** 部分会链接回工件选项卡。

## 远程存储

当您的运行目标为 Hugging Face Space 或自托管服务器（参见 [Track](track)）时，工件文件采用内容寻址方式并仅上传一次：Trackio 会跳过服务器已拥有的 blob，因此重新记录共享文件的开销很低。工件的元数据和 blob 会与其他运行数据一起持久化到已配置的 HF Dataset 或存储桶中。