Exercices › AI Engineering › Pipeline avec branchement conditionnel
🎉

Bravo!

Intermédiaire 🧠 Fondamentaux 20 XP 0 personnes ont réussi

Pipeline avec branchement conditionnel

Un pipeline sequentiel execute toujours les memes etapes dans le meme ordre. Mais en production, tu as souvent besoin de logique conditionnelle. Par exemple : si la question est en francais, utilise le modele francais ; si elle est en anglais, utilise le modele anglais. Ou encore : si la question est simple, passe par un petit modele rapide ; si elle est complexe, utilise un gros modele.

C'est ce qu'on appelle le branchement conditionnel, et c'est le principe central de LangGraph. Tu as un noeud de decision qui evalue une condition et aiguille vers la bonne branche.

Tu vas ecrire une fonction pipeline_branche(donnee, classifier, branches, fallback) qui prend :
- donnee : les donnees d'entree
- classifier : une fonction qui prend la donnee et renvoie une cle de branche (string)
- branches : un dictionnaire ou chaque cle est un nom de branche et chaque valeur est une liste de fonctions (un pipeline)
- fallback : une liste de fonctions a executer si la cle renvoyee par le classifier n'existe pas dans branches

La fonction doit :
1. Appeler le classifier pour determiner la branche
2. Executer le pipeline de la branche choisie (ou le fallback)
3. Renvoyer {"branche": nom_de_la_branche, "resultat": resultat_final}

Exemple :

def classifier(texte):
if "?" in texte:
return "question"
return "affirmation"

branches = {
"question": [lambda t: "Reponse a: " + t],
"affirmation": [lambda t: "Merci pour: " + t],
}

pipeline_branche("Comment ca va ?", classifier, branches, [])
renvoie {"branche": "question", "resultat": "Reponse a: Comment ca va ?"}

Tests (4/5)

Branche correcte choisie
def clf(x): return 'a' if x > 0 else 'b'
branches = {'a': [lambda x: x * 10], 'b': [lambda x: x * -1]}
r = pipeline_branche(5, clf, branches, [])
assert r == {'branche': 'a', 'resultat': 50}
Branche negative
def clf(x): return 'a' if x > 0 else 'b'
branches = {'a': [lambda x: x * 10], 'b': [lambda x: x * -1]}
r = pipeline_branche(-3, clf, branches, [])
assert r == {'branche': 'b', 'resultat': 3}
Fallback quand branche inconnue
def clf(x): return 'inconnu'
branches = {'a': [lambda x: x + 1]}
fallback = [lambda x: x * 0]
r = pipeline_branche(42, clf, branches, fallback)
assert r['branche'] == 'fallback'
assert r['resultat'] == 0
Pipeline multi-etapes dans la branche
def clf(t): return 'upper'
branches = {'upper': [lambda t: t.upper(), lambda t: t + '!']}
r = pipeline_branche('hello', clf, branches, [])
assert r['resultat'] == 'HELLO!'

+ 0 tests cachés

Indices (3 disponibles)

← Pipeline sequentiel Retry avec modele de fallback →
solution.py