Aller au contenu principal

Module 3 — Composition de chaînes en LCEL

Un ticket extrait, une politique à consulter, une décision à formuler : notre assistant de notes de frais enchaîne rarement une seule opération. Le LangChain Expression Language (LCEL) permet de composer ces étapes avec l'opérateur |, de les paralléliser quand c'est possible, et d'obtenir gratuitement le flux, les lots et l'asynchrone. Ce module est purement mécanique — chaque bloc de code ci-dessous se colle dans un fichier et tourne tel quel.

Le contrat Runnable​

Un Runnable est un objet qui expose .invoke, .stream, .batch (plus leurs versions asynchrones). Un modèle chat, un gabarit de consigne, un analyseur, un récupérateur, une fonction Python passée à RunnableLambda : tout est un Runnable. L'opérateur | en enchaîne deux : la sortie du gauche devient l'entrée du droit.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

consigne = ChatPromptTemplate.from_template("Résumez en une phrase : {texte}")
chaine = consigne | ChatOpenAI(temperature=0) | StrOutputParser()

print(chaine.invoke({"texte": "Long ticket de restaurant..."}))

Trois nœuds composés. La chaîne accepte n'importe quel dictionnaire dont les clés correspondent aux variables du gabarit.

RunnablePassthrough : ajouter sans détruire​

Une chaîne perd rapidement l'entrée initiale : après le premier |, seule la sortie du gabarit circule. RunnablePassthrough sert à transporter l'entrée originale à travers le pipeline, ou à l'enrichir sans la remplacer :

from langchain_core.runnables import RunnablePassthrough

chaine = (
    RunnablePassthrough.assign(politique=lambda x: recuperer_politique(x["question"]))
    | consigne_rembo
    | modele
    | StrOutputParser()
)

chaine.invoke({"question": "Le taxi vers l'aéroport est-il remboursable ?"})

RunnablePassthrough.assign(k=fn) ajoute une clé au dictionnaire courant en appelant fn sur l'entrée entière, sans toucher aux clés déjà présentes. C'est le patron de conception du RAG (module 5) : injecter le contexte récupéré avant la consigne finale.

RunnableParallel : lancer plusieurs branches en même temps​

Certaines étapes sont indépendantes et peuvent tourner en parallèle. Décider si un ticket est conforme demande à la fois la politique de remboursement et l'historique de l'agent : deux récupérations différentes, deux appels réseau, la même latence quand on les lance ensemble.

from langchain_core.runnables import RunnableParallel

preparation = RunnableParallel(
    politique=recuperateur_politique,
    historique=recuperateur_historique,
    question=RunnablePassthrough(),
)

chaine = preparation | consigne_arbitrage | modele | StrOutputParser()

Sous le capot, RunnableParallel lance ses branches avec asyncio.gather. Deux appels à 800 ms n'en prennent plus que 800 ms au lieu de 1 600. La sortie est un dictionnaire dont les clés sont les noms des branches, prêt à alimenter un ChatPromptTemplate avec les mêmes variables.

Branchement conditionnel​

Toutes les questions ne méritent pas le même traitement. Une salutation ne demande pas de récupération ; une question sur la politique en demande une ; une demande d'ajout de ligne dans le tableur déclenche un outil. RunnableBranch route selon un prédicat :

from langchain_core.runnables import RunnableBranch

routeur = RunnableBranch(
    (lambda x: "bonjour" in x["question"].lower(), reponse_salutation),
    (lambda x: "ajouter" in x["question"].lower(), chaine_outil),
    chaine_rag_defaut,
)

routeur.invoke({"question": "Bonjour, j'ajoute mon repas d'hier"})

Le dernier argument est la branche par défaut. Un routage plus fin passe généralement par un premier appel LLM de classification qui produit un route: "rag" | "outil" | "salutation", puis un branchement dessus.

Flux, lots et asynchrone​

Le vrai retour sur investissement de LCEL apparaît ici. Une chaîne construite avec | expose sans effort supplémentaire :

  • .stream(...) rend la sortie fragment par fragment. Pour une interface utilisateur, cela transforme les 3 secondes d'attente en un texte qui s'écrit en direct.
  • .batch([...], config={"max_concurrency": 5}) traite plusieurs entrées avec un pool contrôlé. Ré-annoter 500 tickets prend quelques minutes au lieu d'une heure séquentielle.
  • .ainvoke, .astream, .abatch exposent les mêmes verbes en asynchrone pour un serveur FastAPI (cours 40) ou un consommateur de file.
async for fragment in chaine.astream({"question": "Quel est le plafond nuit d'hôtel ?"}):
    print(fragment, end="", flush=True)

Aucune ligne de gestion de tâches asynchrones à écrire ; le pipeline propage le flux jusqu'à l'appel modèle qui, chez OpenAI comme chez Anthropic, sait streamer nativement.

Le piège du parallèle inutile​

RunnableParallel n'accélère que les branches indépendantes. Deux appels au même modèle avec la même entrée ne gagnent rien, sinon un doublement du coût. Avant de paralléliser, se poser la question : est-ce que les branches consomment des ressources différentes ? Un appel modèle et une requête base de données, oui. Deux appels au même modèle, non — c'est un .batch déguisé.

Composer plutôt que sous-classer

Une chaîne devient longue quand on tente d'y encapsuler des embranchements dans une classe maison. LCEL préfère la composition explicite : cinq lignes verticales qui se lisent, chacune un Runnable. Une chaîne qui dépasse dix lignes est presque toujours un signe qu'il faut la découper en sous-chaînes nommées.

En résumé​

  • Tout est Runnable : .invoke, .stream, .batch (et leurs versions asynchrones) sont garantis, l'opérateur | compose deux Runnable en pipeline.
  • RunnablePassthrough.assign(clé=fn) enrichit le dictionnaire courant sans écraser l'entrée initiale — c'est le patron du RAG.
  • RunnableParallel lance des branches indépendantes en concurrence via asyncio.gather ; réserver au vrai parallèle, pas à deux appels du même modèle.
  • Le flux (stream), les lots (batch) et l'asynchrone (a*) sont exposés gratuitement par la composition LCEL.

Module suivant : brancher des documents réels — chargeurs PDF, web et bureautique, puis découpage avec métadonnées.