UQLM

UQLM (Uncertainty Quantification for Language Models), from CVS Health, estimates how trustworthy an LLM response is. Its black-box scorers sample several answers to the same prompt and measure how consistent the original response is with them; its white-box scorers read token log probabilities. Every scorer returns a response-level confidence score in [0, 1], so you can flag risky answers without a reference answer or a judge prompt. This guide shows both ways to wire UQLM into Arize AX: Flow 1 scores existing Arize AX traces with UQLM’s noncontradiction scorer and writes the scores back via client.spans.update_evaluations(...); Flow 2 uploads a small dataset, runs an Arize AX experiment with the same scorer wrapped as an experiment evaluator, 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.

Prerequisites

Python 3.13+
An ARIZE_SPACE_ID and ARIZE_API_KEY from your Arize AX space settings
An OPENAI_API_KEY from OpenAI Platform (used as both the model under trace and the model UQLM samples candidate answers from)
About 3.5 GB of free disk space: the installed packages take about 1.6 GB with CPU-only PyTorch, and the first run downloads the microsoft/deberta-large-mnli natural language inference (NLI) model from Hugging Face, which the noncontradiction scorer runs locally

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

# Python 3.13+ is required: on Python 3.12 and earlier, uqlm pins
# numpy<2, which conflicts with arize>=8.0.0 (numpy>=2).
# Install CPU-only PyTorch first: uqlm needs torch, and the default Linux
# wheel pulls in several GB of NVIDIA CUDA libraries. Drop this line if you
# want GPU support.
pip install torch --index-url https://download.pytorch.org/whl/cpu
pip install uqlm langchain-openai httpx '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>"

Define evaluators

The shared setup: a UQLM BlackBoxUQ scorer, the canonical 2-row hallucination dataset that both flows score, and an Arize SDK client. BlackBoxUQ with the noncontradiction scorer works in two steps. First, generate_candidate_responses(...) asks the LLM the same question several times at temperature=1.0. Then score(...) uses a local NLI model to check whether the response you are grading contradicts each sampled answer. A response that agrees with the samples scores close to 1.0; one that contradicts them scores close to 0.0. UQLM drives the LLM through LangChain, so pass it a ChatOpenAI model. gpt-4.1-mini is used because it accepts the sampling temperature UQLM sets. The two flows run in different asyncio event loops, and an async HTTP connection can’t outlive the loop that opened it. sampler_llm() therefore gives each ChatOpenAI its own httpx.AsyncClient (by default LangChain shares one cached client), and Flow 2 swaps in a fresh sampler before it runs.

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

import httpx
import pandas as pd
from arize import ArizeClient
from langchain_openai import ChatOpenAI
from openai import OpenAI
from uqlm import BlackBoxUQ

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


def sampler_llm() -> ChatOpenAI:
    # A dedicated async HTTP client per event loop — see the note above.
    return ChatOpenAI(
        model="gpt-4.1-mini",
        temperature=1.0,
        http_async_client=httpx.AsyncClient(),
    )


# BlackBoxUQ samples NUM_SAMPLES answers per question, then scores how
# consistent the graded response is with them. Loading the NLI model is
# slow the first time (it downloads ~1.6 GB), so build the scorer once.
NUM_SAMPLES = 5
uq = BlackBoxUQ(llm=sampler_llm(), scorers=["noncontradiction"])


def sampling_prompt(question: str) -> str:
    return f"{question} Answer in one sentence."


# 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.",
    },
]

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"uqlm-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

Flow 1 runs outside any asyncio loop, so wrap the async generate_candidate_responses(...) call in asyncio.run(...). score(...) is synchronous and grades every span in one call. The raw noncontradiction value is continuous and moves slightly between runs because the sampled answers change. The doc applies a 0.5 threshold and normalizes to 1.0 / 0.0 so the printed score column stays stable. To keep the raw confidence instead, assign float(confidence) directly.

prompts = [sampling_prompt(ROWS[i]["input"]) for i in range(len(spans_df))]
sampled = asyncio.run(
    uq.generate_candidate_responses(prompts, num_responses=NUM_SAMPLES)
)
result = uq.score(
    responses=spans_df["output"].tolist(),
    sampled_responses=sampled,
    show_progress_bars=False,
)

scores = []
labels = []
explanations = []
for confidence in result.data["noncontradiction"]:
    is_confident = float(confidence) >= 0.5
    scores.append(1.0 if is_confident else 0.0)
    labels.append("confident" if is_confident else "uncertain")
    explanations.append(
        f"noncontradiction={float(confidence):.3f} across "
        f"{NUM_SAMPLES} sampled answers"
    )

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.

eval_df = pd.DataFrame(
    {
        "context.span_id":                   spans_df["context.span_id"],
        "eval.uqlm_confidence.score":        scores,
        "eval.uqlm_confidence.label":        labels,
        "eval.uqlm_confidence.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 uqlm-tracing-example-<timestamp> (the value printed above) in your Arize AX space. Each ChatCompletion span now carries a uqlm_confidence annotation column showing the normalized 0/1 score, the confident / uncertain label, and the raw noncontradiction value in the explanation.

Flow 2 — Run an experiment

Create a dataset

DATASET_NAME = f"uqlm-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 await generate_candidate_responses(...) directly — asyncio.run(...) fails with asyncio.run() cannot be called from a running event loop. Return an EvaluationResult with score, label, and explanation populated: leaving any of those reserved fields 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

# Flow 1's asyncio.run(...) closed its event loop; give the scorer a
# sampler whose HTTP client belongs to the experiment's loop.
uq.llm = sampler_llm()

# UQLM silences generation by temporarily redirecting stdout. Experiment
# evaluators run concurrently, and overlapping redirects can leave stdout
# pointing at a buffer, so sample one row at a time.
sampling_lock = asyncio.Lock()


async def uqlm_eval(input, output, dataset_row) -> EvaluationResult:
    async with sampling_lock:
        sampled = await uq.generate_candidate_responses(
            [sampling_prompt(dataset_row["input"])],
            num_responses=NUM_SAMPLES,
        )
    result = uq.score(
        responses=[output if isinstance(output, str) else str(output)],
        sampled_responses=sampled,
        show_progress_bars=False,
    )
    confidence = float(result.data["noncontradiction"][0])
    is_confident = confidence >= 0.5
    return EvaluationResult(
        score=1.0 if is_confident else 0.0,
        label="confident" if is_confident else "uncertain",
        explanation=(
            f"noncontradiction={confidence:.3f} across "
            f"{NUM_SAMPLES} sampled answers"
        ),
    )

Run the experiment

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

Expected output

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

Verify in Arize AX

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

Troubleshooting

Because arize>=8.0.0 depends on numpy>=2.0.0 ... requirements are unsatisfiable, or pyarrow requires NumPy 2.0 or newer. You are on Python 3.12 or earlier, where uqlm pins numpy<2. Use Python 3.13+, where uqlm allows numpy>=2.
The first run pauses for several minutes before printing anything. BlackBoxUQ downloads the microsoft/deberta-large-mnli NLI model (about 1.6 GB) the first time it loads. Later runs read it from the Hugging Face cache (~/.cache/huggingface). To use a smaller NLI model, pass nli_model_name= to BlackBoxUQ.
asyncio.run() cannot be called from a running event loop. Your Flow 2 evaluator calls asyncio.run(...). Experiment evaluators already run inside asyncio; await generate_candidate_responses(...) instead.
RuntimeError: Event loop is closed followed by evaluator failed for example id .... The Flow 2 evaluator reused an HTTP client whose connections were opened inside Flow 1’s asyncio.run(...) loop, which has since closed. Build each flow’s ChatOpenAI with its own http_async_client=httpx.AsyncClient() and assign a fresh one (uq.llm = sampler_llm()) before running the experiment, as shown above.
Nothing prints after All evaluators completed, and the script exits cleanly. Two evaluators sampled at once. UQLM redirects stdout while it generates, and overlapping redirects from concurrent evaluators can leave stdout pointing at an in-memory buffer, so every later print is lost. Wrap generate_candidate_responses(...) in async with sampling_lock: as shown in Flow 2.
llm must be provided to generate responses. BlackBoxUQ was created without llm=. Pass a LangChain chat model such as ChatOpenAI(...).
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=...). Arize AX rejects null values in reserved eval columns — 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).
Using other UQLM scorers. BlackBoxUQ also supports semantic_negentropy, entailment, exact_match, cosine_sim, and bert_score; WhiteBoxUQ scores token log probabilities (for example min_probability) without sampling. Each scorer’s values land in result.data["<scorer>"]. See the UQLM documentation.
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

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

UQLM Documentation

UQLM on GitHub

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

UQLM Documentation

UQLM on GitHub

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