아티팩트

소개

artifact은 프로젝트에 연결된 버전 관리가 되는 이름이 있는 파일 묶음으로, 학습된 모델, 데이터셋, 평가 출력 세트 등이 있습니다. 파일을 디스크에 그대로 두는 대신 artifact로 기록하면 다음 이점을 얻을 수 있습니다:

버전 관리 — 동일한 이름으로 로깅할 때마다 Trackio는 새로운 버전(v0, v1, …)을 기록합니다. 동일한 콘텐츠는 중복 제거되므로 변경되지 않은 파일을 다시 로깅하면 기존 버전을 재사용합니다.
별칭 — latest(자동으로 할당됨)이나 사용자 지정(prod, best)과 같이 항상 단일 버전으로 해석되는 포인터를 이동하는 것.
Lineage — 아티팩트는 해당 아티팩트를 생성하고 소비한 실행(run)과 연결되므로, 어떤 실행이 모델을 생성했고 어떤 실행들이 해당 모델을 사용했는지 추적할 수 있습니다.

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

Aliases는 이동하는 포인터입니다: 동일한 alias로 새 버전을 기록하면 해당 alias가 다시 가리키게 됩니다. 버전 지정자(v0, v1, …)는 예약되어 있어 alias로 사용할 수 없습니다.

기존 버전과 동일한 콘텐츠를 기록하면 Trackio는 새 버전을 생성하지 않고 해당 버전을 재사용합니다. 명시적으로 전달하는 aliases는 여전히 해당 버전으로 회전하지만, latest는 현재 위치를 유지합니다. 동일한(또는 이전) 콘텐츠를 다시 기록해도 latest가 뒤로 이동하지 않습니다.

아티팩트 사용

use_artifact()는 아티팩트를 가져와서 현재 run의 input으로 기록합니다. 이를 통해 아티팩트를 소비하는 run과 해당 아티팩트 사이의 계보(lineage)가 구축됩니다:

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()로 파일을 materialize하는 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()는 멱등(idempotent)이며 이미 존재하는 파일은 건너뜁니다. 런이 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가 없는 서버인 경우에도 참조는 기록되며, digest로 URI 자체를 사용하고 경고가 출력됩니다. 존재하지 않는 로컬 file:// 경로는 예외입니다. 로컬 파일이 없는 것은 거의 항상 실수이므로 기록되지 않고 예외가 발생하며, 로컬 파일을 스테이징하려면 add_file를 사용하십시오. 개체별 체크섬을 건너뛰려면 checksum=False를 전달하십시오. 그러면 단일 개체는 URI만 기록합니다. 논리적 경로를 설정하려면 name=를 사용하십시오.

레퍼런스의 버전 동일성은 URI, 크기, digest에서 비롯되므로 동일한 레퍼런스를 다시 기록하면 하나의 버전으로 중복 제거되고, URI, 크기 또는 체크섬이 변경되면 파일 내용과 동일하게 새로운 버전이 생성됩니다. URI가 digest로 사용될 경우, 동일성은 내용 대신 URI를 추적합니다.

URI는 그대로 기록되며, 매니페스트는 Hugging Face Dataset에 동기화할 수 있습니다. 자격 증명을 포함하는 URI, 즉 presigned https:// URL 또는 Azure SAS 토큰은 Dataset을 읽을 수 있는 모든 사람에게 해당 자격 증명을 노출시키므로, add_reference는 URI의 쿼리 문자열이 서명된 것으로 보일 때 경고를 표시합니다. 대신 객체의 정식 서명되지 않은 URI(예: s3://bucket/key)를 사용하는 것이 좋습니다.

Reference bytes는 Space나 bucket에 업로드되지 않지만, 필요 시점에 가져올 수 있습니다. 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가 기본적으로 처리하지 않는 스토어의 객체를 참조하려면 trackio.ReferenceHandler를 서브클래스로 구현하고 trackio.register_reference_handler로 인스턴스를 등록하세요. 핸들러는 다음 네 가지 메서드를 구현합니다:

matches(scheme, uri) — 이 핸들러가 해당 URI를 소유하는지 여부입니다. 등록된 핸들러는 내장 핸들러보다 앞서 참조되므로, 특정 https:// 호스트를 일반 HTTP 핸들러보다 먼저 담당할 수 있습니다(아래 Azure 예시 참조).
resolve(uri, checksum, max_objects) — 소스를 탐색하여 trackio.ResolvedReference 항목의 목록을 반환합니다. 단일 객체인 경우 relkey=None인 항목이 하나 반환되고, URI가 접두사를 나타내는 경우 각 객체에 대해 relkey가 해당 객체의 접두사 상대 경로로 설정된 항목이 하나씩 반환됩니다. 소스를 탐색할 수 있는 경우 size와 digest를 설정합니다(프로바이더의 ETag를 digest로 사용하는 것이 적합합니다. Trackio는 참조 digest를 불투명한 토큰으로 취급합니다). 소스를 탐색할 수 없는 경우 None을 반환하여 checksum이 없는 참조로 기록하며, 이 경우 Trackio는 URI를 digest로 대체하고 경고를 표시합니다.
fetch(uri, dest) — uri에 있는 단일 개체를 로컬 경로 dest로 다운로드합니다(Artifact.download()에서 호출됨).
hint() — resolve가 체크섬을 생성하지 않은 경우 경고에 표시되는 한 줄의 실행 가능한 메시지

prefix expansion 과정에서 발견된 키는 각 항목의 uri를 합성할 때 percent-encoding 처리해야 합니다(그리고 fetch에서 디코딩해야 합니다). manifest 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>)로 지정되므로, 이 핸들러는 호스트 접미사를 기준으로 해당 URL을 처리합니다. 이 핸들러를 등록하면 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 내부에서 프로바이더 오류를 catch하고 [trackio.ResolvedReference(relkey=None, uri=uri)]을 반환하세요. Trackio는 URI를 digest로 폴백하여 경고하며, 이때 hint()를 사용합니다.

해당 참조를 추가하거나 다운로드하는 모든 프로세스에서 핸들러를 등록하세요: Artifact.download()는 바이트를 가져오기 위해 핸들러가 필요하며, 등록된 핸들러가 없는 스킴을 가진 URI는 예외를 발생시킵니다.

대시보드에서 아티팩트 탐색

대시보드의 Artifacts 탭(프로젝트에 아티팩트가 있는 경우에만 표시됨)에는 유형별로 그룹화된 아티팩트가 나열되며, 각 버전의 별칭, 크기, 파일 수를 확인할 수 있습니다. 버전을 선택하면 파일 매니페스트(파일을 클릭하여 다운로드 가능), 메타데이터, 계보(해당 아티팩트를 생성한 실행 및 해당 아티팩트를 사용한 실행)가 표시되며, 각각은 해당 실행의 상세 페이지로 연결됩니다.

각 버전에는 Lineage 섹션도 포함되어 있으며, 해당 버전의 계보를 인터랙티브 그래프로 보여줍니다. 노드는 아티팩트 버전과 런이며, 화살표는 런이 기록한 아티팩트로, 그리고 아티팩트를 사용한 런으로 향합니다. 선택된 버전은 강조 표시됩니다. 큰 그래프의 가독성을 유지하기 위해, 동일한 연결을 가진 유사한 노드가 5개 이상이면(예: 단일 런이 기록한 수많은 체크포인트) 개수를 표시하는 단일 cluster 노드로 그룹화됩니다. 클러스터를 클릭하면 구성 요소를 검색하고 개별 요소를 그래프로 다시 추출할 수 있습니다. 드래그하여 이동하고, Ctrl/Cmd + 스크롤(또는 +/− 버튼)로 확대/축소하며, 노드를 클릭하면 해당 런 또는 아티팩트 버전으로 연결되는 링크가 포함된 요약이 열립니다.

Run detail 페이지는 Output artifacts와 Input artifacts 섹션을 통해 Artifacts 탭으로 되돌아가는 링크를 포함하여, 반대 방향의 동일한 계보를 표시합니다.

원격 저장소

실행이 Hugging Face Space 또는 자체 호스팅 서버를 대상으로 하는 경우(Track 참고), 아티팩트 파일은 콘텐츠 주소 방식으로 저장되며 한 번만 업로드됩니다. Trackio는 서버에 이미 존재하는 블롭을 건너뛰므로 공유 파일을 다시 기록하는 비용이 낮습니다. 아티팩트 메타데이터와 블롭은 설정된 HF Dataset 또는 버킷에 다른 실행 데이터와 함께 영구적으로 저장됩니다.

Update on GitHub