> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# AI model evals with Apache Airflow®
Using AI produces output, and that output needs to be evaluated. This need for evaluation exists no matter where the model runs: a chat application, a local coding agent, a microservice, or [orchestrated in a pipeline](/docs/learn/ai-orchestration-overview).
*AI evals* are systematic, repeatable assessments of an AI system, and there are two types:
* **AI model evals** assess the output itself. Is it factually correct? Did the agent solve the task it was given? Does it fit the task in aspects like helpfulness and conciseness?
* [**AI product evals**](/docs/learn/ai-orchestration-product-evals) assess what happened after the output was used. Do the AI-generated ads convert? Are customers seeing value in the feature?
An AI model eval needs to include both the AI output and the traces of how the output was produced. An evaluation result is only actionable when you also know which model version was used, which tools it called, and which upstream datasets were passed to the model as context in the user prompt. Airflow can collect this information from any [micro orchestration tool](/docs/learn/ai-orchestration-overview#micro-and-macro-orchestration), along with Airflow's own [lineage information](/docs/learn/airflow-openlineage).
For a full explanation of this Dag see [Example: evaluate generated support replies](#example-evaluate-generated-support-replies).
This is a custom [Airflow plugin](/docs/learn/using-airflow-plugins) showing AI model eval results. For a full explanation of the code example see [Example: evaluate generated support replies](#example-evaluate-generated-support-replies).
For a full explanation of the code example see [Example: evaluate generated support replies](#example-evaluate-generated-support-replies).
```python evaluate_support_model_responses.py icon="python" expandable theme={null}
from typing import Literal
from airflow.sdk import dag, task, task_group
from pydantic import BaseModel, Field
from include.eval_records import (
build_eval_record,
check_reply_structure,
load_eval_results,
resolve_scored_runs,
)
from include.otel_traces import read_agent_runs
class DimensionScore(BaseModel):
score: Literal["good", "acceptable", "poor"] = Field(
description=(
"good: fully meets the criterion. acceptable: meets it with a flaw "
"a reviewer would let through. poor: fails the criterion."
)
)
confidence: Literal["low", "medium", "high"] = Field(
description=(
"How certain you are about the score you just gave for this "
"dimension only, not about the reply overall."
)
)
reasoning: str = Field(
description="One sentence supporting the score for this dimension only."
)
class SupportReplyScore(BaseModel):
"""Reference-free criteria, judged from the reply and the customer's message."""
relevance: DimensionScore = Field(
description="Whether the reply addresses what the customer actually asked."
)
helpfulness: DimensionScore = Field(
description="Whether the reply gives the customer something they can act on."
)
conciseness: DimensionScore = Field(
description="Whether the reply says what it needs to without padding or repetition."
)
class ReferenceComparison(BaseModel):
"""Reference-based criteria, judged against the reply the support team wrote."""
correctness: DimensionScore = Field(
description=(
"Whether the reply's factual claims agree with the reference reply. "
"A claim the reference contradicts, or a promise the reference "
"deliberately withholds, is not correct."
)
)
equivalence: Literal["equivalent", "close", "different"] = Field(
description=(
"equivalent: makes the same points as the reference. close: same "
"intent, misses or adds a point. different: a different reply."
)
)
confidence: Literal["low", "medium", "high"] = Field(
description="How certain you are about the equivalence judgement."
)
missing_points: list[str] = Field(
description="Points the reference makes that the generated reply does not."
)
reasoning: str = Field(description="One sentence supporting the judgement.")
@dag
def evaluate_support_model_responses():
@task
def fetch_traces() -> list[dict]:
return read_agent_runs()
@task
def resolve_tickets(agent_runs: list[dict]) -> list[dict]:
return resolve_scored_runs(agent_runs)
@task_group
def evaluate_reply(run: dict):
@task
def check_structure(run: dict) -> dict:
return check_reply_structure(run["generated_reply"])
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You score CosMarket support replies against a fixed rubric. "
"Score each dimension independently, on the reply as written. "
"Confidence applies to the single dimension it sits on, not to "
"the reply as a whole. Do not rewrite the reply."
),
output_type=SupportReplyScore,
serialize_output=True,
)
def score_rubric(run: dict) -> str:
return (
f"Customer message:\n{run['customer_ask']}\n\n"
f"Generated reply:\n{run['generated_reply']}"
)
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You compare a generated support reply against a reference reply "
"written by the support team. The reference is the source of "
"truth: judge correctness against it, not against what sounds "
"plausible. Wording differences are not differences in substance."
),
output_type=ReferenceComparison,
serialize_output=True,
)
def compare_with_reference(run: dict) -> str:
return (
f"Reference reply:\n{run['reference_reply']}\n\n"
f"Generated reply:\n{run['generated_reply']}"
)
@task
def build_record(
run: dict, structure: dict, rubric: dict, reference: dict
) -> dict:
return build_eval_record(run, structure, rubric, reference)
return build_record(
run=run,
structure=check_structure(run),
rubric=score_rubric(run),
reference=compare_with_reference(run),
)
@task
def load_metrics(records: list[dict]) -> None:
load_eval_results(records)
_runs = resolve_tickets(fetch_traces())
load_metrics(evaluate_reply.expand(run=_runs))
evaluate_support_model_responses()
```
This guide covers:
* Why use Airflow for AI model evals.
* Closed-ended and open-ended tasks, the metrics for each, and how to write a rubric.
* How to build a golden dataset.
* How to evaluate output in the context of the traces of a run.
* An example eval Dag that scores [generated support replies](#example-evaluate-generated-support-replies) with a deterministic check and two AI judges.
* An example eval Dag that reports [`pass@k` for ticket classification](#example-pass@k-for-ticket-classification).
## Assumed knowledge
To get the most out of this guide, you should have existing knowledge of:
* Airflow basics. See [Introduction to Apache Airflow®](/docs/learn/intro-to-airflow).
* Basic familiarity with [LLMs and AI agents](/docs/learn/ai-orchestration-overview).
## Why Airflow for AI model evals
AI model evals are performed using pipelines that collect the model output and all traces associated with its production, score it against pre-defined criteria, and write the scores somewhere you can compare them across versions of model choice, model settings, prompts, tools, and more.
* **Access to traces**: every [micro orchestrator](/docs/learn/ai-orchestration-overview#micro-and-macro-orchestration) records what a run did. The [Common AI provider](/docs/learn/airflow-common-ai-provider) emits OpenTelemetry GenAI spans for every model call and tool call, and harnesses such as LangChain, LangGraph, and CrewAI keep their own traces, which you retrieve through their API. Either way, an Airflow task can gather the traces and store the output evaluation results alongside them. See [Evaluate output in the context of AI traces](#evaluate-output-in-the-context-of-ai-traces).
* **Golden dataset creation**: offline evals need datasets of inputs paired with expected outputs. Extracting, transforming, and assembling these datasets from your data foundation with human input can be done with dedicated Airflow Dags. See [Build a golden dataset](#build-a-golden-dataset).
* **AI tasks in eval pipelines**: subjective criteria such as tone or whether an answer addresses the question need a model to score them. Many eval pipelines are themselves [AI orchestration](/docs/learn/ai-orchestration-overview) pipelines using LLMs and agents in an [AI-as-a-judge](/docs/learn/ai-orchestration-overview#review-ai-output) pattern.
* **Dynamic task mapping**: AI eval pipelines need to assess every model output in a defined timeframe. [Dynamic task mapping](/docs/learn/dynamic-tasks) creates one task instance per case at Dag runtime and executes them in parallel.
* **Extensive scheduling options**: model evals run on a schedule, and also in reaction to events such as a new model version or a prompt change. Airflow offers [time-based schedules](/docs/learn/scheduling-in-airflow), [asset-driven](/docs/learn/airflow-datasets), and [event-driven](/docs/learn/airflow-event-driven-scheduling) scheduling.
* **Backfills**: when you change your AI model eval, from adding a new field in an existing evaluation metric to adding an additional AI-as-a-judge, you can use [Airflow backfills](/docs/learn/rerunning-dags) to evaluate historic runs with the new criteria.
* **Dag bundle versioning**: an AI model evaluation is only actionable if you know which prompt and model configuration produced the output. Prompts and model configuration are defined in Dag code, and [versioned Dag bundles](/docs/learn/airflow-dag-versioning) allow you to track which version of that code a Dag run used.
* **Alerting**: a scoring task that compares a rate against a threshold can alert stakeholders through [Airflow notifications](/docs/learn/error-notifications-in-airflow) and [Astro alerts](/docs/astro/alerts).
* **Reliability**: AI-as-a-judge tasks benefit from the same [retry policies](/docs/learn/rerunning-dags#retry-policies) as any other Airflow task.
## Closed-ended and open-ended tasks
How you evaluate an AI model output depends on whether the task has a clear answer. Many use cases sit on a spectrum between two ends:
* **Closed-ended tasks** have one correct answer. For example, a [classification](/docs/learn/ai-orchestration-llm#classification-and-routing) can be compared against a source of truth decided on by a human subject matter expert. Similarly, a field [extracted from a document](/docs/learn/ai-orchestration-llm#extraction-of-structured-data) either matches the value in the source document or it doesn't. Evaluation is deterministic: you can compare the output against the expected value in a golden dataset and report an exact metric, similarly to metrics in traditional machine learning.
* **Open-ended tasks** have many acceptable answers. A [drafted support reply](/docs/learn/ai-orchestration-llm#generation-with-human-approval) or a [summary](/docs/learn/ai-orchestration-llm#summarization) has no single expected string to compare against. For open-ended tasks, an evaluation rubric can add clear criteria to score AI output against, both for human and AI-as-a-judge scorers.
A special case of open-ended tasks is the generation of code. In that case, code can be tested by having unit tests and integration tests, the same way human-written code is tested.
Most use cases have aspects of both types of tasks. For example, [text-to-SQL](/docs/learn/ai-orchestration-llm#text-to-sql) is closed-ended if you compare the result the query returns, and open-ended if you assess the query itself.
There are two modes for AI evals:
*Offline* evaluation means running eval pipelines on a separate schedule from the AI data product. Common patterns for offline evals are running the AI against a fixed test set of inputs and comparing the outputs against a [golden dataset](#build-a-golden-dataset) of best-in-class outputs, or running scheduled batches of [AI product evals](/docs/learn/ai-orchestration-product-evals) after gathering signals further downstream, such as customer behavior over time after interaction with the AI data product.
*Online* evaluation scores AI output during live runs of the pipeline, for example in AI-as-a-judge or human-in-the-loop steps within an AI orchestration pipeline. Online evals offer the possibility to act on the evaluation result in the live pipeline, for example preventing a bad AI output from being sent to a customer.
### Metrics for closed-ended tasks
A closed-ended task is a prediction problem with labeled ground truth, so the metrics are the ones already used for traditional machine learning models.
For example, in the case of [classification](/docs/learn/ai-orchestration-llm#classification-and-routing), accuracy, the fraction of outputs that match the expected value, is often a good overall metric. Precision, recall, and F1 per class, plus a confusion matrix showing which labels the model confuses with which, add additional information about model performance.
For an overview of definitions for metrics in traditional machine learning, see [Metrics and scoring](https://scikit-learn.org/stable/modules/model_evaluation.html#classification-metrics) in the scikit-learn documentation.
### Metrics for open-ended tasks
Two properties of an open-ended output can in part be measured without a (human or AI) judgment call: functional correctness and similarity against reference data.
**Functional correctness** is whether the AI output performs the function that was asked for. If the prompt asked for an email, is what came back a text that can be reasonably described as an email? Whether you can answer this question deterministically depends on the output format:
* **Machine-checkable artifacts** are deterministic to evaluate. Generated code either parses or it doesn't, and once it parses, a test suite evaluates whether the code performs the intended task. Generated SQL either executes and returns the expected result or it doesn't. A generated configuration file either validates against its schema or it doesn't. Each of these checks can be run as an Airflow task. Give that task a read-only connection and a fixed snapshot to run against: on live tables, a reference query and a generated query executed minutes apart can disagree because the data moved rather than because the model was wrong.
* **Prose** is more difficult to evaluate with deterministic code. You can add structural checks like whether a report contains every section that a template requires, whether there are any placeholder characters left, or whether a summary is shorter than its input. Most of the time prose needs an AI-as-a-judge or human evaluation.
Where it applies, functional correctness turns part of an open-ended task back into an exact evaluation, because two outputs that look nothing alike both count as correct when both work.
**`pass@k`** is how a functional correctness evaluation is reported as a single number, and it is the score code-generation benchmarks are ranked by. `k` is the number of solutions the model is allowed to generate per problem, and `pass@k` is the percentage of problems in the dataset where at least one of those `k` solutions passes the tests. A model with a `pass@1` of 60% and a `pass@10` of 85% solves 60% of the problems on its first attempt and 85% of them within ten attempts.
Which `k` score is relevant for your use case depends on whether your pipeline can tell a passing solution from a failing one. With downstream tests the pipeline can: generate `k` candidates in a [mapped](/docs/learn/dynamic-tasks) task, run the test suite against each of them, and only pass a candidate that succeeded to the downstream task. If there is no downstream quality check in your pipeline, every output the model produces is used, so `pass@1` is the relevant quality metric.
**Similarity against reference data** is often used when correctness can't be checked directly. It measures how close the output is to a *reference response*: an output for the same input that is known to be good, also called a *ground truth* or a *canonical response*. A metric that needs one is **reference-based**, and a metric that doesn't is **reference-free**. Functional correctness and a judge scoring against a rubric are reference-free. Similarity is reference-based, so it needs a [golden dataset](#build-a-golden-dataset) where every entry has a reference response alongside its input.
How you compare the output to the reference depends on its length and complexity:
* **Exact match** works for short outputs: a label, an identifier, a single extracted field. For anything longer, two equally good answers rarely match character for character, so an exact match rate reports failures that aren't failures.
* **Lexical similarity** measures overlap in the words themselves. The two most common methods are *fuzzy matching*, which scores how many single-character edits separate the output from the reference, and *n-gram similarity*, which scores how many word sequences of a given length the two share.
* **Semantic similarity**, also called *embedding similarity*, measures overlap in meaning rather than in wording. Embed both the output and the reference and take the cosine distance between the two vectors, so an answer that uses different words for the same concepts still shows close similarity.
* **AI-as-a-judge** is given both the reference and the generated output and scores whether they say the same thing.
Some criteria can be evaluated by neither functional correctness nor similarity. Tone, whether a reply addresses what was asked, and whether a summary left out something important all need a judge scoring against a rubric.
### Write a rubric for subjective criteria
A subjective evaluation gets more exact when you replace the general question, whether the output is any good, with explicit guidelines: a rubric of dimensions, each with defined levels and a confidence score.
Scoring against a rubric is relevant both for human and AI judges, and in cases where AI output is scored against reference data or on its own. A [human-in-the-loop task](/docs/learn/airflow-human-in-the-loop) puts a person in that judge position, waiting for them to approve, edit, or reject the output, and records what they decided.
When you are using LLMs in an AI-as-a-judge pattern, those AI calls also incur token cost. Online evaluation additionally adds latency to your pipeline if done before the result is sent to the stakeholder. In a lot of cases you can use a cheaper `model_id` with lower `usage_limits` in the judge task than in the task being judged.
#### Dimensions
Which dimensions to pick heavily depends on your use case. For the support ticket replies in the example above:
* **Relevance**: whether the reply addresses what the customer actually asked.
* **Helpfulness**: whether the reply gives the customer something they can act on.
* **Conciseness**: whether the reply says what it needs to without padding or repetition.
* **Correctness**: whether the factual claims agree with the reply the support team wrote. This one is reference-based: the reference is either a golden record, or the AI judge needs access to source-of-truth data.
For generated SQL the useful dimensions are different:
* **Correctness**: whether the query answers the question that was asked. Here it can be [checked by execution](#metrics-for-open-ended-tasks) as well as read.
* **Efficiency**: whether the query avoids full scans, cross joins, and columns the question doesn't need.
* **Readability**: whether a reviewer would accept the naming, the CTEs, and the structure.
* **Assumptions**: whether the query guesses at something ambiguous in the question, such as which date column to filter on.
#### Fields
When using the Common AI provider, the rubric is the `output_type` of the judge task, with a list of dimensions, each with at least three fields:
* **Score**, with a list of levels and their meaning. For example, a `Literal["very good", "good", "acceptable", "poor"]` with a description and ideally an example for each level.
* **Confidence**: how confident the judge was in the assessment of this dimension. Confidence scores are often used in deciding which records to route to a human reviewer.
* **Reasoning**: giving a judge the option to explain the reasoning for a score helps when auditing pipelines, and this data can be used to find patterns over time.
The Common AI decorators send the `output_type` schema to the AI model as part of the request, including every field name, type, and description, so it is not necessary to repeat this information in the system prompt.
For a rubric built this way, see `SupportReplyScore` in [Example: evaluate generated support replies](#example-evaluate-generated-support-replies).
Some outputs are easier to evaluate as a black box than piece by piece. For a Dag that drafts sales outreach, the reply rate per generated email is one number that comes out of the CRM, while deciding whether an individual email is a good email needs a rubric, a judge, and reference data. In these cases, teams sometimes build the [AI product eval](/docs/learn/ai-orchestration-product-evals) first: it measures the end-to-end outcome, and the cases flagged during AI product eval can be a good start to define AI model eval criteria.
Eventually, it is recommended to always combine both types of evals, AI model and AI product evals, to avoid situations where the system only optimizes for what is evaluated. For example, sales outreach emails that generate a high response rate, but only because they promise hallucinated features.
Many task types have public benchmarks, and model providers publish how their models score on them. Those numbers are a reasonable way to pick which model to try first. A benchmark measures general capability on someone else's data, so whether a model works for your use case is a separate question that can only be answered by your own AI eval pipelines.
## Build a golden dataset
A golden dataset is a set of inputs paired with the expected output for each one: the labels a closed-ended task should return, or the reference record a reference-based metric compares against. Assembling it and keeping it current is an Airflow pipeline like any other: extract candidate cases from production, produce or collect the expected output, and load the result to a table or object storage.
The expected outputs come from one of three places:
* **Human-written**: a person writes the reference records for each input.
* **AI-generated**: run the most capable model you have access to, without the cost constraints of production, and keep its output as the gold-standard reference. Your production pipeline then runs a cheaper model and the eval asks whether it comes close enough to the expensive model's answer. Note what this measures. The reference is the ceiling a larger model reached on the same input, and it has not been verified as correct, so a case where both models are wrong would score as a pass.
* **Mixed**: generate candidate references with the expensive model and have a person review, edit, and approve them. A [human-in-the-loop task](/docs/learn/airflow-human-in-the-loop) records the approved version, and the edits themselves show where the generating model fell short and can be used to improve the golden dataset generation pipeline in a [context graph](/docs/learn/reference-architecture-context-graph) pattern.
Production cases that a judge or a reviewer scored can be promoted into the next version of the dataset. Version the dataset when you do that, and record which version an eval run used: a score computed against a changed dataset is not comparable to the one before it.
## Evaluate output in the context of AI traces
Your AI model eval pipeline scores AI output. There are many possible reasons for low-quality AI results:
* The model isn't sophisticated enough for the task.
* The prompt did not contain relevant context, or the context given was wrong or outdated.
* The prompt was too long or the context window too small, leading to truncation or compaction and loss of information.
* The output token limit cut the answer off before it was finished.
An agent has more ways to produce a bad result, because it chooses its own steps:
* A tool failed or returned nothing, and the agent answered anyway.
* The agent called the wrong tool, or called the right tool with the wrong arguments.
* A tool the agent needed wasn't available, so it guessed instead.
The trace of the run gives you the diagnostic information to tell these causes apart, including which model version was used, how many reasoning tokens were used, and which tools an agent called.
AI model eval pipelines typically write the LLM/agent trace information into the same record as the evaluation score.
### Common AI and OpenTelemetry
The [Common AI provider](/docs/learn/airflow-common-ai-provider) records traces as [OpenTelemetry](https://opentelemetry.io/) GenAI spans, one per agent run, model call, and tool call, nested under the Airflow task span.
Enable tracing by setting the following environment variables for your exporter and collector:
```text theme={null}
AIRFLOW__TRACES__OTEL_ON=True
AIRFLOW__COMMON_AI__OTEL_EXPORT_ENABLED=True
OTEL_EXPORTER_OTLP_ENDPOINT=
OTEL_SERVICE_NAME=
```
The provider doesn't configure an exporter of its own; it reuses the one core tracing installs with `AIRFLOW__TRACES__OTEL_ON=True`.
`AIRFLOW__COMMON_AI__CAPTURE_CONTENT=True` adds prompt and completion text to the spans. Airflow's secret masking applies to logs and rendered template fields, not to span attributes, so that content is exported unredacted.
An agent run exported to OpenTelemetry by the Common AI provider looks like this (long content values are shortened):
```json expandable wrap theme={null}
{
"resourceSpans": [
{
"resource": {
"attributes": [
{ "key": "service.name", "value": { "stringValue": "orchestrate-everything-patterns" } },
{ "key": "telemetry.sdk.name", "value": { "stringValue": "opentelemetry" } }
]
},
"scopeSpans": [
{
"scope": { "name": "pydantic-ai", "version": "2.28.0" },
"spans": [
{
"name": "invoke_agent agent",
"traceId": "d992c4d1520b3bd20071897f34c9eea1",
"spanId": "04ad168df362edd5",
"parentSpanId": "1ac23bb1ff7b88c0",
"startTimeUnixNano": "1787234596927110073",
"endTimeUnixNano": "1787234636015812279",
"attributes": [
{ "key": "gen_ai.operation.name", "value": { "stringValue": "invoke_agent" } },
{ "key": "model_name", "value": { "stringValue": "gpt-5-mini" } },
{ "key": "gen_ai.aggregated_usage.input_tokens", "value": { "intValue": "9603" } },
{ "key": "gen_ai.aggregated_usage.output_tokens", "value": { "intValue": "2256" } },
{ "key": "gen_ai.aggregated_usage.details.reasoning_tokens", "value": { "intValue": "1600" } },
{ "key": "gen_ai.aggregated_usage.details.cache_read_tokens", "value": { "intValue": "1408" } },
{ "key": "final_result", "value": { "stringValue": "{\"body\":\"Hello Benjamin Sisko,\\n\\nThanks, I can confirm we received your cancellation request for order ORD-89104 ...\"}" } },
{ "key": "pydantic_ai.all_messages", "value": { "stringValue": "[{\"role\":\"user\",\"parts\":[{\"type\":\"text\",\"content\":\"ticket_id: TKT-4004\\ncustomer_id: CUS-457\\n ...\"}]}]" } },
{ "key": "gen_ai.system_instructions", "value": { "stringValue": "[{\"type\":\"text\",\"content\":\"You draft customer support replies for CosMarket ...\"}]" } }
]
}
]
}
]
}
]
}
```
To fetch tracing information from OpenTelemetry you need to connect to your OpenTelemetry backend, then search through the span tree for the relevant attributes.
For other micro orchestrators like LangChain or CrewAI, see the relevant API documentation on how to export agent traces.
## Example: evaluate generated support replies
A support agent drafts replies to incoming tickets, reading the order record and the support policy through its tools, all orchestrated with [Agent orchestration](/docs/learn/ai-orchestration-agent). The AI model eval Dag fetches the agent traces and evaluates the AI output with one deterministic and two LLM judge tasks.
This is a custom [Airflow plugin](/docs/learn/using-airflow-plugins) showing AI model eval results.
```python evaluate_support_model_responses.py icon="python" expandable theme={null}
from typing import Literal
from airflow.sdk import dag, task, task_group
from pydantic import BaseModel, Field
from include.eval_records import (
build_eval_record,
check_reply_structure,
load_eval_results,
resolve_scored_runs,
)
from include.otel_traces import read_agent_runs
class DimensionScore(BaseModel):
score: Literal["good", "acceptable", "poor"] = Field(
description=(
"good: fully meets the criterion. acceptable: meets it with a flaw "
"a reviewer would let through. poor: fails the criterion."
)
)
confidence: Literal["low", "medium", "high"] = Field(
description=(
"How certain you are about the score you just gave for this "
"dimension only, not about the reply overall."
)
)
reasoning: str = Field(
description="One sentence supporting the score for this dimension only."
)
class SupportReplyScore(BaseModel):
"""Reference-free criteria, judged from the reply and the customer's message."""
relevance: DimensionScore = Field(
description="Whether the reply addresses what the customer actually asked."
)
helpfulness: DimensionScore = Field(
description="Whether the reply gives the customer something they can act on."
)
conciseness: DimensionScore = Field(
description="Whether the reply says what it needs to without padding or repetition."
)
class ReferenceComparison(BaseModel):
"""Reference-based criteria, judged against the reply the support team wrote."""
correctness: DimensionScore = Field(
description=(
"Whether the reply's factual claims agree with the reference reply. "
"A claim the reference contradicts, or a promise the reference "
"deliberately withholds, is not correct."
)
)
equivalence: Literal["equivalent", "close", "different"] = Field(
description=(
"equivalent: makes the same points as the reference. close: same "
"intent, misses or adds a point. different: a different reply."
)
)
confidence: Literal["low", "medium", "high"] = Field(
description="How certain you are about the equivalence judgement."
)
missing_points: list[str] = Field(
description="Points the reference makes that the generated reply does not."
)
reasoning: str = Field(description="One sentence supporting the judgement.")
@dag
def evaluate_support_model_responses():
@task
def fetch_traces() -> list[dict]:
return read_agent_runs()
@task
def resolve_tickets(agent_runs: list[dict]) -> list[dict]:
return resolve_scored_runs(agent_runs)
@task_group
def evaluate_reply(run: dict):
@task
def check_structure(run: dict) -> dict:
return check_reply_structure(run["generated_reply"])
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You score CosMarket support replies against a fixed rubric. "
"Score each dimension independently, on the reply as written. "
"Confidence applies to the single dimension it sits on, not to "
"the reply as a whole. Do not rewrite the reply."
),
output_type=SupportReplyScore,
serialize_output=True,
)
def score_rubric(run: dict) -> str:
return (
f"Customer message:\n{run['customer_ask']}\n\n"
f"Generated reply:\n{run['generated_reply']}"
)
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You compare a generated support reply against a reference reply "
"written by the support team. The reference is the source of "
"truth: judge correctness against it, not against what sounds "
"plausible. Wording differences are not differences in substance."
),
output_type=ReferenceComparison,
serialize_output=True,
)
def compare_with_reference(run: dict) -> str:
return (
f"Reference reply:\n{run['reference_reply']}\n\n"
f"Generated reply:\n{run['generated_reply']}"
)
@task
def build_record(
run: dict, structure: dict, rubric: dict, reference: dict
) -> dict:
return build_eval_record(run, structure, rubric, reference)
return build_record(
run=run,
structure=check_structure(run),
rubric=score_rubric(run),
reference=compare_with_reference(run),
)
@task
def load_metrics(records: list[dict]) -> None:
load_eval_results(records)
_runs = resolve_tickets(fetch_traces())
load_metrics(evaluate_reply.expand(run=_runs))
evaluate_support_model_responses()
```
The Dag has seven tasks:
* **Extract**: `fetch_traces` reads the exported spans and returns one record per agent run, with its final output and its tool calls. This example fetches the agent traces from OpenTelemetry with `[common.ai] capture_content` turned on, which adds the prompt and completion text to those spans. See [Observability](/docs/learn/ai-orchestration-llm#observability).
* **Resolve**: `resolve_tickets` matches each agent run back to the ticket it answered. The ticket ID is available from the captured prompt, so the task reads it out of the trace and joins the run to that ticket's customer message and, if available, to its reference reply from the golden dataset.
* **Score**: the `evaluate_reply` task group is [mapped](/docs/learn/dynamic-tasks) over those runs, so each reply is scored by its own set of task instances in parallel. There are three scorers, a deterministic check and the two judges:
* `check_structure` is a deterministic check with no model call, on the format of the reply: it has content, it is within a length band, it has no unfilled placeholders, and it is signed off.
* `score_rubric` is the reference-free judge. It scores relevance, helpfulness, and conciseness, each with its own score, confidence, and reasoning.
* `compare_with_reference` is the reference-based judge. It scores correctness against the reply the support team wrote, and reports which of the reference's points the generated reply left out.
* `build_record` aggregates the three results and the trace metadata into one row.
* **Load**: `load_metrics` writes one row per reply, then the overall evaluation across all scored inputs: the good rate and the low-confidence rate for every dimension, the share equivalent to the reference, and the average tokens, latency, and tool calls.
The results are displayed in a custom Airflow plugin.
## Example: pass\@k for ticket classification
Classifying a support ticket is a [closed-ended task](#closed-ended-and-open-ended-tasks), so the verification task is one equality check against the label in the golden dataset. This Dag classifies every ticket `k` times and reports [`pass@k`](#metrics-for-open-ended-tasks) alongside majority vote accuracy.
```python evaluate_ticket_classification.py icon="python" expandable theme={null}
from typing import Literal
from airflow.sdk import Param, dag, get_current_context, task
from pydantic import BaseModel, Field
from include.ticket_classification import (
CATEGORIES,
labelled_tickets,
sampling_cases,
score_pass_at_k,
)
class TicketClassification(BaseModel):
category: Literal[CATEGORIES] = Field(
description="The single category that best fits what the customer is asking for."
)
reasoning: str = Field(description="One sentence supporting the category.")
@dag(
params={
"k": Param(
5,
type="integer",
minimum=1,
maximum=20,
description="How many times each ticket is classified in total.",
)
},
)
def evaluate_ticket_classification():
@task
def fetch_labelled_tickets() -> list[dict]:
return labelled_tickets()
@task
def build_sampling_cases(tickets: list[dict]) -> list[dict]:
return sampling_cases(tickets, attempts=get_current_context()["params"]["k"])
@task.llm(
llm_conn_id="pydanticai_default",
model_id="openai:gpt-4.1-nano",
system_prompt=(
"You categorise incoming CosMarket support tickets. Pick the single "
"category that best fits what the customer is asking for, based only "
"on their message."
),
output_type=TicketClassification,
serialize_output=True,
agent_params={"model_settings": {"temperature": 1.0}},
map_index_template="{{ task.op_kwargs.case.ticket_id }} #{{ task.op_kwargs.case.attempt }}",
)
def classify_ticket(case: dict) -> str:
return case["customer_ask"]
@task
def report_pass_at_k(cases: list[dict], classifications: list[dict]) -> dict:
return score_pass_at_k(cases, classifications)
_cases = build_sampling_cases(fetch_labelled_tickets())
report_pass_at_k(
cases=_cases,
classifications=classify_ticket.expand(case=_cases),
)
evaluate_ticket_classification()
```
`k` can be set as a [Dag param](/docs/learn/airflow-params). `build_sampling_cases` returns one case per ticket and attempt, and `classify_ticket` is mapped over that list, so 9 tickets at `k=5` is 45 task instances.
`report_pass_at_k` groups the samples by ticket and reports three numbers:
* **`pass@1`**: the share of tickets where the first sample matched the label.
* **`pass@k`**: the share where any of the `k` samples matched.
* **Majority vote accuracy**: the share where the most frequent label matched.
## Next steps
* Assess what happened after the output was used with [AI product evals](/docs/learn/ai-orchestration-product-evals).
* Run a single model inference as a task with [LLM orchestration with Apache Airflow®](/docs/learn/ai-orchestration-llm).
* Give a model tools and multi-step reasoning with [Agent orchestration](/docs/learn/ai-orchestration-agent).
* Add human decisions and capture their output with [Human-in-the-loop workflows with Apache Airflow®](/docs/learn/airflow-human-in-the-loop).