Orchestration de pipelines d’IA avec Airflow

Démarrer un prototype d’IA est facile. Un notebook, une clé d’API et quelques lignes de Python peuvent suffire pour classer du texte, résumer des documents ou générer des recommandations. 

En revanche, faire tourner ce prototype de manière fiable au quotidien est un tout autre défi.

Que se passe-t-il lorsque l’API du modèle atteint sa limite de débit ? Et si une requête expire, si la sortie ne respecte pas le format attendu, ou si une base de données aval est temporairement indisponible ?

Quand un pipeline d’IA devient un workflow récurrent, la fiabilité, le monitoring et la reprise deviennent aussi importants que le modèle lui-même.

Dans cet article, je vous montre comment construire un pipeline d’IA complet avec Apache Airflow, une plateforme open source pour développer, planifier et superviser des workflows. 

Notre pipeline s’exécutera chaque jour, récupérera des avis clients, classera leur tonalité à l’aide d’un grand modèle de langage (LLM) et stockera les résultats. Le tout, de façon fiable, dans un seul workflow !

Pourquoi Airflow pour les pipelines d’IA ?

Un notebook Jupyter est parfait pour l’expérimentation. Nous aimons tous tester des idées, écrire du code rapidement et visualiser les résultats au fil de l’eau.

Un cron suffit souvent pour lancer un script simple chaque nuit. Même si, soyons honnêtes, beaucoup de scripts sont exécutés manuellement pendant des semaines avant qu’on ne leur assigne enfin une vraie planification.

Les problèmes commencent lorsqu’un script devient plusieurs scripts, et que ces scripts dépendent les uns des autres. À ce stade, ce n’est plus un programme autonome : c’est un workflow.

Ni les notebooks ni les tâches cron ne sont conçus pour gérer des workflows complexes de manière fiable. 

Les dépendances sont souvent cachées dans le code, les échecs nécessitent une investigation manuelle, et relancer le pipeline peut impliquer de répéter des étapes déjà réussies.

Nous rencontrons déjà une partie de cette complexité dans les pipelines ETL (Extract-Transform-Load) traditionnels. Un workflow ETL peut devoir récupérer des données, les transformer, les valider, puis les charger dans un autre système. 

Même lorsque les entrées et transformations sont prévisibles, chaque étape supplémentaire ajoute un point de défaillance potentiel. Aujourd’hui, les pipelines d’IA accentuent encore ce défi. 

Si vous vous intéressez spécifiquement aux pipelines ETL, je vous recommande ce cours sur les pipelines ETL dans Airflow.

Pipelines d’IA

Considérez un workflow d’IA typique : collecter des données, les nettoyer, appeler l’API d’un modèle, valider la réponse, stocker le résultat et prévenir quelqu’un en cas de problème. 

Chaque étape dépend de la précédente, et chacune peut échouer différemment : une API peut renvoyer une erreur de rate limit. Une requête modèle peut expirer. Une réponse générée peut être un JSON valide mais échouer un contrôle qualité. 

Les sorties peuvent aussi être incomplètes, incohérentes, ou formatées différemment de ce qu’attend l’étape suivante.

Airflow n’élimine pas ces échecs, mais en facilite la gestion. 

Avec Airflow, une tâche qui appelle un LLM peut réessayer après une réponse temporaire de rate limit sans relancer l’ingestion de données. Une tâche de validation peut rejeter un résultat mal formé ou de faible qualité avant qu’il n’atteigne la production. Des alertes peuvent prévenir l’équipe seulement après l’épuisement des tentatives configurées.

Airflow rend cela possible en représentant le workflow sous forme de graphe orienté acyclique (DAG). 

Au lieu de cacher tout le pipeline dans un long script, un DAG définit chaque opération comme une tâche distincte avec des dépendances explicites, des politiques de réessai, des logs et un statut d’exécution. 

C’est la raison principale pour laquelle Airflow convient si bien aux pipelines d’IA : il considère les échecs comme une composante attendue du workflow, pas comme une surprise exceptionnelle. 

À mesure qu’une expérimentation devient un processus de production récurrent, Airflow fournit la couche d’orchestration nécessaire pour le rendre observable, récupérable et plus simple à maintenir.

Avant de construire notre propre pipeline, configurons l’environnement.

Mise en place de l’environnement

Trois étapes rapides suffisent pour préparer l’environnement. 

Installer Airflow 3 et le SDK Airflow AI

Pour ce tutoriel, nous exécuterons Airflow en local. 

Cette configuration est pratique pour l’apprentissage et le développement, car une seule commande initialise la base de métadonnées et démarre les services nécessaires à l’utilisation d’Airflow.

Commençons par créer et activer un environnement virtuel pour isoler Airflow et ses dépendances du reste de votre installation Python :

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

Ensuite, installons Airflow 3 avec un verrouillage de version et le fichier de contraintes officiel :

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

Notez qu’Airflow dépend de nombreux paquets Python, qui ont eux-mêmes leurs propres dépendances. Sans le fichier de contraintes, pip peut choisir des versions plus récentes ou incompatibles. 

L’installation peut sembler fonctionner, mais Airflow peut ensuite échouer lors de l’import des providers, du démarrage des services ou de l’exécution des tâches. Le guide d’installation officiel d’Airflow recommande d’utiliser des fichiers de contraintes pour installer un ensemble de dépendances connu comme compatible.

Comme notre workflow appellera un LLM, nous avons aussi besoin du provider Common AI d’Airflow et de la dépendance OpenAI pour Pydantic AI :

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

Configurons ensuite la connexion OpenAI. Airflow utilise des connexions pour stocker les identifiants et paramètres de connexion en dehors du code du DAG. Cela rend le fichier Python plus propre et évite de coder en dur des secrets dans le pipeline.

Une fois l’installation validée, il est temps de définir la clé OpenAI pour permettre la communication avec le modèle. Plusieurs options existent. 

Mon option préférée est la suivante :

1. Exporter la clé d’API OpenAI dans la session de terminal courante :

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

2.  Créer une connexion Airflow nommée pydanticai_default :

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

Le provider Common AI utilise le type de connexion pydanticai pour communiquer avec les fournisseurs de LLM via Pydantic AI. 

Cette commande enregistre la connexion dans la base de métadonnées d’Airflow : ensuite, vous n’avez plus besoin de conserver le fichier .env ni de repasser la clé à chaque démarrage d’Airflow.

3. Enfin, nous pouvons démarrer l’environnement Airflow local avec :

airflow standalone

Gardez ce terminal ouvert pendant qu’Airflow s’exécute. Le mode standalone est idéal pour le développement local et les tutoriels, mais il n’est pas destiné à la production.

Pour accéder à l’interface Airflow, ouvrez maintenant http://localhost:8080 dans le navigateur. La première chose que vous verrez est la page de connexion. Mais quel nom d’utilisateur et mot de passe utiliser ?

Écran de connexion d’Apache Airflow.

Lorsque Airflow démarre en mode standalone, il génère automatiquement des identifiants de connexion. Le nom d’utilisateur et le mot de passe sont affichés dans la sortie du terminal et également stockés dans $AIRFLOW_HOME/simple_auth_manager_passwords.json.generated.

Par défaut, AIRFLOW_HOME vaut ~/airflow, donc nous pouvons récupérer les identifiants avec :

cat ~/airflow/simple_auth_manager_passwords.json.generated

Ces identifiants vous permettront enfin de vous connecter à l’interface d’Airflow.

Découvrir l’interface

Après connexion, Airflow s’ouvre sur la page d’accueil. Cette vue offre un aperçu rapide de l’environnement Airflow local.

Vue de l’interface Airflow après la première connexion.

En haut, les cartes Statistiques résument l’état actuel de vos DAGs, y compris les workflows en échec, en cours et actifs. 

Comme nous n’avons pas encore créé notre pipeline, ces valeurs peuvent encore être à zéro. 

La section Santé indique si les principaux services Airflow sont en fonctionnement. 

En dessous, la barre Pool Slots montre la capacité d’exécution disponible. Nous ne configurerons pas de pools personnalisés ici, mais cette zone devient utile pour contrôler combien de tâches peuvent s’exécuter en parallèle. 

Enfin, la section Historique résume les exécutions récentes des DAGs. Une fois que nous aurons déclenché notre pipeline d’analyse de sentiment des avis, cette zone affichera les exécutions en file d’attente, en cours, réussies et échouées.

Structure du projet

Maintenant qu’Airflow tourne, créons les fichiers nécessaires pour notre pipeline :

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

Dans une installation locale en standalone, Airflow utilise ~/airflow comme AIRFLOW_HOME par défaut. Nous placerons notre fichier DAG dans le répertoire dags et garderons un fichier requirements.txt pour documenter les dépendances du projet. 

Le projet aura donc la structure suivante :

Structure minimale d’un projet Airflow.

Le fichier eview_sentiment_pipeline.py contiendra la définition du workflow, y compris les tâches de récupération des avis, de classification du sentiment et de stockage des résultats. 

Airflow surveille le répertoire dags configuré et recherche les définitions de DAG valides dans les fichiers Python. Une fois notre fichier pourvu d’un DAG valide, le workflow apparaîtra sur la page Dags de l’interface Airflow.

Si vous rencontrez des difficultés au démarrage, je vous conseille le tutoriel Getting started with Apache Airflow. .

Construire votre premier pipeline Airflow

Pour passer à la pratique avec Airflow, nous allons construire une tâche quotidienne qui traite des avis clients et classifie leur sentiment. 

Imaginez que nous venions de lancer une plateforme e-commerce. Les clients laissent déjà des avis, et nous voulons les analyser automatiquement chaque jour. 

Plutôt que d’exécuter un script manuellement, utilisons Airflow pour orchestrer le workflow.

Étape 1 : récupérer les avis

Imaginons que les avis bruts soient stockés dans un fichier .jsonl. 

Chaque ligne contient un avis au format JSON, et de nouveaux avis sont ajoutés en fin de fichier au fil de la journée. En production, ces données pourraient venir d’une API, d’une base de données, d’un objet storage ou d’une file de messages. Pour ce tutoriel, le fichier JSONL nous fournit une source simple et réaliste de données brutes.

Dans ce scénario, la première étape du pipeline consiste à charger ces avis dans Airflow. Nous commençons par une fonction Python classique, comme nous le ferions dans un notebook ou un 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

Cette fonction ouvre le fichier JSONL, le lit ligne par ligne, ignore les lignes vides et convertit chaque objet JSON en dictionnaire Python. 

Si une ligne contient un JSON invalide, la tâche échoue avec un message d’erreur explicite indiquant le numéro de ligne.

Jusqu’ici, c’est du Python pur. L’étape suivante consiste à en faire une tâche Airflow et à l’intégrer dans un DAG.

Encapsuler la tâche dans un DAG

Pour intégrer la fonction à un workflow Airflow, il nous faut trois éléments :

1. D’abord, décorer la fonction avec @task. Cela indique à Airflow que la fonction est une tâche.
2. Ensuite, définir un DAG avec @dag. Le DAG donne au workflow un nom, une planification, une date de début et d’autres paramètres.
3. Enfin, appeler la tâche à l’intérieur de la fonction du DAG et instancier le DAG en fin de fichier.

Voici la version minimale :

from airflow.sdk import dag

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

def fetch_reviews_pipeline():
   fetch_reviews()

fetch_reviews_pipeline()

Notez que ce DAG ne sera pas encore exécuté automatiquement. 

En l’état, nous devrions le déclencher manuellement depuis l’interface Airflow. Pour configurer le moment d’exécution du workflow, ajoutons des paramètres de planification à la définition du DAG. Une option possible :

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

Décomposons cela !

• dag_id donne au DAG un nom unique dans Airflow.

• schedule="@daily" exécute le DAG une fois par jour.
• start_date indique à partir de quand Airflow peut commencer la planification.
• catchup=False empêche Airflow de créer des exécutions pour les dates passées.
• tags aide à organiser et filtrer le DAG dans l’interface.

Une fois le DAG défini, passons dans l’interface pour voir le rendu :

Vue Airflow de notre pipeline review_sentiment_pipeline. Notez les deux types d’exécutions (planifiées et manuelles) et le bouton Trigger manuel.

Dans l’interface, on voit qu’Airflow a découvert automatiquement le DAG nommé review_sentiment_pipeline. 

On voit aussi qu’il s’est exécuté selon sa planification, et que nous avons déclenché une exécution supplémentaire manuellement via le bouton Trigger en haut à droite.

Une fonctionnalité appréciable est la vue calendrier, où Airflow journalise les exécutions du pipeline avec le statut de chaque exécution :

Vue calendrier de notre pipeline Airflow avec les exécutions planifiées et manuelles.

N’hésitez pas à faire une pause ici pour explorer l’interface Airflow !

Étape 2 : classification par LLM

Une fois les avis chargés, l’étape suivante consiste à classer leur sentiment. Pour cela, nous allons utiliser @task.llm, un décorateur qui transforme un appel LLM en tâche Airflow native.

Nous pourrions appeler l’API OpenAI manuellement dans une @task classique, mais @task.llm offre une intégration plus propre. 

Il gère les aspects spécifiques au LLM comme la connexion au modèle, le system prompt, la sortie structurée et la sérialisation du résultat. En parallèle, la tâche bénéficie toujours des fonctionnalités standards d’Airflow : réessais, logs, monitoring et gestion des dépendances.

Cette deuxième étape a quelques éléments de plus que la tâche précédente, construisons-la progressivement :

1. D’abord, définissons la structure de sortie attendue du modèle :

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

Cela indique à la tâche LLM que chaque réponse doit contenir l’identifiant de l’avis et l’un des trois labels de sentiment : positive, neutral ou negative.

2. Ensuite, créons la fonction qui construit le prompt pour un avis donné :

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

Le corps de la fonction retourne un prompt sous forme de chaîne, tandis que le décorateur @task.llm se charge d’envoyer ce prompt au modèle via la connexion pydanticai_default que nous avons configurée plus tôt. 

Le system_prompt donne l’instruction au modèle, tandis que output_type=ReviewClassification demande une réponse structurée. Nous définissons aussi serialize_output=True pour qu’Airflow stocke le résultat en dictionnaire simple, plus facile à transmettre aux tâches suivantes.

Intéressé par des tâches nécessitant de l’inférence LLM ? Découvrez ce cours LangChain pour des applications plus spécialisées !

Visualiser la classification du modèle

La tâche de classification renvoie un résultat structuré, mais pendant le développement, il est utile de voir rapidement ce que le modèle a produit sans attendre l’étape finale de stockage. 

En effet, une étape du pipeline peut être composée de plusieurs tâches Airflow. 

Dans notre cas, ajoutons une petite tâche qui imprime chaque classification dans les logs 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

Cette tâche ne modifie pas le résultat. Elle se contente de journaliser la sortie du modèle et de renvoyer le même dictionnaire, ce qui permet de transmettre la classification à l’étape suivante du pipeline.

 3. Enfin, connectons les nouvelles tâches dans la fonction du 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()

Remarque : dans ce pipeline, nous utilisons le mapping dynamique de tâches avec .expand(review=reviews). Cela indique à Airflow de créer une instance de tâche classify_review par avis, donc chaque avis est envoyé au LLM via une requête distincte.

C’est généralement préférable à l’envoi de toute la liste d’avis à une seule tâche LLM. Un appel du type classify_reviews(reviews) traiterait tous les avis ensemble, ce qui complique le débogage des échecs, rend les réessais moins ciblés, et augmente le risque de dépasser le contexte du modèle ou les limites de l’API.

Avant de boucler cette deuxième étape, revenons dans l’interface Airflow pour visualiser l’évolution du workflow. 

À ce stade, le DAG contient trois tâches : fetch_reviews, classify_review et show_classification :

Vue Airflow des trois tâches de notre DAG review_sentiment_pipeline.

Dépannage : gérer les échecs temporaires des LLM

Comme nous avons utilisé le mapping dynamique, Airflow envoie une requête LLM par avis. C’est utile, car chaque avis est suivi indépendamment :

L’interface Airflow montre des avis classés indépendamment. Certaines tâches sont terminées ou en échec, tandis que d’autres tournent encore.

Les requêtes LLM peuvent échouer pour des raisons comme des limites d’API, des problèmes réseau ou des timeouts du modèle. 

Plutôt que d’échouer immédiatement l’ensemble du pipeline, nous pouvons demander à Airflow de relancer automatiquement la tâche concernée.

Ajoutons cette fonctionnalité !

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

Ici, retries=3 autorise jusqu’à trois tentatives supplémentaires après le premier échec. 

Le premier réessai attend 30 secondes, et la backoff exponentielle augmente le délai entre les tentatives suivantes jusqu’à un maximum de cinq minutes.

L’agréable dans Airflow, c’est qu’il ne relance que la tâche mappée en échec. 

Si 49 avis ont été classés avec succès et qu’une requête a échoué, seul cet avis échoué est replanifié. Le reste du pipeline n’a pas besoin de répéter du travail déjà réussi. 

Après un moment, les classifications d’avis échouées seront de nouveau prêtes à être relancées :

L’interface Airflow montrant toutes les tâches de classification d’avis en échec en attente de réessai.

Une fois la classification de l’avis réussie, chaque instance mappée de show_classification() affiche la réponse du modèle pour un avis dans ses propres logs. 

Cette étape d’affichage est donc indépendante aussi : vous pouvez ouvrir, inspecter ou déboguer la sortie d’un avis sans dépendre de l’état des autres.

Vue Airflow de la classification de sentiment par LLM pour l’avis 47.

Étape 3 : stocker les résultats

Imprimer chaque classification est utile lors du développement du DAG, car cela permet d’inspecter rapidement les réponses du modèle dans les logs Airflow. 

Cependant, les logs ne sont pas un endroit fiable pour stocker les résultats d’un pipeline. Ils sont plus difficiles à requêter, à réutiliser et à intégrer aux systèmes aval.

La dernière étape consiste donc à rassembler les classifications et les écrire dans une sortie structurée. 

Pour cet article, nous allons retirer l’appel à show_classification() et stocker les résultats dans un fichier JSON local. 

En production, cette tâche écrirait typiquement dans une base de données, un data warehouse ou un object store. 

Une implémentation possible de cette nouvelle @task serait :

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)

Cette tâche est notre étape de chargement. 

Elle prend les classifications produites par le LLM, crée le dossier de sortie si nécessaire et écrit le tout dans un fichier JSON local. 

La petite conversion list(results) est utile car les classifications proviennent de tâches mappées. 

Airflow collecte ces sorties mappées pour nous, et cette tâche les convertit en liste classique avant de les sauvegarder. Il ne reste plus qu’à connecter la tâche dans le DAG en appelant store_results(classifications).

Avec cette ligne, Airflow sait que store_results() dépend de l’étape de classification. Autrement dit, le fichier de sortie n’est écrit qu’après la fin réussie de toutes les classifications.

Exécuter le pipeline

Une fois les trois étapes connectées, le DAG complet récupère les avis, crée une tâche LLM par avis et stocke les résultats combinés. 

Dans la fonction du DAG, le pipeline ressemble à ceci :

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

Après avoir enregistré le fichier, Airflow prendra quelques secondes pour l’analyser et découvrir le DAG complet. Une fois review_sentiment_pipeline visible dans la liste, activez-le avec le bouton et cliquez sur Trigger pour lancer une exécution manuelle. 

Cela nous permet de tester immédiatement le pipeline complet, sans attendre la prochaine exécution quotidienne planifiée. Une fois l’exécution réussie, le fichier JSON final sera disponible localement à l’emplacement ~/airflow/output/classified_reviews.json.

Le fichier JSON est la sortie durable du workflow, réutilisable en dehors du DAG ou transmissible à des systèmes aval ultérieurement. 

En parallèle, l’interface Airflow nous offre la vue opérationnelle du pipeline : quelles tâches ont tourné, lesquelles ont échoué ou réussi, et où inspecter les logs. 

Le pipeline final est affiché ci-dessous :

Vue Airflow des trois tâches de notre pipeline. Le panneau en bas à droite affiche les propriétés de chaque tâche, dont son nom, sa règle de déclenchement, son opérateur ou décorateur, et le nombre de réessais.

Enfin, au fil des jours, l’interface Airflow se remplira d’exécutions quotidiennes automatiques, nous offrant un historique clair des exécutions et de leur succès :

Vue Airflow des exécutions du pipeline et de leurs états. Après l’étape 3, le pipeline n’utilise plus show_classification et exécute store_results à la place.

Aller plus loin : fonctionnalités IA d’Airflow 3 à explorer

Notre pipeline récupère désormais des avis, les classe avec un LLM et stocke les résultats sur une base quotidienne. 

Airflow 3 prend aussi en charge des schémas plus avancés qui rendent les workflows d’IA plus réactifs et plus faciles à superviser.

Planification pilotée par événement

Plutôt que d’exécuter le pipeline à heure fixe chaque jour, Airflow peut le démarrer lorsqu’un événement externe survient. 

Par exemple, le pipeline pourrait s’exécuter à chaque fois qu’un nouveau fichier d’avis est déposé dans un object storage ou lorsqu’un message arrive dans une file.

Airflow 3 prend en charge la planification orientée événement via des assets et des déclencheurs compatibles. Cela évite de sonder en boucle l’arrivée de nouvelles données et permet au pipeline de réagir dès qu’un événement est détecté.

Human-in-the-loop avec HITLOperator

Toute décision d’un modèle ne doit pas être acceptée automatiquement. 

Certains avis peuvent être ambigus, sensibles ou nécessiter un jugement humain. Les opérateurs human-in-the-loop d’Airflow peuvent mettre en pause un workflow en cours et demander une saisie via l’interface. 

Un relecteur peut approuver ou rejeter un résultat, fournir des informations, ou choisir quelle branche du workflow doit continuer.

Le provider Common AI prend aussi en charge des schémas de revue interactive pour des workflows agentiques, où un relecteur peut approuver, rejeter ou demander des modifications à un résultat généré par LLM.

Ensemble, la planification pilotée par événement et la revue humaine permettent de créer des pipelines d’IA réactifs sans supprimer la supervision humaine là où elle est essentielle. 

Pour en savoir plus sur les fonctionnalités d’Airflow 3.0, je vous recommande notre guide dédié. 

Conclusion

Dans cet article, nous avons construit un pipeline d’IA complet avec Apache Airflow. 

Le pipeline récupère des avis clients, classe chaque avis avec un LLM et stocke les résultats finaux dans un fichier JSON structuré.

Au passage, nous avons vu comment Airflow transforme des fonctions Python en tâches planifiées et observables. C’est particulièrement utile pour les workflows d’IA, où les limites d’API, les timeouts et des sorties inattendues peuvent survenir. 

Avec les dépendances de tâches, le mapping dynamique, les réessais, les logs et l’interface Airflow, nous pouvons rendre les pipelines d’IA plus faciles à superviser, déboguer et réparer.

L’exemple utilisait des fichiers locaux pour rester simple, mais la même structure s’étend aux APIs, bases de données, workflows orientés événement et revue human-in-the-loop. 

Si vous êtes prêt à poursuivre, commencez par notre cours Introduction to Apache Airflow in Python, puis enchaînez avec Building Data Pipelines with Airflow pour davantage de pratique.

Pour un parcours plus large, la carrière Data Engineer in Python couvre l’ensemble des compétences nécessaires pour créer et gérer des pipelines de données en production.