成果物

はじめに

artifactとは、プロジェクトに紐付けられた、バージョン管理された名前付きのファイルのバンドルです。学習済みモデル、データセット、評価出力のセットなどが該当します。ファイルをディスク上に残すのではなく、artifactとしてログに記録することで、以下のような利点があります。

バージョン管理 — 同じ名前でログを記録するたびに、Trackio は新しいバージョン（v0、v1、…）を記録します。同一の内容は重複排除されるため、変更されていないファイルを再記録すると、既存のバージョンが再利用されます。
エイリアス — latest（自動で割り当てられます）や、自分で定義したもの（prod、best など）のように、常に単一のバージョンに解決されるポインターを移動します。
系譜 — 成果物は、それらを生成・消費した実行とリンクされるため、どの実行がモデルを作成し、どの実行がそれを使用したかを追跡できます。

Artifactsはローカルストレージに対して完全にオフラインで動作し、設定されている場合、メトリクスと同様にHugging Face Space（またはセルフホストサーバー）に同期されます。

アーティファクトのログ記録

最も簡単な方法は、ファイルやディレクトリを直接ログに記録することです。これにより、新しいアーティファクトが作成され、パスが追加され、現在のランの出力として記録されます。

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")

アーティファクトの種類

Artifactの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 はその位置のまま変更されません。同一（または古い）内容を再ログしても、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 は、download() でファイルをマテリアライズする Artifact を返します:

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>/に書き込まれます。プロジェクトごとにキーが設定されるため、異なるプロジェクトからの同名のアーティファクトが衝突することはありません。別のディレクトリを選択するにはrootを渡します。download()は冪等です。すでに存在するファイルはスキップされます。また、ランがSpaceによってバックエンドされている場合、ローカルに存在しないファイルはリモートから取得されます。

外部データの参照

一部のデータはコピーが大きすぎるか、重複を避けたい永続ストレージに既に存在しています。add_reference は、Trackio のストレージに バイトをステージングせずに、マニフェストに URI としてそのようなデータを記録します。これにより、データが移動しない場合でも、バージョン管理、重複排除、エイリアス、系譜管理を利用できます:

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 は、展開され、各オブジェクトごとに1つの参照エントリが生成されます。各エントリは、プレフィックスからの相対パスで名前付けられます (name= が指定されている場合は、その下にネストされます)。 max_objects (デフォルト 10,000) は展開の上限を設け、超過した場合は例外を発生させます。

チェックサムを取得できない場合（ETagのないサーバーなど）は、URI 自体をdigestとして参照が記録され、警告が出力されます。存在しないローカルのfile://パスは例外であり、ローカルファイルの欠落はほぼ間違いであるため、記録されるのではなく例外がスローされます。ローカルファイルをステージングするにはadd_fileを使用してください。checksum=Falseを渡すと、オブジェクトごとのチェックサムをスキップし、そのオブジェクトは URI のみを記録します。name=で論理パスを設定できます。

参照のバージョン同一性は、そのURI、サイズ、およびdigestから決定されるため、同一の参照を再ログすると1つのバージョンに重複排除されます。一方、URI、サイズ、またはチェックサムが変更されると新しいバージョンが作成されます。これはファイルの内容の場合と全く同じです。URIがdigestとして使用される場合、同一性は内容ではなくURIを追跡することになります。

URI はそのまま記録され、マニフェストは Hugging Face Dataset に同期できます。認証情報を埋め込んだ URI（プレサインされた https:// URL や Azure SAS トークンなど）は、Dataset を読み取れる誰にでもその認証情報を露出させることになるため、add_reference は URI のクエリ文字列が署名付きのように見える場合に警告を発します。署名なしの正規 URI（例: s3://bucket/key）の使用を推奨します。

Reference bytes は 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"

取得できない参照（ローカルファイルの欠落、到達できない URL、登録されていないハンドラーを持つスキーム）にdownload()が到達すると、例外が発生します。

他のストア（S3、GCS、Azure）への参照

Trackioは意図的にクラウドベンダーのSDKを同梱していません。ネイティブで処理しないストア内のオブジェクトを参照するには、trackio.ReferenceHandlerをサブクラス化し、trackio.register_reference_handlerでインスタンスを登録します。ハンドラーは4つのメソッドを実装します:

matches(scheme, uri) — このハンドラーがURIを管理しているかどうかを示します。登録されたハンドラーは組み込みのハンドラーより先に参照されるため、汎用のHTTPハンドラーに先立って、特定のhttps://ホスト（下のAzureの例のように）を管理することもできます。
resolve(uri, checksum, max_objects) — ソースをプローブし、trackio.ResolvedReference エントリのリストを返します。単一のオブジェクトの場合は relkey=None のエントリが 1 つ、URI がプレフィックスを指す場合はオブジェクトごとに 1 つのエントリ（relkey はプレフィックスからの相対パス）が返されます。ソースをプローブできる場合は size と digest を設定します（プロバイダーの ETag をダイジェストとして使用しても問題ありません。Trackio は参照のダイジェストを不透明なトークンとして扱います）。プローブできない場合は None を返してチェックサムなしで参照を記録し、この場合 Trackio は URI をダイジェストとしてフォールバックし、警告を発します。
fetch(uri, dest) — uri にある単一のオブジェクトをローカルパス dest にダウンロードします（Artifact.download() から呼び出されます）。
hint() — resolveがチェックサムを生成しなかった場合に、警告内に表示される1行のアクション可能なメッセージ。

プレフィックス展開の過程で発見されたキーは、各エントリの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 blobs はプレーンな 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内でプロバイダーの例外をキャッチし、[trackio.ResolvedReference(relkey=None, uri=uri)]を返してください。Trackio はダイジェストとして URI にフォールバックし、警告を出力します。この際、hint()が使用されます。

追加またはダウンロードするすべてのプロセスでハンドラーを登録してください: Artifact.download()はバイトを取得するためにハンドラーを必要とし、スキームにハンドラーが登録されていないURIは例外を発生させます。

ダッシュボードで成果物を閲覧する

ダッシュボードの Artifacts タブ（プロジェクトに成果物がある場合に表示されます）には、成果物が種類別にグループ化されて一覧表示され、各バージョンのエイリアス、サイズ、ファイル数が表示されます。バージョンを選択すると、そのファイルマニフェスト（ファイルをダウンロードするにはファイルをクリックします）と、メタデータおよび系譜が表示されます。系譜には、その成果物を生成した実行と、その成果物を使用した実行が含まれ、それぞれが実行の詳細ページにリンクしています。

各バージョンには、Lineageセクションも含まれており、バージョンの系譜を示すインタラクティブなグラフが表示されます。このグラフでは、アーティファクトのバージョンとランがノードとして表され、ランからそのランが記録したアーティファクトへ、またアーティファクトからそれを使用したランへ向かう矢印が引かれています。選択されたバージョンはハイライト表示されます。大規模なグラフの可読性を保つため、接続が同一の類似ノードが5つ以上ある場合（例えば、単一のランが記録した多数のチェックポイントなど）は、件数を表示する単一のclusterノードにグループ化されます。このクラスターをクリックすると、そのメンバーを検索し、個々のノードをグラフに復元できます。ドラッグでパンし、Ctrl/Cmd + スクロール（または +/− ボタン）でズームし、任意のノードをクリックすると、そのランまたはアーティファクトバージョンへのリンク付きのサマリーが開きます。

Runの詳細ページでは、逆方向の系譜が表示され、Output artifactsセクションとInput artifactsセクションからArtifactsタブにリンクできます。

リモートストレージ

ランが Hugging Face Space またはセルフホストされたサーバーをターゲットにしている場合（Trackを参照）、アーティファクトファイルはコンテンツアドレス方式で管理され、一度だけアップロードされます。Trackio はサーバーが既に保持しているブロブをスキップするため、共有ファイルの再ログ記録は低コストです。アーティファクトのメタデータとブロブは、設定された HF Dataset またはバケットに、他のランデータとともに永続化されます。

Update on GitHub