> ## 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 product evals with Apache Airflow®
AI in production creates AI data products. This is true no matter how you orchestrate the AI inference itself, whether your team is using a generic chat application, a local coding agent, a microservice, or [AI orchestrated in a pipeline](/docs/learn/ai-orchestration-overview). An AI data product might be an AI-generated email you send to a customer, the answers from a support chatbot on your website, or a report sent to your executive team about customer sentiment. In short, a [data product](/docs/learn/data-products) is an asset produced by one or more pipelines that a stakeholder cares about. An AI data product is any such asset where AI was involved in the creation of the final output.
AI evals are systematic, repeatable assessments of those AI data products. There are two different types of AI evals:
* **AI model evals**: evaluate the direct output of the AI. Is a produced asset factually correct? Did the agent solve its task successfully? See [AI model evals](/docs/learn/ai-orchestration-model-evals).
* **AI product evals**: these evaluations are closer to the business outcome and measure the results of the entire AI data product. Does using the AI-generated ads lead to more conversions? Are customers using our AI features and seeing value in them?
Both types of evals depend on signals gathered over time from different locations and systems. On a regular schedule you need to collect AI output and downstream results, transform that data, score it, and associate it with the AI aspects that created it (model, model settings, context used, token cost). At its core this is ETL, and it can all be orchestrated by [Apache Airflow®](https://airflow.apache.org/).
For a full explanation of this Dag see [Example: evaluate a support ticket agent](#example-evaluate-a-support-ticket-agent).
For a full explanation of the code example see [Example: evaluate a support ticket agent](#example-evaluate-a-support-ticket-agent).
```python evaluate_support_threads.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.metrics_store import write_eval_records, write_metrics
from include.support_systems import (
CSAT_DIMENSIONS,
fetch_customer_replies,
fetch_satisfaction_scores,
fetch_support_replies,
)
class SupportReplyScore(BaseModel):
accurate: bool = Field(
description="True if every claim in the support response is supported by the ticket or the order record."
)
addresses_question: bool = Field(
description="True if the support response answers what the customer asked."
)
tone: Literal["good", "acceptable", "poor"] = Field(
description="Fit of the tone for a customer on this account tier."
)
reasoning: str = Field(description="One sentence supporting the scores.")
class CustomerReplyScore(BaseModel):
sentiment: Literal["positive", "neutral", "negative"] = Field(
description="Overall sentiment the customer expresses in their reply."
)
satisfied: bool = Field(
description="True if the customer considers their request handled."
)
escalation_risk: Literal["low", "medium", "high"] = Field(
description="Risk that this customer escalates or churns without follow-up."
)
reasoning: str = Field(description="One sentence supporting the scores.")
@dag(schedule="@daily")
def evaluate_support_threads():
@task
def fetch_replies_from_helpdesk() -> list[dict]:
return fetch_support_replies()
@task
def fetch_replies_from_inbox() -> list[dict]:
return fetch_customer_replies()
@task
def fetch_ratings_from_survey_tool() -> list[dict]:
return fetch_satisfaction_scores()
@task
def join_by_ticket(
support_replies: list[dict],
customer_replies: list[dict],
satisfaction_scores: list[dict],
) -> list[dict]:
customer_by_ticket = {r["ticket_id"]: r for r in customer_replies}
survey_by_ticket = {s["ticket_id"]: s for s in satisfaction_scores}
return [
{
"ticket_id": support["ticket_id"],
"customer_ask": support["customer_ask"],
"order_record": support["order_record"],
"support_response": support["support_response"],
"customer_response": customer_by_ticket[support["ticket_id"]][
"customer_response"
],
"ratings": survey_by_ticket[support["ticket_id"]]["ratings"],
}
for support in support_replies
if support["ticket_id"] in customer_by_ticket
and support["ticket_id"] in survey_by_ticket
]
@task_group
def evaluate_thread(thread: dict):
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You score customer support responses against the evidence "
"provided. Judge only what the evidence supports. Do not "
"rewrite the response."
),
output_type=SupportReplyScore,
serialize_output=True,
)
def score_support_reply(thread: dict) -> str:
return (
f"Customer ask:\n{thread['customer_ask']}\n\n"
f"Order record:\n{thread['order_record']}\n\n"
f"Support response:\n{thread['support_response']}"
)
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You score how a customer reacted to a support response. Judge "
"only the customer's reply, using the earlier messages as "
"context."
),
output_type=CustomerReplyScore,
serialize_output=True,
)
def score_customer_reply(thread: dict) -> str:
return (
f"Customer ask:\n{thread['customer_ask']}\n\n"
f"Support response:\n{thread['support_response']}\n\n"
f"Customer reply:\n{thread['customer_response']}"
)
@task
def score_satisfaction(thread: dict) -> dict:
ratings = thread["ratings"]
return {
**{f"csat_{d}": ratings[d] for d in CSAT_DIMENSIONS},
"csat_average": sum(ratings[d] for d in CSAT_DIMENSIONS)
/ len(CSAT_DIMENSIONS),
"detractor": any(ratings[d] <= 2 for d in CSAT_DIMENSIONS),
}
@task
def build_record(
thread: dict,
reply_score: dict,
customer_score: dict,
satisfaction_score: dict,
) -> dict:
return {
"ticket_id": thread["ticket_id"],
"reply_accurate": reply_score["accurate"],
"reply_addresses_question": reply_score["addresses_question"],
"reply_tone": reply_score["tone"],
"reply_reasoning": reply_score["reasoning"],
"customer_sentiment": customer_score["sentiment"],
"customer_satisfied": customer_score["satisfied"],
"escalation_risk": customer_score["escalation_risk"],
"customer_reasoning": customer_score["reasoning"],
**satisfaction_score,
}
return build_record(
thread=thread,
reply_score=score_support_reply(thread),
customer_score=score_customer_reply(thread),
satisfaction_score=score_satisfaction(thread),
)
@task
def load_metrics(records: list[dict]) -> None:
write_eval_records(records)
total = len(records)
write_metrics(
tickets_scored=total,
accuracy_rate=sum(r["reply_accurate"] for r in records) / total,
relevance_rate=sum(r["reply_addresses_question"] for r in records) / total,
poor_tone_rate=sum(r["reply_tone"] == "poor" for r in records) / total,
positive_sentiment_rate=sum(
r["customer_sentiment"] == "positive" for r in records
)
/ total,
satisfied_rate=sum(r["customer_satisfied"] for r in records) / total,
high_escalation_risk_rate=sum(
r["escalation_risk"] == "high" for r in records
)
/ total,
detractor_rate=sum(r["detractor"] for r in records) / total,
**{
f"avg_csat_{d}": sum(r[f"csat_{d}"] for r in records) / total
for d in CSAT_DIMENSIONS
},
)
_threads = join_by_ticket(
support_replies=fetch_replies_from_helpdesk(),
customer_replies=fetch_replies_from_inbox(),
satisfaction_scores=fetch_ratings_from_survey_tool(),
)
load_metrics(evaluate_thread.expand(thread=_threads))
evaluate_support_threads()
```
This guide covers:
* Why to use Airflow for AI product evals
* What AI outputs you can evaluate with Airflow
* AI product evals with Airflow
* An example AI product evals Airflow pipeline for a support ticket agent
## 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 product evals
Evaluating AI means extracting AI output and feedback signals from various systems, transforming structured signals, using LLMs to assess signals in unstructured data, computing a score, associating the outcome with the input settings of your AI application, and writing the results of the evaluation to a central location. This type of pipeline, in essence extracting, transforming, and loading data, is exactly what Airflow excels at.
* **Tool agnostic**: AI eval signals need to be collected from different external systems. Airflow can connect to any system that has an API, with many pre-built modules available in [Airflow providers](https://airflow.apache.org/registry/).
* **Extensive scheduling options**: AI evals need to run at regular intervals as well as in reaction to events like switching to a new model version. Airflow offers extensive scheduling options including [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**: AI eval signals are created over time. Some (AI model eval signals) appear as soon as the AI task finishes, while others might only appear weeks or months later (like a customer response). With Airflow you can [backfill](/docs/learn/rerunning-dags) an eval Dag to compute newly discovered eval signals for historic data.
* **Dynamic task mapping**: In many cases, you'll want to evaluate many AI outputs in parallel. [Dynamic task mapping](/docs/learn/dynamic-tasks) lets you create one parallel task instance per AI output you are assessing.
* **AI tasks in eval pipelines**: Signals in unstructured data, like the sentiment of a customer's reply, benefit from an LLM to score them. Airflow can orchestrate that model call in the same pipeline. See [Classification and routing](/docs/learn/ai-orchestration-llm#classification-and-routing).
* **Alerting**: AI evaluations are run on a constant schedule to monitor AI data product performance. If the evaluation shows a decline in outcome, you can use [Airflow notifications](/docs/learn/error-notifications-in-airflow) and [Astro alerts](/docs/astro/alerts) to send a notification to the correct team.
* **Dag bundle versioning**: An eval result is only actionable if you know which model, prompt, and settings produced it. 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.
* **Reliability**: Evals need to keep running over long periods of time. Retries, alerting, and monitoring apply to eval pipelines the same as to any other Dag.
* **One place for related pipelines**: Eval pipelines read from the same sources as the rest of your data pipelines. ETL, [context engineering](https://www.astronomer.io/ebooks/ai-context-engineering-with-apache-airflow/), AI, and MLOps pipelines already run in Airflow and use the same connections and warehouses.
## What AI outputs can you evaluate with Airflow?
Any AI output, wherever the model runs, should be evaluated on a regular cadence against business metrics and KPIs, and ultimately to assess ROI. AI data products can generally be classified into:
* LLM-derived AI data products:
* **Text drafted for a person to send**: support replies, sales outreach, release notes. See [Generation with human approval](/docs/learn/ai-orchestration-llm#generation-with-human-approval).
* **Classifications and routing decisions**: ticket categories, priorities, lead scores, content moderation labels. See [Classification and routing](/docs/learn/ai-orchestration-llm#classification-and-routing).
* **Extracted structured data**: fields pulled from contracts, invoices, or call transcripts. See [Extraction of structured data](/docs/learn/ai-orchestration-llm#extraction-of-structured-data).
* **Generated code and queries**: pull requests, SQL, configuration, migrations. See [Text-to-SQL](/docs/learn/ai-orchestration-llm#text-to-sql).
* **Summaries and reports**: meeting notes, account summaries, incident write-ups. See [Summarization](/docs/learn/ai-orchestration-llm#summarization).
* **Translations and localized content**. See [Transformation](/docs/learn/ai-orchestration-llm#transformation).
* **Recommendations and rankings**: personalized product suggestions, search result ordering, content feeds.
* Agent-derived data products:
* **Answers to open questions about your data**: a report created based on several queries, for example for ad-hoc data questions. See [Data exploration agent](/docs/learn/ai-orchestration-agent#data-exploration-agent), and [Building Kepler, Astronomer's internal data assistant](https://www.astronomer.io/blog/building-kepler-astronomer-internal-data-assistant) for examples.
* **Retrieval quality**: which chunks a query returned, and whether they contained the answer.
* **Actions an agent took in an external system**: an account tier upgrade, a refund, a support ticket escalation. See [Support ticket agent](/docs/learn/ai-orchestration-agent#support-ticket-agent), and [AI-powered education operations with Apache Airflow®](/docs/learn/reference-architecture-ai-education-operations) for examples.
* **Code changes proposed by a group of agents**: pull requests from a bug report or feature request. See [Agentic software development](/docs/learn/ai-orchestration-agent#agentic-software-development), and the [Together AI case study](https://www.astronomer.io/case-studies/together-ai/) for agents authoring dbt models and Dags.
* **Diagnoses**: root-cause analyses and suggested fixes. See [Root-cause analysis and self-healing pipelines](/docs/learn/ai-orchestration-agent#root-cause-analysis-and-self-healing-pipelines) and [How to use Otto to automatically investigate Dag failures and PR a fix](/docs/learn/airflow-otto-rca-auto-fix).
* **Complex, customer-facing AI products**: several agents behind one AI data product, each with its own output and requiring its own AI eval pipeline. See the [Booking.com case study](https://www.astronomer.io/case-studies/booking/).
You don't need to run AI inside Airflow to run evals in Airflow. A chat app, a coding assistant, and an agent deployed in a microservice all produce output and downstream effects that a Dag can collect and score. [Orchestrating the AI in Airflow as well](/docs/learn/ai-orchestration-overview) gives you easier access to information about model input, settings, and output.
## AI product evals with Airflow
A product eval scores what happened after the output was used, so the signals come from the systems your stakeholders work in: ticketing, CRM, source control, product analytics, or billing. Some are available as structured data, such as ratings, click-through rate, or even emoji reactions on a social post. Other signals are unstructured, for example, whether a person edited the output before using it, the sentiment of a customer reply, or downstream reviews.
Those signals arrive minutes to months after the AI produced the output, which is why you need to gather them continuously using Airflow Dags with similar structures to ETL use cases, extracting, transforming, and loading this data into a central location for further processing and scoring over time.
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](/docs/learn/ai-orchestration-model-evals#build-a-golden-dataset) of best-in-class outputs, or running scheduled batches of AI 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.
## Example: evaluate a support ticket agent
An AI agent drafts replies to incoming tickets, a human reviews them, and the approved reply is sent to the customer, all orchestrated with an Airflow Dag using [Agent orchestration](/docs/learn/ai-orchestration-agent). This pipeline creates an AI data product, the response to the customer, that needs to be evaluated regularly to ensure correctness, quality, and high customer satisfaction.
The AI orchestration Dag already includes some initial online evaluation steps: an AI-as-a-judge and a human-in-the-loop task. Both are [AI model evals](/docs/learn/ai-orchestration-model-evals), evaluations only concerned with the model output itself. To evaluate the AI data product end to end, you need an additional Dag that retrieves downstream signals: the customer's response, satisfaction rating, and scoring of the AI-generated ticket reply after model eval steps.
These AI product evals are run in a second Dag, once a day on all tickets that have received a customer reply and customer satisfaction rating:
```python evaluate_support_threads.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.metrics_store import write_eval_records, write_metrics
from include.support_systems import (
CSAT_DIMENSIONS,
fetch_customer_replies,
fetch_satisfaction_scores,
fetch_support_replies,
)
class SupportReplyScore(BaseModel):
accurate: bool = Field(
description="True if every claim in the support response is supported by the ticket or the order record."
)
addresses_question: bool = Field(
description="True if the support response answers what the customer asked."
)
tone: Literal["good", "acceptable", "poor"] = Field(
description="Fit of the tone for a customer on this account tier."
)
reasoning: str = Field(description="One sentence supporting the scores.")
class CustomerReplyScore(BaseModel):
sentiment: Literal["positive", "neutral", "negative"] = Field(
description="Overall sentiment the customer expresses in their reply."
)
satisfied: bool = Field(
description="True if the customer considers their request handled."
)
escalation_risk: Literal["low", "medium", "high"] = Field(
description="Risk that this customer escalates or churns without follow-up."
)
reasoning: str = Field(description="One sentence supporting the scores.")
@dag(schedule="@daily")
def evaluate_support_threads():
@task
def fetch_replies_from_helpdesk() -> list[dict]:
return fetch_support_replies()
@task
def fetch_replies_from_inbox() -> list[dict]:
return fetch_customer_replies()
@task
def fetch_ratings_from_survey_tool() -> list[dict]:
return fetch_satisfaction_scores()
@task
def join_by_ticket(
support_replies: list[dict],
customer_replies: list[dict],
satisfaction_scores: list[dict],
) -> list[dict]:
customer_by_ticket = {r["ticket_id"]: r for r in customer_replies}
survey_by_ticket = {s["ticket_id"]: s for s in satisfaction_scores}
return [
{
"ticket_id": support["ticket_id"],
"customer_ask": support["customer_ask"],
"order_record": support["order_record"],
"support_response": support["support_response"],
"customer_response": customer_by_ticket[support["ticket_id"]][
"customer_response"
],
"ratings": survey_by_ticket[support["ticket_id"]]["ratings"],
}
for support in support_replies
if support["ticket_id"] in customer_by_ticket
and support["ticket_id"] in survey_by_ticket
]
@task_group
def evaluate_thread(thread: dict):
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You score customer support responses against the evidence "
"provided. Judge only what the evidence supports. Do not "
"rewrite the response."
),
output_type=SupportReplyScore,
serialize_output=True,
)
def score_support_reply(thread: dict) -> str:
return (
f"Customer ask:\n{thread['customer_ask']}\n\n"
f"Order record:\n{thread['order_record']}\n\n"
f"Support response:\n{thread['support_response']}"
)
@task.llm(
llm_conn_id="pydanticai_default",
system_prompt=(
"You score how a customer reacted to a support response. Judge "
"only the customer's reply, using the earlier messages as "
"context."
),
output_type=CustomerReplyScore,
serialize_output=True,
)
def score_customer_reply(thread: dict) -> str:
return (
f"Customer ask:\n{thread['customer_ask']}\n\n"
f"Support response:\n{thread['support_response']}\n\n"
f"Customer reply:\n{thread['customer_response']}"
)
@task
def score_satisfaction(thread: dict) -> dict:
ratings = thread["ratings"]
return {
**{f"csat_{d}": ratings[d] for d in CSAT_DIMENSIONS},
"csat_average": sum(ratings[d] for d in CSAT_DIMENSIONS)
/ len(CSAT_DIMENSIONS),
"detractor": any(ratings[d] <= 2 for d in CSAT_DIMENSIONS),
}
@task
def build_record(
thread: dict,
reply_score: dict,
customer_score: dict,
satisfaction_score: dict,
) -> dict:
return {
"ticket_id": thread["ticket_id"],
"reply_accurate": reply_score["accurate"],
"reply_addresses_question": reply_score["addresses_question"],
"reply_tone": reply_score["tone"],
"reply_reasoning": reply_score["reasoning"],
"customer_sentiment": customer_score["sentiment"],
"customer_satisfied": customer_score["satisfied"],
"escalation_risk": customer_score["escalation_risk"],
"customer_reasoning": customer_score["reasoning"],
**satisfaction_score,
}
return build_record(
thread=thread,
reply_score=score_support_reply(thread),
customer_score=score_customer_reply(thread),
satisfaction_score=score_satisfaction(thread),
)
@task
def load_metrics(records: list[dict]) -> None:
write_eval_records(records)
total = len(records)
write_metrics(
tickets_scored=total,
accuracy_rate=sum(r["reply_accurate"] for r in records) / total,
relevance_rate=sum(r["reply_addresses_question"] for r in records) / total,
poor_tone_rate=sum(r["reply_tone"] == "poor" for r in records) / total,
positive_sentiment_rate=sum(
r["customer_sentiment"] == "positive" for r in records
)
/ total,
satisfied_rate=sum(r["customer_satisfied"] for r in records) / total,
high_escalation_risk_rate=sum(
r["escalation_risk"] == "high" for r in records
)
/ total,
detractor_rate=sum(r["detractor"] for r in records) / total,
**{
f"avg_csat_{d}": sum(r[f"csat_{d}"] for r in records) / total
for d in CSAT_DIMENSIONS
},
)
_threads = join_by_ticket(
support_replies=fetch_replies_from_helpdesk(),
customer_replies=fetch_replies_from_inbox(),
satisfaction_scores=fetch_ratings_from_survey_tool(),
)
load_metrics(evaluate_thread.expand(thread=_threads))
evaluate_support_threads()
```
The Dag consists of nine tasks, grouped by function:
* **Extract**: `fetch_replies_from_helpdesk` retrieves the reply the first Dag sent to the customer, `fetch_replies_from_inbox` the customer's answer, and `fetch_ratings_from_survey_tool` the survey ratings. Each task connects to a different system.
* **Join**: `join_by_ticket` inner-joins the three sources on `ticket_id`, so a ticket without both a customer reply and a survey response is dropped.
* **Score**: the `evaluate_thread` task group is [mapped](/docs/learn/dynamic-tasks) over the joined threads with `.expand()`, to score each thread in parallel. The task group contains three AI product evals:
* `score_support_reply`: a `@task.llm` judge scores the agent's reply for accuracy, whether it answers what the customer asked, and tone.
* `score_customer_reply`: a second AI judge reads the customer's answer for sentiment, whether they consider the request handled, and escalation risk.
* `score_satisfaction` is an AI product eval that needs no model call, because the survey ratings are already numeric. A plain `@task` computes a per-dimension score, an average, and whether the customer counts as a detractor.
* `build_record` combines the thread and the three scores into one row.
* **Load**: `load_metrics` writes one row per ticket with `write_eval_records`, then the daily rates with `write_metrics`: accuracy, relevance, poor tone, positive sentiment, satisfied, high escalation risk, detractor, and average satisfaction per dimension.
Because the join drops tickets that are missing either signal, every rate `load_metrics` computes describes customers who responded, which is a biased subset. Report each metric alongside the response rate over the total number of tickets.
Two more signals could be added to the same Dag:
* **Reviewer edits**: edited text shows what was wrong with the original. A [human-in-the-loop task](/docs/learn/airflow-human-in-the-loop) records both the decision and the edited output, which an eval Dag can read from XCom.
* **Reopens and escalations**: whether the ticket was reopened or escalated to a human agent after the reply was sent. Signals like this can appear days or even weeks later and are best handled by a separate Dag that checks for any new information relevant to existing records on a regular cadence.
## Next steps
* Evaluate the output of the AI itself with [AI model evals](/docs/learn/ai-orchestration-model-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).