Cleanlab

Cleanlab TLM (Trustworthy Language Model) scores how trustworthy an LLM response is. Given a prompt and a response, it returns a trustworthiness_score between 0 and 1, plus an optional explanation, so you can flag hallucinated or unreliable answers. This guide shows both ways to wire TLM into Arize AX: Flow 1 grades existing Arize AX traces with TLM and writes the scores back via client.spans.update_evaluations(...); Flow 2 uploads a small dataset, runs an Arize AX experiment with a TLM-backed evaluator function, and surfaces the scores in Datasets+Experiments. Both flows share the same setup. Run the code blocks below in order inside a single Python session — each block builds on imports and variables from earlier ones.

Cleanlab TLM Evaluation Tutorial (Google Colab)

Prerequisites

Python 3.11+
An ARIZE_SPACE_ID and ARIZE_API_KEY from your Arize AX space settings
An OPENAI_API_KEY from OpenAI Platform (used for the model under trace)
A CLEANLAB_TLM_API_KEY from your Cleanlab account (used for the TLM trustworthiness scores)

Launch Arize AX

If you don’t already have an Arize AX account, sign up at arize.com and grab your ARIZE_SPACE_ID and ARIZE_API_KEY from Settings → Space Settings.

Install

pip install cleanlab-tlm 'arize>=8.0.0' openai \
  openinference-instrumentation-openai opentelemetry-sdk \
  opentelemetry-exporter-otlp-proto-grpc pandas

Configure credentials

export ARIZE_SPACE_ID="<your-space-id>"
export ARIZE_API_KEY="<your-api-key>"
export OPENAI_API_KEY="<your-openai-api-key>"
export CLEANLAB_TLM_API_KEY="<your-cleanlab-tlm-api-key>"

Define evaluators

The shared setup: a Cleanlab TLM client that also returns an explanation for each score, the canonical 2-row hallucination dataset that both flows score, and an Arize SDK client. TLM() reads CLEANLAB_TLM_API_KEY from the environment.

# combined.py
import os
import time
from datetime import datetime, timedelta, timezone

import pandas as pd
from arize import ArizeClient
from cleanlab_tlm import TLM
from openai import OpenAI

SPACE_ID = os.environ["ARIZE_SPACE_ID"]
API_KEY = os.environ["ARIZE_API_KEY"]
TIMESTAMP = int(time.time())

# TLM scores each prompt/response pair between 0 and 1 — higher means the
# response is more likely to be correct. Ask for an explanation as well.
tlm = TLM(options={"log": ["explanation"]})

# Canonical 2-row dataset — row 0 is factual, row 1 is hallucinated. Both
# flows grade these same rows.
ROWS = [
    {
        "input":  "What is the capital of France?",
        "output": "Paris is the capital of France.",
    },
    {
        "input":  "What is the capital of France?",
        "output": "Berlin is the capital of France.",
    },
]

# TLM scores are continuous, so binarize at 0.5 for a stable pass/fail.
THRESHOLD = 0.5

arize = ArizeClient(api_key=API_KEY)

Flow 1 — Evaluate existing traces

Source the spans

Instrument OpenAI with OpenInference, make two calls (each forced to echo a known answer so the trace contains predictable text), then pull the resulting spans back from Arize AX.

from openinference.instrumentation.openai import OpenAIInstrumentor
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter,
)
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

PROJECT_NAME = f"cleanlab-tracing-example-{TIMESTAMP}"

resource = Resource.create(
    {
        "service.name":                PROJECT_NAME,
        "openinference.project.name":  PROJECT_NAME,
        "model_id":                    PROJECT_NAME,
    }
)
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(
            endpoint="https://otlp.arize.com:443",
            headers={
                "authorization":   API_KEY,
                "arize-space-id":  SPACE_ID,
                "arize-interface": "python",
            },
        )
    )
)
trace.set_tracer_provider(provider)
OpenAIInstrumentor().instrument(tracer_provider=provider)

sync_oai = OpenAI()
for row in ROWS:
    sync_oai.chat.completions.create(
        model="gpt-6-luna",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a fact-recall assistant. The user states the "
                    "exact answer to use; reply with that verbatim."
                ),
            },
            {
                "role": "user",
                "content": (
                    f"Question: {row['input']}\n"
                    f"Answer (reply verbatim): {row['output']}"
                ),
            },
        ],
    )

provider.force_flush(timeout_millis=10_000)
print(f"Project: {PROJECT_NAME}")

# Poll defensively: Arize's OTLP ingest and Flight export use different
# catalogs and the new project can briefly appear "unauthorized" to the
# export endpoint while still accepting span writes via OTLP, so swallow
# transient errors and retry.
start = datetime.now(timezone.utc) - timedelta(minutes=5)
end = datetime.now(timezone.utc) + timedelta(minutes=1)
spans_df = None
last_err: Exception | None = None
for _ in range(12):
    time.sleep(5)
    try:
        spans_df = arize.spans.export_to_df(
            space_id=SPACE_ID,
            project_name=PROJECT_NAME,
            start_time=start,
            end_time=end,
        )
    except Exception as e:
        last_err = e
        continue
    if spans_df is not None and len(spans_df) >= len(ROWS):
        break
else:
    raise RuntimeError(
        f"Spans never appeared after 60s (last error: {last_err})"
    )

spans_df = spans_df.sort_values("start_time").reset_index(drop=True)

Run the evaluators

get_trustworthiness_score(...) accepts lists of prompts and responses and scores them in one batched call. Pair each span’s response text (the output column) with the original question rather than the echo instruction the trace sent, so TLM judges the answer on its own merits.

prompts = [ROWS[i]["input"] for i in range(len(spans_df))]
responses = list(spans_df["output"])
results = tlm.get_trustworthiness_score(prompts, responses)

scores = []
labels = []
explanations = []
for result in results:
    trustworthy = result["trustworthiness_score"] >= THRESHOLD
    scores.append(1.0 if trustworthy else 0.0)
    labels.append("trustworthy" if trustworthy else "untrustworthy")
    explanations.append(result["log"]["explanation"])

Log evaluations to Arize AX

update_evaluations(...) requires a context.span_id column (which export_to_df already provides) plus the reserved eval.<name>.{score,label,explanation} columns. Each TLM score becomes one row in this DataFrame.

eval_df = pd.DataFrame(
    {
        "context.span_id":                    spans_df["context.span_id"],
        "eval.trustworthiness.score":         scores,
        "eval.trustworthiness.label":         labels,
        "eval.trustworthiness.explanation":   explanations,
    }
)
arize.spans.update_evaluations(
    space_id=SPACE_ID,
    project_name=PROJECT_NAME,
    dataframe=eval_df,
)

flow1_display = pd.DataFrame(
    {
        "input":  [r["input"]  for r in ROWS],
        "output": [r["output"] for r in ROWS],
        "score":  scores,
    }
)
print("Flow 1 results:")
print(flow1_display.to_string())

Expected output

Flow 1 results:
                            input                            output  score
0  What is the capital of France?   Paris is the capital of France.    1.0
1  What is the capital of France?  Berlin is the capital of France.    0.0

Verify in Arize AX

Open the project named cleanlab-tracing-example-<timestamp> (the value printed above) in your Arize AX space. Each ChatCompletion span now carries a trustworthiness annotation column showing the score, the trustworthy / untrustworthy label, and TLM’s explanation.

Flow 2 — Run an experiment

Create a dataset

The dataset is the same two rows. The space= / examples= kwarg names match the v8 SDK exactly (note: not space_id= and not dataframe=).

DATASET_NAME = f"cleanlab-experiment-example-ds-{TIMESTAMP}"
dataset_df = pd.DataFrame(ROWS)
arize.datasets.create(
    name=DATASET_NAME,
    space=SPACE_ID,
    examples=dataset_df,
)
print(f"Dataset: {DATASET_NAME}")

Define the task

The task function receives the dataset row and returns whatever the experiment should grade. A real workflow would call an LLM here; this passthrough keeps the example deterministic.

def task(dataset_row):
    return dataset_row["output"]

Wrap the evaluators

Experiment evaluators run inside an asyncio loop, so use async def and TLM’s get_trustworthiness_score_async(...). The sync get_trustworthiness_score(...) drives its own event loop and fails when one is already running. Return an EvaluationResult with score, label, and explanation populated — leaving any of those as None triggers unsupported cast from null to <type>: reserved column cannot be coerced to canonical type at upload time.

from arize.experiments.evaluators.types import EvaluationResult


async def trustworthiness_eval(
    input, output, dataset_row
) -> EvaluationResult:
    result = await tlm.get_trustworthiness_score_async(
        dataset_row["input"],
        output if isinstance(output, str) else str(output),
    )
    trustworthy = result["trustworthiness_score"] >= THRESHOLD
    return EvaluationResult(
        score=1.0 if trustworthy else 0.0,
        label="trustworthy" if trustworthy else "untrustworthy",
        explanation=result["log"]["explanation"] or "no explanation",
    )

Run the experiment

EXPERIMENT_NAME = f"cleanlab-experiment-example-{TIMESTAMP}"
experiment, runs_df = arize.experiments.run(
    space=SPACE_ID,
    name=EXPERIMENT_NAME,
    dataset=DATASET_NAME,
    task=task,
    evaluators={"trustworthiness": trustworthiness_eval},
)
print(f"Experiment: {EXPERIMENT_NAME}")
print("Flow 2 results:")
flow2_display = runs_df[
    [
        "output",
        "eval.trustworthiness.score",
        "eval.trustworthiness.label",
    ]
].rename(
    columns={
        "eval.trustworthiness.score": "score",
        "eval.trustworthiness.label": "label",
    }
)
print(flow2_display.to_string())

Expected output

Flow 2 results:
                             output  score          label
0   Paris is the capital of France.    1.0    trustworthy
1  Berlin is the capital of France.    0.0  untrustworthy

Verify in Arize AX

Open the Datasets + Experiments tab in Arize AX. The dataset cleanlab-experiment-example-ds-<timestamp> and the experiment cleanlab-experiment-example-<timestamp> (names printed above) appear with one run per dataset row, each carrying the trustworthiness score and label columns.

Troubleshooting

MissingApiKeyError: No API key provided. TLM() raised MissingApiKeyError at construction. Export CLEANLAB_TLM_API_KEY in the same shell that runs the script.
This event loop is already running or Cannot run the event loop while another loop is running from Flow 2. Your experiment evaluator calls the sync get_trustworthiness_score(...), which drives TLM’s own event loop with run_until_complete. Experiment evaluators run inside asyncio; await tlm.get_trustworthiness_score_async(...) instead. Flow 1 calls the sync method because it runs outside any loop.
KeyError: 'log' when reading the explanation. The TLM client was created without options={"log": ["explanation"]}, so results carry only trustworthiness_score. Pass the option when constructing TLM.
Scores near the threshold flip between runs. TLM’s score is continuous. The doc binarizes at 0.5 so the printed score column stays stable; for borderline responses, inspect the raw trustworthiness_score and pick a threshold that fits your use case.
column "eval.<name>.label": unsupported cast from null to string: reserved column cannot be coerced to canonical type. Your evaluator returned a bare number or string instead of a fully populated EvaluationResult(score=..., label=..., explanation=...). Populate all three fields.
Spans never appear after 60s. Span flush + ingest typically takes 5–15s. If the loop times out, check that ARIZE_SPACE_ID + ARIZE_API_KEY are right and that you’re connecting to the correct region’s OTLP endpoint (otlp.arize.com for US, otlp.eu.arize.com for EU).
Experiment re-runs collide. Both names embed TIMESTAMP = int(time.time()) so a single re-run produces unique names. If you re-execute the same combined.py quickly, regenerate TIMESTAMP first or call arize.experiments.delete(...) / arize.datasets.delete(...) on the prior run’s names.

OpenTelemetry

LLM Providers

Agent Frameworks

Coding Agents

Platforms

Orchestration

Vector Databases

Evaluation

Cleanlab TLM Evaluation Tutorial (Google Colab)

Prerequisites

Launch Arize AX

Install

Configure credentials

Define evaluators

Flow 1 — Evaluate existing traces

Source the spans

Run the evaluators

Log evaluations to Arize AX

Expected output

Verify in Arize AX

Flow 2 — Run an experiment

Create a dataset

Define the task

Wrap the evaluators

Run the experiment

Expected output

Verify in Arize AX

Troubleshooting

Resources

Cleanlab TLM Documentation

cleanlab-tlm on PyPI

Logging evaluations to Arize AX

Ragas evaluators in Arize AX

Cleanlab TLM Evaluation Tutorial (Google Colab)

​Prerequisites

​Launch Arize AX

​Install

​Configure credentials

​Define evaluators

​Flow 1 — Evaluate existing traces

​Source the spans

​Run the evaluators

​Log evaluations to Arize AX

​Expected output

​Verify in Arize AX

​Flow 2 — Run an experiment

​Create a dataset

​Define the task

​Wrap the evaluators

​Run the experiment

​Expected output

​Verify in Arize AX

​Troubleshooting

​Resources

Cleanlab TLM Documentation

cleanlab-tlm on PyPI

Logging evaluations to Arize AX

Ragas evaluators in Arize AX

Prerequisites

Launch Arize AX

Install

Configure credentials

Define evaluators

Flow 1 — Evaluate existing traces

Source the spans

Run the evaluators

Log evaluations to Arize AX

Expected output

Verify in Arize AX

Flow 2 — Run an experiment

Create a dataset

Define the task

Wrap the evaluators

Run the experiment

Expected output

Verify in Arize AX

Troubleshooting

Resources