Orchestrating AI Pipelines With Airflow

Yapay zekâ prototiplerine başlamak kolaydır. Bir not defteri, bir API anahtarı ve birkaç satır Python; metin sınıflandırmak, belgeleri özetlemek veya öneriler üretmek için yeterli olabilir. 

Ancak, bu prototipi her gün güvenilir şekilde çalışır halde tutmak bambaşka bir zorluktur.

Model API’si hız sınırına ulaştığında ne olur? Bir istek zaman aşımına uğrarsa, çıktı beklenen biçimi takip etmezse veya aşağı akıştaki bir veritabanı geçici olarak kullanılamazsa ne yapacağız?

Bir yapay zekâ boru hattı yinelenen bir iş akışına dönüştükçe, güvenilirlik, izleme ve kurtarma en az modelin kendisi kadar hızlı biçimde önemli hâle gelir.

Bu yazıda, iş akışları geliştirmek, zamanlamak ve izlemek için açık kaynaklı bir platform olan Apache Airflow ile uçtan uca bir yapay zekâ boru hattını nasıl kuracağınızı göstereceğim. 

Boru hattımız her gün çalışacak; müşteri yorumlarını getirecek, bunların duyarlılığını büyük dil modeli (LLM) kullanarak sınıflandıracak ve sonuçları depolayacak. Hem de tek bir iş akışında güvenilir şekilde!

Yapay Zekâ Boru Hatları için Neden Airflow?

Jupyter not defteri, deneyler için harikadır. Fikirleri test etmek, hızlı kod yazmak ve ilerledikçe sonuçları görselleştirmek için not defterlerini hepimiz severiz.

Basit bir betiği her gece çalıştırmak için bir cron görevi çoğu zaman yeterlidir. Kabul edelim, pek çok betik, sonunda birisi düzgün bir zamanlama tanımlayana kadar haftalarca elle çalıştırılır.

Sorunlar, bir betik birkaç betiğe dönüştüğünde ve bu betikler birbirine bağımlı olmaya başladığında başlar. O noktada artık bağımsız bir programımız yoktur: bir iş akışımız vardır.

Ne not defterleri ne de cron görevleri, karmaşık iş akışlarını güvenilir şekilde yönetmek için tasarlanmıştır. 

Bağımlılıklar çoğu zaman kodun içinde gizlidir, hatalar elle araştırma gerektirir ve boru hattını yeniden çalıştırmak, zaten başarıyla tamamlanmış adımların tekrarlanması anlamına gelebilir.

Bu karmaşıklığın bir kısmıyla, geleneksel Çıkar-Dönüştür-Yükle (ETL) boru hatlarında zaten karşılaşıyoruz. ETL iş akışları; veriyi alma, dönüştürme, doğrulama ve başka bir sisteme yükleme gerektirebilir. 

Girdiler ve dönüşümler öngörülebilir olsa bile, her ek adım ilave bir olası hata noktası oluşturur. Günümüzde, yapay zekâ boru hatları bu zorluğu daha da belirgin hâle getiriyor. 

Özellikle ETL boru hatlarıyla ilgileniyorsanız, Airflow’da ETL boru hatları konulu bu kursu öneririm.

Yapay zekâ boru hatları

Tipik bir yapay zekâ iş akışını düşünün: veriyi topla, temizle, bir model API’sini çağır, yanıtı doğrula, sonucu depola ve bir sorun olduğunda birilerini bilgilendir. 

Her adım bir öncekine bağlıdır ve her biri farklı bir şekilde başarısız olabilir: Bir API hız sınırı hatası döndürebilir. Bir model isteği zaman aşımına uğrayabilir. Üretilen bir yanıt geçerli JSON olabilir ancak yine de bir kalite kontrolünden geçemeyebilir. 

Çıktılar ayrıca eksik, tutarsız olabilir veya bir sonraki adımın beklediğinden farklı biçimlendirilmiş olabilir.

Airflow bu hataları ortadan kaldırmaz, ancak yönetmelerini kolaylaştırır. 

Airflow ile, bir LLM’i çağıran görev, veri alma adımını yeniden çalıştırmadan geçici bir hız sınırı yanıtından sonra yeniden deneyebilir. Bir doğrulama görevi, üretim ortamına ulaşmadan önce bozuk biçimli veya düşük kaliteli bir sonucu reddedebilir. Uyarılar, yalnızca yapılandırılmış yeniden denemeler tükendiğinde ekibi bilgilendirebilir.

Airflow bunu, iş akışını bir Yönlendirilmiş Çevrimsiz Grafik (DAG) olarak temsil ederek mümkün kılar. 

Tüm boru hattını tek bir uzun betiğin içine gizlemek yerine, bir DAG her işlemi, açık bağımlılıkları, yeniden deneme ilkeleri, günlükler ve yürütme durumlarıyla ayrı bir görev olarak tanımlar. 

Airflow’un yapay zekâ boru hatlarına bu kadar iyi uymasının ana nedeni budur: hataları beklenmedik bir sürpriz değil, iş akışının beklenen bir parçası olarak ele alır. 

Bir deney tekrarlanan bir üretim sürecine dönüştükçe, Airflow bunu gözlemlenebilir, kurtarılabilir ve daha kolay bakım yapılabilir kılmak için gereken orkestrasyon katmanını sağlar.

Kendi boru hattımızı kurmaya başlamadan önce, ortamı kuralım.

Ortamı Hazırlama

Ortamı kurmak için üç kısa adım uygulamamız gerekiyor. 

Airflow 3 ve Airflow AI SDK’nın kurulması

Bu öğreticide Airflow’u yerel olarak çalıştıracağız. 

Bu kurulum, öğrenme ve geliştirme için uygundur; çünkü tek bir komut, meta veri veritabanını başlatır ve Airflow’u kullanmak için gereken hizmetleri çalıştırır.

Önce, Airflow ve bağımlılıklarını Python kurulumunuzun geri kalanından yalıtmak için bir sanal ortam oluşturalım ve etkinleştirelim:

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

Sonra, sürüm sabitleme ve resmi kısıt dosyasını kullanarak Airflow 3’ü kurmamız gerekiyor:

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}"

Airflow’un birçok Python paketine bağımlı olduğunu ve bu paketlerin de kendi bağımlılıkları olduğunu unutmayın. Kısıt dosyası olmadan pip daha yeni veya uyumsuz paket sürümleri seçebilir. 

Kurulum çalışıyor gibi görünse de Airflow daha sonra sağlayıcıları içe aktarırken, hizmetleri başlatırken veya görevleri çalıştırırken başarısız olabilir. Airflow’un resmi kurulum kılavuzu, bilinen-uyumlu bir bağımlılık setini kurmak için kısıt dosyalarının kullanılmasını önerir.

Boru hattımız bir LLM çağıracağı için, Airflow’un Common AI sağlayıcısına ve Pydantic AI için OpenAI bağımlılığına da ihtiyacımız var:

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

Şimdi OpenAI bağlantısını yapılandıralım. Airflow, kimlik bilgilerini ve bağlantı ayarlarını DAG kodunun dışında saklamak için bağlantıları kullanır. Bu, Python dosyasını daha temiz tutar ve boru hattında gizli bilgileri sabitlemekten kaçınır.

Kurulumu doğruladıktan sonra, modelle gerçek iletişimi sağlamak için OpenAI anahtarını ayarlama zamanı. Bunu yapmanın birçok yolu var. 

Benim tercih ettiğim seçenek şudur:

1. Geçerli terminal oturumunda OpenAI API anahtarımızı dışa aktarın:

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

2.  pydanticai_default adlı bir Airflow bağlantısı oluşturun:

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

Common AI sağlayıcısı, LLM sağlayıcılarıyla

Bu komut bağlantıyı Airflow’un meta veri veritabanında saklar; bu nedenle bundan sonra .env dosyasını tutmanıza veya Airflow’u her başlattığınızda anahtarı yeniden vermenize gerek kalmaz.

3. Son olarak, yerel Airflow ortamını şu komutla başlatabiliriz:

airflow standalone

Airflow çalışırken bu terminali açık tutun. Standalone modu, yerel geliştirme ve öğreticiler için idealdir; ancak üretim dağıtımları için tasarlanmamıştır.

Airflow arayüzüne erişmek için artık tarayıcıda http://localhost:8080 adresini açabiliriz. İlk göreceğimiz şey oturum açma sayfası olacaktır. Peki hangi kullanıcı adı ve şifreyi kullanmalıyız?

Apache Airflow’un oturum açma görünümü.

Airflow standalone modda başlatıldığında, oturum açma kimlik bilgilerini otomatik olarak üretir. Kullanıcı adı ve şifre terminal çıktısına yazdırılır ve ayrıca $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated dosyasında saklanır.

Varsayılan olarak AIRFLOW_HOME = ~/airflow’dur; bu nedenle kimlik bilgilerini şu komutla alabiliriz:

cat ~/airflow/simple_auth_manager_passwords.json.generated

Son olarak, bu kimlik bilgileri Airflow arayüzüne giriş yapmamıza olanak tanıyacaktır.

Arayüzü keşfetme

Giriş yaptıktan sonra Airflow Ana sayfası açılır. Bu görünüm, yerel Airflow ortamının hızlı bir özetini sunar.

İlk oturum açmadan sonraki Airflow arayüzü görünümü.

Üst kısımdaki İstatistik kartları, başarısız, çalışan ve etkin iş akışları dâhil olmak üzere DAG’lerinizin mevcut durumunu özetler. 

Boru hattımızı henüz oluşturmadığımız için bu değerler hâlâ sıfır olabilir. 

Sağlık bölümü, temel Airflow hizmetlerinin çalışıp çalışmadığını gösterir. 

Aşağıda, Havuz Yuvaları çubuğu kullanılabilir yürütme kapasitesini gösterir. Bu yazıda özel havuzları yapılandırmayacağız; ancak aynı anda kaç görevin çalışabileceğini kontrol ederken bu alan kullanışlı hâle gelir. 

Son olarak, Geçmiş bölümü son DAG çalıştırmalarını özetler. Yorum duyarlılığı boru hattımızı tetiklediğimizde, bu alan kuyruğa alınan, çalışan, başarılı ve başarısız çalıştırmaları göstermeye başlayacaktır.

Proje yapısı

Artık Airflow çalıştığına göre, boru hattımız için gerekli dosyaları oluşturalım:

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

Yerel standalone bir kurulumda Airflow, varsayılan olarak ~/airflow dizinini AIRFLOW_HOME olarak kullanır. DAG dosyamızı dags dizininin içine yerleştirecek ve proje bağımlılıklarını belgelemek için bir requirements.txt dosyası tutacağız. 

Dolayısıyla proje aşağıdaki yapıya sahip olacaktır:

Minimal Airflow proje yapısı.

eview_sentiment_pipeline.py dosyası; yorumları getirme, bunların duyarlılığını sınıflandırma ve sonuçları depolama görevleri dâhil iş akışı tanımını içerecektir. 

Airflow, yapılandırılmış dags dizinini izler ve Python dosyalarında geçerli DAG tanımları arar. Dosyamız geçerli bir DAG içerir içermez, iş akışı Airflow arayüzündeki Dags sayfasında görünecektir.

Uygulamaya başlarken sorun yaşıyorsanız, Apache Airflow’a Giriş öğreticisine göz atmanızı öneririm. .

İlk Airflow Boru Hattınızı Kurma

Airflow ile uygulamalı çalışmaya başlamak için, müşteri yorumlarını işleyen ve duyarlılıklarını sınıflandıran günlük bir iş oluşturacağız. 

Bir e-ticaret platformu kurduğumuzu hayal edin. Müşteriler şimdiden yorum bırakıyor ve biz bunları her gün otomatik olarak analiz etmek istiyoruz. 

Betiği elle çalıştırmak yerine, iş akışını orkestre etmek için Airflow’u kullanabiliriz.

Adım 1: Yorumları getirme

Ham yorumların bir .jsonl dosyasında saklandığını varsayalım. 

Her satırda bir yorum bir JSON nesnesi olarak bulunur ve gün boyunca yeni yorumlar dosyanın sonuna eklenir. Üretim sisteminde bu veri bir API, veritabanı, nesne deposu veya mesaj kuyruğundan gelebilir. Bu öğretici için JSONL dosyası bize ham yorum verisi için basit ve gerçekçi bir kaynak sağlar.

Bu senaryoda, boru hattındaki ilk adım bu yorumları Airflow’a yüklemektir. Bir not defterinde veya betikte yapacağımız gibi, sıradan bir Python işleviyle başlarız:

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

Bu işlev JSONL dosyasını açar, satır satır okur, boş satırları atlar ve her JSON nesnesini bir Python sözlüğüne dönüştürür. 

Bir satır geçersiz JSON içeriyorsa, görev satır numarasını gösteren net bir hata mesajıyla başarısız olur.

Şimdiye kadar bu, sade Python’du. Sıradaki adım, bunu bir Airflow görevine dönüştürmek ve bir DAG içine yerleştirmektir.

Görevi bir DAG içine sarmalama

İşlevi bir Airflow iş akışının parçasına dönüştürmek için üç parçaya ihtiyacımız var:

1. Önce işlevi @task ile süsleriz. Bu, Airflow’a işlevin bir görev olduğunu söyler.
2. İkinci olarak, @dag ile bir DAG tanımlarız. DAG, iş akışına bir ad, zamanlama, başlangıç tarihi ve diğer yapılandırmaları verir.
3. Üçüncü olarak, görevi DAG işlevi içinde çağırır ve dosyanın sonunda DAG’ı örnekleriz.

İşte en minimal sürüm:

from airflow.sdk import dag

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

def fetch_reviews_pipeline():
   fetch_reviews()

fetch_reviews_pipeline()

Bu DAG henüz otomatik olarak çalıştırılmayacaktır. 

Yazıldığı hâliyle, Airflow arayüzünden elle tetiklememiz gerekir. İş akışının ne zaman çalışacağını yapılandırmak için DAG tanımına zamanlama parametreleri ekleyebiliriz. Olası bir seçenek şudur:

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

Haydi parçalayalım!

• dag_id DAG’e Airflow içinde benzersiz bir ad verir.

• schedule="@daily" DAG’i günde bir kez çalıştırır.
• start_date Airflow’a zamanlamanın ne zaman başlayabileceğini söyler.
• catchup=False Airflow’un geçmiş tarihler için çalıştırmalar oluşturmasını engeller.
• tags DAG’i Airflow arayüzünde düzenlemeye ve filtrelemeye yardımcı olur.

DAG tanımlandıktan sonra, arayüze geçip bunun pratikte nasıl göründüğüne bakabiliriz:

Airflow arayüzünde review_sentiment_pipeline boru hattımızın görünümü. Zamanlanmış ve manuel olmak üzere iki tür çalıştırmaya ve manuel Tetikle düğmesine dikkat edin.

Arayüzde, Airflow’un review_sentiment_pipeline adlı DAG’i otomatik olarak keşfettiğini görebiliriz. 

Ayrıca zamanlamasına göre çalıştığını ve arayüzün sağ üst köşesindeki Trigger düğmesini kullanarak bir ek çalıştırmayı manuel olarak tetiklediğimizi de görebiliriz.

Güzel özelliklerden biri, Airflow’un boru hattı çalıştırmalarını her çalıştırmanın durumuyla birlikte kaydettiği takvim tarzı görünümdür:

Zamanlanmış ve manuel çalıştırmalarla Airflow boru hattımızın takvim görünümü.

Makalenin bu noktasında durup Airflow arayüzünü keşfetmekte özgürsünüz!

Adım 2: LLM sınıflandırması

Yorumlar yüklendikten sonra sıradaki adım, bunların duyarlılığını sınıflandırmaktır. Bunun için, bir LLM çağrısını yerel bir Airflow görevine dönüştüren @task.llm süsleyicisini kullanacağız.

OpenAI API’sini normal bir @task içinde elle çağırabilirdik; ancak @task.llm bize daha temiz bir entegrasyon sunar. 

Model bağlantısı, sistem komutu, yapılandırılmış çıktı ve sonuç serileştirme gibi LLM’e özgü ayrıntıları ele alır. Aynı zamanda görev, yeniden denemeler, günlükleme, izleme ve bağımlılık yönetimi gibi standart Airflow özelliklerinden yararlanmaya devam eder.

Bu ikinci adım, tam görevden biraz daha fazla parçaya sahiptir; bu yüzden aşamalı olarak inşa edelim:

1. Önce, modelden beklediğimiz çıktı yapısını tanımlarız:

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"]

Bu, LLM görevine her yanıtta yorum kimliğinin ve üç duyarlılık etiketinden birinin: positive, neutral veya negative bulunması gerektiğini söyler.

2. Sonra, tek bir yorum için istemi oluşturan işlevi yazarız:

@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']}"
   )

İşlev gövdesi bir dize istemi döndürürken, @task.llm süsleyicisi, bu istemi daha önce yapılandırdığımız pydanticai_default bağlantısını kullanarak modele göndermekle ilgilenir. 

system_prompt modele talimatını verirken, output_type=ReviewClassification yapılandırılmış bir yanıt ister. Ayrıca serialize_output=True ayarlar, böylece Airflow sonucu düz bir sözlük olarak depolar; bu da aşağı akıştaki görevlere aktarmayı kolaylaştırır.

Interested in tasks requiring LLM inference? Consider this LangChain course for more specialized applications!

Visualizing the Model Classification

The classification task returns a structured result, but during development, it is useful to see what the model produced without waiting until the final storage step. 

Indeed, a pipeline step can be composed of more than one Airflow task. 

In this case, we will add a small task that prints each classification to the Airflow logs:

@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

This task does not change the result. It simply logs the model output and returns the same dictionary, so the classification can still be passed to the next step of the pipeline.

 3. Finally, we need to connect the new tasks inside the DAG function:

@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()

Note: In this pipeline, we use dynamic task mapping with .expand(review=reviews). This tells Airflow to create one classify_review task instance for each review, so each review is sent to the LLM in a separate request.

This is usually preferable to sending the entire review list to one LLM task. A call such as classify_reviews(reviews) would process all reviews together, which can make failures harder to debug, retries less targeted, and large inputs more likely to hit model context or API limits.

Before finishing this second step of the pipeline, we can return to the Airflow UI and visualize how the workflow has evolved. 

At this point, the DAG contains three tasks: fetch_reviews, classify_review, and show_classification :

Airflow UI view of the three tasks in our review_sentiment_pipeline DAG.

Troubleshooting :  Handling temporary LLM failures

Because we used dynamic task mapping, Airflow sends one LLM request per review. This is useful because each review is tracked independently:

Airflow UI showing reviews being classified independently. Some tasks have already completed or failed, while others are still running.

LLM requests can fail for reasons such as API rate limits, network issues, or model timeouts. 

Instead of failing the whole pipeline immediately, we can ask Airflow to retry the affected task automatically.

Let’s add this feature!

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

Here, retries=3 allows up to three additional attempts after the first failure. 

The first retry waits 30 seconds, and exponential backoff increases the delay between later attempts, up to a maximum of five minutes.

The nice part is that Airflow retries only the failed mapped task. 

If 49 reviews were classified successfully and one request failed, only that failed review is scheduled for retry. The rest of the pipeline does not need to repeat work that has already succeeded. 

After some time, the failed review classifications will again be up for retry:

Airflow UI showing all failed review-classification tasks waiting to be retried.

Once the review classification succeeds, each mapped show_classification() task instance displays the model response for one review in its own logs. 

This means the display step is independent too: you can open, inspect, or debug the output of one review regardless of the state of the others.

Airflow UI view of the LLM sentiment classification for review 47.

Step 3: Store results

Printing each classification is useful while developing the DAG because it lets us quickly inspect the model responses in the Airflow logs. 

However, logs are not a reliable place to store pipeline results. They are harder to query, reuse, and integrate with downstream systems.

The final step is therefore to collect the classifications and write them to a structured output. 

For this article, we will remove the call to show_classification() and store the results in a local JSON file. 

In a production pipeline, this task would typically write to a database, data warehouse, or object store. 

One possible implementation for this new @task would be:

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)

This task is our load step. 

It takes the classifications produced by the LLM, creates an output folder if needed, and writes everything to a local JSON file. 

The small list(results) conversion is useful because the classifications come from mapped tasks. 

Airflow collects those mapped outputs for us, and this task turns them into a regular list before saving them. Now we only need to connect the task inside the DAG by calling store_results(classifications).

With this line, Airflow knows that store_results() depends on the classification step. In other words, the output file is written only after all review classifications have finished successfully.

Running the pipeline

With all three steps connected, the complete DAG now fetches the reviews, creates one LLM task for each review, and stores the combined results. 

Inside the DAG function, the pipeline looks like this:

@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()

After you save the file, Airflow will take a few seconds to parse it and discover the complete DAG. Once review_sentiment_pipeline appears in the DAG list, enable it with the toggle and click Trigger to start a manual run. 

This lets us test the full pipeline immediately, without waiting for the next scheduled daily run. Once the run succeeds, the final JSON file will be locally available at ~/airflow/output/classified_reviews.json.

The output JSON file is the durable output of the workflow, which can be reused outside the DAG or passed to downstream systems later. 

In contrast, the Airflow UI gives us the operational view of the pipeline: which tasks ran, which ones failed or succeeded, and where to inspect logs. 

The final pipeline is displayed below:

Airflow UI view of the three tasks in our pipeline. The bottom-right panel displays each task’s properties, including its name, trigger rule, operator or decorator, and number of retries.

Finally, over the next few days, the Airflow UI will start filling with automatic daily runs, giving us a clear history of when the pipeline ran and whether each execution succeeded:

Airflow UI view of the pipeline runs and their states. After Step 3, the pipeline stops using show_classification and runs store_results instead.

Going Further: Airflow 3 AI Features to Explore Next

Our pipeline now fetches reviews, classifies them with an LLM, and stores the results on a daily schedule. 

Airflow 3 also supports more advanced patterns that can make AI workflows more responsive and easier to supervise.

Event-driven scheduling

Instead of running the pipeline at a fixed time every day, Airflow can start it when an external event occurs. 

For example, the review pipeline could run whenever a new review file is uploaded to object storage or a message arrives in a queue.

Airflow 3 supports event-driven scheduling through assets and event-compatible triggers. This removes the need to repeatedly check for new data and allows the pipeline to react as soon as an event is detected.

Human-in-the-loop with HITLOperator

Not every model decision should be accepted automatically. 

Some reviews may be ambiguous, sensitive, or require human judgment. Airflow’s human-in-the-loop operators can pause a running workflow and request input through the Airflow interface. 

A reviewer can approve or reject a result, provide information, or choose which branch of the workflow should continue.

The Common AI provider also supports interactive review patterns for agentic workflows, where a reviewer can approve, reject, or request changes to an LLM-generated result.

Together, event-driven scheduling and human review make it possible to build AI pipelines that react quickly without removing human oversight where it matters. 

To learn more about the features of Airflow 3.0, I recommend checking out our separate guide. 

Final Thoughts

In this article, we built a complete AI pipeline with Apache Airflow. 

The pipeline fetches customer reviews, classifies each review with an LLM, and stores the final results in a structured JSON file.

Along the way, we saw how Airflow turns Python functions into scheduled, observable tasks. This is especially useful for AI workflows, where API limits, model timeouts, and unexpected outputs can happen. 

With task dependencies, dynamic task mapping, retries, logs, and the Airflow UI, we can make AI pipelines easier to monitor, debug, and recover.

The example used local files to keep things simple, but the same structure can be extended to APIs, databases, event-driven workflows, and human-in-the-loop review. 

If you are ready to continue exploring, I recommend starting with our Introduction to Apache Airflow in Python course, then moving on to Building Data Pipelines with Airflow for more hands-on practice. 

For a broader learning path, the Data Engineer in Python career track covers the wider set of skills needed to build and manage production data pipelines.