Triage support tickets¶

A queue of free-text support tickets arrives every hour and someone has to read each one, decide how urgent it is, and hand it to the right team. This Dag has the model do the reading and produce a typed record per ticket: priority, category, a one-line summary and a suggested next action. Airflow runs one task per ticket, so a bad ticket retries alone, every result is in XCom, and one schedule argument makes it hourly.

What this demonstrates¶

Single prompts: LLMOperator and @task.llm – @task.llm returns a prompt string; the operator makes the call and pushes the result.
Structured output and XCom – output_type=TicketAnalysis gives the downstream task a Pydantic instance, not a string to parse.
Dynamic Task Mapping – .expand() fans one model call out per ticket.

Run it¶

Install the provider with the extra for your vendor:

pip install "apache-airflow-providers-common-ai[openai]"

Trigger the Dag:

airflow dags test example_llm_analysis_pipeline

The store_results log has one [PRIORITY] category: summary line per ticket; each analyze_ticket map index holds a TicketAnalysis XCom.

The Dag¶

airflow/providers/common/ai/example_dags/example_llm_analysis_pipeline.py[source]

# Pydantic output classes must be defined at module scope so they can be
# imported by name when downstream tasks deserialize the XCom payload.
class TicketAnalysis(BaseModel):
    """Structured analysis of a single support ticket."""

    priority: str
    category: str
    summary: str
    suggested_action: str


@dag(tags=["example"])
def example_llm_analysis_pipeline():
    """Triage a queue of support tickets: one model call per ticket, typed results, ready for a schedule."""

    @task
    def get_support_tickets():
        """Fetch unprocessed support tickets."""
        return [
            (
                "Our nightly ETL pipeline has been failing for the past 3 days. "
                "The error shows a connection timeout to the Postgres source database. "
                "This is blocking our daily financial reports."
            ),
            (
                "We'd like to add a new connection type for our internal ML model registry. "
                "Is there documentation on creating custom hooks?"
            ),
            (
                "After upgrading to the latest version, the Grid view takes over "
                "30 seconds to load for DAGs with more than 500 tasks. "
                "Previously it loaded in under 5 seconds."
            ),
        ]

    @task.llm(
        llm_conn_id="pydanticai_default",
        system_prompt=(
            "Analyze the support ticket and extract: "
            "priority (critical/high/medium/low), "
            "category (bug/feature_request/question/performance), "
            "a one-sentence summary, and a suggested next action."
        ),
        output_type=TicketAnalysis,
    )
    def analyze_ticket(ticket: str):
        return f"Analyze this support ticket:\n\n{ticket}"

    @task
    def store_results(analyses: list[TicketAnalysis]):
        """Store ticket analyses. In production, this would write to a database or ticketing system."""
        for analysis in analyses:
            print(f"[{analysis.priority.upper()}] {analysis.category}: {analysis.summary}")

    tickets = get_support_tickets()
    analyses = analyze_ticket.expand(ticket=tickets)
    store_results(analyses)

Adapting it¶

Replace the list in get_support_tickets with a query against your ticketing system, for example an SQLExecuteQueryOperator or an HttpOperator upstream.
Set schedule="@hourly" on the @dag and use {{ data_interval_start }} in the query so each run picks up only new tickets.
Make priority and category Literal types on TicketAnalysis so the model cannot invent a label. Route pipeline failures to a fix or a person shows how to branch on the result.