OpenRath v2.0.0 Evaluation, Observability, and Artifacts#
OpenRath v2.0.0 includes three supporting planes around durable execution: content-addressed artifacts, revision-bound evaluation, and correlated telemetry.
+--> ArtifactStore
| durable outputs by SHA-256
Run + revision + trace --+--> EvaluationRunner
| datasets, results, regression gates
+--> Telemetry
spans, counters, structured logs
Each plane keeps tenant, revision, or trace identity explicit instead of embedding operational state in an opaque Agent transcript.
Content-addressed artifacts#
LocalArtifactStore provides an atomic filesystem store for embedded and
single-node use. Objects are tenant scoped and addressed by their SHA-256
digest.
from pathlib import Path
from rath.artifacts import LocalArtifactStore
store = LocalArtifactStore(
Path("openrath-artifacts"),
max_bytes=128 * 1024 * 1024,
)
artifact = store.put(
"tenant-a",
b'{"answer":"durable"}',
media_type="application/json",
metadata={"run_id": "run-1"},
)
print(artifact.uri)
assert artifact.uri == f"artifact://tenant-a/{artifact.digest}"
assert store.get("tenant-a", artifact.digest) == b'{"answer":"durable"}'
assert store.stat("tenant-a", artifact.digest) == artifact
put(...) accepts bytes or a binary file object. Both local and S3 stores
enforce max_bytes; get(...) verifies the content digest before returning
the payload.
For an S3-compatible production store, install the s3 extra:
python -m pip install \
"openrath[s3] @ https://github.com/Rath-Team/OpenRath/releases/download/v2.0.0/openrath-2.0.0-py3-none-any.whl"
Create the store with standard Boto3 client options:
from rath.artifacts import S3ArtifactStore
store = S3ArtifactStore(
"openrath-artifacts",
prefix="openrath",
endpoint_url="https://objects.example.com",
region_name="us-east-1",
max_bytes=128 * 1024 * 1024,
)
The S3 implementation stores the payload and its manifest under a tenant prefix. The manifest records digest, size, media type, creation time, and JSON-safe metadata.
Versioned evaluation#
An evaluation Dataset contains immutable examples. An Experiment binds
the resulting scores to the evaluated deployment revision. Evaluators are
async protocols, so they can inspect a completed durable Run and perform
local or remote checks.
import asyncio
from pathlib import Path
from uuid import uuid4
from rath.eval import (
Dataset,
EvaluationResult,
EvaluationRunner,
Example,
GateDecision,
SQLiteEvaluationStore,
regression_gate,
)
from rath.runtime import Run, RunStatus, SQLiteRunStore
revision_id = uuid4()
dataset = Dataset(
id=uuid4(),
name="uppercase",
version="1",
examples=(
Example.create(
{"text": "openrath"},
{"answer": "OPENRATH"},
),
),
)
class ExactAnswer:
name = "exact-answer"
async def evaluate(self, example, run):
passed = run.state["answer"] == example.expected["answer"]
return EvaluationResult(
evaluator=self.name,
score=1.0 if passed else 0.0,
passed=passed,
reason="actual answer matches expected answer" if passed else "mismatch",
)
async def execute(example):
return Run.create(
plan_id=uuid4(),
revision_id=revision_id,
session_id=uuid4(),
tenant_id="local",
status=RunStatus.SUCCEEDED,
state={"answer": str(example.inputs["text"]).upper()},
)
async def evaluate():
return await EvaluationRunner().run(
dataset,
revision_id=revision_id,
execute=execute,
evaluators=(ExactAnswer(),),
)
experiment = asyncio.run(evaluate())
run_store = SQLiteRunStore(Path("evaluation.db"))
evaluation_store = SQLiteEvaluationStore(run_store)
evaluation_store.save_dataset(dataset)
evaluation_store.save_experiment(experiment)
decision = regression_gate(
experiment,
baseline=experiment,
maximum_regression=0.02,
minimum_score=0.8,
)
assert decision is GateDecision.PASS
print(experiment.mean_score, decision.value)
run_store.close()
regression_gate(...) returns FAIL when the candidate mean falls below
minimum_score or below the baseline by more than maximum_regression.
PostgresEvaluationStore provides the matching production persistence
contract over a PostgresRunStore.
Tracing and counters#
InMemoryTelemetry is useful for embedded diagnostics and tests. Every span
uses an explicit TraceContext, and counters can carry bounded string
attributes.
from rath.context import TraceContext
from rath.observability import InMemoryTelemetry
trace = TraceContext.new()
telemetry = InMemoryTelemetry()
with telemetry.span(
"openrath.run",
context=trace,
attributes={"run.status": "running"},
):
telemetry.increment(
"openrath.run.started",
attributes={"tenant": "tenant-a"},
)
assert telemetry.spans[0].trace_id == trace.trace_id
assert telemetry.spans[0].status == "ok"
For an OpenTelemetry SDK bridge, install the otel extra:
python -m pip install \
"openrath[otel] @ https://github.com/Rath-Team/OpenRath/releases/download/v2.0.0/openrath-2.0.0-py3-none-any.whl"
OpenTelemetry uses the configured OpenTelemetry tracer and meter providers.
Wrap it with GuardedTelemetry when exporter failures must remain isolated
from the application result:
from rath.observability import GuardedTelemetry, OpenTelemetry
telemetry = GuardedTelemetry(
OpenTelemetry(service_name="openrath"),
)