Use the open-source phoenix-evals library to grade Arize AX traces and as evaluators in Arize AX experiments.
arize-phoenix-evals is Arize’s open-source, backend-agnostic evaluation library — the same library the LLM-as-a-judge provider guides wire a judge model into. It ships composable primitives (LLM, create_classifier, evaluate_dataframe) and prebuilt metric evaluators (hallucination, correctness, relevance, and more), and runs entirely in-process: it needs no separate server, and its results feed Arize AX through the arize SDK.This guide shows both ways to wire phoenix-evals into Arize AX: Flow 1 grades existing Arize AX traces with a classifier and writes the scores back via client.spans.update_evaluations(...); Flow 2 uploads a small dataset, runs an Arize AX experiment with the same classifier wrapped as an evaluator, and surfaces the scores in Datasets+Experiments. For the concepts behind the offline pattern, see Offline evaluation; for running these evals on a schedule inside the platform, see Run evals on traces.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.
The shared setup: a create_classifier correctness evaluator backed by a GPT-4.1 judge, the canonical 2-row dataset both flows score, and an Arize SDK client. create_classifier relies on the judge’s tool-calling / structured-output support, so use a non-reasoning model such as GPT-4.1. The {input}, {output}, and {reference} placeholders in the template are filled from dataframe columns of the same name.
# combined.pyimport osimport timefrom datetime import datetime, timedelta, timezoneimport pandas as pdfrom arize import ArizeClientfrom phoenix.evals import LLM, create_classifier, evaluate_dataframefrom phoenix.evals.utils import to_annotation_dataframeSPACE_ID = os.environ["ARIZE_SPACE_ID"]API_KEY = os.environ["ARIZE_API_KEY"]TIMESTAMP = int(time.time())# The judge model. GPT-4.1 is a non-reasoning model that supports the# tool-calling / structured output create_classifier relies on.judge = LLM(provider="openai", model="gpt-4.1")TEMPLATE = ( "You are grading whether a response is factually correct.\n" "[Question]: {input}\n" "[Response]: {output}\n" "[Reference]: {reference}\n\n" "Is the response factually correct given the reference? " "Answer 'correct' or 'incorrect'.")correctness = create_classifier( name="correctness", prompt_template=TEMPLATE, llm=judge, choices={"correct": 1.0, "incorrect": 0.0}, direction="maximize",)# 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)
create_classifier builds a custom categorical evaluator. For common metrics, phoenix.evals.metrics ships prebuilt evaluators — HallucinationEvaluator, CorrectnessEvaluator, DocumentRelevanceEvaluator, RefusalEvaluator, and others — that you can pass to evaluate_dataframe in place of the classifier above.
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 OpenAIInstrumentorfrom opentelemetry import tracefrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporterfrom opentelemetry.sdk.resources import Resourcefrom opentelemetry.sdk.trace import TracerProviderfrom opentelemetry.sdk.trace.export import BatchSpanProcessorfrom openai import OpenAIPROJECT_NAME = f"phoenix-evals-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-5.4-mini", 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}")# Spans take ~5–15s to be queryable after flush. 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 = Nonelast_err: Exception | None = Nonefor _ 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): breakelse: raise RuntimeError( f"Spans never appeared after 60s (last error: {last_err})" )spans_df = spans_df.sort_values("start_time").reset_index(drop=True)
Map the span attributes to the template variables, then run the classifier over the whole dataframe in one pass. evaluate_dataframe returns the input dataframe plus a correctness_score column (a dict holding the label, score, and explanation) and a correctness_execution_details column. The reference value isn’t on the span, so it’s sourced from ROWS by position.
spans_df["input"] = spans_df["attributes.input.value"]spans_df["output"] = spans_df["attributes.output.value"]spans_df["reference"] = [r["reference"] for r in ROWS]results = evaluate_dataframe(dataframe=spans_df, evaluators=[correctness])
For larger batches, async_evaluate_dataframe(dataframe=..., evaluators=[...]) is the async equivalent and runs the judge calls concurrently.
to_annotation_dataframe flattens the nested correctness_score dict into score, label, and explanation columns while preserving context.span_id (the join key export_to_df already provides). Rename those to the reserved eval.<name>.{score,label,explanation} columns Arize AX expects, then upload. For trace or session evals use the trace_eval.<name> / session_eval.<name> prefixes instead.
annotations = to_annotation_dataframe(dataframe=results)upload_df = annotations.rename( columns={ "label": "eval.correctness.label", "score": "eval.correctness.score", "explanation": "eval.correctness.explanation", })[[ "context.span_id", "eval.correctness.label", "eval.correctness.score", "eval.correctness.explanation",]]arize.spans.update_evaluations( space_id=SPACE_ID, project_name=PROJECT_NAME, dataframe=upload_df,)# Print the scores so they appear in stdout for verification.print("Flow 1 results:")print(upload_df[["eval.correctness.label", "eval.correctness.score"]].to_string())
Open the project named phoenix-evals-example-<timestamp> (the value printed above) in your Arize AX space. Each ChatCompletion span now carries a correctness annotation column showing the score and label written by update_evaluations(...).
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.
Experiment evaluators run inside an asyncio loop, so use async def and the classifier’s async_evaluate(...) — it returns a list of Score objects, one per evaluator. Convert the first Score to 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 EvaluationResultasync def correctness_eval(output, dataset_row) -> EvaluationResult: scores = await correctness.async_evaluate( { "input": dataset_row["input"], "output": output if isinstance(output, str) else str(output), "reference": dataset_row["reference"], } ) s = scores[0] return EvaluationResult( score=s.score, label=s.label, explanation=s.explanation or "no explanation", )
EvaluationResult’s constructor order is (score, label, explanation, metadata) — not(label, score, ...) as the alphabetical reading might suggest. Always use keyword arguments; positional calls silently swap label and score.
Open the Datasets + Experiments tab in Arize AX. The dataset phoenix-evals-example-ds-<timestamp> and the experiment phoenix-evals-example-<timestamp> (names printed above) appear with one run per dataset row, each carrying the correctness score and label columns.
create_classifier fails or returns empty labels. The judge model must support tool calling / structured output. Reasoning models (GPT-5-family or o-series ids) can also reject standard sampling params — use a non-reasoning judge such as gpt-4.1.
column "eval.<name>.label": unsupported cast from null to string: reserved column cannot be coerced to canonical type. In Flow 2 your evaluator returned a bare number or string instead of a fully-populated EvaluationResult(score=..., label=..., explanation=...). Arize AX’s Flight server rejects null values in reserved eval columns — populate all three fields.
Cannot call sync ... from an async context. Your Flow 2 evaluator is calling the classifier’s sync evaluate(...) instead of async_evaluate(...). Experiment evaluators run inside asyncio; use the async API. Flow 1 uses evaluate_dataframe(...) because it runs outside any loop.
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). Evals can only be applied to spans up to 14 days old; for older spans contact support@arize.com.
task failed for example id .... Your task function’s parameter name isn’t one of the recognized names (input, output, metadata, dataset_row). Rename it to dataset_row if you want the whole row, or pick the field you actually need.
Experiment runs duplicate or the dataset already exists. 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.