6. Pipelines déclaratifs avec LCEL

Dans les chapitres précédents, nous avons écrit du code impératif pour orchestrer les interactions avec les LLM : créer un prompt, invoquer le modèle, parser la réponse. Cela fonctionne, mais à mesure que les applications IA deviennent plus complexes, cette approche devient verbeuse et plus difficile à maintenir. Vous vous retrouvez avec des appels de fonctions profondément imbriqués, une gestion manuelle des erreurs à chaque étape, et des difficultés à comprendre le flux de données global.

Le LangChain Expression Language (LCEL) résout ce problème en vous permettant de déclarer ce que vous voulez qu'il se passe, et non comment le faire. Au lieu d'écrire du code procédural qui appelle des fonctions en séquence, vous composez des composants en utilisant un simple opérateur pipe (|) qui se lit comme un pipeline Unix. Le résultat est un code plus propre, plus maintenable, qui exprime clairement le flux de données à travers votre système IA.

Ce chapitre introduit LCEL pour construire des workflows linéaires - des séquences d'opérations où les données circulent du début à la fin sans branchement ni boucle. Nous verrons quand utiliser LCEL, comment composer des pipelines, et comment les exécuter de manière synchrone et avec sortie en streaming.

6.1) Pourquoi LCEL ?

Le problème que LCEL résout

Commençons par un exemple concret. Supposons que vous construisiez un assistant de support client qui doit :

1. Prendre une question utilisateur
2. La formater dans un prompt avec des instructions système et le message utilisateur
3. L'envoyer à un LLM
4. Parser la réponse pour extraire uniquement le contenu texte

Voici comment vous pourriez écrire cela de manière impérative (sans LCEL) :

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # Étape 1 : Formater le prompt avec la question
    messages = prompt.invoke({"question": question})
    
    # Étape 2 : Invoquer le LLM
    response = llm.invoke(messages)
    
    # Étape 3 : Parser la sortie pour extraire le contenu texte
    result = parser.invoke(response)
    
    return result
 
# Utilisation
result = answer_question("How do I reset my password?")
print(result)

Sortie :

Pour réinitialiser votre mot de passe, veuillez suivre ces étapes :
1. Allez sur la page de connexion
2. Cliquez sur "Mot de passe oublié"
3. Entrez votre adresse e-mail
4. Consultez votre e-mail pour un lien de réinitialisation
5. Suivez le lien et créez un nouveau mot de passe

Cela fonctionne, mais remarquez les problèmes :

Verbosité : Chaque étape nécessite une affectation de variable explicite et des appels de fonction. La logique réelle (formater → invoquer → parser) est enfouie dans du code répétitif.

Structure rigide : Si vous voulez ajouter une étape (comme valider la question ou logger la réponse), vous devez insérer du code au milieu de la fonction, augmentant la complexité.

Pas de streaming intégré : Pour streamer les tokens au fur et à mesure qu'ils arrivent, vous devriez réécrire toute la fonction pour utiliser llm.stream() et gérer l'itération asynchrone manuellement.

Flux de données peu clair : En lisant le code, il n'est pas immédiatement évident qu'il s'agit d'un simple pipeline. Vous devez suivre les affectations de variables pour comprendre le flux.

Voyons maintenant la version LCEL :

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# Définir les composants (identiques à avant)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Les composer en une chaîne en utilisant l'opérateur pipe
chain = prompt | llm | parser
 
# Utilisation
result = chain.invoke({"question": "How do I reset my password?"})
print(result)

Sortie :

Pour réinitialiser votre mot de passe, veuillez suivre ces étapes :
1. Allez sur la page de connexion
2. Cliquez sur "Mot de passe oublié"
3. Entrez votre adresse e-mail
4. Consultez votre e-mail pour un lien de réinitialisation
5. Suivez le lien et créez un nouveau mot de passe

La sortie est identique, mais le code est radicalement différent :

Déclaratif : chain = prompt | llm | parser exprime tout le flux en une ligne. Lisez-le de gauche à droite : prompt → LLM → parser.

Composable : Chaque composant (prompt, llm, parser) est indépendant et réutilisable. Vous pouvez échanger des composants sans réécrire le pipeline.

Streaming intégré : Basculez entre chain.invoke() et chain.stream() sans changer la définition du pipeline.

Intention claire : L'opérateur | rend le flux de données évident d'un coup d'œil.

Où LCEL s'intègre : Workflows linéaires

LCEL est conçu pour les workflows linéaires - des séquences où les données circulent dans une direction du début à la fin sans branchement ni boucle.

Ce pattern linéaire couvre de nombreuses applications IA. Considérez un système de Q&A sur documents : vous recevez une question → récupérez les documents pertinents → formatez un prompt → envoyez au LLM → parsez la réponse. Chaque étape est une séquence claire où la sortie d'une étape devient l'entrée de la suivante.

Mais que se passe-t-il si votre workflow doit :

• Faire en sorte que le LLM décide quel outil appeler en fonction de la question
• Appeler un outil, observer le résultat, puis décider quoi faire ensuite
• Réessayer les opérations échouées avec différentes approches

Ces scénarios nécessitent des boucles et des branchements conditionnels - des choses que LCEL ne peut pas gérer. C'est pourquoi LangGraph existe (nous l'introduirons au Chapitre 15).

Voici une comparaison visuelle des deux approches :

LCEL est parfait pour les pipelines simples où chaque étape traite la sortie de l'étape précédente. Les données circulent dans une seule direction.

LangGraph est pour quand vous avez besoin de prise de décision et de boucles. L'agent peut agir, observer les résultats, et réfléchir à nouveau.

Pour ce chapitre, nous nous concentrons sur LCEL. Pourquoi maîtriser d'abord les pipelines linéaires ?

• Fondation : L'opérateur pipe (|) est la syntaxe centrale de LangChain. Comprendre cela facilite tout le reste.
• Prérequis pour LangGraph : Les agents LangGraph utilisent extensivement les chaînes LCEL à l'intérieur de leurs nœuds
• Pattern du monde réel : Les agents complexes sont construits en combinant des chaînes LCEL

Le reste de ce chapitre vous montrera comment construire ces pipelines linéaires avec l'opérateur pipe (|).

6.2) Composer des pipelines avec l'opérateur |

Comment fonctionnent les pipes LCEL

L'opérateur pipe (|) de LCEL fonctionne grâce à l'interface Runnable.

Qu'est-ce qu'un Runnable ?

Runnable est l'interface standard de LangChain. Quand un composant implémente l'interface Runnable, il peut être chaîné avec d'autres composants en utilisant l'opérateur |.

Chaque Runnable fournit ces méthodes :

• .invoke(input) - Exécuter une fois et obtenir le résultat complet
• .stream(input) - Exécuter et recevoir chaque mot au fur et à mesure que le LLM le génère
• .batch(inputs) - Exécuter plusieurs fois avec différentes entrées et obtenir tous les résultats

Pour ce chapitre, nous nous concentrerons sur .invoke() et .stream() (nous couvrirons .batch() plus tard si nécessaire).

Pourquoi | fonctionne-t-il ?

Parce que la classe Runnable implémente l'opérateur | en utilisant la surcharge d'opérateur de Python. Quand vous écrivez prompt | llm, cela crée un nouveau Runnable qui exécute les deux composants séquentiellement.

La plupart des composants LangChain sont des Runnables

C'est pourquoi vous pouvez chaîner tant de composants différents :

• ChatPromptTemplate est un Runnable
• ChatOpenAI est un Runnable
• StrOutputParser est un Runnable
• Même les chaînes personnalisées que vous créez avec | sont elles-mêmes des Runnables !

Cela signifie que vous pouvez construire des pipelines complexes en combinant des pipelines plus simples.

Connecter les composants : Types d'entrée/sortie

Lors de la connexion de composants avec |, vous devez vous assurer que le type de sortie d'un composant correspond au type d'entrée du suivant.

Signatures des composants clés :

Exemple de flux :

chain = prompt | llm | parser

Voici comment le type de données se transforme en passant par chaque composant :

• prompt : Prend dict en entrée et convertit en list[BaseMessage]
• llm : Prend list[BaseMessage] en entrée et convertit en AIMessage
• parser : Prend AIMessage en entrée et convertit en str

Voyons cela en action :

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "What is 2+2?"})
print(result)  # "2+2 equals 4."

Ce qui se passe à chaque étape :

Que se passe-t-il si les types ne correspondent pas ?

Si vous essayez de connecter des composants incompatibles, vous obtiendrez une erreur :

# ERREUR : Cela ne fonctionnera pas
chain = llm | prompt  # Le LLM produit AIMessage, mais prompt nécessite dict en entrée

Le message d'erreur vous indiquera quel type d'entrée le composant suivant attend par rapport au type d'entrée qu'il a réellement reçu.

6.3) Exécuter une chaîne : .invoke() et .stream()

Exécuter un pipeline

Une fois que vous avez composé une chaîne, vous l'exécutez en utilisant la méthode .invoke(). C'est la manière synchrone d'exécuter un pipeline - elle attend la réponse complète avant de retourner le résultat.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Invocation synchrone
result = chain.invoke({"question": "What is 2+2?"})
print(result)

Sortie :

2+2 égale 4.

La méthode .invoke() est simple : passez le paramètre d'entrée que le premier composant attend, et obtenez la sortie produite par le dernier composant.

Streamer les tokens depuis la même chaîne

Bien que .invoke() soit simple, elle a une limitation pour les applications orientées utilisateur : les utilisateurs ne voient rien jusqu'à ce que la réponse complète soit terminée. Pour les réponses longues (10-20 secondes), cela crée une mauvaise expérience utilisateur.

Le streaming affiche les tokens immédiatement au fur et à mesure qu'ils sont générés, plutôt que d'attendre la réponse complète. C'est l'effet de frappe que vous voyez dans ChatGPT.

Voyons le streaming en action :

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Streamer les chunks au fur et à mesure qu'ils arrivent
for chunk in chain.stream({"question": "Explain what Python is in one sentence"}):
    print(chunk, end="", flush=True)

Sortie (affichée en temps réel, token par token) :

Python est un langage de programmation polyvalent de haut niveau, connu pour sa simplicité et sa lisibilité, largement utilisé dans le développement web, la science des données et l'automatisation.

Remarquez que la définition de la chaîne est identique à l'exemple .invoke() ci-dessus. Nous n'avons pas eu besoin de la reconstruire - nous avons simplement appelé .stream() au lieu de .invoke().

La méthode .stream() retourne les résultats sous forme de chunks pendant que le LLM génère les tokens. Le print(chunk, end="", flush=True) affiche chaque chunk immédiatement à l'écran, créant l'effet de frappe en direct.

Quand utiliser .invoke() vs .stream()

Utilisez .invoke() quand :

• Vous avez seulement besoin du résultat final (analyse, traduction, classification)
• La réponse est courte et le temps d'attente n'est pas un problème
• Vous avez besoin de la sortie complète avant de passer à l'étape suivante

Utilisez .stream() quand :

• Les utilisateurs ont besoin de voir la progression (chat, génération de contenu)
• La réponse est longue et le temps d'attente serait perceptible
• Vous construisez une UI où le feedback en temps réel est important

Les deux méthodes fonctionnent sur la même chaîne. Définissez la chaîne une fois, puis choisissez le mode d'exécution en fonction de vos besoins.

Dans ce chapitre, vous avez appris LCEL - la syntaxe de pipeline déclarative de LangChain :

• Construire des pipelines avec l'opérateur pipe : prompt | llm | parser
• Exécuter avec flexibilité : Utilisez .invoke() pour des résultats complets ou .stream() pour une sortie en temps réel
• Sécurité des types : Faites correspondre les types de sortie aux types d'entrée lors du chaînage des composants

La même chaîne fonctionne pour les deux modes d'exécution - définissez une fois, utilisez partout.

Suivant : Le Chapitre 7 couvre la sortie structurée avec Pydantic, vous permettant d'extraire des données JSON validées à partir des réponses LLM.