Orchestrare pipeline di IA con Airflow

I prototipi di IA sono facili da avviare. Un notebook, una chiave API e poche righe di Python possono bastare per classificare testi, riassumere documenti o generare raccomandazioni. 

Tenerli in esecuzione in modo affidabile tutti i giorni, però, è tutta un'altra sfida.

Cosa succede quando l'API del modello raggiunge il rate limit? E se una richiesta va in timeout, l'output non segue il formato atteso o un database a valle non è temporaneamente disponibile?

Quando una pipeline di IA diventa un flusso di lavoro ricorrente, affidabilità, monitoraggio e ripristino diventano rapidamente importanti quanto il modello stesso.

In questo articolo, ti mostrerò come costruire una pipeline di IA completa con Apache Airflow, una piattaforma open source per sviluppare, pianificare e monitorare flussi di lavoro. 

La nostra pipeline verrà eseguita ogni giorno, recuperando le recensioni dei clienti, classificandone il sentiment con un large language model (LLM) e salvando i risultati. Tutto in modo affidabile all'interno di un unico workflow!

Perché Airflow per le pipeline di IA?

Un notebook Jupyter è perfetto per la sperimentazione. Li adoriamo per testare idee, scrivere codice al volo e visualizzare i risultati man mano.

Un cron job spesso basta per eseguire un semplice script ogni notte. Anche se, a essere onesti, molti script vengono eseguiti manualmente per settimane prima che qualcuno dia loro una pianificazione adeguata.

I problemi iniziano quando uno script diventa diversi script, e questi iniziano a dipendere l'uno dall'altro. A quel punto non abbiamo più un programma standalone: abbiamo un workflow.

Né i notebook né i cron job sono progettati per gestire in modo affidabile flussi di lavoro complessi. 

Le dipendenze sono spesso nascoste nel codice, i fallimenti richiedono indagini manuali e rieseguire la pipeline può significare ripetere passaggi già completati con successo.

Questo tipo di complessità lo incontriamo già nelle pipeline ETL (Extract-Transform-Load) tradizionali. I workflow ETL possono dover recuperare dati, trasformarli, validarli e caricarli in un altro sistema. 

Anche quando input e trasformazioni sono prevedibili, ogni passaggio aggiuntivo crea un nuovo possibile punto di errore. Oggi, le pipeline di IA rendono questa sfida ancora più marcata. 

Se ti interessano in particolare le pipeline ETL, ti consiglio questo corso sulle pipeline ETL in Airflow.

Pipeline di IA

Considera un tipico workflow di IA: raccogliere i dati, pulirli, chiamare un'API di modello, validare la risposta, archiviare il risultato e avvisare qualcuno quando qualcosa va storto. 

Ogni passaggio dipende dal precedente e ognuno può fallire in modo diverso: un'API può restituire un errore di rate limit. Una richiesta al modello può andare in timeout. Una risposta generata può essere JSON valido ma non superare comunque un controllo qualità. 

Gli output possono anche essere incompleti, incoerenti o formattati in modo diverso da quanto si aspetta il passaggio successivo.

Airflow non elimina questi fallimenti, ma li rende più facili da gestire. 

Con Airflow, un task che chiama un LLM può riprovare dopo una risposta temporanea di rate limit senza rieseguire lo step di ingestion dei dati. Un task di validazione può rifiutare un risultato malformato o di bassa qualità prima che arrivi in produzione. Gli avvisi possono notificare il team solo dopo che i tentativi configurati sono stati esauriti.

Airflow rende tutto questo possibile rappresentando il workflow come un grafo aciclico diretto (DAG). 

Invece di nascondere l'intera pipeline in un unico lungo script, un DAG definisce ogni operazione come un task separato con dipendenze esplicite, politiche di retry, log e stato di esecuzione. 

Questo è il motivo principale per cui Airflow si adatta così bene alle pipeline di IA: tratta i fallimenti come una parte prevista del workflow, non come una sorpresa eccezionale. 

Quando un esperimento cresce fino a diventare un processo ricorrente di produzione, Airflow fornisce lo strato di orchestrazione necessario per renderlo osservabile, ripristinabile e più facile da mantenere.

Prima di iniziare a costruire la nostra pipeline, impostiamo l'ambiente.

Configurazione dell'ambiente

Ci sono tre brevi passaggi da eseguire per configurare l'ambiente. 

Installare Airflow 3 e l'SDK Airflow AI

Per questo tutorial, eseguiremo Airflow in locale. 

Questa configurazione è comoda per l'apprendimento e lo sviluppo perché un singolo comando inizializza il database di metadati e avvia i servizi necessari per usare Airflow.

Per prima cosa, creiamo e attiviamo un ambiente virtuale per mantenere Airflow e le sue dipendenze isolati dal resto dell'installazione di Python:

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

Poi, dobbiamo installare Airflow 3 usando il pinning di versione e il file di vincoli ufficiale:

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

Nota che Airflow dipende da molti pacchetti Python, e questi pacchetti hanno a loro volta dipendenze. Senza il file di vincoli, pip potrebbe scegliere versioni più recenti o incompatibili. 

L'installazione potrebbe sembrare riuscita, ma Airflow può poi fallire durante l'import dei provider, l'avvio dei servizi o l'esecuzione dei task. La guida ufficiale di installazione di Airflow raccomanda di usare i file di vincoli per installare un set di dipendenze noto e compatibile.

Poiché il nostro workflow chiamerà un LLM, ci serve anche il provider Common AI di Airflow e la dipendenza OpenAI per Pydantic AI:

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

Ora configuriamo la connessione a OpenAI. Airflow usa le connessioni per memorizzare credenziali e impostazioni di connessione al di fuori del codice del DAG. Questo mantiene il file Python più pulito ed evita di hardcodare i segreti nella pipeline.

Una volta verificata l'installazione, è il momento di impostare la chiave OpenAI per consentire la comunicazione effettiva con il modello. Ci sono molte opzioni per farlo. 

La mia opzione preferita è:

1. Esportare la nostra chiave API di OpenAI nella sessione terminale corrente:

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

2.  Creare una connessione Airflow chiamata pydanticai_default:

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

Il provider Common AI usa il tipo di connessione pydanticai per comunicare con i provider LLM tramite Pydantic AI. 

Questo comando memorizza la connessione nel database dei metadati di Airflow, quindi dopo non devi più mantenere il file .env o passare di nuovo la chiave ogni volta che avvii Airflow.

3. Infine, possiamo avviare l'ambiente Airflow locale con:

airflow standalone

Mantieni questo terminale aperto mentre Airflow è in esecuzione. La modalità standalone è ideale per sviluppo locale e tutorial, ma non è pensata per ambienti di produzione.

Per accedere all'interfaccia di Airflow, ora possiamo aprire http://localhost:8080 nel browser. La prima cosa che vedremo è la pagina di login. Ma quale username e password dobbiamo usare?

Schermata di login di Apache Airflow.

Quando Airflow si avvia in modalità standalone, genera automaticamente le credenziali di accesso. Username e password vengono stampati nell'output del terminale e salvati anche in $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated.

Per impostazione predefinita, AIRFLOW_HOME è ~/airflow, quindi possiamo recuperare le credenziali con:

cat ~/airflow/simple_auth_manager_passwords.json.generated

Infine, quelle credenziali ci permetteranno di effettuare l'accesso all'interfaccia di Airflow.

Esplorare la UI

Dopo il login, Airflow si apre sulla Home. Questa vista ti offre una panoramica rapida dell'ambiente Airflow locale.

UI di Airflow dopo il primo accesso.

In alto, le schede Stats riassumono lo stato corrente dei tuoi DAG, inclusi workflow falliti, in esecuzione e attivi. 

Dato che non abbiamo ancora creato la nostra pipeline, questi valori potrebbero essere ancora a zero. 

La sezione Health mostra se i principali servizi di Airflow sono in esecuzione. 

Più sotto, la barra Pool Slots indica la capacità di esecuzione disponibile. Non configureremo pool personalizzati in questo articolo, ma quest'area diventa utile quando vuoi controllare quante attività possono essere eseguite contemporaneamente. 

Infine, la sezione History riassume le esecuzioni recenti dei DAG. Una volta attivata la nostra pipeline per il sentiment delle recensioni, quest'area inizierà a mostrare esecuzioni in coda, in corso, riuscite e fallite.

Struttura del progetto

Ora che Airflow è in esecuzione, creiamo i file necessari per la nostra pipeline:

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

In un'installazione locale standalone, Airflow usa ~/airflow come AIRFLOW_HOME per impostazione predefinita. Inseriremo il nostro file DAG nella directory dags e manterremo un file requirements.txt per documentare le dipendenze del progetto. 

Pertanto, il progetto avrà la seguente struttura:

Struttura minima di un progetto Airflow.

Il file eview_sentiment_pipeline.py conterrà la definizione del workflow, inclusi i task per recuperare le recensioni, classificarne il sentiment e archiviare i risultati. 

Airflow monitora la directory dags configurata e cerca definizioni valide di DAG all'interno dei file Python. Una volta che il nostro file conterrà un DAG valido, il workflow apparirà nella pagina Dags dell'UI di Airflow.

Se riscontri problemi con l'onboarding, ti suggerisco di consultare il tutorial Getting Started with Apache Airflow. .

Costruire la tua prima pipeline Airflow

Per iniziare a lavorare concretamente con Airflow, costruiremo un job giornaliero che elabora le recensioni dei clienti e ne classifica il sentiment. 

Immagina di aver appena creato una piattaforma e-commerce. I clienti stanno già lasciando recensioni e vogliamo analizzarle automaticamente ogni giorno. 

Invece di eseguire uno script manualmente, possiamo usare Airflow per orchestrare il workflow.

Passo 1: recupero delle recensioni

Immaginiamo che le recensioni grezze siano archiviate in un file .jsonl. 

Ogni riga contiene una recensione come oggetto JSON e durante la giornata le nuove recensioni vengono aggiunte in coda al file. In un sistema di produzione, questi dati potrebbero provenire da un'API, un database, un object store o una coda di messaggi. Per questo tutorial, il file JSONL ci offre una sorgente semplice e realistica di dati grezzi sulle recensioni.

In questo scenario, il primo step della pipeline è caricare quelle recensioni in Airflow. Partiamo con una normale funzione Python, proprio come faremmo in un notebook o in uno script:

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

Questa funzione apre il file JSONL, lo legge riga per riga, salta le righe vuote e converte ogni oggetto JSON in un dizionario Python. 

Se una riga contiene JSON non valido, il task fallisce con un messaggio di errore chiaro che mostra il numero di riga.

Finora è puro Python. Il passo successivo è trasformarla in un task Airflow e inserirla in un DAG.

Inserire il task in un DAG

Per trasformare la funzione in parte di un workflow Airflow, ci servono tre elementi:

1. Per prima cosa, decoriamo la funzione con @task. Questo dice ad Airflow che la funzione è un task.
2. Poi, definiamo un DAG con @dag. Il DAG fornisce al workflow un nome, una pianificazione, una data di inizio e altre configurazioni.
3. Infine, chiamiamo il task dentro la funzione del DAG e istanziamo il DAG alla fine del file.

Ecco la versione minima:

from airflow.sdk import dag

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

def fetch_reviews_pipeline():
   fetch_reviews()

fetch_reviews_pipeline()

Nota che questo DAG non verrà ancora eseguito automaticamente. 

Così com'è, dovremmo attivarlo manualmente dalla UI di Airflow. Per configurare quando deve essere eseguito il workflow, possiamo aggiungere parametri di schedulazione alla definizione del DAG. Una possibile opzione è:

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

Vediamoli nel dettaglio!

• dag_id assegna al DAG un nome univoco in Airflow.

• schedule="@daily" esegue il DAG una volta al giorno.
• start_date indica ad Airflow quando può iniziare la schedulazione.
• catchup=False impedisce ad Airflow di creare esecuzioni per le date passate.
• tags aiutano a organizzare e filtrare il DAG nella UI di Airflow.

Una volta definito il DAG, possiamo passare alla UI e vedere come appare in pratica:

Vista della UI di Airflow della nostra pipeline review_sentiment_pipeline. Nota i due tipi di esecuzioni (pianificate e manuali) e il pulsante Trigger manuale.

Nella UI possiamo vedere che Airflow ha scoperto automaticamente il DAG chiamato review_sentiment_pipeline. 

Possiamo anche vedere che è stato eseguito secondo la sua pianificazione e che abbiamo attivato un'esecuzione aggiuntiva manualmente usando il pulsante Trigger nell'angolo in alto a destra dell'interfaccia.

Una funzione utile è la vista a calendario, in cui Airflow registra le esecuzioni della pipeline insieme allo stato di ciascuna esecuzione:

Vista calendario della nostra pipeline Airflow con esecuzioni sia pianificate sia manuali.

Prenditi pure una pausa dall'articolo per esplorare la UI di Airflow!

Passo 2: classificazione con LLM

Con le recensioni caricate, il passo successivo è classificarne il sentiment. Per farlo useremo @task.llm, un decorator che trasforma una chiamata a un LLM in un task nativo di Airflow.

Potremmo chiamare manualmente l'API di OpenAI dentro un normale @task, ma @task.llm ci offre un'integrazione più pulita. 

Gestisce dettagli specifici degli LLM come la connessione al modello, il system prompt, l'output strutturato e la serializzazione del risultato. Allo stesso tempo, il task beneficia comunque delle funzionalità standard di Airflow come retry, logging, monitoraggio e gestione delle dipendenze.

Questo secondo step ha qualche elemento in più rispetto al task in sé, quindi costruiamolo gradualmente:

1. Per prima cosa, definiamo la struttura dell'output che ci aspettiamo dal modello:

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

Questo indica al task LLM che ogni risposta deve contenere l'ID della recensione e una tra tre etichette di sentiment: positive, neutral o negative.

2. Poi, creiamo la funzione che costruisce il prompt per una singola recensione:

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

Il corpo della funzione restituisce una stringa di prompt, mentre il decorator @task.llm si occupa di inviare quel prompt al modello usando la connessione pydanticai_default che abbiamo configurato prima. 

Il system_prompt fornisce al modello l'istruzione, mentre output_type=ReviewClassification richiede una risposta strutturata. Impostiamo anche serialize_output=True così Airflow memorizza il risultato come un semplice dizionario, più facile da passare ai task a valle.

Ti interessano task che richiedono inferenza LLM? Valuta questo corso su LangChain per applicazioni più specialistiche!

Visualizzare la classificazione del modello

Il task di classificazione restituisce un risultato strutturato, ma durante lo sviluppo è utile vedere cosa ha prodotto il modello senza aspettare lo step finale di salvataggio. 

Infatti, uno step di pipeline può essere composto da più task Airflow. 

In questo caso, aggiungeremo un piccolo task che stampa ogni classificazione nei log di Airflow:

@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

Questo task non modifica il risultato. Si limita a registrare l'output del modello e restituisce lo stesso dizionario, così la classificazione può comunque essere passata allo step successivo della pipeline.

 3. Infine, dobbiamo collegare i nuovi task dentro la funzione del DAG:

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

Nota: in questa pipeline usiamo il dynamic task mapping con .expand(review=reviews). Questo dice ad Airflow di creare un'istanza di task classify_review per ogni recensione, così ciascuna recensione viene inviata all'LLM in una richiesta separata.

Questo di solito è preferibile rispetto a inviare l'intera lista di recensioni a un unico task LLM. Una chiamata come classify_reviews(reviews) elaborerebbe tutte le recensioni insieme, rendendo più difficile il debug dei fallimenti, meno mirati i retry e più probabile che input grandi incorrano in limiti di contesto del modello o dell'API.

Prima di concludere questo secondo step della pipeline, possiamo tornare alla UI di Airflow e visualizzare come si è evoluto il workflow. 

A questo punto, il DAG contiene tre task: fetch_reviews, classify_review e show_classification:

Vista UI di Airflow dei tre task nel DAG review_sentiment_pipeline.

Risoluzione dei problemi: gestire i fallimenti temporanei dell'LLM

Poiché abbiamo usato il dynamic task mapping, Airflow invia una richiesta LLM per recensione. Questo è utile perché ogni recensione viene tracciata in modo indipendente:

UI di Airflow che mostra le recensioni classificate in modo indipendente. Alcuni task sono già completati o falliti, mentre altri sono ancora in esecuzione.

Le richieste LLM possono fallire per motivi come rate limit dell'API, problemi di rete o timeout del modello. 

Invece di far fallire subito l'intera pipeline, possiamo chiedere ad Airflow di ritentare automaticamente il task interessato.

Aggiungiamo questa funzionalità!

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

Qui, retries=3 consente fino a tre tentativi aggiuntivi dopo il primo fallimento. 

Il primo retry attende 30 secondi e il backoff esponenziale aumenta l'intervallo tra i tentativi successivi, fino a un massimo di cinque minuti.

La parte interessante è che Airflow ritenta solo il task mappato che è fallito. 

Se 49 recensioni sono state classificate con successo e una richiesta è fallita, solo quella recensione fallita viene pianificata per il retry. Il resto della pipeline non deve ripetere lavoro già riuscito. 

Dopo un po', le classificazioni delle recensioni fallite torneranno in attesa di retry:

UI di Airflow che mostra tutti i task di classificazione delle recensioni falliti in attesa di retry.

Una volta che la classificazione della recensione riesce, ogni istanza mappata di show_classification() mostra la risposta del modello per una recensione nei propri log. 

Questo significa che anche lo step di visualizzazione è indipendente: puoi aprire, ispezionare o fare debug dell'output di una recensione indipendentemente dallo stato delle altre.

Vista UI di Airflow della classificazione del sentiment LLM per la recensione 47.

Passo 3: salvare i risultati

Stampare ogni classificazione è utile durante lo sviluppo del DAG perché ci permette di ispezionare rapidamente le risposte del modello nei log di Airflow. 

Tuttavia, i log non sono un posto affidabile dove archiviare i risultati della pipeline. Sono più difficili da interrogare, riutilizzare e integrare con i sistemi a valle.

Lo step finale è quindi raccogliere le classificazioni e scriverle in un output strutturato. 

Per questo articolo, rimuoveremo la chiamata a show_classification() e salveremo i risultati in un file JSON locale. 

In una pipeline di produzione, questo task di solito scriverebbe su un database, un data warehouse o un object store. 

Una possibile implementazione per questo nuovo @task è:

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)

Questo task è il nostro step di load. 

Prende le classificazioni prodotte dall'LLM, crea una cartella di output se necessario e scrive tutto in un file JSON locale. 

La piccola conversione list(results) è utile perché le classificazioni provengono da task mappati. 

Airflow raccoglie per noi quegli output mappati e questo task li trasforma in una lista normale prima di salvarli. Ora dobbiamo solo collegare il task dentro il DAG chiamando store_results(classifications).

Con questa riga, Airflow sa che store_results() dipende dallo step di classificazione. In altre parole, il file di output viene scritto solo dopo che tutte le classificazioni delle recensioni sono terminate con successo.

Eseguire la pipeline

Con tutti e tre gli step collegati, il DAG completo ora recupera le recensioni, crea un task LLM per ciascuna recensione e salva i risultati combinati. 

Dentro la funzione del DAG, la pipeline appare così:

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

Dopo aver salvato il file, Airflow impiegherà alcuni secondi per analizzarlo e scoprire il DAG completo. Una volta che review_sentiment_pipeline appare nell'elenco dei DAG, abilitalo con l'interruttore e fai clic su Trigger per avviare un'esecuzione manuale. 

Questo ci permette di testare subito l'intera pipeline, senza aspettare la prossima esecuzione giornaliera pianificata. Una volta che l'esecuzione va a buon fine, il file JSON finale sarà disponibile localmente in ~/airflow/output/classified_reviews.json.

Il file JSON di output è l'output durevole del workflow, che può essere riutilizzato al di fuori del DAG o passato successivamente ai sistemi a valle. 

Al contrario, la UI di Airflow ci offre la vista operativa della pipeline: quali task sono stati eseguiti, quali sono falliti o riusciti e dove ispezionare i log. 

La pipeline finale è mostrata qui sotto:

Vista UI di Airflow dei tre task della nostra pipeline. Il pannello in basso a destra mostra le proprietà di ciascun task, incluso nome, regola di attivazione, operatore o decorator e numero di retry.

Infine, nei giorni successivi, la UI di Airflow inizierà a riempirsi di esecuzioni giornaliere automatiche, dandoci una cronologia chiara di quando la pipeline è stata eseguita e se ogni esecuzione è riuscita:

Vista UI di Airflow delle esecuzioni della pipeline e dei loro stati. Dopo il Passo 3, la pipeline smette di usare show_classification ed esegue invece store_results.

Oltre: funzionalità di IA in Airflow 3 da esplorare

La nostra pipeline ora recupera le recensioni, le classifica con un LLM e salva i risultati su base giornaliera. 

Airflow 3 supporta anche pattern più avanzati che possono rendere i workflow di IA più reattivi e facili da supervisionare.

Schedulazione event-driven

Invece di eseguire la pipeline a un'ora fissa ogni giorno, Airflow può avviarla quando si verifica un evento esterno. 

Per esempio, la pipeline delle recensioni potrebbe essere eseguita quando un nuovo file di recensioni viene caricato nell'object storage o quando arriva un messaggio in una coda.

Airflow 3 supporta la schedulazione event-driven tramite asset e trigger compatibili con gli eventi. Questo elimina la necessità di controllare ripetutamente la presenza di nuovi dati e permette alla pipeline di reagire non appena un evento viene rilevato.

Human-in-the-loop con HITLOperator

Non tutte le decisioni del modello dovrebbero essere accettate automaticamente. 

Alcune recensioni possono essere ambigue, sensibili o richiedere giudizio umano. Gli operatori human-in-the-loop di Airflow possono mettere in pausa un workflow in esecuzione e richiedere input tramite l'interfaccia di Airflow. 

Un revisore può approvare o rifiutare un risultato, fornire informazioni o scegliere quale ramo del workflow debba continuare.

Il provider Common AI supporta anche pattern di revisione interattiva per workflow agentici, in cui un revisore può approvare, rifiutare o richiedere modifiche a un risultato generato da un LLM.

Insieme, schedulazione event-driven e revisione umana consentono di costruire pipeline di IA che reagiscono rapidamente senza rimuovere la supervisione umana dove conta. 

Per saperne di più sulle funzionalità di Airflow 3.0, ti consiglio di consultare la nostra guida dedicata. 

Considerazioni finali

In questo articolo, abbiamo costruito una pipeline di IA completa con Apache Airflow. 

La pipeline recupera le recensioni dei clienti, classifica ciascuna recensione con un LLM e salva i risultati finali in un file JSON strutturato.

Lungo il percorso, abbiamo visto come Airflow trasformi funzioni Python in task pianificati e osservabili. Questo è particolarmente utile per i workflow di IA, dove possono verificarsi limiti di API, timeout del modello e output inattesi. 

Con dipendenze tra task, dynamic task mapping, retry, log e la UI di Airflow, possiamo rendere le pipeline di IA più facili da monitorare, fare debug e ripristinare.

L'esempio ha usato file locali per mantenere la semplicità, ma la stessa struttura può essere estesa ad API, database, workflow event-driven e revisione human-in-the-loop. 

Se sei pronto a continuare l'esplorazione, ti consiglio di partire dal nostro corso Introduction to Apache Airflow in Python, per poi passare a Building Data Pipelines with Airflow per esercitarti in modo più pratico. 

Per un percorso di apprendimento più ampio, il Data Engineer in Python career track copre il set più ampio di competenze necessarie per costruire e gestire pipeline di dati di produzione.