Record runs from your own harness

Recording runs from your own harness gives an existing benchmark runner, simulator, or job in another language the same datasets, summaries, CI gate, and comparisons as Eval; the same recording works for the atlanai SDK’s Eval, pytest, Vitest, the Node.js test runner, and custom harnesses. Use it when you already have a harness and do not want to rewrite it around Eval. Eval performs these same steps internally, so prefer it for new evals: see Run your first eval.

Before you begin

Before you start, make sure you have:

atlanai SDK with tracing: Installed as in Run your first eval, or an OTLP exporter for another language.
Dataset: A record-backed or CSV-backed dataset in the workspace. See Create eval datasets.
API key and workspace ID: A key that can create scorers, experiments, and results in the workspace.

Create scorer and start experiment

Every score needs a scorer ID and version, or its score span is dropped. Create a code scorer once and reuse it. It stores no definition, so give it a new name when its logic changes.start_experiment (startExperiment, or POST /eval/v1/experiments) starts the experiment in running and pins the dataset snapshot. Experiment names are unique per workspace.

my_runner.py

import os
import time
from atlanai import AtlanClient, start_experiment
from atlanai.tracing import init_logger

WORKSPACE = os.environ["ATLAN_WORKSPACE_ID"]
client = AtlanClient(
    os.environ.get("ATLAN_BASE_URL", "https://api.atlan.com"),
    bearer_token=os.environ["ATLAN_API_KEY"],
    workspace=WORKSPACE,
)
logger = init_logger(project_name="my-harness")
tracer = logger.client


def my_harness(question: str) -> str:   # your existing runner
    return "processed" if "refund" in question else "Friday"


found = client.scorers.list(limit=500).items
scorer = next((s for s in found if s.name == "my-harness-exact-match"), None) or client.scorers.create(
    {"workspace_id": WORKSPACE, "name": "my-harness-exact-match", "scorer_kind": "code"})

run = start_experiment(client, "support-agent-cases",
                       {"name": f"my-harness-{int(time.time())}", "config": {"runner": "my-harness", "model": "example-model"}})

Trace and score each case

Run each case inside run.trace() in Python, or propagateAttributes(run.traceOptions, fn) in TypeScript. Every span is then stamped with the experiment ID, so it appears in the experiment’s trace views. Use one root span per case.

items = []
for record in run.experiment.dataset_snapshot.records:
    question, expected = record.input["question"], record.expected["value"]
    with run.trace(), tracer.start_as_current_span("case", as_type="task") as root:
        trace_id = root.trace_id
        root.update(input=question)
        output = my_harness(question)
        root.update(output=output)
        score = float(output == expected)
        root.score_trace("exact_match", value=score,
                         scorer_id=scorer.id, scorer_version=scorer.version_ordinal)
    items.append({
        "workspace_id": WORKSPACE, "name": f"{run.id}-{record.id}",
        "case_id": record.id, "dataset_record_id": record.id,
        "input": record.input, "expected": record.expected, "output": {"value": output},
        "scores": {"exact_match": score}, "trace_id": trace_id,
    })

trace_id: The 32-character lowercase hex ID of that case’s trace. Read it from the root span, as above. Do not invent one.
dataset_record_id: Required on every result of a record-backed experiment, and it must be a record in the pinned snapshot. Omit it for inline cases.
case_id: Unique within the experiment. A second result with the same case_id is refused with 409 duplicate_case_id, which makes an upload retry safe.

Upload results and finalize

Traces export in the background, so flush them before you finalize. Upload at most 100 results per call: each item is accepted or refused on its own, so check every status. Each item’s workspace_id must match the experiment’s workspace, so a result reported for a different workspace is refused. Results are create-only.

logger.flush()
for i in range(0, len(items), 100):
    report = client.experiments.results.create_bulk(run.id, {"items": items[i:i + 100]})
    failed = [r for r in report.items if r.status_code >= 300]
    if failed:
        raise SystemExit(f"{len(failed)} results were refused: {failed[0].error}")

done = client.experiments.update(run.id, {"experiment_status": "completed",
                                          "summary": {"metrics": {"cases": len(items)}}})
print(done.id, done.summary["scores"])

Completing the experiment computes summary.scores and the counts result_count, scored_results, and unscored_results from the results and freezes its status and summary. A result is scored when any of its scores is a finite number, zero included; empty runs report three zero counts. Set experiment_status to failed for a run you abandon.

Verify experiment evidence

To check the setup, complete a small experiment, then inspect its result and confirm each case links to its trace. Traces become queryable a few seconds after flush(). Then confirm the evidence is complete:

from atlanai import verify_experiment

verify_experiment(client, run.id).raise_for_status()

The run is recorded when the check passes and done.summary["scores"] has an entry for each score. If a result is refused, see Eval errors.

Run CSV-backed dataset (optional)

A dataset created from a CSV file pins the file rather than records. Read the pinned version’s rows, run them, and name each result by its case_id column. Do not set dataset_record_id.

import csv
import io
import urllib.request

snapshot = run.experiment.dataset_snapshot          # dataset_id, version_ordinal, file pin
url = (f"{os.environ.get('ATLAN_BASE_URL', 'https://api.atlan.com')}"
       f"/eval/v1/datasets/{snapshot.dataset_id}/versions/{snapshot.version_ordinal}/content")
request = urllib.request.Request(url, headers={
    "Authorization": f"Bearer {os.environ['ATLAN_API_KEY']}",
    "X-Atlan-Workspace-Id": WORKSPACE,
})
rows = list(csv.DictReader(io.StringIO(urllib.request.urlopen(request).read().decode())))

for row in rows:
    ...  # run and trace the case as above, then:
    items.append({"workspace_id": WORKSPACE, "name": f"{run.id}-{row['case_id']}",
                  "case_id": row["case_id"], "input": {"question": row["question"]},
                  "expected": {"value": row["expected"]}, "output": {"value": output},
                  "scores": {"exact_match": score}, "trace_id": trace_id})

Every case_id must exist in the pinned CSV. The CI action and comparisons pair CSV-backed runs on case_id.

Record from another language (optional)

The same steps are plain HTTP. Send Authorization: Bearer <token> and X-Atlan-Workspace-Id on every call, and include workspace_id in each create body.

POST  /eval/v1/experiments                      {workspace_id, name, dataset_id, config}      → 201, experiment_status "running"
      ... run each case; export its spans over OTLP to /otel/v1/traces with the attribute
          atlan.eval.experiment_id = <experiment id> on every span
POST  /eval/v1/experiments/{id}/results/bulk    {items: [{workspace_id, name, case_id, dataset_record_id,
                                                 input, expected, output, scores, trace_id, duration_ms, error}]}
                                               → 207, one {index, status_code, record | error} per item
PATCH /eval/v1/experiments/{id}                 {experiment_status: "completed", summary: {metrics: {...}}}
                                               → 200, with summary.scores computed

scores values must be numbers or null. A result for a completed or failed experiment is refused with 409. To make scores visible in trace statistics too, emit one score span per score: a child span with atlan.span.type = "score" and the attributes atlan.score.name, atlan.score.value, atlan.score.scorer_id, and atlan.score.scorer_version.

Troubleshooting

Upload failures include a trace_id that is not 32 lowercase hex characters, a dataset_record_id outside the pinned snapshot, and a result sent after the experiment finished. See Eval errors.

Next steps

Gate pull requests with evals: Gate pull requests on the experiments your harness records.

Register and manage agents

Give agents business context

Observe agent runs

Evaluate and improve agents

Track agent cost

Record runs from your own harness

Before you begin

Troubleshooting

Next steps

See also