Metadata-Version: 2.4
Name: zvondb
Version: 0.6.0
Summary: Python SDK for the ML Artifact Registry
Author-email: Zvonimir Bednarcik <zvonimir.bednarcik@prosus.com>
License-Expression: MIT
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.0.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0.0; extra == "dev"
Requires-Dist: pytest-httpx>=0.34.0; extra == "dev"
Requires-Dist: ruff>=0.8.0; extra == "dev"
Dynamic: license-file

# zvondb

Python SDK for the ML Artifact Registry — track artifacts, their lineage, and versions across ML experiments.

## Installation

```bash
pip install zvondb
```

## Quick start

```python
from zvondb import Registry

with Registry(project="my-project") as reg:
    exp = reg.ensure_experiment("my-experiment", parent="baseline")

    with exp.start_job(name="train", git_sha="abc123") as job:
        # Resolve input artifacts (walks parent chain)
        data = job.resolve("TrainingData")
        print(f"Using {data.name} v{data.version} from {data.path}")

        # ... train your model ...

        # Register output artifact
        job.create(
            "TrainedModel",
            type="model",
            storage_name="s3",
            description="Model trained on latest data",
            metadata={"val_loss": 0.15},
            tags=["v1"],
        )
        # Or use create_with_info() to get version and other details
        artifact = job.create_with_info(
            "TrainedModel",
            type="model",
            storage_name="s3",
            description="Model trained on latest data",
            metadata={"val_loss": 0.15},
        )
        print(f"Created {artifact.name} v{artifact.version} at {artifact.path}")

        job.complete()
```

## Configuration

| Environment variable | Constructor param | Default | Description |
|---|---|---|---|
| `ZVONDB_URL` | `base_url` | `http://localhost:8000` | Registry server URL |
| `ZVONDB_API_KEY` | `api_key` | `""` | API authentication token |
| `ZVONDB_TIMEOUT` | `timeout` | `30` | HTTP timeout in seconds; raise it for jobs that ship rows in bulk |

## Key concepts

- **Registry** — entry point; manages projects and experiments
- **Experiment** — a branch of work that can inherit artifacts from a parent experiment
- **Job** — a unit of work that consumes and produces artifacts
- **Artifact** — a versioned, typed piece of data with metadata, tags, and lineage

## Registering external data

```python
exp = reg.get_experiment("baseline")
exp.register("TrainingData", type="dataset", path="s3://bucket/data/v3", description="External dataset")

# Jobs can now resolve it
with exp.start_job(name="train", git_sha="abc") as job:
    data = job.resolve("TrainingData")
```

## Skipping work that is already done

Check for an artifact instead of catching an exception. `artifact_exists()` never
records a lineage edge, so asking the question does not add a fake dependency to
your job:

```python
with exp.start_job(name="build-embeddings", git_sha="abc") as job:
    if job.artifact_exists("EmbeddingSet"):
        job.complete()  # nothing to do
    else:
        path = job.create("EmbeddingSet", type="embeddings", storage_name="s3")
        write_embeddings_to(path)
        job.complete()
```

Or let `get_or_create()` make the step idempotent in one call:

```python
artifact, created = job.get_or_create("EmbeddingSet", type="embeddings", storage_name="s3")
if created:
    write_embeddings_to(artifact.path)
```

`Experiment` has `artifact_exists()` and `find_artifact()` too, so you can decide
whether to start a job at all. See the [SDK reference](../docs/sdk.md#artifact_existsname-versionnone-inheritedtrue---bool).

## Benchmark metrics

`log_metrics()` takes plain floats, and that keeps working unchanged. When an
evaluation also produces an error bar, a sample count, or a breakdown by cohort,
record those too — they are what make a difference between two models judgeable
rather than merely visible:

```python
from zvondb import MetricValue

# Value with its uncertainty.
job.log_metrics({"recall@10": MetricValue(0.4213, sem=0.0044, n=12909)})

# The same metric, measured per cohort.
job.log_metrics(evaluate(seen), slice="history_seen")
job.log_metrics(evaluate(unseen), slice="history_unseen")
```

Declare once per project how a key should be shown and which direction wins, so
dashboards can format it and mark the best result:

```python
from zvondb import MetricDefinitionInfo

reg.define_metrics([
    MetricDefinitionInfo(key="recall@10", label="Recall@10", fmt="pct", best="max"),
    MetricDefinitionInfo(key="latency_p99", label="p99 latency", fmt="ms", best="min"),
    # A harness that reports 75.4 rather than 0.754. Everything is stored over
    # 0..1, including what is already there, so the two can share a column.
    MetricDefinitionInfo(key="immlu/acc", scale=100, best="max"),
])
```

Declare which job settings decide whether two results can be compared, so a
dashboard warns instead of quietly putting incomparable numbers side by side:

```python
from zvondb import ParamDefinitionInfo

reg.define_params([
    ParamDefinitionInfo(key="scoring", comparability_critical=True),
    ParamDefinitionInfo(key="chat_template", comparability_critical=True),
])

# Values are ordinary job metadata; nothing about logging changes.
with exp.start_job(
    name="evaluate",
    git_sha=sha,
    metadata={"scoring": "raw_max_delta", "chat_template": True, "iteration": 509},
) as job:
    job.log_metrics({"recall@10": 0.41})
```

Recording a step or token count in metadata is also what lets a dashboard plot a
metric as a curve across checkpoints.

See the [SDK reference](../docs/sdk.md#log_metricsmetrics-sliceall) for details.

## W&B integration

```python
with exp.start_job(name="train", git_sha="abc123") as job:
    job.inject_wandb_env()  # Sets WANDB_API_KEY, WANDB_PROJECT, WANDB_RUN_ID etc.
    # Tracking URL auto-set to https://wandb.ai/{entity}/{project}/runs/{job_id}

    # Frameworks like Axolotl that respect W&B env vars just work.
    # For manual usage:
    import wandb
    wandb.init()
    wandb.log({"loss": 0.5})
```

## License

MIT
