opengrade

Python SDK for OpenGrade: OTLP/HTTP export with optional OpenInference instrumentation, plus a control-plane Client for datasets, experiments, and evaluation.

Install

uv add opengrade-ai
uv add 'opengrade-ai[openai]'
uv add 'opengrade-ai[ui]'   # also installs opengrade-ai-server (API + packaged UI)

Pip remains supported:

pip install opengrade-ai
pip install 'opengrade-ai[openai]'
pip install 'opengrade-ai[ai]'   # all tier-1 OpenInference instrumentors
pip install 'opengrade-ai[ui]'  # also installs opengrade-ai-server (API + packaged UI)

Individual extras: openai, anthropic, langchain, llama-index, bedrock, ui.

Import path remains import opengrade. Local platform (SQLite + UI):

uv add 'opengrade-ai[ui]'
uv run opengrade ui   # http://127.0.0.1:8000 — creates ./opengrade.db

The opengrade command ships with the SDK. Without the [ui] extra it exits with a hint to install it.

Develop from this monorepo:

cd sdk && uv sync
uv sync --extra openai   # add OpenAI instrumentor
# UI packaging (server wheel): python scripts/package_ui.py
# then: cd backend && uv sync --all-extras && uv run opengrade ui

Quickstart

One tracking URI, one active project, one init. A @trace wrapper for your app logic, and provider calls become child spans when the matching instrumentor is installed:

import opengrade
from openai import OpenAI

opengrade.set_tracking_uri("http://localhost:8000")  # or OPENGRADE_TRACKING_URI
opengrade.set_project("demo")  # created if it does not exist
opengrade.init(instrument="auto")  # OTLP → <tracking_uri>/v1/traces

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")


@opengrade.trace
def ask(q: str) -> str:
    r = client.chat.completions.create(
        model="gemma4:e2b",
        messages=[{"role": "user", "content": q}],
    )
    return r.choices[0].message.content or ""


ask("ping")
opengrade.flush()

Runnable examples: examples/sdk_hello/ (Ollama via OpenAI-compatible API) and the guided examples/sdk_tour/ (tracing, RAG verification, architecture comparison, production metrics and alerts; runs offline).

API summary

App code should use these helpers instead of importing OpenTelemetry directly.

Spans inside a function (opengrade.span)

@opengrade.trace covers a whole function; opengrade.span traces one block inside it, so each stage of a pipeline shows up as its own step in the trace timeline:

@opengrade.trace
def answer(question: str) -> str:
    with opengrade.span("retrieve", kind="RETRIEVER", input=question) as s:
        docs = retriever.search(question)
        s.set_output([d.id for d in docs])
    with opengrade.span("rerank", kind="RERANKER", attributes={"rerank.top_k": 3}) as s:
        docs = reranker.rank(question, docs)[:3]
        s.set_output([d.id for d in docs])
    opengrade.set_retrieval_documents(docs)  # still feeds retrieval metrics / judges
    return generate(question, docs)

• kind is an OpenInference span kind: CHAIN (default), LLM, RETRIEVER, RERANKER, EMBEDDING, TOOL, AGENT, GUARDRAIL, EVALUATOR, PROMPT (case-insensitive; other values are sent as-is with a warning).
• The handle has set_input, set_output, set_attribute(s), set_error, trace_id, span_id. set_output on a span does not change the experiment output (use opengrade.set_output for that).
• An exception in the block marks the span ERROR and propagates. async with works the same way.
• Outside any traced call, the span is a root span; with init(live=True) it becomes a live interaction like an outermost @opengrade.trace (live=False opts out).

Users, sessions, tags and metadata

@opengrade.trace
def answer(question: str, user, chat_id: str) -> str:
    opengrade.set_user(user.id)
    opengrade.set_session(chat_id)  # all turns of a conversation share it
    opengrade.set_tags("beta", "rerank-v2")  # accumulates across calls
    opengrade.set_metadata(tenant=user.tenant, prompt_version="v3")
    ...


# Or for every span opened in a block (blocks nest; tags and metadata accumulate):
with opengrade.using_attributes(user_id=user.id, session_id=chat_id, tags=["beta"]):
    answer(question)

• They use the OpenInference keys user.id, session.id, tag.tags and metadata (JSON), so any span of the trace works and auto-instrumented spans inside using_attributes get them too (when openinference-instrumentation is installed).
• The setters act on the current span and are no-ops outside a traced call.
• Metadata values should be scalars (others are stored as JSON text); keys starting with _ are reserved. Limits: 50 tags, 50 metadata keys, 500 chars per value.
• In the UI (Production → Traces) filter with tag:beta, tenant:acme, user:… or session:…; click a tag, user, session or metadata value on a trace to see all matching traces. API: GET /projects/{id}/traces?tag=beta&metadata=tenant:acme&user_id=…&session_id=… (repeat tag / metadata; all must match), GET /projects/{id}/traces/facets for the values seen.

Experiments (evaluate / start_experiment)

One call runs a task over a dataset, writes the outputs, scores them and compares against a baseline:

import opengrade

opengrade.set_project("demo")
opengrade.init()  # optional: adds trace context (tokens, cost, tool calls) to each item


def answer_question(item):  # item.input, item.expected_output, item.metadata
    docs = retrieve(item.input)
    opengrade.set_retrieval_documents(docs)  # feeds hit_at_k / recall_at_k / groundedness
    return generate(item.input, docs)


def short(item, output) -> bool:  # local metric: float | bool | None
    return len(output) < 400


result = opengrade.evaluate(
    dataset="gold-v1",  # name/id, path, list/iterable of dicts, pandas/Polars DataFrame
    task=answer_question,
    metrics=["groundedness", "hit_at_k", short],
    # metrics_set_id="…",  # reuse an existing set (e.g. from_experiment in the CLI); XOR with metrics kinds
    name="gemma-prompt-v2",
    params={"model": "gemma4"},  # stored as the experiment's model_config
    baseline="prod",  # experiment name (latest), id, or "alias:main" (promoted baseline)
    # version="v1.4.0",  # default: short git SHA of the code under test
    # git=False,  # the commit/branch/dirty state is recorded automatically; False skips it
    max_concurrency=8,
    # repeats=3,  # run each item 3 times: per-item mean, flaky items counted
    # group_by="category",  # also report means / baseline deltas per metadata value
)
print(result.summary(), result.url)

# Promote the run as the new baseline (the CLI does this with --promote when the gate passes):
opengrade.get_client().experiments.set_alias("main", result.id, note="accepted")

Manual loop, same plumbing (binding, flush, output upload, scoring on exit):

with opengrade.start_experiment("v2", dataset="gold-v1", metrics=["hit_at_k"]) as exp:
    for item in exp.items():
        with exp.run(item) as run:  # root span bound to (experiment, item)
            run.log(my_app(item.input))  # or opengrade.set_output(...) anywhere inside
print(exp.result.summary())

How it behaves:

• Metrics: a string is a server metric kind (hit_at_k, recall_at_k, mrr, must_contain, groundedness, correctness, answer_relevance, context_precision, context_recall, citation_accuracy, refusal, latency). A dict sets threshold/config, for example {"kind": "hit_at_k", "config": {"k": 3}}. The SDK finds or creates a metrics set that holds exactly those kinds, using the project's defaults for anything you leave out. metrics=None uses the project default set. A Python metric runs locally (see Custom metrics below), and its scores are submitted to the platform (POST /experiments/{id}/client-results). From there they are aggregated, shown in the UI and compared with the baseline like server metrics.
• Datasets: pass a registered name/id, a path to CSV/JSON/JSONL, a list or iterable of dicts, a pandas DataFrame, or a Polars DataFrame. Rows need an input column (or question). expected_output (or expected_answer), context and metadata map to item fields; any other column goes into metadata (so expected_doc_ids and must_contain work). With the default task_type="rag_qa" every row needs expected_output and expected_doc_ids, except rows labelled unanswerable (metadata.answerable: false or expected_refusal: true), which may omit both. A missing input/question raises opengrade.DatasetColumnError (CSV headers are checked up front). Inline rows are hashed into a dataset name, so re-running on the same rows reuses the same dataset. Generators are materialised before upload (no lazy streaming).
• Outputs are uploaded with PUT /experiments/{id}/outputs, so this works without opengrade.init(). With tracing on, finish() waits for the OTLP traces to be bound (up to 30s) and merges their context in. Your explicit output always wins.
• Repeats: repeats=N runs the task N times per item (each run is its own trace and output, repeat 0..N-1). Scores are averaged per item before aggregating, so every item weighs once and the baseline comparison still pairs items; the bootstrap CI then reflects run-to-run noise less. Items whose PASS/FAIL changes between repeats count as flaky (result.flaky, the flaky column in summary(), the UI scorecard). In a manual loop pass exp.run(item, repeat=i). Max 100.
• Groups: group_by="category" (a dataset-item metadata key, dotted paths allowed) adds per-group means and baseline deltas: result.groups and a by category: block in summary(). A drop on one kind of question shows up even when the overall mean holds. client.experiments.summary(id, group_by=...) / compare(..., group_by=...) return the same data; the release gate can enforce it with a groups policy block (see CLI and CI).
• Errors: when a task raises, evaluate records it on the item (result.errors, metadata.error) and moves on. Pass raise_on_error=True to stop instead. When the with block raises, nothing is scored. Bad dataset columns raise DatasetColumnError before the run starts.

Citations and refusals

Two server judges check behaviours that answer-quality scores miss:

opengrade.evaluate(
    dataset=[
        {
            "input": "How many PTO days?",
            "expected_output": "20 days.",
            "metadata": {"expected_doc_ids": ["pto"]},
        },
        # unanswerable from the knowledge base: the assistant should decline. No gold
        # documents and no reference answer needed (retrieval metrics skip the item).
        {"input": "What is the CEO's salary?", "metadata": {"answerable": False}},
    ],
    task=answer_question,  # answers like "You get 20 days [pto]."
    metrics=["citation_accuracy", "refusal"],
)

• citation_accuracy: parses the answer's citation markers ([pto], [1], [1, 3], [^2]; another style via {"kind": "citation_accuracy", "config": {"pattern": r"\(source: ([^)]+)\)"}}) and checks that every cited source was actually retrieved and supports the sentence that cites it. Score = supported cited sentences / cited sentences; an answer citing nothing fails (require_citations: false to skip instead), min_coverage: 0.8 also fails answers where too few sentences carry a citation. Documents come from set_retrieval_documents / retriever spans and need an id.
• refusal: did the assistant decline exactly when it should? Mark items with metadata.answerable (or expected_refusal), or give a reference answer that declines ("The documents do not say"). Without either (e.g. on live traffic) the judge reads the retrieved documents. metadata.outcome is correct_answer, correct_refusal, over_refusal or missed_refusal. {"kind": "refusal", "config": {"method": "phrases"}} matches refusal phrases (English/Italian, or your own refusal_phrases) without an LLM call.

Both run on live traffic too (enable them in the project's metrics set; they ship disabled).

Custom metrics

@opengrade.metric(threshold=0.5, version=2, requires=("documents",))
def cites_source(s: opengrade.Sample) -> opengrade.Score:
    hit = any(d["id"] in s.output for d in s.documents)
    return opengrade.Score(1.0 if hit else 0.0, explanation="cited" if hit else "no citation")


opengrade.evaluate(dataset="gold-v1", task=answer_question, metrics=["hit_at_k", cites_source])

• Sample fields: input, output, expected (dataset expected_output), documents (from set_retrieval_documents), context, metadata, trace_id, item_id.
• Return opengrade.Score(value, label=, explanation=, metadata=), a float, a bool, None or {"score": ...}. The explanation shows per item in the UI; metadata (JSON-serializable, no NaN/inf) is stored with the result. A non-finite score is an ERROR.
• label can only fail, skip or error an item ("FAIL", "SKIPPED", "ERROR"): PASS/FAIL otherwise comes from the threshold, so label="PASS" does not make a score below the threshold pass.
• To change the threshold or name of an existing metric, wrap it again: opengrade.LocalMetric(cites_source, threshold=0.8) keeps its function, requires and config.
• fn(item, output) (two parameters) still works and receives the dataset item. async def metrics work too.
• requires: Sample fields that must be non-empty (output, expected, documents, context). When one is missing the item is SKIPPED, not failed.
• version is stored in the metric config: bump it when the code changes, so compare reports a config mismatch instead of a misleading delta.
• When the task fails on an item, that item counts as FAIL (score 0). When the metric raises, the item is ERROR. A local metric name must not match a server evaluator name (409). If the submission fails, result.metrics falls back to the in-process mean.

Custom LLM judges

When "good" is a quality an LLM has to judge (tone, format, policy, completeness), describe it in plain language instead of writing a prompt and a parser. The server runs the judge, so its scores are stored, shown in the UI, compared with the baseline and gated by the release check like built-in metrics.

tone = opengrade.LlmJudge(
    "tone",
    criteria="The answer is polite and professional, uses no slang and never blames the user.",
    scale="binary",  # or "1-5" (score 0, .25, .5, .75, 1)
    fields=("input", "output"),  # + "expected", "documents", "metadata"
    threshold=0.5,
    # levels={5: "warm and formal", 1: "rude"}, examples=[{"output": "yo", "verdict": "fail"}],
    # model="gpt-4o-mini",  # default: the server's LLM_MODEL
)

print(tone.preview(input="Can I carry over PTO?", output="lol no").explanation)  # try it, nothing stored
opengrade.evaluate(dataset="gold-v1", task=answer_question, metrics=["groundedness", tone])

• The judge is stored as a project evaluator (custom_judge); evaluate reuses it while the config is unchanged and creates a new version when it changes, so different criteria are never compared as the same metric.
• live=True also scores production traffic from init(live=True), next to the built-in judges (a live judge cannot read expected). evaluate registers the judge; for production only, call tone.register() once (e.g. at startup). You can then alert on its score_fail_rate / score_mean.
• Missing inputs (no answer, no reference for expected, no documents for documents) SKIP the item. Judge failures are ERROR with the reason.
• UI: Metrics → Custom judges (with a "Grade sample" preview), and "Add a custom judge" on a metrics set.

Params and run metrics

with opengrade.start_experiment("v2", dataset="gold-v1", params={"model": "gemma4"}) as exp:
    exp.log_params(temperature=0.2)  # merged into the experiment's model_config
    exp.run_task(answer_question)
    opengrade.log_metrics({"cost_usd": total_cost})  # experiment-level scalar, last value wins
print(exp.result.experiment_metrics)

• log_params merges into model_config until an evaluation starts. After that, or when the experiment was created from an app config, the server answers 409: start a new experiment instead.
• log_metrics stores one final number per key (no steps). Keys use letters, digits and _ . - /. They appear in result.summary() with the delta against the baseline, and can gate a release via the release-check policy: {"experiment_metrics": {"cost_usd": {"max": 2.0}}}.
• opengrade.log_params / opengrade.log_metrics act on the active experiment: inside with start_experiment(...) and inside any item run (exp.run(...), run_task, evaluate() tasks). Elsewhere they raise RuntimeError. With concurrency > 1, call log_metrics/log_params once outside per-item code: per-item calls overwrite each other (last write wins).
• Every finish() logs opengrade.error_rate, opengrade.latency_p50_ms and opengrade.latency_p95_ms (the task's wall-clock latency as measured by the SDK, over the items that did not fail, nearest-rank percentiles).

Comparing configurations (sweep)

Run the same test set over a grid of configurations and rank them in one call:

def answer(item, params):  # each configuration's params come in as the second argument
    docs = retrieve(item.input, top_k=params["top_k"])
    opengrade.set_retrieval_documents(docs)
    return generate(item.input, docs, model=params["model"])


result = opengrade.sweep(
    dataset="gold-v1",
    task=answer,
    grid={"model": ["gemma4", "llama3.3"], "top_k": [3, 5]},  # 4 experiments
    # variants=[{"model": "gemma4", "top_k": 3}, ...],         # or an explicit list
    metrics=["groundedness", "hit_at_k", "correctness"],
    params={"prompt": "v3"},  # stored on every experiment
    baseline="alias:main",  # optional: each configuration is compared with it
)
print(result.summary())  # leaderboard, best first, with latency p95 and error rate
best = result.best("correctness")  # SweepRun: .params, .result (ExperimentResult)
rows = result.to_records()  # list of dicts → pandas.DataFrame(rows)
print(result.url)  # Compare view pinned to the sweep's experiments

• Every configuration is a normal experiment named <sweep> [model=gemma4, top_k=3], with its configuration plus sweep=<name> stored as params. In the UI (Experiments → Compare all) filter on the sweep parameter, color the charts by any parameter, and plot quality against cost or latency.
• leaderboard(metric) / best(metric) rank on a scored metric or a run metric (opengrade.latency_p95_ms, cost_usd from log_metrics); latency, cost, error and token metrics rank lower-is-better (override with lower_is_better=).
• The test set is resolved (inline rows uploaded) once. Configurations run one after another; max_concurrency and repeats apply within each. A configuration that fails to run is reported in result.errors and the others still run (raise_on_error=True stops instead).
• retriever= instead of task= sweeps a retrieval-only evaluation (see below), e.g. chunk sizes or embedding models: opengrade.sweep(dataset=..., retriever=search, grid={"chunk_size": [256, 512]}, k=5).

Retrieval-only evaluation (evaluate_retrieval)

Score the retrieval step without generating answers: cheap enough to run on every change to chunking, embeddings or reranking.

def search(query: str):  # or search(item); return the ranked documents
    return vector_store.similarity_search(query, k=10)


result = opengrade.evaluate_retrieval(dataset="gold-v1", retriever=search, k=5)
print(result.summary())  # hit_at_k, recall_at_k, mrr at cutoff 5 + retrieval latency

• The retriever gets the dataset item, or the question text when its first parameter is named query, question, text or q. Sync or async.
• Documents can be dicts (id, text, …), strings, LangChain Document (id / metadata["id"|"doc_id"|"source"]) or LlamaIndex NodeWithScore (node id). Pass doc_id=lambda doc: ... to choose the id. Gold ids come from the item's expected_doc_ids; documents without an id get a text hash, which never matches them.
• Default metrics: hit_at_k, recall_at_k, mrr with cutoff k (deterministic, no LLM). Add "context_precision" / "context_recall" to have the LLM judge rate relevance and coverage against expected_output. Metrics that grade an answer (groundedness, correctness, …) are rejected.
• The experiment is stored with params["eval_mode"] = "retrieval"; opengrade.latency_p50_ms / _p95_ms measure the retriever alone.

Control plane (Client)

HTTP client for the platform JSON API (uses httpx2, a core SDK dependency). Without base_url it uses the tracking URI. In app code prefer the shared instance, opengrade.get_client(), over building a Client() per function:

import opengrade

opengrade.set_project("demo")
client = opengrade.get_client()
datasets = client.datasets.list()  # project_id defaults to the active project
exp = client.experiments.create(name="baseline", dataset_id=datasets[0].id)

Project-scoped methods (datasets.create/list/create_with_items, experiments.create/list, metrics_packs.ensure/get, live_runs.submit/list, projects.get) take project_id as an optional first argument: explicit value > init() / set_project() id > OPENGRADE_PROJECT_ID > lazy slug resolve (init(project_slug=…) / OPENGRADE_PROJECT_SLUG), else ValueError.

Explicit, self-managed client:

from opengrade import Client

with Client("http://localhost:8000") as client:
    project = client.projects.create(name="Demo", slug="demo")
    ds = client.datasets.create_with_items(
        project.id,
        name="gold-v1",
        task_type="rag_qa",
        items=[{"input": "Q?", "expected_output": "A"}],
    )
    # or CSV/JSONL via multipart import:
    # client.datasets.import_items(ds.id, path="gold.csv", format="csv")

    exp = client.experiments.create(
        project.id,
        name="baseline",
        dataset_id=ds.id,
        model_config={"model": "gemma4:e2b"},
    )
    client.metrics_packs.ensure(project.id)
    # after OTLP-bound runs:
    client.experiments.evaluate_pack(exp.id)
    summary = client.experiments.summary(exp.id)

Control-plane methods return Pydantic models (see opengrade.models); use attributes or .model_dump() for a dict. Prefer with Client(...) / client.close() so the shared httpx2 connection pool is released.

Namespaces: projects, datasets (create, create_with_items, add_item, get, list, import_items), experiments (create, get, list_outputs, upsert_outputs, evaluate_pack, summary, compare), metrics_packs (ensure, get), metrics_sets (list, get, create), evaluators (list, create, preview), alerts, live_runs (submit, list, get, promote, rescore), app_configs (create_app_config, list_app_configs, set_alias, get_aliases). Non-2xx responses raise OpengradeAPIError (with parsed FastAPI detail when present).

Async control plane (AsyncClient)

For FastAPI / async RAG apps, use AsyncClient so control-plane calls do not block the event loop. Prefer opengrade.get_async_client() and await opengrade.aset_project(...). MVP surface: health, projects, live_runs (full), trace_scores. Datasets, experiments, metrics packs/sets, and app configs remain on the sync Client.

import opengrade


# FastAPI lifespan
async def lifespan(app):
    opengrade.set_tracking_uri("http://localhost:8000")
    await opengrade.aset_project("my-rag")
    opengrade.init(instrument=False)
    yield
    await opengrade.get_async_client().aclose()


@opengrade.trace_async(name="answer")
async def answer(q: str) -> str:
    text = "..."
    client = opengrade.get_async_client()
    await client.live_runs.submit(question=q, answer=text, documents=[])
    return text

Or manage the client yourself: async with AsyncClient(...) as client: .... Changing set_tracking_uri resets both shared clients; if an event loop is already running, aclose the async client before changing the URI (otherwise the shared async client is dropped with a warning).

Releases and environments

Tag production traffic with the release and deployment that served it, so a deploy can be compared with the previous one:

opengrade.init(
    version=os.environ["GIT_SHA"], environment="prod"
)  # or OPENGRADE_VERSION / OPENGRADE_ENVIRONMENT

They become the OTel resource attributes service.version / deployment.environment (when opengrade attaches to a host TracerProvider, they are stamped on each span as opengrade.version / opengrade.environment). Production → Health then filters by environment and version and shows a By version table (error rate, latency, cost per trace, failing scores side by side); the Traces list filters the same way, and alert rules can be scoped to an environment / version. API: GET /projects/{id}/monitoring?environment=&version=, /monitoring/versions, /monitoring/facets, GET /projects/{id}/traces?environment=&version=.

Live runs from tracing (production)

With live=True, the outermost @opengrade.trace call outside an experiment becomes a live interaction at OTLP ingestion; no separate live_runs.submit call is needed.

opengrade.init(
    project_slug="acme", live=True, live_sample_rate=0.1
)  # or OPENGRADE_LIVE=1, OPENGRADE_LIVE_SAMPLE_RATE=0.1


@opengrade.trace
def answer(question: str) -> str:
    docs = retrieve(question)  # documents from RETRIEVER spans or set_retrieval_documents()
    return llm(question, docs)

• Every call is recorded; only live_sample_rate of them (default 1.0) are sent to the judge. The rest show judge_status="unsampled" and can still be judged with live_runs.rescore. Sampling is deterministic on the trace id.
• Question = the only string argument (or a question/query/input/prompt kwarg); answer = the return value. Override with opengrade.set_input(...) / opengrade.set_output(...).
• @opengrade.trace(live=False) excludes a function; live=True opts one in when init(live=...) is off.
• The server stores span text by default; with CONTENT_CAPTURE_ENABLED=false input/output are redacted and no live interaction is created.
• Idempotent per trace (external_id="trace:<trace_id>").

Production scores and feedback

Attach scores and end-user feedback to a production trace. They are stored per trace and show up on the Live page (next to the judge verdicts), in the "Failing" filter and on the overview.

opengrade.init(live=True)


@opengrade.metric(threshold=0.5, requires=("documents",))
def cites_source(s: opengrade.Sample) -> float:
    return float(any(d["id"] in s.output for d in s.documents))


@opengrade.trace(
    metrics=[cites_source], metrics_sample_rate=0.2
)  # 20% of traces, scored off the request path
def answer(question: str) -> str:
    docs = retrieve(question)
    opengrade.set_retrieval_documents(docs)
    return generate(question, docs)


@opengrade.trace
def handle(question: str) -> dict:
    out = answer(question)
    opengrade.log_score("answer_chars", len(out))  # current trace
    return {"answer": out, "trace_id": opengrade.current_trace_id()}  # send the id to your frontend


# later, from the thumbs button (trace_id is the one handle() returned)
opengrade.log_feedback(True, trace_id=trace_id, user_id="user-42", comment="great")

• log_score(name, value, *, trace_id=None, label=None, explanation=None, threshold=None, metadata=None): without trace_id it uses the current trace. PASS/FAIL follows the same rule as experiment metrics: a FAIL label fails; a value under threshold fails; otherwise, with a threshold a value at or above it passes, and without one only a value of 0 fails.
• log_feedback(positive, *, trace_id=None, comment=None, user_id=None): stored as user_feedback (1 or 0); positive must be a bool (or 0/1). One value per user per trace: a user changing their mind replaces it. Without user_id the feedback is anonymous and all anonymous feedback on a trace shares one row (last write wins). A thumbs-down counts as a failure in the Live "Failing" filter.
• opengrade.score_trace(metric, sample, trace_id=...) runs one metric now, on the calling thread (only the sending is in the background), and stores the result. The trace is trace_id, else sample.trace_id, else the current trace. A raising metric queues label="ERROR" with the exception as explanation (returns True when that score is queued); missing requires skips with False.
• @opengrade.trace(metrics=[...], metrics_sample_rate=1.0) runs your metrics after the function returns successfully (not when it raises, not inside experiments, which score their own metrics). Needs opengrade.init(). Sampling is per trace and configured separately from the judge's live_sample_rate; both use the same hash of the trace id, so when both are set the sampled traces are nested (the lower rate's traces are a subset of the higher rate's).
  • Sample.input is what you passed to opengrade.set_input() inside the call, else the only argument, else {"args", "kwargs"}; Sample.output is what you passed to opengrade.set_output() inside the call, else the return value (so a handler can return a response object and still have its metrics see the answer); Sample.documents is the last set_retrieval_documents() inside the call (including nested @opengrade.trace helpers on the same thread); expected is None, so metrics with requires=("expected",) are skipped. Input, output and documents are deep-copied when the job is queued (so later mutation is invisible); samples over 256 KiB serialized are skipped with a warning. Capture uses a ContextVar: a worker thread your function starts does not inherit it, so set_retrieval_documents (or record_documents) from that thread does not feed the outer sample — call it on the traced thread before return, or pass documents into the metric yourself.
  • Metrics run on a background pool (default 2 threads; init(metrics_workers=...) or OPENGRADE_METRICS_WORKERS) fed by a bounded queue; when it is full the call is not scored and a warning is logged. Async metrics time out after 60 s; a sync metric cannot be interrupted. A raising or timed-out metric stores label="ERROR" with the exception as explanation (never breaks the traced call). For LLM judges raise metrics_workers or lower metrics_sample_rate.
  • Metrics run with instrumentation suppressed: LLM calls a judge makes and @opengrade.trace helpers it calls do not create production traces (spans started manually with an OpenTelemetry tracer still do).
  • trace(metrics=...) rejects async and generator functions; use trace_async for coroutines. trace_async also accepts functools.partial of a coroutine function; other async wrappers need inspect.markcoroutinefunction.
• These calls never raise. log_score, log_feedback and trace(metrics=...) never block the caller; score_trace() blocks for as long as the metric takes. Invalid input (bad value, label or trace id, or metadata over 16 KB serialized) logs a warning and returns False; setup problems (no project id, no active trace) warn at most once a minute.
• Scores are sent in the background (batches every 2 s, 100 scores or about 512 KB). A 413 (body over the server's 1 MiB limit) splits the batch in half and resends, dropping only a single score that is still too large. A batch rejected with 400/422 is resent item by item so only the bad scores are dropped; 401/403/404 and other 4xx drop the batch at once; only network errors and 5xx are retried once after a 1 s backoff, then dropped with a warning.
• opengrade.flush(timeout_millis) is one total budget: spans, then pending metrics, then scores. At exit, spans, pending metrics, and scores share one ~5 s budget (span flush is capped at ~2 s of that).

Alerts

The server checks alert rules every minute on production data and notifies a webhook (Slack or any URL) when one fires or resolves. A rule compares the metric with a fixed threshold (default), with the previous period (condition="change": drift, threshold is a relative change) or with its recent history (condition="anomaly": threshold is a z-score). Create them in the UI (Production → Alerts) or as code:

client = opengrade.get_client()
client.alerts.create(
    name="Groundedness failing",
    metric="score_fail_rate",  # also error_rate, latency_p90_ms, traffic, cost_usd, ...
    score_name="groundedness",
    threshold=0.2,  # rates are 0-1
    window_minutes=60,
    webhook_url="https://hooks.slack.com/services/...",
    webhook_format="slack",
)
# Drift: groundedness mean down 20% vs the previous week.
client.alerts.create(
    name="Groundedness drift",
    metric="score_mean",
    score_name="groundedness",
    condition="change",
    operator="lt",
    threshold=-0.2,
    window_minutes=60,
    baseline_minutes=7 * 24 * 60,
)
# Anomaly: p90 latency 3 standard deviations away from the last 24 hours (hourly windows).
client.alerts.create(
    name="Latency anomaly",
    metric="latency_p90_ms",
    condition="anomaly",
    operator="abs",
    threshold=3,
    window_minutes=60,
)
client.alerts.list()  # with state: ok / firing / no_data
client.alerts.events()  # notification log

Live runs → TestSet loop

from opengrade import Client

with Client() as client:
    project_id = "..."  # existing project
    live = client.live_runs.submit(
        project_id,
        question="How many PTO days?",
        answer="Unlimited.",
        documents=[{"id": "pto", "text": "Full-time staff get 20 PTO days."}],
    )
    detail = client.live_runs.get(live.id)  # or list(..., failed_only=True)
    # After review, promote with reviewer-authored gold (never copy the prod answer):
    ds = client.datasets.create(project_id, name="from-live", task_type="rag_qa")
    client.live_runs.promote(
        live.id,
        dataset_id=ds.id,
        expected_output="Full-time employees get 20 PTO days per year.",
        expected_doc_ids=["pto"],
    )
    exp = client.experiments.create(project_id, name="after-promote", dataset_id=ds.id)
    client.experiments.evaluate_pack(exp.id)

See also scripts/live_run_demo.py (--promote).

Eval binding still uses tracing attributes (no REST call):

@opengrade.trace
def run_item(question: str) -> str:
    opengrade.bind_evaluation(experiment_id=exp_id, dataset_item_id=item_id)
    ...

Registry (app configs)

Versioned prompt + model configs live under client.app_configs:

import opengrade

project_id = opengrade.set_project("demo").id
registry = opengrade.get_client().app_configs
created = registry.create_app_config(
    project_id,
    "rag-faq",
    prompt={"system": "You are helpful."},
    model={"model_id": "demo"},
)
latest = registry.list_app_configs(project_id, latest=True)
registry.set_alias(project_id, "baseline", created.id)

Methods: create_app_config, list_app_configs (name, latest), set_alias, get_aliases. Constructing AppConfigClient(...) directly still works but is deprecated (DeprecationWarning).

End-to-end prompt A/B (same model, two App Config prompts): uv run python 05_prompt_ab.py in examples/sdk_tour.

Tier-1 compatibility matrix

Install the extra for the providers you use, then init(instrument="auto") activates every tier-1 entry whose target library and instrumentor package are importable. Use init(instrument=["openai", "langchain"]) to restrict keys, or instrument=False for decorator-only tracing.

Environment variables

Project precedence for init(): arguments > set_project() > env. If both OPENGRADE_PROJECT_ID and OPENGRADE_PROJECT_SLUG are set, id wins. init(project_slug=…) also wires control-plane calls (log_score, production metrics) via lazy slug→id resolution.

When the UI is served on a different host than the API, set OPENGRADE_UI_URL so result links open the UI instead of the API origin.

HTTP stacks (proxy / TLS)

Control-plane calls (Client, log_score, experiments) use httpx2. OTLP span export uses OpenTelemetry’s HTTP exporter (requests / urllib3). Proxy and TLS settings can therefore differ: configure httpx2 for the API client, and HTTP_PROXY / HTTPS_PROXY / REQUESTS_CA_BUNDLE (or equivalent) for trace export. If the two stacks disagree in your environment, set OPENGRADE_DISABLED=1 to stop tracing and score posts, or align both proxy/CA configs.

Failure behavior

Tracing never breaks the host app:

• init() never raises. Without a project it logs a warning on the opengrade logger and stays a no-op; a broken instrumentor is skipped with a warning.
• Pending spans are flushed at interpreter exit (atexit, sharing the ~5 s exit budget with metrics/scores; span stage capped at ~2 s), so short scripts don't need opengrade.flush().
• If the platform is unreachable, spans are dropped: one warning when export starts failing, one info when it recovers. After a failure, export pauses for 30s, so a platform that is down adds no network calls and doesn't slow down exit.

Content capture

The SDK may set OpenInference input / output attributes on decorated spans (capture_input / capture_output, default on). The platform API applies its own policy: it stores them by default (CONTENT_CAPTURE_ENABLED=true); with CONTENT_CAPTURE_ENABLED=false it drops input/output values, LLM prompts/messages and completions (including OpenInference flattened keys such as llm.input_messages.0.message.content) and the text of retrieved documents at ingestion. Timings, tokens, cost and document ids/titles/scores are kept, so retrieval metrics (hit@k, recall, MRR) still work, while judges that need document text (groundedness, context precision/recall) are SKIPPED. Turn capture off on either side when spans may carry data you must not store.

Monorepo note

• SDK PyPI / import / CLI: opengrade-ai / opengrade / opengrade
• Server PyPI / import / CLI: opengrade-ai-server / opengrade_server / opengrade-server (share one venv via opengrade-ai[ui])

Manual smoke (not CI-required)

With the platform running and a project created:

Anthropic

cd sdk && uv sync --extra anthropic
uv pip install anthropic
uv run python -c "
import opengrade
import anthropic
opengrade.init(project_slug='demo', instrument=['anthropic'])
c = anthropic.Anthropic()
@opengrade.trace
def go():
    c.messages.create(model='claude-3-5-haiku-20241022', max_tokens=16, messages=[{'role':'user','content':'hi'}])
go(); opengrade.flush()
"

LangChain

cd sdk && uv sync --extra langchain
# install langchain + your LLM backend as needed
uv run python -c "
import opengrade
opengrade.init(project_slug='demo', instrument=['langchain'])
# invoke your LangChain chain here
opengrade.flush()
"

See platform tracing docs: docs/04-tracing-and-otel.md.