工件

简介

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

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

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

记录工件

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

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 可显式设置它们：

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

工件类型

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

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

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

显式构建工件

如需更精细的控制——例如多个文件、自定义逻辑路径、描述或元数据——请构造一个 Artifact，向其添加文件，然后记录它：

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 将被填充：

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

版本和别名

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

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

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

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

使用工件

use_artifact() 会获取一个工件，并将其记录为当前运行的输入，这正是用于在消费该工件的运行与工件之间建立血缘关系的方式：

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 以断言工件的类型，如果不匹配则引发异常：

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

下载文件

use_artifact 返回一个 Artifact，你可以使用 download() 来实例化其中的文件：

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 的存储中，因此您仍然可以对从未移动的数据进行版本控制、去重、设置别名并追踪其血缘关系：

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 示例请参见下文中的引用其他存储。使用 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 或您注册的处理器），这可能会传输大量数据。在获取之前，会重新探测每个对象的来源，并将其当前校验和与记录的校验和进行比较——如果不匹配，会发出警告提示来源可能已更改，但下载仍会继续进行。每个对象都是原子写入的，如果磁盘上已存在该对象，则会跳过而不重新检查来源，因此重复调用成本低且是幂等的：

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

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

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）

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）

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 处理器：

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）时，工件文件采用内容寻址方式并仅上传一次：Trackio 会跳过服务器已拥有的 blob，因此重新记录共享文件的开销很低。工件的元数据和 blob 会与其他运行数据一起持久化到已配置的 HF Dataset 或存储桶中。

Update on GitHub