DVBench client¶

dvbench Python client.

Client ¶

Client(endpoint: str | None = None, *, silent: bool = False)

Source code in dvbench/client.py

def __init__(
    self,
    endpoint: str | None = None,
    *,
    silent: bool = False,
) -> None:
    self.endpoint = (endpoint or DEFAULT_API_ENDPOINT).rstrip("/")
    self.silent = silent
    self._session = requests.Session()
    self._upload_sessions = [requests.Session() for _ in range(8)]
    self._available_upload_sessions: SimpleQueue[requests.Session] = SimpleQueue()
    for session in self._upload_sessions:
        self._available_upload_sessions.put(session)
    self._key = Ed25519PrivateKey.generate()
    self._cache = tempfile.TemporaryDirectory(prefix="dvbench-vectors-")
    self._uploaded_vectors: set[str] = set()

create_benchmark ¶

create_benchmark(name: str, description: str, measurement_ids: Iterable[int], key_measurement_id: int, *, sort_direction: Literal['asc', 'desc'] = 'desc', long_description: str | None = None) -> dict[str, Any]

Create a leaderboard from existing measurement definitions.

Source code in dvbench/client.py

def create_benchmark(
    self,
    name: str,
    description: str,
    measurement_ids: Iterable[int],
    key_measurement_id: int,
    *,
    sort_direction: Literal["asc", "desc"] = "desc",
    long_description: str | None = None,
) -> dict[str, Any]:
    """Create a leaderboard from existing measurement definitions."""
    body: dict[str, Any] = {
        "name": name.strip(),
        "description": description.strip(),
        "measurement_ids": list(measurement_ids),
        "key_measurement_id": key_measurement_id,
        "sort_direction": sort_direction,
    }
    if long_description is not None:
        body["long_description"] = long_description.strip()
    return self._post(
        "benchmarks",
        body,
    )

define_measurement ¶

define_measurement(name: str, description: str, *, long_description: str | None = None) -> dict[str, Any]

Create or update a measurement definition.

Source code in dvbench/client.py

def define_measurement(
    self,
    name: str,
    description: str,
    *,
    long_description: str | None = None,
) -> dict[str, Any]:
    """Create or update a measurement definition."""
    body = {"name": name.strip(), "description": description.strip()}
    if long_description is not None:
        body["long_description"] = long_description.strip()
    return self._post(
        "measurement-definitions",
        body,
    )

measurement_status ¶

measurement_status(request_id: str) -> dict[str, Any] | None

Return the worker result, or None while the request is still in flight.

Source code in dvbench/client.py

def measurement_status(self, request_id: str) -> dict[str, Any] | None:
    """Return the worker result, or None while the request is still in flight."""
    response = self._session.get(f"{self.endpoint}/measurements/{request_id}")
    if response.status_code == 404:
        return None
    response.raise_for_status()
    response_body = response.json()
    payload = response_body.get("payload")
    if not isinstance(payload, dict):
        raise ValueError("dvbench measurement status response has an invalid shape")
    return payload

DVBenchProof ¶

Add upload and cleanup behavior to a concrete ZKDV proof.

Experiment.proof combines this mixin with the selected backend's ZKDV class. The returned object must itself be the ZKDV driver: integrations such as zkdv.jax.contrib.TrainState use object identity to ensure that compiled and applied updates belong to the same proof.

Experiment ¶

Experiment(record: dict[str, Any], client: Client)

Source code in dvbench/experiment.py

def __init__(self, record: dict[str, Any], client: Client) -> None:
    self.id = int(record["id"])
    self.name = str(record["name"])
    self.client = client
    self.checkpoints = list(record.get("checkpoints", []))
    self._proof: DVBenchProof | None = None

measure_opening ¶

measure_opening(inner_root: str, *, transcript_id: str | None = None) -> Measurement

Measure a checkpoint with a compact projection opening computed externally.

Source code in dvbench/experiment.py

def measure_opening(
    self,
    inner_root: str,
    *,
    transcript_id: str | None = None,
) -> Measurement:
    """Measure a checkpoint with a compact projection opening computed externally."""
    sketch = self.client.opening(self.checkpoints, inner_root, transcript_id)
    return Measurement(self.client, self.id, sketch)

client ¶

HTTP client for the dvbench API.

Client ¶

Client(endpoint: str | None = None, *, silent: bool = False)

Source code in dvbench/client.py

def __init__(
    self,
    endpoint: str | None = None,
    *,
    silent: bool = False,
) -> None:
    self.endpoint = (endpoint or DEFAULT_API_ENDPOINT).rstrip("/")
    self.silent = silent
    self._session = requests.Session()
    self._upload_sessions = [requests.Session() for _ in range(8)]
    self._available_upload_sessions: SimpleQueue[requests.Session] = SimpleQueue()
    for session in self._upload_sessions:
        self._available_upload_sessions.put(session)
    self._key = Ed25519PrivateKey.generate()
    self._cache = tempfile.TemporaryDirectory(prefix="dvbench-vectors-")
    self._uploaded_vectors: set[str] = set()

create_benchmark ¶

create_benchmark(name: str, description: str, measurement_ids: Iterable[int], key_measurement_id: int, *, sort_direction: Literal['asc', 'desc'] = 'desc', long_description: str | None = None) -> dict[str, Any]

Create a leaderboard from existing measurement definitions.

Source code in dvbench/client.py

def create_benchmark(
    self,
    name: str,
    description: str,
    measurement_ids: Iterable[int],
    key_measurement_id: int,
    *,
    sort_direction: Literal["asc", "desc"] = "desc",
    long_description: str | None = None,
) -> dict[str, Any]:
    """Create a leaderboard from existing measurement definitions."""
    body: dict[str, Any] = {
        "name": name.strip(),
        "description": description.strip(),
        "measurement_ids": list(measurement_ids),
        "key_measurement_id": key_measurement_id,
        "sort_direction": sort_direction,
    }
    if long_description is not None:
        body["long_description"] = long_description.strip()
    return self._post(
        "benchmarks",
        body,
    )

define_measurement ¶

define_measurement(name: str, description: str, *, long_description: str | None = None) -> dict[str, Any]

Create or update a measurement definition.

Source code in dvbench/client.py

def define_measurement(
    self,
    name: str,
    description: str,
    *,
    long_description: str | None = None,
) -> dict[str, Any]:
    """Create or update a measurement definition."""
    body = {"name": name.strip(), "description": description.strip()}
    if long_description is not None:
        body["long_description"] = long_description.strip()
    return self._post(
        "measurement-definitions",
        body,
    )

measurement_status ¶

measurement_status(request_id: str) -> dict[str, Any] | None

Return the worker result, or None while the request is still in flight.

Source code in dvbench/client.py

def measurement_status(self, request_id: str) -> dict[str, Any] | None:
    """Return the worker result, or None while the request is still in flight."""
    response = self._session.get(f"{self.endpoint}/measurements/{request_id}")
    if response.status_code == 404:
        return None
    response.raise_for_status()
    response_body = response.json()
    payload = response_body.get("payload")
    if not isinstance(payload, dict):
        raise ValueError("dvbench measurement status response has an invalid shape")
    return payload

experiment ¶

Experiment training and measurement lifecycle.

Experiment ¶

Experiment(record: dict[str, Any], client: Client)

Source code in dvbench/experiment.py

def __init__(self, record: dict[str, Any], client: Client) -> None:
    self.id = int(record["id"])
    self.name = str(record["name"])
    self.client = client
    self.checkpoints = list(record.get("checkpoints", []))
    self._proof: DVBenchProof | None = None

measure_opening ¶

measure_opening(inner_root: str, *, transcript_id: str | None = None) -> Measurement

Measure a checkpoint with a compact projection opening computed externally.

Source code in dvbench/experiment.py

def measure_opening(
    self,
    inner_root: str,
    *,
    transcript_id: str | None = None,
) -> Measurement:
    """Measure a checkpoint with a compact projection opening computed externally."""
    sketch = self.client.opening(self.checkpoints, inner_root, transcript_id)
    return Measurement(self.client, self.id, sketch)

measurement ¶

Checkpoint-bound measurement submission.

trace ¶

Stateless evaluation and aggregation source tracing.

vector ¶

Canonical local vector objects uploaded with measurement submissions.

proof ¶

Lifecycle shared by the concrete JAX and Torch dvbench proofs.

DVBenchProof ¶

Add upload and cleanup behavior to a concrete ZKDV proof.

Experiment.proof combines this mixin with the selected backend's ZKDV class. The returned object must itself be the ZKDV driver: integrations such as zkdv.jax.contrib.TrainState use object identity to ensure that compiled and applied updates belong to the same proof.

sketch ¶

Incremental checkpoint projection against transcript probe vectors.

uploader ¶

Live streaming for dvbench proofs.