KI-Pipelines mit Airflow orchestrieren

Mit KI-Prototypen legst du schnell los. Ein Notebook, ein API-Schlüssel und ein paar Zeilen Python reichen oft, um Texte zu klassifizieren, Dokumente zu zusammenfassen oder Empfehlungen zu generieren. 

Doch diesen Prototypen dann täglich stabil am Laufen zu halten, ist eine ganz andere Herausforderung.

Was passiert, wenn die Modell-API ihr Ratenlimit erreicht? Was, wenn eine Anfrage in den Timeout läuft, das Output-Format nicht passt oder eine nachgelagerte Datenbank kurzzeitig nicht erreichbar ist?

Sobald aus einer KI-Pipeline ein wiederkehrender Workflow wird, sind Zuverlässigkeit, Monitoring und Recovery mindestens so wichtig wie das Modell selbst.

In diesem Artikel zeige ich dir, wie du mit Apache Airflow eine vollständige KI-Pipeline aufbaust – einer Open-Source-Plattform zum Entwickeln, Planen und Überwachen von Workflows. 

Unsere Pipeline läuft täglich, holt Kundenbewertungen ab, klassifiziert deren Stimmung mit einem Large Language Model (LLM) und speichert die Ergebnisse. Das alles zuverlässig in einem einzigen Workflow!

Warum Airflow für KI-Pipelines?

Ein Jupyter-Notebook ist perfekt fürs Experimentieren. Wir lieben Notebooks, um Ideen zu testen, schnell Code zu schreiben und Ergebnisse direkt zu visualisieren.

Ein Cron-Job reicht oft, um nachts ein einfaches Skript laufen zu lassen. Wenn wir ehrlich sind, werden viele Skripte aber erst mal wochenlang manuell gestartet, bevor sie einen sauberen Zeitplan bekommen.

Die Probleme beginnen, wenn aus einem Skript mehrere werden – und diese Skripte voneinander abhängen. Dann haben wir kein einzelnes Programm mehr, sondern einen Workflow.

Weder Notebooks noch Cron-Jobs sind dafür gebaut, komplexe Workflows zuverlässig zu steuern. 

Abhängigkeiten verstecken sich oft im Code, Fehler erfordern manuelle Analyse, und ein erneuter Lauf wiederholt unter Umständen Schritte, die bereits erfolgreich abgeschlossen waren.

Vieles davon kennen wir schon aus klassischen Extract-Transform-Load-(ETL)-Pipelines. ETL-Workflows müssen Daten abrufen, transformieren, validieren und in ein anderes System laden. 

Selbst wenn Eingaben und Transformationen berechenbar sind, erzeugt jeder zusätzliche Schritt einen weiteren potenziellen Fehlerpunkt. Heute verstärken KI-Pipelines diese Herausforderung noch einmal deutlich. 

Wenn dich speziell ETL-Pipelines interessieren, empfehle ich diesen Kurs zu ETL-Pipelines in Airflow.

KI-Pipelines

Denk an einen typischen KI-Workflow: Daten sammeln, bereinigen, eine Modell-API aufrufen, die Antwort validieren, das Ergebnis speichern und bei Problemen benachrichtigen. 

Jeder Schritt hängt vom vorherigen ab – und jeder kann auf andere Weise scheitern: Eine API meldet ein Ratenlimit, eine Anfrage läuft in den Timeout, eine generierte Antwort ist zwar gültiges JSON, fällt aber bei der Qualitätsprüfung durch. 

Outputs können außerdem unvollständig, inkonsistent oder anders formatiert sein, als der nächste Schritt erwartet.

Airflow verhindert diese Fehler nicht, macht sie aber leichter handhabbar. 

Mit Airflow kann ein Task, der ein LLM aufruft, nach einer temporären Rate-Limit-Antwort erneut versuchen, ohne den Dateningest neu auszuführen. Ein Validierungstask kann fehlerhafte oder minderwertige Ergebnisse abfangen, bevor sie die Produktion erreichen. Alarme informieren das Team erst, wenn die konfigurierten Wiederholungen ausgeschöpft sind.

Airflow ermöglicht das, indem es den Workflow als gerichteten azyklischen Graphen (DAG) abbildet. 

Statt die gesamte Pipeline in einem langen Skript zu verstecken, definiert ein DAG jede Operation als eigenen Task mit expliziten Abhängigkeiten, Retry-Strategien, Logs und Ausführungsstatus. 

Genau deshalb passt Airflow so gut zu KI-Pipelines: Fehler sind kein Ausnahmefall, sondern ein erwartbarer Teil des Workflows.

Wenn aus einem Experiment ein wiederkehrender Produktionsprozess wird, liefert Airflow die Orchestrierungsebene, die Sichtbarkeit, Recovery und Wartbarkeit ermöglicht.

Bevor wir unsere Pipeline bauen, richten wir die Umgebung ein.

Die Umgebung einrichten

Dafür sind drei kurze Schritte nötig. 

Airflow 3 und das Airflow AI SDK installieren

Für dieses Tutorial starten wir Airflow lokal. 

Diese Einrichtung ist zum Lernen und Entwickeln praktisch, denn ein einziger Befehl initialisiert die Metadatenbank und startet die Services, die Airflow benötigt.

Zuerst erstellen und aktivieren wir eine virtuelle Umgebung, damit Airflow und seine Abhängigkeiten vom restlichen Python-Setup getrennt sind:

python3 -m venv ~/airflow-venv
source ~/airflow-venv/bin/activate

Als Nächstes installieren wir Airflow 3 mit Versions-Pinning und der offiziellen Constraints-Datei:

AIRFLOW_VERSION=3.3.0
PYTHON_VERSION="$(python -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')"

CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

pip install "apache-airflow==${AIRFLOW_VERSION}" \
 --constraint "${CONSTRAINT_URL}"

Beachte: Airflow hängt von vielen Python-Paketen ab, die wiederum eigene Abhängigkeiten haben. Ohne Constraints-Datei kann pip neuere oder inkompatible Paketversionen wählen. 

Die Installation scheint dann vielleicht zu funktionieren, aber später kann Airflow beim Import von Providern, beim Starten von Services oder beim Ausführen von Tasks scheitern. In der offiziellen Installationsanleitung empfiehlt Airflow deshalb dringend, mit Constraints-Dateien einen bekannten, kompatiblen Abhängigkeitsstand zu installieren.

Da unser Workflow ein LLM aufruft, benötigen wir außerdem den Common-AI-Provider von Airflow und die OpenAI-Abhängigkeit für Pydantic AI:

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

Als Nächstes konfigurieren wir die OpenAI-Verbindung. Airflow nutzt Connections, um Zugangsdaten und Verbindungseinstellungen außerhalb des DAG-Codes zu speichern. Das hält die Python-Datei sauber und verhindert, dass Secrets hart codiert werden.

Sobald die Installation validiert ist, setzen wir den OpenAI-Key, damit die Kommunikation mit dem Modell funktioniert. Dafür gibt es mehrere Optionen. 

Meine bevorzugte Vorgehensweise ist:

1. Exportiere den OpenAI-API-Key in der aktuellen Terminal-Session:

export OPENAI_API_KEY="your-openai-api-key"

2.  Erstelle eine Airflow-Connection mit dem Namen pydanticai_default:

airflow connections add pydanticai_default \
 --conn-type pydanticai \
 --conn-password "$OPENAI_API_KEY" \
 --conn-extra '{"model":"openai:gpt-4o-mini"}'

Der Common-AI-Provider verwendet den Connection-Typ pydanticai, um über Pydantic AI mit LLM-Providern zu kommunizieren. 

Dieser Befehl speichert die Connection in der Metadatenbank von Airflow. Danach brauchst du die .env-Datei nicht mehr und musst den Key beim Start von Airflow nicht erneut übergeben.

3. Zum Schluss starten wir die lokale Airflow-Umgebung mit:

airflow standalone

Lass dieses Terminal geöffnet, solange Airflow läuft. Der Standalone-Modus ist ideal für lokale Entwicklung und Tutorials, aber nicht für den produktiven Betrieb gedacht.

Für den Zugriff auf die Airflow-Oberfläche öffnest du jetzt http://localhost:8080 im Browser. Als Erstes erscheint die Login-Seite. Welche Zugangsdaten nutzen wir?

Anmeldeansicht von Apache Airflow.

Wenn Airflow im Standalone-Modus startet, werden die Zugangsdaten automatisch erzeugt. Benutzername und Passwort stehen in der Terminalausgabe und in $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated.

Standardmäßig ist AIRFLOW_HOME = ~/airflow, daher können wir die Credentials so auslesen:

cat ~/airflow/simple_auth_manager_passwords.json.generated

Mit diesen Daten können wir uns in der Airflow-Oberfläche anmelden.

Die UI erkunden

Nach dem Login öffnet Airflow die Startseite. Diese Ansicht gibt dir einen schnellen Überblick über die lokale Airflow-Umgebung.

Airflow-UI nach dem ersten Login.

Oben fassen die Stat-Karten den aktuellen Zustand deiner DAGs zusammen, inklusive fehlgeschlagener, laufender und aktiver Workflows. 

Da wir unsere Pipeline noch nicht erstellt haben, stehen hier vermutlich noch Nullen. 

Der Health-Bereich zeigt, ob die wichtigsten Airflow-Services laufen. 

Darunter zeigt die Pool-Slots-Leiste die verfügbare Ausführungskapazität. Eigene Pools konfigurieren wir hier nicht, aber dieser Bereich ist hilfreich, um gleichzeitige Tasks zu steuern. 

Im Abschnitt History findest du Zusammenfassungen der letzten DAG-Runs. Sobald wir unsere Sentiment-Pipeline triggern, siehst du hier eingereihte, laufende, erfolgreiche und fehlgeschlagene Runs.

Projektstruktur

Jetzt, da Airflow läuft, legen wir die benötigten Dateien für unsere Pipeline an:

mkdir -p ~/airflow/dags
touch ~/airflow/dags/review_sentiment_pipeline.py
touch ~/airflow/requirements.txt

In einer lokalen Standalone-Installation verwendet Airflow standardmäßig ~/airflow als AIRFLOW_HOME. Unsere DAG-Datei legen wir im Verzeichnis dags ab und pflegen eine requirements.txt, um die Projektabhängigkeiten zu dokumentieren. 

So sieht die Struktur aus:

Minimale Airflow-Projektstruktur.

Die Datei eview_sentiment_pipeline.py enthält die Workflow-Definition mit den Tasks zum Abrufen der Bewertungen, zur Sentiment-Klassifikation und zum Speichern der Ergebnisse. 

Airflow überwacht das konfigurierte dags-Verzeichnis und sucht in Python-Dateien nach gültigen DAG-Definitionen. Sobald unsere Datei einen gültigen DAG enthält, erscheint der Workflow auf der Seite Dags in der Airflow-UI.

Wenn beim Onboarding etwas hakt, schau dir das Tutorial Getting Started with Apache Airflow an.

Deine erste Airflow-Pipeline bauen

Für den praktischen Einstieg bauen wir einen täglichen Job, der Kundenbewertungen verarbeitet und deren Stimmung klassifiziert. 

Stell dir vor, wir haben eine E-Commerce-Plattform gebaut. Kundinnen und Kunden hinterlassen bereits Bewertungen, und wir wollen sie jeden Tag automatisch analysieren. 

Statt ein Skript manuell auszuführen, orchestrieren wir den Workflow mit Airflow.

Schritt 1: Bewertungen abrufen

Nehmen wir an, die Rohbewertungen liegen in einer .jsonl-Datei. 

Jede Zeile enthält eine Bewertung als JSON-Objekt, und neue Bewertungen werden über den Tag hinweg ans Dateiende angehängt. In der Produktion könnten diese Daten aus einer API, Datenbank, einem Object-Store oder einer Message-Queue kommen. Für dieses Tutorial ist die JSONL-Datei eine einfache und realistische Quelle für Rohdaten.

Der erste Schritt der Pipeline lädt diese Bewertungen in Airflow. Wir starten mit einer normalen Python-Funktion – so, wie wir es auch in einem Notebook oder Skript tun würden:

REVIEWS_FILE = Path.home() / "airflow" / "data" / "reviews.jsonl"

def fetch_reviews() -> list[dict]:
   """Load raw customer reviews from the JSONL landing file."""
   reviews = []

   with REVIEWS_FILE.open("r", encoding="utf-8") as file:
       for line_number, line in enumerate(file, start=1):
           if not line.strip():
               continue

           try:
               reviews.append(json.loads(line))
           except json.JSONDecodeError as error:
               raise ValueError(
                   f"Invalid JSON on line {line_number}"
               ) from error

   print(f"Fetched {len(reviews)} reviews")
   return reviews

Die Funktion öffnet die JSONL-Datei, liest sie zeilenweise, überspringt leere Zeilen und wandelt jedes JSON-Objekt in ein Python-Dict um. 

Enthält eine Zeile ungültiges JSON, schlägt der Task mit einer klaren Fehlermeldung inklusive Zeilennummer fehl.

Bis hierhin ist das pures Python. Als Nächstes machen wir daraus einen Airflow-Task und hängen ihn in einen DAG.

Den Task in einen DAG einbetten

Damit die Funktion Teil eines Airflow-Workflows wird, brauchen wir drei Bausteine:

1. Zuerst dekorieren wir die Funktion mit @task. So weiß Airflow, dass es sich um einen Task handelt.
2. Dann definieren wir mit @dag den DAG. Der DAG bekommt einen Namen, einen Zeitplan, ein Startdatum und weitere Einstellungen.
3. Schließlich rufen wir den Task innerhalb der DAG-Funktion auf und instanziieren den DAG am Ende der Datei.

Hier die Minimalversion:

from airflow.sdk import dag

@dag(    fetch_reviews()
   dag_id="fetch_reviews_pipeline",
)

def fetch_reviews_pipeline():
   fetch_reviews()

fetch_reviews_pipeline()

Beachte: Dieser DAG wird so noch nicht automatisch ausgeführt. 

In dieser Form müssten wir ihn in der Airflow-UI manuell triggern. Um festzulegen, wann der Workflow laufen soll, fügen wir dem DAG Zeitplan-Parameter hinzu. Eine mögliche Konfiguration ist:

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)

Schauen wir uns das an!

• dag_id vergibt einen eindeutigen Namen für den DAG in Airflow.

• schedule="@daily" führt den DAG einmal pro Tag aus.
• start_date legt fest, ab wann Airflow mit der Planung beginnen darf.
• catchup=False verhindert Backfills für vergangene Daten.
• tags helfen, den DAG in der UI zu organisieren und zu filtern.

Sobald der DAG definiert ist, wechseln wir in die UI und sehen uns das Ergebnis an:

Airflow-UI-Ansicht unserer Pipeline review_sentiment_pipeline. Beachte die beiden Run-Typen (geplant und manuell) und den Trigger-Button.

In der UI sehen wir, dass Airflow den DAG review_sentiment_pipeline automatisch entdeckt hat. 

Außerdem siehst du, dass er nach Zeitplan gelaufen ist und wir zusätzlich einen manuellen Run über den Trigger-Button oben rechts gestartet haben.

Praktisch ist die Kalenderansicht, in der Airflow alle Pipeline-Runs inklusive Status protokolliert:

Kalenderansicht unserer Airflow-Pipeline mit geplanten und manuellen Runs.

Mach hier gern eine kurze Pause und erkunde die Airflow-UI!

Schritt 2: LLM-Klassifikation

Sind die Bewertungen geladen, klassifizieren wir im nächsten Schritt deren Stimmung. Dafür nutzen wir @task.llm – ein Decorator, der einen LLM-Aufruf in einen nativen Airflow-Task verwandelt.

Wir könnten die OpenAI-API auch manuell in einem normalen @task aufrufen, aber @task.llm sorgt für eine saubere Integration. 

Der Decorator übernimmt LLM-spezifische Details wie Modellverbindung, System-Prompt, strukturiertes Output und Serialisierung. Gleichzeitig profitieren wir von Airflow-Standards wie Retries, Logging, Monitoring und Abhängigkeitsmanagement.

Dieser zweite Schritt hat ein paar Bausteine mehr, daher gehen wir ihn schrittweise durch:

1. Zuerst definieren wir die erwartete Ausgabestruktur des Modells:

from typing import Literal
from pydantic import BaseModel

class ReviewClassification(BaseModel):
   """Structured result returned by the LLM."""

   review_id: str
   sentiment: Literal["positive", "neutral", "negative"]

Das sagt dem LLM-Task, dass jede Antwort die Review-ID und eines von drei Stimmungslabels enthalten soll: positive, neutral oder negative.

2. Als Nächstes erstellen wir die Funktion, die den Prompt für eine einzelne Bewertung baut:

@task.llm(
   llm_conn_id="pydanticai_default",
   system_prompt=(
       "Classify the customer review as positive, neutral, or negative. "
       "Use both the written review and the star rating as context. "
       "Return the review ID and sentiment."
   ),
   output_type=ReviewClassification,
   serialize_output=True,
   retries=3,
)
def classify_review(review: dict) -> str:
   """Build the prompt sent to the LLM."""
   return (
       f"Review ID: {review['review_id']}\n"
       f"Product: {review['product']}\n"
       f"Rating: {review['rating']} out of 5 stars\n"
       f"Review: {review['review_text']}"
   )

Der Funktionsrumpf gibt einen Prompt-String zurück, während der @task.llm-Decorator den Prompt mit der zuvor konfigurierten Connection pydanticai_default an das Modell sendet. 

Der system_prompt gibt die Instruktion vor, output_type=ReviewClassification fordert eine strukturierte Antwort an. Mit serialize_output=True speichert Airflow das Ergebnis als Dictionary – so lässt es sich leichter an nachgelagerte Tasks übergeben.

Interessiert an Tasks mit LLM-Inferenz? Dann sieh dir diesen LangChain-Kurs für spezialisierte Anwendungen an!

Die Modellklassifikation sichtbar machen

Der Klassifikationstask liefert ein strukturiertes Ergebnis. Während der Entwicklung ist es praktisch, zu sehen, was das Modell ausgibt, ohne bis zum Speicherschritt zu warten. 

Ein Pipeline-Schritt kann aus mehreren Airflow-Tasks bestehen. 

Wir fügen daher einen kleinen Task hinzu, der jede Klassifikation in die Airflow-Logs schreibt:

@task
def show_classification(classification: dict) -> dict:
   """Print one model classification to the Airflow task logs."""
   print(
       f"Review {classification['review_id']}: "
       f"{classification['sentiment']}"
   )

   return classification

Dieser Task verändert das Ergebnis nicht. Er loggt lediglich die Modellausgabe und gibt dasselbe Dictionary zurück, sodass die Klassifikation an den nächsten Schritt übergeben werden kann.

 3. Zum Schluss verbinden wir die neuen Tasks in der DAG-Funktion:

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)
def review_sentiment_pipeline():
   reviews = fetch_reviews()

   classifications = classify_review.expand(
       review=reviews
   )

   displayed_classifications = show_classification.expand(
       classification=classifications
   )

review_sentiment_pipeline()

Hinweis: In dieser Pipeline nutzen wir Dynamic Task Mapping mit .expand(review=reviews). Das sagt Airflow, dass pro Bewertung eine eigene classify_review-Instanz angelegt wird – jede Bewertung wird also in einer separaten Anfrage an das LLM gesendet.

Das ist meist besser, als die ganze Liste in einem LLM-Task zu verarbeiten. Ein Aufruf wie classify_reviews(reviews) würde alles zusammen bearbeiten – Fehler sind schwerer zu debuggen, Retries sind ungezielter, und große Inputs laufen eher in Kontext- oder API-Limits.

Bevor wir Schritt 2 abschließen, schauen wir in der Airflow-UI, wie sich der Workflow entwickelt hat. 

Jetzt enthält der DAG drei Tasks: fetch_reviews, classify_review und show_classification:

Airflow-UI-Ansicht der drei Tasks in unserem review_sentiment_pipeline-DAG.

Troubleshooting: Temporäre LLM-Fehler handhaben

Durch Dynamic Task Mapping sendet Airflow pro Bewertung genau eine LLM-Anfrage. Das ist hilfreich, weil jede Bewertung unabhängig getrackt wird:

Airflow-UI zeigt die unabhängige Klassifikation je Bewertung. Einige Tasks sind bereits fertig oder fehlgeschlagen, andere laufen noch.

LLM-Anfragen können z. B. wegen API-Ratenlimits, Netzwerkproblemen oder Model-Timeouts fehlschlagen. 

Statt die gesamte Pipeline direkt abbrechen zu lassen, können wir Airflow anweisen, den betroffenen Task automatisch zu wiederholen.

Fügen wir das hinzu!

from datetime import timedelta

@task.llm(
   llm_conn_id="pydanticai_default",
   system_prompt=(
       "Classify the customer review as positive, neutral, or negative. "
       "Use both the written review and the star rating as context. "
       "Return the review ID and sentiment."
   ),
   output_type=ReviewClassification,
   serialize_output=True,
   retries=3,
   retry_delay=timedelta(seconds=30),
   retry_exponential_backoff=True,
   max_retry_delay=timedelta(minutes=5),
)

Hier erlaubt retries=3 bis zu drei zusätzliche Versuche nach dem ersten Fehlschlag. 

Der erste Retry wartet 30 Sekunden, mit Exponential Backoff vergrößert sich die Wartezeit zwischen weiteren Versuchen – bis maximal fünf Minuten.

Das Schöne: Airflow wiederholt nur den fehlgeschlagenen gemappten Task. 

Wenn 49 Bewertungen erfolgreich klassifiziert wurden und eine Anfrage scheitert, wird nur diese eine erneut geplant. Der Rest muss keine bereits erfolgreichen Arbeitsschritte wiederholen. 

Nach einiger Zeit stehen die fehlgeschlagenen Klassifikationen wieder zum Retry an:

Airflow-UI mit allen fehlgeschlagenen Klassifikations-Tasks, die auf einen Retry warten.

Sobald die Klassifikation erfolgreich ist, zeigt jede gemappte Instanz von show_classification() die Modellantwort für genau eine Bewertung in ihren Logs. 

Heißt: Auch der Anzeige-Schritt ist unabhängig – du kannst die Ausgabe einer Bewertung öffnen, prüfen oder debuggen, unabhängig vom Zustand der anderen.

Airflow-UI mit der LLM-Sentimentklassifikation für Bewertung 47.

Schritt 3: Ergebnisse speichern

Während der DAG-Entwicklung ist das Ausgeben jeder Klassifikation in die Logs hilfreich, um schnell Modellantworten zu prüfen. 

Logs sind jedoch kein verlässlicher Ort, um Pipeline-Ergebnisse zu speichern. Sie sind schwerer abzufragen, wiederzuverwenden und in nachgelagerte Systeme zu integrieren.

Im letzten Schritt sammeln wir daher die Klassifikationen und schreiben sie in eine strukturierte Ausgabe. 

Für diesen Artikel entfernen wir den Aufruf von show_classification() und speichern die Ergebnisse in einer lokalen JSON-Datei. 

In einer produktiven Pipeline würde dieser Task typischerweise in eine Datenbank, ein Data Warehouse oder einen Object-Store schreiben. 

Eine mögliche Implementierung für diesen neuen @task wäre:

OUTPUT_FILE = (
   Path.home()
   / "airflow"
   / "output"
   / "classified_reviews.json"
)

@task
def store_results(results: list[dict]) -> str:
   """Write all classified reviews to a local JSON file."""
   OUTPUT_FILE.parent.mkdir(parents=True, exist_ok=True)

   classifications = list(results)

   with OUTPUT_FILE.open("w", encoding="utf-8") as file:
       json.dump(
           classifications,
           file,
           indent=2,
           ensure_ascii=False,
       )

   print(
       f"Stored {len(classifications)} classifications "
       f"in {OUTPUT_FILE}"
   )

   return str(OUTPUT_FILE)

Dieser Task ist unser Load-Schritt. 

Er nimmt die LLM-Klassifikationen, legt falls nötig den Ausgabepfad an und schreibt alles in eine lokale JSON-Datei. 

Die kleine Konvertierung mit list(results) ist nützlich, da die Klassifikationen aus gemappten Tasks stammen. 

Airflow sammelt diese gemappten Outputs, und dieser Task wandelt sie vor dem Speichern in eine normale Liste um. Jetzt müssen wir den Task nur noch im DAG verbinden, indem wir store_results(classifications) aufrufen.

Damit weiß Airflow, dass store_results() vom Klassifikationsschritt abhängt – die Ausgabedatei wird also erst geschrieben, wenn alle Klassifikationen erfolgreich sind.

Die Pipeline ausführen

Mit allen drei Schritten holt der komplette DAG jetzt die Bewertungen, erstellt pro Bewertung einen LLM-Task und speichert die kombinierten Ergebnisse. 

Innerhalb der DAG-Funktion sieht die Pipeline so aus:

@dag(
   dag_id="review_sentiment_pipeline",
   schedule="@daily",
   start_date=datetime(2026, 1, 1),
   catchup=False,
   tags=["ai", "sentiment"],
)
def review_sentiment_pipeline():
   reviews = fetch_reviews()
   classifications = classify_review.expand(
       review=reviews
   )
   store_results(classifications)

review_sentiment_pipeline()

Nach dem Speichern der Datei braucht Airflow ein paar Sekunden, um sie zu parsen und den vollständigen DAG zu entdecken. Sobald review_sentiment_pipeline in der DAG-Liste erscheint, aktiviere ihn per Toggle und klicke auf Trigger, um einen manuellen Run zu starten. 

So testen wir die gesamte Pipeline sofort – ohne auf den nächsten täglichen Lauf zu warten. Nach einem erfolgreichen Run findest du die finale JSON-Datei lokal unter ~/airflow/output/classified_reviews.json.

Die JSON-Ausgabedatei ist das dauerhafte Ergebnis des Workflows. Sie kann außerhalb des DAGs wiederverwendet oder an nachgelagerte Systeme übergeben werden. 

Die Airflow-UI liefert dazu die operative Sicht: Welche Tasks liefen, welche sind fehlgeschlagen oder erfolgreich, und wo finde ich die Logs. 

So sieht die finale Pipeline aus:

Airflow-UI mit den drei Tasks unserer Pipeline. Rechts unten siehst du die Eigenschaften jedes Tasks: Name, Trigger-Regel, Operator/Decorator und Anzahl Retries.

In den nächsten Tagen füllt sich die Airflow-UI mit automatischen Daily Runs – inklusive Historie, wann die Pipeline lief und ob die Ausführung jeweils erfolgreich war:

Airflow-UI mit den Pipeline-Runs und ihren Zuständen. Nach Schritt 3 nutzt die Pipeline nicht mehr show_classification, sondern store_results.

Noch weiter gehen: Airflow-3-KI-Features für den nächsten Schritt

Unsere Pipeline holt Bewertungen ab, klassifiziert sie mit einem LLM und speichert die Ergebnisse täglich. 

Airflow 3 unterstützt außerdem fortgeschrittene Muster, die KI-Workflows reaktiver und leichter überwachbar machen.

Ereignisgesteuertes Scheduling

Statt die Pipeline jeden Tag zu einer festen Zeit zu starten, kann Airflow sie bei einem externen Ereignis loslaufen lassen. 

Beispielsweise könnte die Bewertungs-Pipeline starten, sobald eine neue Datei im Object-Storage landet oder eine Nachricht in einer Queue ankommt.

Airflow 3 unterstützt ereignisgesteuerte Planung über Assets und event-kompatible Trigger. So entfällt das wiederholte Pollen auf neue Daten, und die Pipeline reagiert, sobald ein Ereignis erkannt wird.

Human-in-the-Loop mit HITLOperator

Nicht jede Modellentscheidung sollte automatisch akzeptiert werden. 

Manche Bewertungen sind mehrdeutig, sensibel oder erfordern menschliches Urteil. Die Human-in-the-Loop-Operatoren von Airflow können einen laufenden Workflow pausieren und Eingaben über die Airflow-Oberfläche anfordern. 

Reviewer können ein Ergebnis freigeben oder ablehnen, Informationen ergänzen oder entscheiden, welcher Branch des Workflows fortgesetzt wird.

Der Common-AI-Provider unterstützt außerdem interaktive Review-Muster für agentische Workflows, in denen Reviewer LLM-Ergebnisse freigeben, ablehnen oder Änderungen anfordern können.

Zusammen ermöglichen ereignisgesteuertes Scheduling und menschliche Reviews KI-Pipelines, die schnell reagieren – ohne dort auf Aufsicht zu verzichten, wo sie wichtig ist. 

Mehr zu den Features von Airflow 3.0 findest du in unserem separaten Guide. 

Fazit

In diesem Artikel haben wir eine vollständige KI-Pipeline mit Apache Airflow aufgebaut. 

Die Pipeline holt Kundenbewertungen ab, klassifiziert jede Bewertung mit einem LLM und speichert die finalen Ergebnisse in einer strukturierten JSON-Datei.

Unterwegs hast du gesehen, wie Airflow Python-Funktionen in geplante, beobachtbare Tasks verwandelt. Das ist besonders bei KI-Workflows hilfreich, in denen API-Limits, Timeouts und unerwartete Outputs vorkommen können. 

Mit Task-Abhängigkeiten, Dynamic Task Mapping, Retries, Logs und der Airflow-UI lassen sich KI-Pipelines leichter überwachen, debuggen und wiederherstellen.

Im Beispiel haben wir zur Einfachheit lokale Dateien verwendet – dieselbe Struktur lässt sich aber auf APIs, Datenbanken, ereignisgesteuerte Workflows und Human-in-the-Loop-Reviews übertragen. 

Wenn du weitermachen willst, starte mit unserem Kurs Introduction to Apache Airflow in Python und wechsle dann zu Building Data Pipelines with Airflow für noch mehr Praxis. 

Für einen breiteren Lernpfad deckt der Data Engineer in Python-Lernpfad das gesamte Kompetenzspektrum, das du für produktive Datenpipelines brauchst.