Skip to content

operonx.telemetry.scores

Where experiments, their items and every score live: an eval's checks, a judge's verdicts, a human's review, an online rule's results — one score row with a target and an id derived from what it judges. Backends are the run stores': files (the default), SQLite, and ClickHouse (schema version 3, in the runs' database). The guide: Evals.

scores

Score stores — where experiments, their items, and every score live.

An eval run is an experiment; each case × repeat is an item; each judgement — an eval's check, a judge, a human review, an online rule, a pairwise preference — is a score, one row type with a target (item, trace, op, session, pair) and an id derived from what it judges, so the same judgement written twice is one row. A :class:ScoreStore keeps them on the run stores' backends:

  • files — JSONL files plus a SQLite index (the default, zero setup)
  • sqlite — one file
  • clickhouse — the runs' database, schema version 3

score_store: in resources.yaml declares one; :func:open_score_store opens one from a mapping; :func:project_score_store says which one a project uses ([evals], else its [tracing] sinks). The contract is in :mod:.base.

The contract

ScoreStore

Bases: ABC

See the module docstring.

Methods:

put_experiment abstractmethod 

put_experiment(experiment: Experiment) -> None

Write experiment, replacing any earlier row of its id.

Source code in operonx/telemetry/scores/base.py 
44
45
46
@abstractmethod
def put_experiment(self, experiment: Experiment) -> None:
    """Write *experiment*, replacing any earlier row of its id."""

put_items abstractmethod 

put_items(items: Sequence[ExperimentItem]) -> None

Write items, each replacing its (experiment, case, repeat).

Source code in operonx/telemetry/scores/base.py 
48
49
50
@abstractmethod
def put_items(self, items: Sequence[ExperimentItem]) -> None:
    """Write *items*, each replacing its (experiment, case, repeat)."""

list_experiments abstractmethod 

list_experiments(
    where: Optional[ExperimentFilter] = None,
    limit: int = 50,
    cursor: Optional[str] = None,
) -> ExperimentPage

Experiments matching where, newest first, one page at a time.

Source code in operonx/telemetry/scores/base.py 
52
53
54
55
56
57
58
59
@abstractmethod
def list_experiments(
    self,
    where: Optional[ExperimentFilter] = None,
    limit: int = 50,
    cursor: Optional[str] = None,
) -> ExperimentPage:
    """Experiments matching *where*, newest first, one page at a time."""

get_experiment abstractmethod 

get_experiment(experiment_id: str) -> Optional[ExperimentRecord]

One experiment with its items, or None.

Source code in operonx/telemetry/scores/base.py 
61
62
63
@abstractmethod
def get_experiment(self, experiment_id: str) -> Optional[ExperimentRecord]:
    """One experiment with its items, or ``None``."""

put_scores abstractmethod 

put_scores(scores: Sequence[Score]) -> None

Write scores, each replacing any earlier row of its score_id.

Source code in operonx/telemetry/scores/base.py 
67
68
69
@abstractmethod
def put_scores(self, scores: Sequence[Score]) -> None:
    """Write *scores*, each replacing any earlier row of its ``score_id``."""

scores abstractmethod 

scores(where: Optional[ScoreFilter] = None, limit: int = 10000) -> List[Score]

Scores matching where, oldest first (created_at, score_id). An online score past its retention is not returned.

Source code in operonx/telemetry/scores/base.py 
71
72
73
74
@abstractmethod
def scores(self, where: Optional[ScoreFilter] = None, limit: int = 10000) -> List[Score]:
    """Scores matching *where*, oldest first (``created_at``, ``score_id``).
    An online score past its retention is not returned."""

score_series abstractmethod 

score_series(where: Optional[ScoreFilter], bucket_s: float) -> List[Bucket]

Scores matching where in buckets of bucket_s seconds of created_at, per score name: oldest bucket first.

Source code in operonx/telemetry/scores/base.py 
76
77
78
79
@abstractmethod
def score_series(self, where: Optional[ScoreFilter], bucket_s: float) -> List[Bucket]:
    """Scores matching *where* in buckets of *bucket_s* seconds of
    ``created_at``, per score name: oldest bucket first."""

cache_get abstractmethod 

cache_get(key: str) -> Optional[Dict[str, Any]]

The verdict cached under key, or None (missing or expired).

Source code in operonx/telemetry/scores/base.py 
83
84
85
@abstractmethod
def cache_get(self, key: str) -> Optional[Dict[str, Any]]:
    """The verdict cached under *key*, or ``None`` (missing or expired)."""

cache_put abstractmethod 

cache_put(
    key: str, verdict: Dict[str, Any], ttl_s: Optional[float] = None
) -> None

Cache verdict under key; ttl_s None keeps it until replaced.

Source code in operonx/telemetry/scores/base.py 
87
88
89
@abstractmethod
def cache_put(self, key: str, verdict: Dict[str, Any], ttl_s: Optional[float] = None) -> None:
    """Cache *verdict* under *key*; ``ttl_s`` ``None`` keeps it until replaced."""

refresh 

refresh() -> int

Pick up what other writers left (the files backend reads lines other processes appended). Returns how many rows; most backends have nothing to do.

Source code in operonx/telemetry/scores/base.py 
93
94
95
96
97
def refresh(self) -> int:
    """Pick up what other writers left (the files backend reads lines
    other processes appended). Returns how many rows; most backends
    have nothing to do."""
    return 0

close 

close() -> None

Release connections. Idempotent.

Source code in operonx/telemetry/scores/base.py 
 99
100
def close(self) -> None:
    """Release connections. Idempotent."""

The data

Experiment dataclass 

Experiment(
    experiment_id: str,
    eval: str,
    project: str = "",
    dataset: str = "",
    dataset_version: str = "",
    split: Optional[str] = None,
    graph: str = "",
    code_version: Optional[str] = None,
    version_dirty: Optional[bool] = None,
    graph_hash: str = "",
    config_hash: str = "",
    evaluators_hash: str = "",
    operonx_version: str = "",
    variant: Optional[str] = None,
    repeats: int = 1,
    baseline_id: Optional[str] = None,
    status: str = "running",
    started_at: float = 0.0,
    ended_at: Optional[float] = None,
    cases: int = 0,
    errored: int = 0,
    cost_usd: Optional[float] = None,
    judge_cost_usd: Optional[float] = None,
    p50_ms: Optional[float] = None,
    p95_ms: Optional[float] = None,
    metrics: Dict[str, Any] = dict(),
    gate: Dict[str, Any] = dict(),
    metadata: Dict[str, Any] = dict(),
)

One eval run. experiment_id is the job's run_id; times are epoch seconds; metrics and gate are run.json["eval"]'s.

ExperimentItem dataclass 

ExperimentItem(
    experiment_id: str,
    case_id: str,
    repeat: int = 0,
    case_hash: str = "",
    trace_id: Optional[str] = None,
    status: str = "ok",
    ms: float = 0.0,
    cost_usd: Optional[float] = None,
    tokens_in: int = 0,
    tokens_out: int = 0,
    passed: Optional[bool] = None,
    tags: List[str] = list(),
    cluster: Optional[str] = None,
    output: Any = None,
    error: Optional[str] = None,
)

One case × repeat of an experiment. output is the clipped value the job record holds; cost_usd and tokens are the case run's own.

Score dataclass 

Score(
    score_name: str,
    target: str = "item",
    source: str = "code",
    data_type: str = "bool",
    value: Optional[float] = None,
    passed: Optional[bool] = None,
    label: Optional[str] = None,
    reason: str = "",
    trace_id: Optional[str] = None,
    op_id: Optional[str] = None,
    session_id: Optional[str] = None,
    experiment_id: Optional[str] = None,
    pair_experiment_id: Optional[str] = None,
    case_id: Optional[str] = None,
    repeat: Optional[int] = None,
    origin: str = "",
    name: str = "",
    evaluator_version: str = "",
    judge_trace_id: Optional[str] = None,
    cost_usd: Optional[float] = None,
    author: Optional[str] = None,
    rule: Optional[str] = None,
    queue: Optional[str] = None,
    snapshot: Any = None,
    metadata: Dict[str, Any] = dict(),
    created_at: float = 0.0,
    score_id: str = "",
)

One judgement. target says what was judged and which ids say it (see :data:TARGETS); source who judged. A bool score also holds value 0/1, so means work across sources. score_id and created_at fill themselves in when left empty.

score_id_of 

score_id_of(s: Score) -> str

The id a score's own fields give it — so the same judgement is always the same row:

  • pairwise: (experiment, pair experiment, case, score name);
  • human: (what was judged, score name, author) — an edit replaces;
  • an experiment's item: (experiment, case, repeat, score name);
  • a run, an op, a session (online, rescore), an evaluator (its alignment): (trace, op, session, score name, evaluator version) — a new evaluator is a new score, the same one's record is replaced.
Source code in operonx/telemetry/scores/model.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def score_id_of(s: Score) -> str:
    """The id a score's own fields give it — so the same judgement is
    always the same row:

    * pairwise: (experiment, pair experiment, case, score name);
    * human: (what was judged, score name, author) — an edit replaces;
    * an experiment's item: (experiment, case, repeat, score name);
    * a run, an op, a session (online, rescore), an evaluator (its
      alignment): (trace, op, session, score name, evaluator version) — a
      new evaluator is a new score, the same one's record is replaced.
    """
    if s.target == "pair":
        return _sha("pair", s.experiment_id, s.pair_experiment_id, s.case_id, s.score_name)
    ids = (s.target, s.trace_id, s.op_id, s.session_id, s.experiment_id, s.case_id, s.repeat)
    if s.source == "human":
        return _sha("human", *ids, s.score_name, s.author)
    if s.target == "item":
        return _sha("item", s.experiment_id, s.case_id, s.repeat or 0, s.score_name)
    return _sha(s.target, s.trace_id, s.op_id, s.session_id, s.score_name, s.evaluator_version)

ExperimentRecord dataclass 

ExperimentRecord(experiment: Experiment, items: List[ExperimentItem] = list())

One experiment and its items (by case, then repeat).

ExperimentPage dataclass 

ExperimentPage(
    items: List[Experiment],
    next_cursor: Optional[str] = None,
    total: Optional[int] = None,
)

One page of experiments, newest first; next_cursor None at the end.

ExperimentFilter dataclass 

ExperimentFilter(
    project: Optional[str] = None,
    eval: Optional[str] = None,
    dataset: Optional[str] = None,
    status: Optional[str] = None,
    code_version: Optional[str] = None,
    since: Optional[float] = None,
    until: Optional[float] = None,
    experiment_ids: Optional[Sequence[str]] = None,
)

Which experiments — a small data object, as :class:RunFilter is. Times are epoch seconds over started_at, since inclusive.

ScoreFilter dataclass 

ScoreFilter(
    experiment_id: Optional[str] = None,
    trace_id: Optional[str] = None,
    op_id: Optional[str] = None,
    session_id: Optional[str] = None,
    case_id: Optional[str] = None,
    score_name: Optional[str] = None,
    origin: Optional[str] = None,
    name: Optional[str] = None,
    source: Optional[str] = None,
    target: Optional[str] = None,
    rule: Optional[str] = None,
    since: Optional[float] = None,
    until: Optional[float] = None,
    score_ids: Optional[Sequence[str]] = None,
)

Which scores. Times are epoch seconds over created_at.

Bucket dataclass 

Bucket(
    start: float,
    score_name: str,
    n: int,
    mean: Optional[float],
    passed: Optional[float],
)

One time bucket of one score: how many, their mean value, and the share that passed (of those with a passed).

Configuration

ScoreStoreConfig

Bases: YamlModel

YAML-configurable :class:ScoreStore.

open_score_store 

open_score_store(spec: Optional[Dict[str, Any]] = None) -> ScoreStore

A store from a plain mapping (the YAML block's fields).

Source code in operonx/telemetry/scores/config.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def open_score_store(spec: Optional[Dict[str, Any]] = None) -> ScoreStore:
    """A store from a plain mapping (the YAML block's fields)."""
    spec = dict(spec or {})
    backend = str(spec.get("backend") or "files")
    ttl = float(spec.get("online_ttl_days") or ONLINE_TTL_DAYS)
    if backend == "files":
        from .files import FilesScoreStore

        return FilesScoreStore(root=spec.get("root") or "", online_ttl_days=ttl)
    if backend == "sqlite":
        from .sqlite import SqliteScoreStore

        return SqliteScoreStore(path=spec.get("path") or "", online_ttl_days=ttl)
    if backend == "clickhouse":
        from operonx.telemetry.runs.config import _flag

        from .clickhouse import ClickHouseScoreStore

        if not spec.get("host"):
            raise ValueError("score_store backend 'clickhouse' needs host")
        return ClickHouseScoreStore(
            host=spec["host"],
            port=int(spec.get("port") or 0),
            user=spec.get("user") or "default",
            password=spec.get("password") or "",
            database=spec.get("database") or "operonx",
            secure=_flag(spec.get("secure")),
            timeout=float(spec.get("timeout") or 10.0),
            online_ttl_days=ttl,
        )
    raise ValueError(f"unknown score_store backend {backend!r}; one of {', '.join(BACKENDS)}")

Backends

FilesScoreStore 

FilesScoreStore(
    root: Any = "",
    refresh_every: float = 2.0,
    online_ttl_days: float = ONLINE_TTL_DAYS,
)

Bases: SqliteScoreStore

See the module docstring. root unset → <runs root>/scores ($OPERONX_RUNS_DIR, else <project>/.operonx/runs); reads look for other writers' lines at most every refresh_every seconds.

Source code in operonx/telemetry/scores/files.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    root: Any = "",
    refresh_every: float = 2.0,
    online_ttl_days: float = ONLINE_TTL_DAYS,
):
    self.root = Path(root) if root else resolve_root("") / "scores"
    if not self.root.is_absolute():
        self.root = resolve_root("") / self.root
    self.root.mkdir(parents=True, exist_ok=True)
    self.refresh_every = float(refresh_every)
    self.online_ttl_days = float(online_ttl_days)
    self.path = self.root / _INDEX
    self.index = ScoreIndex(self.path)
    self._append_lock = threading.Lock()
    self._last_refresh = 0.0
    self.refresh()

Methods:

refresh 

refresh() -> int

Index lines appended since the last read; return how many rows.

Source code in operonx/telemetry/scores/files.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def refresh(self) -> int:
    """Index lines appended since the last read; return how many rows."""
    known = self.index.offsets()
    added = 0
    for rel, kind in self._files():
        path = self.root / rel
        start = known.get(rel, 0)
        try:
            size = path.stat().st_size
        except OSError:
            continue
        if size < start:  # rewritten from scratch: read it all
            start = 0
        if size == start:
            continue
        with path.open("rb") as fh:
            fh.seek(start)
            data = fh.read(size - start)
        end = data.rfind(b"\n")
        if end < 0:
            continue  # a line still being written
        rows = self._parse(kind, data[: end + 1])
        self.index.upsert(kind, rows)
        self.index.set_offset(rel, start + end + 1)
        added += len(rows)
    self._last_refresh = time.monotonic()
    return added

SqliteScoreStore 

SqliteScoreStore(path: Any = '', online_ttl_days: float = ONLINE_TTL_DAYS)

Bases: ScoreStore

See the module docstring. path defaults to scores.sqlite under the runs root (<project>/.operonx/runs); an online score (one with a rule) is kept online_ttl_days.

Source code in operonx/telemetry/scores/sqlite.py 
32
33
34
35
36
37
def __init__(self, path: Any = "", online_ttl_days: float = ONLINE_TTL_DAYS):
    self.path = Path(path) if path else resolve_root("") / "scores.sqlite"
    if not self.path.is_absolute():
        self.path = resolve_root("") / self.path
    self.online_ttl_days = float(online_ttl_days)
    self.index = ScoreIndex(self.path)

ClickHouseScoreStore 

ClickHouseScoreStore(
    host: str = "localhost",
    port: int = 0,
    user: str = "default",
    password: str = "",
    database: str = "operonx",
    secure: bool = False,
    timeout: float = 10.0,
    online_ttl_days: float = ONLINE_TTL_DAYS,
    client: Any = None,
)

Bases: ClickHouseConnection, ScoreStore

See the module docstring. The connection fields are the run store's (host, port, user, password, database, secure, timeout); client= takes a ready client (tests hand in a fake).

Source code in operonx/telemetry/scores/clickhouse.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def __init__(
    self,
    host: str = "localhost",
    port: int = 0,
    user: str = "default",
    password: str = "",
    database: str = "operonx",
    secure: bool = False,
    timeout: float = 10.0,
    online_ttl_days: float = ONLINE_TTL_DAYS,
    client: Any = None,
):
    self._connect_init(
        host=host,
        port=port,
        user=user,
        password=password,
        database=database,
        secure=secure,
        timeout=timeout,
        client=client,
    )
    self.online_ttl_days = float(online_ttl_days)