Store¶
Persistence of traces, experiments and evaluation results (ADR 0003, FR-T4).
Typical use:
import agentlens
from agentlens.store import SQLiteStore
with SQLiteStore("runs.db") as store:
with agentlens.run("support-agent", input="hi", sink=store.save_trace):
...
recent = store.list_traces(name="support-agent", limit=10)
export_jsonl(store, "fixtures/support.jsonl", name="support-agent")
MetadataScalar
module-attribute
¶
Values accepted by the metadata filter of Store.list_traces.
SCHEMA_VERSION
module-attribute
¶
The schema version this AgentLens writes.
DEFAULT_PATH
module-attribute
¶
Where the CLI keeps its database unless told otherwise (project-local).
SchemaVersionError ¶
Bases: StoreError
The database was written by a newer AgentLens with an unknown schema version.
Store ¶
Bases: Protocol
Persistence for traces, experiments and evaluation results.
Saving is an upsert: saving an object whose id already exists replaces it, so re-running a step (e.g. re-evaluating a trace) is idempotent.
save_trace ¶
save_trace(trace: Trace, *, experiment_id: str | None = None) -> None
Persist trace, optionally linking it to an existing experiment.
Usable directly as a tracing sink: agentlens.run(..., sink=store.save_trace).
Re-saving a trace without experiment_id keeps its existing link.
get_trace ¶
get_trace(trace_id: str) -> Trace | None
The trace with id trace_id, or None if it is not stored.
list_traces ¶
list_traces(*, name: str | None = None, experiment_id: str | None = None, metadata: Mapping[str, MetadataScalar] | None = None, has_error: bool | None = None, limit: int | None = None, offset: int = 0) -> list[Trace]
Stored traces matching all given filters, most recent first.
metadata matches traces whose metadata has each key with exactly that
scalar value ({"case_id": "refund-simple", "repetition": 0}).
has_error=True keeps the traces where the run or one of its spans
recorded an error (False: the others). offset skips that many
matches before limit applies (pagination).
save_experiment ¶
save_experiment(experiment: Experiment) -> None
Persist experiment (upsert by id).
get_experiment ¶
get_experiment(experiment_id: str) -> Experiment | None
The experiment with id experiment_id, or None.
list_experiments ¶
list_experiments(*, name: str | None = None, suite: str | None = None, is_baseline: bool | None = None) -> list[Experiment]
Stored experiments matching all given filters, most recent first.
save_eval_results ¶
save_eval_results(trace_id: str, results: Sequence[EvalResult]) -> None
Replace the evaluation results of a stored trace with results.
get_eval_results ¶
get_eval_results(trace_id: str) -> list[EvalResult]
The evaluation results of one trace, in the order they were saved.
list_eval_results ¶
list_eval_results(experiment_id: str) -> dict[str, list[EvalResult]]
Evaluation results of every trace in an experiment, keyed by trace id.
StoreError ¶
Bases: Exception
A store operation failed (e.g. a reference to an unknown experiment).
JSONLFormatError ¶
Bases: StoreError
A line of a JSONL file is not a valid trace.
The message names the file and the (1-based) line number.
SQLiteStore ¶
SQLiteStore(path: str | PathLike[str] = DEFAULT_PATH)
A Store backed by a SQLite database file.
Objects are stored as JSON documents next to a few indexed columns used for
filtering, so reading a trace back yields exactly the trace that was saved.
The schema is created/upgraded on open (agentlens.store.migrations).
One connection is shared behind a lock, so a store can be used from several threads (e.g. as the sink of runs executed in a thread pool). It is not meant for many concurrent writer processes (ADR 0003).
Example:
with SQLiteStore("runs.db") as store, agentlens.run("demo", sink=store.save_trace):
...
save_trace ¶
save_trace(trace: Trace, *, experiment_id: str | None = None) -> None
Persist trace (upsert), optionally linking it to an experiment.
Raises StoreError if experiment_id is not a stored experiment.
list_traces ¶
list_traces(*, name: str | None = None, experiment_id: str | None = None, metadata: Mapping[str, MetadataScalar] | None = None, has_error: bool | None = None, limit: int | None = None, offset: int = 0) -> list[Trace]
Traces matching all given filters, most recent first.
has_error=True keeps traces where the run or any of its spans
recorded an error (False: neither did). offset skips that many
matches first (pagination together with limit).
save_experiment ¶
save_experiment(experiment: Experiment) -> None
Persist experiment (upsert by id).
get_experiment ¶
get_experiment(experiment_id: str) -> Experiment | None
The experiment with id experiment_id, or None.
list_experiments ¶
list_experiments(*, name: str | None = None, suite: str | None = None, is_baseline: bool | None = None) -> list[Experiment]
Experiments matching all given filters, most recent first.
save_eval_results ¶
save_eval_results(trace_id: str, results: Sequence[EvalResult]) -> None
Replace the evaluation results of a stored trace.
Raises StoreError if the trace is not stored.
get_eval_results ¶
get_eval_results(trace_id: str) -> list[EvalResult]
The evaluation results of one trace, in saved order (empty if none).
list_eval_results ¶
list_eval_results(experiment_id: str) -> dict[str, list[EvalResult]]
Evaluation results of every evaluated trace of an experiment, by trace id.
export_jsonl ¶
export_jsonl(store: Store, path: str | PathLike[str], *, name: str | None = None, experiment_id: str | None = None, metadata: Mapping[str, MetadataScalar] | None = None, limit: int | None = None) -> int
Export the traces of store matching the filters to path; return how many.
The filters are those of Store.list_traces (limit keeps the most
recent ones). Traces are written oldest first, so files exported at different
times diff cleanly and read like a timeline.
import_jsonl ¶
import_jsonl(store: Store, path: str | PathLike[str], *, experiment_id: str | None = None) -> int
Import every trace of a JSONL file into store; return how many.
The whole file is validated before anything is saved, so a malformed line
(JSONLFormatError) leaves the store untouched. Saving is an upsert:
importing the same file twice is a no-op the second time. experiment_id
links all imported traces to that (already stored) experiment.
read_jsonl ¶
read_jsonl(path: str | PathLike[str]) -> Iterator[Trace]
Yield the traces stored in a JSONL file, in file order.
Blank lines are skipped. A line that is not a valid trace raises
JSONLFormatError (when the iterator reaches it).
write_jsonl ¶
write_jsonl(traces: Iterable[Trace], path: str | PathLike[str]) -> int
Write traces to path, one JSON document per line; return how many.
The file is written to a temporary sibling and then atomically renamed, so an existing file is never left half-overwritten if serialisation fails midway. Missing parent directories are created.