Construire un graphe avec LangGraph

A l'exercice precedent, tu as construit une boucle agent "a la main". Ca marche, mais des que le flux devient complexe (plusieurs chemins possibles, des conditions, des retours en arriere), le code devient un plat de spaghettis. LangGraph resout ce probleme en modelisant le flux de ton agent comme un graphe.

Un graphe, c'est un ensemble de noeuds connectes par des aretes. Chaque noeud est une etape (raisonner, appeler un outil, formuler la reponse), et les aretes definissent l'enchainement. Certaines aretes sont conditionnelles : "si le LLM veut un outil, va au noeud appeler_outil, sinon va au noeud formuler_reponse".

Pour travailler sans installer LangGraph (on le fera dans un exercice futur), on va construire notre propre mini-graphe en Python pur. Ca te fera comprendre exactement ce qui se passe sous le capot de LangGraph.

Ecris une classe GrapheAgent qui modelise un flux de travail :

ajouter_noeud(nom, fonction) : ajoute un noeud au graphe. La fonction prend un etat (dictionnaire) et renvoie l'etat modifie.

ajouter_arete(source, destination) : ajoute une arete simple (toujours suivi ce chemin).

ajouter_arete_conditionnelle(source, fonction_condition) : la fonction_condition prend l'etat et renvoie le nom du noeud suivant.

definir_entree(nom) : definit le noeud de depart.

definir_sortie(nom) : definit le noeud de fin (le graphe s'arrete quand on atteint ce noeud).

executer(etat_initial) : execute le graphe a partir de l'entree, en suivant les aretes, jusqu'a atteindre la sortie ou un maximum de 20 etapes. Renvoie l'etat final.

Exemple :

g = GrapheAgent()
g.ajouter_noeud("debut", lambda etat: {**etat, "etape": "debut fait"})
g.ajouter_noeud("fin", lambda etat: {**etat, "terminee": True})
g.ajouter_arete("debut", "fin")
g.definir_entree("debut")
g.definir_sortie("fin")
resultat = g.executer({"question": "test"})
resultat["etape"] vaut "debut fait"
resultat["terminee"] vaut True

Tests (4/5)

Graphe lineaire simple

g = GrapheAgent()
g.ajouter_noeud("a", lambda e: {**e, "a": True})
g.ajouter_noeud("b", lambda e: {**e, "b": True})
g.ajouter_arete("a", "b")
g.definir_entree("a")
g.definir_sortie("b")
r = g.executer({})
assert r.get("a") == True and r.get("b") == True, "Les deux noeuds doivent avoir modifie l'etat"

Arete conditionnelle

g = GrapheAgent()
g.ajouter_noeud("check", lambda e: e)
g.ajouter_noeud("oui", lambda e: {**e, "resultat": "oui"})
g.ajouter_noeud("non", lambda e: {**e, "resultat": "non"})
g.ajouter_arete_conditionnelle("check", lambda e: "oui" if e.get("valeur") else "non")
g.definir_entree("check")
g.definir_sortie("oui")

r1 = g.executer({"valeur": True})
assert r1["resultat"] == "oui", f"Avec valeur=True, attendu 'oui', obtenu {r1.get('resultat')}"

Graphe avec boucle (3 iterations)

g = GrapheAgent()
g.ajouter_noeud("incrementer", lambda e: {**e, "compteur": e.get("compteur", 0) + 1})
g.ajouter_noeud("fin", lambda e: {**e, "fini": True})
g.ajouter_arete_conditionnelle("incrementer",
    lambda e: "fin" if e["compteur"] >= 3 else "incrementer")
g.definir_entree("incrementer")
g.definir_sortie("fin")
r = g.executer({"compteur": 0})
assert r["compteur"] == 3, f"Le compteur doit valoir 3 apres 3 passages, pas {r['compteur']}"
assert r.get("fini") == True

Protection contre les boucles infinies

g = GrapheAgent()
g.ajouter_noeud("boucle", lambda e: {**e, "tours": e.get("tours", 0) + 1})
g.ajouter_arete("boucle", "boucle")
g.definir_entree("boucle")
g.definir_sortie("jamais_atteint")
r = g.executer({})
assert r["tours"] <= 20, f"Le graphe doit s'arreter apres max 20 etapes, pas {r['tours']}"

+ 0 tests cachés

Indices (3 disponibles)

Solution officielle

class GrapheAgent:
    def __init__(self):
        self.noeuds = {}
        self.aretes = {}
        self.aretes_conditionnelles = {}
        self.entree = None
        self.sortie = None

    def ajouter_noeud(self, nom, fonction):
        self.noeuds[nom] = fonction

    def ajouter_arete(self, source, destination):
        self.aretes[source] = destination

    def ajouter_arete_conditionnelle(self, source, fonction_condition):
        self.aretes_conditionnelles[source] = fonction_condition

    def definir_entree(self, nom):
        self.entree = nom

    def definir_sortie(self, nom):
        self.sortie = nom

    def executer(self, etat_initial):
        etat = dict(etat_initial)
        noeud_actuel = self.entree
        max_etapes = 20

        for _ in range(max_etapes):
            if noeud_actuel not in self.noeuds:
                break
            fonction = self.noeuds[noeud_actuel]
            etat = fonction(etat)

            if noeud_actuel == self.sortie:
                break

            if noeud_actuel in self.aretes_conditionnelles:
                noeud_actuel = self.aretes_conditionnelles[noeud_actuel](etat)
            elif noeud_actuel in self.aretes:
                noeud_actuel = self.aretes[noeud_actuel]
            else:
                break

        return etat

← Memoire de l'agent Routing intelligent des questions →

solution.py

Bravo!

Construire un graphe avec LangGraph

Tests (4/5)

Indices (3 disponibles)

Solution officielle