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. 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 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.
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®.

Dag graph
Code

For a full explanation of this Dag see Example: evaluate a support ticket agent.

Airflow graph view of an eval Dag. Three parallel extract tasks, fetch_replies_from_inbox, fetch_replies_from_helpdesk, and fetch_ratings_from_survey_tool, feed a join_by_ticket task. Its output is dynamically mapped over an evaluate_thread task group containing two @task.llm scoring tasks, score_support_reply and score_customer_reply, plus a deterministic score_satisfaction task, all feeding a build_record task. A final load_metrics task writes the results.

For a full explanation of the code example see Example: evaluate a support ticket agent.

evaluate_support_threads.py

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®.
Basic familiarity with LLMs and AI agents.

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.
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, asset-driven, and event-driven 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 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 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.
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 and 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 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, 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.
- Classifications and routing decisions: ticket categories, priorities, lead scores, content moderation labels. See Classification and routing.
- Extracted structured data: fields pulled from contracts, invoices, or call transcripts. See Extraction of structured data.
- Generated code and queries: pull requests, SQL, configuration, migrations. See Text-to-SQL.
- Summaries and reports: meeting notes, account summaries, incident write-ups. See Summarization.
- Translations and localized content. See 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, and Building Kepler, Astronomer’s 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, and AI-powered education operations with Apache Airflow® for examples.
- Code changes proposed by a group of agents: pull requests from a bug report or feature request. See Agentic software development, and the Together AI case study for agents authoring dbt models and Dags.
- Diagnoses: root-cause analyses and suggested fixes. See Root-cause analysis and self-healing pipelines and How to use Otto to automatically investigate Dag failures and PR a 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.

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 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 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. 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.

Two Airflow Dags around a support ticket agent. The first Dag extracts an incoming ticket and customer data, runs an agent that makes multiple LLM API calls through any AI harness, verifies the draft with an AI-as-a-judge and a human-in-the-loop step, then formats and sends the response to the customer. The customer's response and support rating start the second Dag, which extracts the response and customer information, runs an agent to assess it, embeds insights and calculates a score, and loads the result to a database. That database is one of the data sources the first agent reads through tools and MCPs on later runs.

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, 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:

Dag graph
Code

evaluate_support_threads.py

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 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 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.
Run a single model inference as a task with LLM orchestration with Apache Airflow®.
Give a model tools and multi-step reasoning with Agent orchestration.
Add human decisions and capture their output with Human-in-the-loop workflows with Apache Airflow®.

Overview

Get started

Airflow and AI

Airflow concepts

Tutorials

Integrations & connections

Reference Architectures

Observability concepts

Glossary

AI product evals with Apache Airflow®

Assumed knowledge

Why Airflow for AI product evals

What AI outputs can you evaluate with Airflow?

AI product evals with Airflow

Example: evaluate a support ticket agent

Next steps

​Assumed knowledge

​Why Airflow for AI product evals

​What AI outputs can you evaluate with Airflow?

​AI product evals with Airflow

​Example: evaluate a support ticket agent

​Next steps

Assumed knowledge

Why Airflow for AI product evals

What AI outputs can you evaluate with Airflow?

AI product evals with Airflow

Example: evaluate a support ticket agent

Next steps