Création d’un pipeline de données d’auto-réparation qui corrige ses propres erreurs Python

Je suis un mardi (enfin, techniquement mercredi, je suppose), lorsque mon téléphone a sonné avec cette notification PagerDuty familière et redoutée.

Je n’ai même pas eu besoin d’ouvrir mon ordinateur portable pour savoir que le daily_ingest.py le script avait échoué. Encore.

Cela continue d’échouer car notre fournisseur de données modifie toujours son format de fichier sans avertissement. Je veux dire, ils pourraient passer au hasard des virgules aux tuyaux ou même gâcher les dates du jour au lendemain.

Habituellement, le correctif proprement dit ne me prend qu’une trentaine de secondes : j’ouvre simplement le script, j’échange sep=',' pour sep='|'et appuyez sur Exécuter.

Je sais que cela a été rapide, mais en toute honnêteté, le coût réel n’est pas le temps de codage, mais plutôt le sommeil interrompu et la difficulté de faire fonctionner son cerveau à 2 heures du matin.

Cette routine m’a fait réfléchir : si la solution est si évidente que je peux la trouver simplement en jetant un coup d’œil au texte brut, pourquoi un modèle ne pourrait-il pas le faire ?

Nous entendons souvent parler de « Agentic AI » remplaçant les ingénieurs logiciels, ce qui, honnêtement, me semble quelque peu exagéré.

Mais ensuite, l’idée d’utiliser un petit LLM rentable pour agir en tant que développeur junior de garde s’occupant des tâches ennuyeuses pandas des exceptions ?

Cela semblait être un projet qui valait la peine d’être essayé.

J’ai donc construit un pipeline « d’auto-guérison ». Bien que ce ne soit pas magique, cela m’a protégé avec succès d’au moins trois réveils nocturnes ce mois-ci.

Et personnellement, tout ce qui (aussi petit soit-il) peut améliorer ma santé pendant mon sommeil est définitivement une grande victoire pour moi.

Voici comment je l’ai fait afin que vous puissiez le construire vous-même.

L’architecture : une boucle « Essayer-Réparer-Réessayer »

Le concept de base est relativement simple. La plupart des pipelines de données sont fragiles car ils supposent que le monde est parfait, et lorsque les données d’entrée changent, même légèrement, ils échouent.

Au lieu d’accepter ce crash, j’ai conçu mon script pour intercepter l’exception, capturer les « preuves sur la scène du crime », qui sont essentiellement le traçage et les premières lignes du fichier, puis les transmettre à un LLM.

Plutôt sympa, non ?

Le LLM agit désormais comme un outil de diagnostic, analysant les preuves pour restituer le correct paramètres, que le script utilise ensuite pour réessayer automatiquement l’opération.

Pour rendre ce système robuste, je me suis appuyé sur trois outils spécifiques :

1. Pandas : Pour le chargement réel des données (évidemment).
2. Pydantique : Pour garantir que le LLM renvoie du JSON structuré plutôt qu’un remplissage conversationnel.
3. Ténacité: Une bibliothèque Python qui rend l’écriture d’une logique de nouvelle tentative complexe incroyablement propre.

Étape 1 : Définir le « correctif »

Le principal défi lié à l’utilisation de grands modèles de langage pour la génération de code est leur tendance à halluciner. D’après mon expérience, si vous demandez un paramètre simple, vous recevez souvent en retour un paragraphe de texte conversationnel.

Pour arrêter cela, j’ai exploité des sorties structurées via l’API Pydantic et OpenAI.

Cela oblige le modèle à remplir une forme stricte, agissant comme un filtre entre le raisonnement désordonné de l’IA et notre code Python propre.

Voici le schéma sur lequel j’ai opté, en me concentrant strictement sur les arguments qui provoquent le plus souvent read_csv échouer :

from pydantic import BaseModel, Field
from typing import Optional, Literal

# We need a strict schema so the LLM doesn't just yap at us.
# I'm only including the params that actually cause crashes.
class CsvParams(BaseModel):
    sep: str = Field(description="The delimiter, e.g. ',' or '|' or ';'")
    encoding: str = Field(default="utf-8", description="File encoding")
    header: Optional[int | str] = Field(default="infer", description="Row for col names")
    
    # Sometimes the C engine chokes on regex separators, so we let the AI switch engines
    engine: Literal["python", "c"] = "python"

En définissant cela BaseModelnous disons effectivement au LLM : « Je ne veux pas de conversation ou d’explication. Je veux que ces quatre variables soient remplies, et rien d’autre. »

Étape 2 : La fonction de guérisseur

Cette fonction est le cœur du système, conçue pour fonctionner uniquement lorsque les choses tournent mal.

Obtenir la bonne invite a nécessité quelques essais et erreurs. Et c’est parce qu’au départ, je n’ai fourni que le message d’erreur, ce qui a obligé le modèle à deviner aveuglément le problème.

J’ai rapidement réalisé que pour identifier correctement les problèmes tels que les incompatibilités de délimiteurs, le modèle devait réellement « voir » un échantillon des données brutes.

Voici maintenant le gros problème. Vous ne pouvez pas réellement lire l’intégralité du fichier.

Si vous essayez de transmettre un CSV de 2 Go dans l’invite, vous ferez exploser votre fenêtre contextuelle et apparemment votre portefeuille.

Heureusement, j’ai découvert que le simple fait d’extraire les premières lignes donne au modèle juste assez d’informations pour résoudre le problème dans 99 % des cas.

import openai
import json

client = openai.OpenAI()

def ask_the_doctor(fp, error_trace):
    """
    The 'On-Call Agent'. It looks at the file snippet and error, 
    and suggests new parameters.
    """
    print(f"🔥 Crash detected on {fp}. Calling LLM...")

    # Hack: Just grab the first 4 lines. No need to read 1GB.
    # We use errors='replace' so we don't crash while trying to fix a crash.
    try:
        with open(fp, "r", errors="replace") as f:
            head = "".join([f.readline() for _ in range(4)])
    except Exception:
        head = "<<FILE UNREADABLE>>"

    # Keep the prompt simple. No need for complex "persona" injection.
    prompt = f"""
    I'm trying to read a CSV with pandas and it failed.
    
    Error Trace: {error_trace}
    
    Data Snippet (First 4 lines):
    ---
    {head}
    ---
    
    Return the correct JSON params (sep, encoding, header, engine) to fix this.
    """

    # We force the model to use our Pydantic schema
    completion = client.chat.completions.create(
        model="gpt-4o", # gpt-4o-mini is also fine here and cheaper
        messages=[{"role": "user", "content": prompt}],
        functions=[{
            "name": "propose_fix",
            "description": "Extracts valid pandas parameters",
            "parameters": CsvParams.model_json_schema()
        }],
        function_call={"name": "propose_fix"}
    )

    # Parse the result back to a dict
    args = json.loads(completion.choices[0].message.function_call.arguments)
    print(f"💊 Prescribed fix: {args}")
    return args

Je passe en quelque sorte sous silence la configuration de l’API ici, mais vous voyez l’idée. Il prend les « symptômes » et prescrit une « pilule » (les arguments).

Étape 3 : La boucle de réessai (là où la magie opère)

Nous devons maintenant connecter cet outil de diagnostic à notre chargeur de données actuel.

Dans le passé, j’écrivais moche while True boucles avec imbriquées try/except des blocs qui étaient un cauchemar à lire.

Puis j’ai trouvé tenacityqui vous permet de décorer une fonction avec une logique de nouvelle tentative propre.

Et le meilleur, c’est que tenacity vous permet également de définir un « rappel » personnalisé qui s’exécute entre tentatives.

C’est exactement là que nous injectons notre fonction Healer.

import pandas as pd
from tenacity import retry, stop_after_attempt, retry_if_exception_type

# A dirty global dict to store the "fix" between retries.
# In a real class, this would be self.state, but for a script, this works.
fix_state = {} 

def apply_fix(retry_state):
    # This runs right after the crash, before the next attempt
    e = retry_state.outcome.exception()
    fp = retry_state.args[0]
    
    # Ask the LLM for new params
    suggestion = ask_the_doctor(fp, str(e))
    
    # Update the state so the next run uses the suggestion
    fix_state[fp] = suggestion

@retry(
    stop=stop_after_attempt(3), # Give it 3 strikes
    retry_if_exception_type(Exception), # Catch everything (risky, but fun)
    before_sleep=apply_fix # <--- This is the hook
)
def tough_loader(fp):
    # Check if we have a suggested fix for this file, otherwise default to comma
    params = fix_state.get(fp, {"sep": ","})
    
    print(f"🔄 Trying to load with: {params}")
    df = pd.read_csv(fp, **params)
    return df

Est-ce que ça marche réellement ?

Pour tester cela, j’ai créé un fichier délibérément cassé appelé messy_data.csv. Je l’ai fait délimité par des barres verticales (|) mais je n’ai pas dit le script.

Quand j’ai couru tough_loader('messy_data.csv')le script s’est écrasé, s’est arrêté un instant pendant qu’il « réfléchissait », puis s’est corrigé automatiquement.

Il est étonnamment satisfaisant de voir le code échouer, se diagnostiquer et récupérer sans aucune intervention humaine.

Les « Gotchas » (Parce que rien n’est parfait)

Je ne veux pas exagérer cette solution, car elle comporte certainement des risques.

Le coût

Tout d’abord, n’oubliez pas qu’à chaque fois que votre pipeline tombe en panne, vous effectuez un appel API.

Cela peut convenir pour quelques erreurs, mais si vous avez un traitement de travail massif, disons environ 100 000 fichiers, et qu’un mauvais déploiement provoque tous Si l’un d’entre eux se brise d’un coup, vous pourriez vous réveiller avec une très mauvaise surprise sur votre facture OpenAI.

Si vous exécutez cela à grande échelle, je vous recommande fortement de mettre en œuvre un disjoncteur ou de passer à un modèle local comme Llama-3 via Ollama pour réduire vos coûts.

Sécurité des données

Bien que j’envoie uniquement les quatre premières lignes du fichier au LLM, vous devez faire très attention à ce qui est dans ces lignes. Si vos données contiennent des informations personnelles identifiables (PII), vous envoyez effectivement ces données sensibles à une API externe.

Si vous travaillez dans un secteur réglementé comme la santé ou la finance, veuillez utiliser un modèle local.

Sérieusement.

N’envoyez pas les données du patient à GPT-4 uniquement pour corriger une erreur de virgule.

Le « Garçon qui criait au loup »

Enfin, il arrive parfois que les données échouent.

Si un fichier est vide ou corrompu, vous ne voulez pas que l’IA ait de toute façon des hallucinations sur un moyen de le charger, remplissant potentiellement votre DataFrame de déchets.

Pydantic filtre les mauvaises données, mais ce n’est pas magique. Vous devez faire attention à ne pas cacher de véritables erreurs que vous devez réellement corriger vous-même.

---

Conclusion et points à retenir

Vous pourriez affirmer que l’utilisation d’une IA pour corriger les CSV est exagérée, et techniquement, vous avez peut-être raison.

Mais dans un domaine aussi en évolution rapide que la science des données, les meilleurs ingénieurs ne sont pas ceux qui s’accrochent aux méthodes qu’ils ont apprises il y a cinq ans ; ce sont eux qui expérimentent constamment de nouveaux outils pour résoudre d’anciens problèmes.

Honnêtement, ce projet n’était qu’un rappel pour rester flexible.

Nous ne pouvons pas continuer à garder nos vieux pipelines ; nous devons continuer à trouver des moyens de les améliorer. Dans ce secteur, la compétence la plus précieuse n’est pas d’écrire du code plus rapidement ; il s’agit plutôt d’avoir la curiosité d’essayer une toute nouvelle façon de travailler.