Pydantic Evals

Pydantic Evals is the evaluation library from the Pydantic team. It models test suites as a Dataset of Case objects and scores them with deterministic evaluators or with LLMJudge, an LLM-as-judge evaluator driven by a plain-language rubric. This guide shows both ways to wire Pydantic Evals into Arize AX: Flow 1 grades existing Arize AX traces with an LLMJudge and writes the scores back via client.spans.update_evaluations(...); Flow 2 uploads a small dataset, runs an Arize AX experiment with the same rubric 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.11+
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 LLMJudge judge model)

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 pydantic-evals '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 Pydantic Evals LLMJudge that checks each answer against a reference answer, the canonical 2-row hallucination dataset that both flows score, and an Arize SDK client. include_input=True and include_expected_output=True send the question and the reference answer to the judge along with the output. The assertion config names the result factuality and keeps the judge’s reason, so the pass/fail verdict comes back with an explanation.

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

import pandas as pd
from arize import ArizeClient
from pydantic_evals.evaluators import LLMJudge
from pydantic_evals.evaluators.common import OutputConfig

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

JUDGE_MODEL = "openai:gpt-6-luna"
RUBRIC = (
    "The output answers the question with the same fact as the expected "
    "output. Fail the output if it contradicts the expected output."
)

# LLMJudge returns a pass/fail assertion plus the judge's reason.
factuality = LLMJudge(
    rubric=RUBRIC,
    model=JUDGE_MODEL,
    include_input=True,
    include_expected_output=True,
    assertion=OutputConfig(evaluation_name="factuality", include_reason=True),
)

# Canonical 2-row dataset — row 0 is factual (answer matches the reference),
# 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.",
        "reference": "Paris is the capital and most populous city of France.",
    },
    {
        "input":     "What is the capital of France?",
        "output":    "Berlin is the capital of France.",
        "reference": "Paris is the capital and most populous city 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 openai import OpenAI
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"pydantic-evals-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)
instrumentor = OpenAIInstrumentor()
instrumentor.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)
# Stop tracing so the judge's own OpenAI calls don't land in the project.
instrumentor.uninstrument()
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

Turn each exported span into a Pydantic Evals Case, with the span’s question and ID as the inputs and the reference answer as the expected output. The task looks up the span’s recorded output by ID, so the judge grades exactly what the traced model returned. evaluate_sync(...) runs the LLMJudge over every case and returns a report whose assertions hold the pass/fail verdicts.

from pydantic_evals import Case, Dataset

span_outputs = dict(zip(spans_df["context.span_id"], spans_df["output"]))

cases = [
    Case(
        name=span_id,
        inputs={"question": ROWS[i]["input"], "span_id": span_id},
        expected_output=ROWS[i]["reference"],
    )
    for i, span_id in enumerate(spans_df["context.span_id"])
]


def recorded_output(inputs: dict) -> str:
    return span_outputs[inputs["span_id"]]


dataset = Dataset(
    name="arize-spans", cases=cases, evaluators=[factuality]
)
report = dataset.evaluate_sync(recorded_output, progress=False)

verdicts = {case.name: case.assertions["factuality"] for case in report.cases}
scores = []
labels = []
explanations = []
for span_id in spans_df["context.span_id"]:
    verdict = verdicts[span_id]
    scores.append(1.0 if verdict.value else 0.0)
    labels.append("factual" if verdict.value else "hallucinated")
    explanations.append(verdict.reason or "no 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 LLMJudge verdict becomes one row in this DataFrame.

eval_df = pd.DataFrame(
    {
        "context.span_id":              spans_df["context.span_id"],
        "eval.factuality.score":        scores,
        "eval.factuality.label":        labels,
        "eval.factuality.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 pydantic-evals-tracing-example-<timestamp> (the value printed above) in your Arize AX space. Each ChatCompletion span now carries a factuality annotation column showing the score, the factual / hallucinated label, and the judge’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"pydantic-evals-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. The parameter name must be one of input, output, metadata, or dataset_row — a single-arg task with an unrecognized name is bound to dataset_row by default. 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 the wrapper is an async def that awaits judge_input_output_expected(...) — the same judge call LLMJudge makes, with the same rubric and model. Calling Dataset.evaluate_sync(...) here instead would fail, because it starts its own event loop. Return an EvaluationResult with score and 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
from pydantic_evals.evaluators.llm_as_a_judge import (
    judge_input_output_expected,
)


async def factuality_eval(input, output, dataset_row) -> EvaluationResult:
    grading = await judge_input_output_expected(
        inputs={"question": dataset_row["input"]},
        output=output if isinstance(output, str) else str(output),
        expected_output=dataset_row["reference"],
        rubric=RUBRIC,
        model=JUDGE_MODEL,
    )
    return EvaluationResult(
        score=1.0 if grading.pass_ else 0.0,
        label="factual" if grading.pass_ else "hallucinated",
        explanation=grading.reason or "no explanation",
    )

Run the experiment

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

Expected output

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

Verify in Arize AX

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

Troubleshooting

This event loop is already running or asyncio.run() cannot be called from a running event loop. Dataset.evaluate_sync(...) starts its own event loop, so it fails inside a Jupyter notebook or an experiment evaluator. In a notebook, await dataset.evaluate(task) instead. In an experiment evaluator, await judge_input_output_expected(...) directly as Flow 2 does.
KeyError: 'factuality' when reading case.assertions. The LLMJudge failed for that case, so Pydantic Evals recorded it in case.evaluator_failures instead of case.assertions. Print report.cases[0].evaluator_failures to see the error — usually a missing OPENAI_API_KEY or an unknown judge model.
Judge calls appear as extra spans in the Flow 1 project. The judge talks to OpenAI through the same openai client the instrumentor patches. Call instrumentor.uninstrument() before evaluating, as shown in Source the spans.
column "eval.factuality.label": unsupported cast from null to string: reserved column cannot be coerced to canonical type. Your experiment evaluator returned a bare bool or number instead of a fully-populated EvaluationResult(score=..., label=..., explanation=...). Populate all three fields.
Using a different judge provider. JUDGE_MODEL takes any Pydantic AI model string, such as anthropic:<model> or google-gla:<model>. Install that provider’s extra (for example pip install 'pydantic-ai-slim[anthropic]') and export its API key.
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

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

Pydantic Evals Documentation

Pydantic Evals on GitHub

Logging evaluations to Arize AX

Pydantic AI tracing 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

Pydantic Evals Documentation

Pydantic Evals on GitHub

Logging evaluations to Arize AX

Pydantic AI tracing 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