> ## Documentation Index > Fetch the complete documentation index at: /docs/llms.txt > Use this file to discover all available pages before exploring further. # Pydantic Evals > Use Pydantic Evals LLMJudge evaluators to grade Arize AX traces and as evaluators in Arize AX experiments. [Pydantic Evals](https://ai.pydantic.dev/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](https://platform.openai.com/api-keys) (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](https://arize.com/) and grab your `ARIZE_SPACE_ID` and `ARIZE_API_KEY` from Settings → Space Settings. ## Install ```bash theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} pip install pydantic-evals 'arize>=8.0.0' openai \ openinference-instrumentation-openai opentelemetry-sdk \ opentelemetry-exporter-otlp-proto-grpc pandas ``` ## Configure credentials ```bash theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} export ARIZE_SPACE_ID="" export ARIZE_API_KEY="" export 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. ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} # 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. ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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. ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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..{score,label,explanation}` columns. Each `LLMJudge` verdict becomes one row in this DataFrame. ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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 ```text wrap theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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-` (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=`). ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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. ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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 : reserved column cannot be coerced to canonical type` at upload time. ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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 ```python theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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 ```text wrap theme={"theme":{"light":"github-light-default","dark":"github-dark-default"}} 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-` and the experiment `pydantic-evals-experiment-example-` (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:` or `google-gla:`. 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. ## Resources > **[Pydantic Evals Documentation](https://ai.pydantic.dev/evals/)** > **[Pydantic Evals on GitHub](https://github.com/pydantic/pydantic-ai/tree/main/pydantic_evals)** > **[Logging evaluations to Arize AX](/docs/api-clients/python/version-8/client-resources/spans.md#update-evaluations)** > **[Pydantic AI tracing in Arize AX](/docs/ax/integrations/python-agent-frameworks/pydantic/pydantic-ai-tracing.md)**