vertebrae.monitoring

Framework-neutral monitoring for repeated labeled benchmark evaluations.

Classes

`EvaluationContext`	Normalized caller-provided coordinates for one benchmark evaluation.
`EvaluationHistoryConfig`	Storage settings for representation-monitoring history.
`EvaluationRecord`	One committed monitoring evaluation and its operational summary rows.
`EvaluationReporter`	Observer protocol invoked after a monitoring record is committed.
`ConsoleReporter`	Print compact monitoring results without third-party console dependencies.
`EvaluationHistory`	Memory- or local JSONL-backed representation-monitoring history.
`RepresentationMonitor`	Repeatedly run a fresh labeled benchmark over live extractor objects.

Module Contents

class vertebrae.monitoring.EvaluationContext[source]

Normalized caller-provided coordinates for one benchmark evaluation.

identity_payload()[source]

Return the explicitly populated fields used for duplicate detection.

Return type:: Dict[str, Any]

identity_signature()[source]

Return a deterministic signature for the evaluation coordinates.

Return type:: str

to_dict()[source]

Serialize the context to strict JSON-compatible data.

Return type:: Dict[str, Any]

classmethod from_dict(value)[source]

Restore a context from persisted history data.

Parameters:: value (Mapping[str, Any])
Return type:: EvaluationContext

class vertebrae.monitoring.EvaluationHistoryConfig[source]: Storage settings for representation-monitoring history.

class vertebrae.monitoring.EvaluationRecord[source]

One committed monitoring evaluation and its operational summary rows.

to_dict()[source]

Serialize this record to strict JSON-compatible data.

Return type:: Dict[str, Any]

classmethod from_dict(value)[source]

Restore a record from persisted history data.

Parameters:: value (Mapping[str, Any])
Return type:: EvaluationRecord

class vertebrae.monitoring.EvaluationReporter[source]

Bases: Protocol

Observer protocol invoked after a monitoring record is committed.

report(record)[source]

Report one committed evaluation.

Parameters:: record (EvaluationRecord)
Return type:: None

class vertebrae.monitoring.ConsoleReporter(stream=None, precision=4)[source]

Print compact monitoring results without third-party console dependencies.

Parameters:

stream (Optional[TextIO])
precision (int)

report(record)[source]

Print one compact line per result row.

Parameters:: record (EvaluationRecord)
Return type:: None

class vertebrae.monitoring.EvaluationHistory(config=None, *, monitor_metadata=None, _read_only=False)[source]

Memory- or local JSONL-backed representation-monitoring history.

Parameters:

config (Optional[EvaluationHistoryConfig])
monitor_metadata (Optional[Mapping[str, Any]])
_read_only (bool)

property detail: str[source]

Return the configured history detail level.

Return type:: str

property next_evaluation_index: int[source]

Return the index that must be assigned to the next record.

Return type:: int

contains_context(context)[source]

Return whether these evaluation coordinates were already recorded.

Parameters:: context (EvaluationContext)
Return type:: bool

append(record)[source]

Commit one record to the configured history.

Parameters:: record (EvaluationRecord)
Return type:: None

iter_records()[source]

Iterate records, streaming disk-backed histories.

Return type:: Iterator[EvaluationRecord]

to_dataframe()[source]

Return one tidy row per evaluation result variant.

Return type:: Any

latest_dataframe()[source]

Return rows from only the most recently committed evaluation.

Return type:: Any

classmethod load(path)[source]

Open an existing disk history for read-only inspection.

Parameters:: path (Union[str, pathlib.Path])
Return type:: EvaluationHistory

class vertebrae.monitoring.RepresentationMonitor(dataset, extractors, *, history_config=None, reporters=(), error_policy='raise', cache_config=None, **benchmark_options)[source]

Repeatedly run a fresh labeled benchmark over live extractor objects.

Parameters:

dataset (Any)
extractors (Iterable[Any])
history_config (Optional[EvaluationHistoryConfig])
reporters (Iterable[EvaluationReporter])
error_policy (str)
cache_config (Optional[vertebrae.config.CacheConfig])
benchmark_options (Any)

evaluate(*, snapshot_id=None, epoch=None, global_step=None, timestamp=None, checkpoint=None, metadata=None)[source]

Run and record one fresh benchmark evaluation.

Parameters:

snapshot_id (Optional[str])
epoch (Optional[int])
global_step (Optional[int])
timestamp (Optional[Union[str, datetime.datetime]])
checkpoint (Optional[Union[str, pathlib.Path]])
metadata (Optional[Mapping[str, Any]])

Return type:

Optional[vertebrae.results.BenchmarkResult]