Pipeline sequentiel

Dans un systeme LLM reel, la question de l'utilisateur ne va jamais directement au modele. Elle passe d'abord par une serie d'etapes : nettoyage du texte, enrichissement avec du contexte, formatage du prompt, appel au LLM, puis post-traitement de la reponse. C'est un pipeline sequentiel, ou chaque etape prend le resultat de la precedente.

Pense a une chaine de montage en usine : la piece brute arrive, passe a la decoupe, puis au polissage, puis a la peinture, puis a l'emballage. Chaque poste fait une chose et passe le resultat au suivant.

Voici le principe :

def nettoyer(texte):
return texte.strip().lower()

def enrichir(texte):
return f"Contexte: Tu es un expert. Question: {texte}"

pipeline = [nettoyer, enrichir]
# " BONJOUR " -> "bonjour" -> "Contexte: Tu es un expert. Question: bonjour"

Tu vas ecrire une fonction executer_pipeline(donnee, etapes) qui prend une donnee initiale et une liste de fonctions (les etapes du pipeline). Chaque fonction prend un argument et renvoie un resultat. La fonction les execute dans l'ordre, en passant le resultat de chaque etape comme argument de la suivante.

La fonction renvoie un dictionnaire avec :
- "resultat" : le resultat final
- "etapes" : une liste de dictionnaires, chacun avec "nom" (le nom de la fonction, via __name__) et "sortie" (le resultat de cette etape)

Si une etape leve une exception, la fonction s'arrete et renvoie :
- "erreur" : le message de l'exception
- "etape_echouee" : le nom de la fonction qui a echoue
- "etapes" : les etapes qui ont reussi avant l'echec

Exemple :

def upper(t): return t.upper()
def exclaim(t): return t + "!"

executer_pipeline("bonjour", [upper, exclaim])
renvoie {
"resultat": "BONJOUR!",
"etapes": [
{"nom": "upper", "sortie": "BONJOUR"},
{"nom": "exclaim", "sortie": "BONJOUR!"}
]
}

Tests (4/5)

Pipeline simple

def double(n): return n * 2
def plus_un(n): return n + 1
r = executer_pipeline(3, [double, plus_un])
assert r['resultat'] == 7
assert len(r['etapes']) == 2

Noms des etapes enregistres

def strip_it(t): return t.strip()
def upper_it(t): return t.upper()
r = executer_pipeline('  hello  ', [strip_it, upper_it])
assert r['etapes'][0]['nom'] == 'strip_it'
assert r['etapes'][1]['sortie'] == 'HELLO'

Pipeline vide

r = executer_pipeline('test', [])
assert r['resultat'] == 'test'
assert r['etapes'] == []

Gestion d'erreur en cours de pipeline

def ok(x): return x + 1
def boom(x): raise ValueError('Erreur etape')
def jamais(x): return x
r = executer_pipeline(0, [ok, boom, jamais])
assert 'erreur' in r
assert r['etape_echouee'] == 'boom'
assert len(r['etapes']) == 1

+ 0 tests cachés

Indices (3 disponibles)

Solution officielle

def executer_pipeline(donnee, etapes):
    resultat = donnee
    historique = []
    for etape in etapes:
        try:
            resultat = etape(resultat)
            historique.append({"nom": etape.__name__, "sortie": resultat})
        except Exception as e:
            return {
                "erreur": str(e),
                "etape_echouee": etape.__name__,
                "etapes": historique,
            }
    return {"resultat": resultat, "etapes": historique}

← Tool use : executer un … Pipeline avec branchement conditionnel →

solution.py

Bravo!

Pipeline sequentiel

Tests (4/5)

Indices (3 disponibles)

Solution officielle