LangChain LCEL en pratique : de la chaîne traditionnelle à la réponse en streaming — un paradigme moderne

L’année dernière, j’ai repris un vieux projet. En ouvrant le code, j’ai été sidéré : une seule chaîne de dialogue faisait plus de 200 lignes. Initialiser PromptTemplate, configurer LLMChain, mapper manuellement entrées et sorties, écrire des callbacks pour le streaming. Pire : personne dans l’équipe n’osait y toucher — « tant que ça tourne » était le consensus.

C’est l’héritage des premières versions de LangChain. LLMChain, SequentialChain et consorts étaient la norme en 2023 ; aujourd’hui, l’équipe officielle les a marqués obsolètes. Problème : la plupart des tutos en ligne utilisent encore l’ancienne syntaxe.

Cet article est le 13ᵉ de la série Développement IA en pratique. Avec des comparaisons de code réelles, je montre pourquoi LCEL (LangChain Expression Language) réduit le code de 70 % pour la même fonctionnalité, et comment il gère automatiquement streaming, async et batch — autrefois des montagnes de boilerplate.

Si vous construisez un système RAG ou une application Agent, cet article se lit avec l’optimisation RAG en pratique et la gestion d’état avec LangGraph — des piliers incontournables de l’écosystème LangChain.

Chapitre 1 : Qu’est-ce que LCEL ? Pourquoi l’utiliser ?

Si vous avez appris LangChain sur des tutos de 2023, vous avez probablement écrit ceci :

# Approche LLMChain traditionnelle (obsolète)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

# 1. Initialiser le modèle
llm = OpenAI(temperature=0.7)

# 2. Définir le template de prompt
template = """Vous êtes un(e) {role} expérimenté(e).
Question utilisateur : {question}
Réponse professionnelle :"""
prompt = PromptTemplate(
    template=template,
    input_variables=["role", "question"]
)

# 3. Créer la chaîne
chain = LLMChain(llm=llm, prompt=prompt)

# 4. Appeler la chaîne (notez le passage des paramètres)
result = chain.run(role="ingénieur frontend", question="React ou Vue ?")
print(result)

Ça semble correct ? Ajoutez streaming, batch ou plusieurs chaînes en série, et le code explose. Trois problèmes majeurs des chaînes traditionnelles :

Premier : streaming médiocre. LLMChain ne stream pas par défaut. Il faut des callbacks pour écouter chaque token — code lourd, async fragile.

Deuxième : composition lourde. Deux chaînes en série ? SequentialChain. En parallèle ? Une autre API. Chaque mode de composition = une interface à apprendre.

Troisième : mapping entrée/sortie explicite et fastidieux. Chaque chaîne déclare input_variables et output_variables ; il faut aligner manuellement les noms entre étapes.

LCEL vise exactement ces problèmes. Même fonctionnalité en LCEL :

# Approche LCEL (recommandée LangChain v0.3+)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# 1. Définir le modèle
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 2. Définir le prompt
prompt = ChatPromptTemplate.from_template(
    "Vous êtes un(e) {role} expérimenté(e).\nQuestion utilisateur : {question}\nRéponse professionnelle :"
)

# 3. Connecter avec le pipe
chain = prompt | model

# 4. Appeler (mapping automatique)
result = chain.invoke({"role": "ingénieur frontend", "question": "React ou Vue ?"})
print(result.content)

De 15 à 9 lignes — mais surtout :

• Streaming automatique : invoke → stream, sans autre changement
• Async automatique : ainvoke ou astream en une ligne
• Batch automatique : batch avec une liste d’entrées

L’opérateur pipe | s’inspire des pipes Linux. cat log.txt | grep error | wc -l enchaîne trois commandes ; LCEL fait pareil : prompt | model | output_parser, les données coulent de gauche à droite.

Au début, j’ai trouvé la syntaxe déroutante — en Python, | n’est-il pas un OR bit à bit ? C’est en fait du sucre syntaxique Python 3.10 via __or__. Élégant.

Chapitre 2 : Fonctionnement de l’opérateur pipe

Le pipe paraît simple ; le design est complet. Expérience :

Alibaba Cloud - Choix n°1 des développeurs, 15 % de réduction exclusive sur Lighthouse

Obtenir -15 % →Pub

from langchain_core.runnables import RunnableLambda

def add_one(x: int) -> int:
    return x + 1

def multiply_two(x: int) -> int:
    return x * 2

add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)

chain = add_one_runnable | multiply_two_runnable

result = chain.invoke(3)  # 3 -> 4 -> 8
print(result)  # Sortie : 8

chain = add_one_runnable | multiply_two_runnable appelle add_one_runnable.__or__(multiply_two_runnable).

Runnable implémente __or__ et renvoie un RunnableSequence qui exécute chaque composant en séquence, en passant la sortie au suivant.

Runnable est l’abstraction centrale de LCEL. Quatre méthodes unifient tous les composants :

Versions async : astream, abatch, abatch_as_completed.

PromptTemplate, ChatModel, OutputParser ou RunnableLambda — même interface :

chain.invoke({"input": "hello"})
chain.stream({"input": "hello"})
chain.batch([{"input": "a"}, {"input": "b"}])

await chain.ainvoke({"input": "hello"})
async for chunk in chain.astream({"input": "hello"}):
    print(chunk, end="", flush=True)

Flux de données dans un pipe un peu plus complexe :

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

prompt = ChatPromptTemplate.from_template("Traduisez en anglais : {text}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser

result = chain.invoke({"text": "你好世界"})
print(result)  # Sortie : Hello World

Flux :

{"text": "你好世界"}
       ↓
   [prompt] → ChatPromptValue(messages=[HumanMessage("Traduisez en anglais : 你好世界")])
       ↓
   [model]  → AIMessage(content="Hello World")
       ↓
   [parser] → "Hello World" (str)

Chaque composant a des types d’entrée/sortie convenus. Prompt : dict → ChatPromptValue ; Model : PromptValue → AIMessage ; Parser : Message → str.

Si l’ordre est faux (model | prompt), erreur à l’exécution — l’IDE peut aussi alerter via les hints de type.

Chapitre 3 : Streaming en pratique

Le streaming est ce qui m’a le plus impressionné avec LCEL.

Imaginez un bot SAV. Question complexe : « Analysez les avantages et inconvénients de ce produit et comparez aux concurrents. » GPT-4o-mini met 5 à 8 secondes.

Sans streaming, écran blanc 8 secondes. L’utilisateur se demande : plantage ? réseau ? rafraîchir ?

Avec streaming, le premier caractère apparaît tout de suite, mot après mot — comme une réponse en direct. L’attente psychologique disparaît.

Avant, le streaming LangChain exigeait beaucoup de code :

# Streaming traditionnel (ère LLMChain)
from langchain.chains import LLMChain
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = OpenAI(
    temperature=0.7,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)
chain = LLMChain(llm=llm, prompt=prompt)
chain.run(role="SAV", question="...")

Problèmes :

1. Callbacks pénibles — pour envoyer les tokens au frontend, hériter de BaseCallbackHandler.
2. Bascule stream/non-stream — streaming=True/False à l’init, pas à l’exécution.
3. Async stream — AsyncCallbackHandler, code doublé.

LCEL intègre le streaming :

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "Vous êtes un(e) {role} professionnel(le). Répondez : {question}"
)
chain = prompt | model

result = chain.invoke({"role": "SAV", "question": "Analysez les avantages et inconvénients de ce produit"})
print(result.content)

for chunk in chain.stream({"role": "SAV", "question": "Analysez les avantages et inconvénients de ce produit"}):
    print(chunk.content, end="", flush=True)

invoke → stream, c’est tout.

Exemple complet de chat temps réel :

import asyncio
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
    ("system", "Vous êtes un assistant IA amical, expert en questions techniques."),
    ("human", "{input}")
])
parser = StrOutputParser()

chain = prompt | model | parser
memory = ChatMessageHistory()

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=lambda session_id: memory,
    input_messages_key="input",
    history_messages_key="chat_history"
)

async def chat_stream(user_input: str):
    print("IA : ", end="", flush=True)
    async for chunk in chain_with_history.astream(
        {"input": user_input},
        config={"configurable": {"session_id": "demo"}}
    ):
        print(chunk, end="", flush=True)
    print("\n")

async def main():
    print("=== Assistant IA (démo streaming) ===")
    await chat_stream("Qu'est-ce que LangChain ?")
    await chat_stream("À quoi sert-il ?")
    await chat_stream("Quel lien avec LCEL ?")

if __name__ == "__main__":
    asyncio.run(main())

Résultat :

=== Assistant IA (démo streaming) ===
IA : LangChain est un framework open source pour construire des applications basées sur les grands modèles de langage...
IA : Il sert à construire chatbots, systèmes RAG, applications Agent, etc...
IA : LCEL est LangChain Expression Language, composant central de LangChain...

Chaque caractère en temps réel.

Comparaison mesurée stream vs non-stream :

Railway - Plateforme de déploiement moderne, sans opérations complexes, en ligne en quelques minutes

Commencer →Pub

Sans streaming, latence du premier token = temps total. Avec streaming, c’est le temps du premier token — souvent < 1 s.

Mécanisme : stream appelle récursivement le stream de chaque composant. Le modèle utilise l’API streaming OpenAI ; prompt et parser renvoient souvent un bloc complet. LCEL coordonne tout — un composant non-streamable est traité en retour unique sans casser le flux.

Chapitre 4 : Composants Runnable en détail

Le pipe enchaîne ; les projets réels exigent parallélisme, résultats intermédiaires, transformations custom. LangChain fournit des Runnable dédiés.

RunnableParallel : exécution parallèle

En RAG, on interroge souvent vecteur, mots-clés et graphe en même temps :

from langchain_core.runnables import RunnableParallel

def vector_search(query: str) -> str:
    return f"Résultat vectoriel : 3 documents pour {query}"

def keyword_search(query: str) -> str:
    return f"Résultat mots-clés : 5 enregistrements pour {query}"

def graph_search(query: str) -> str:
    return f"Résultat graphe : 2 entités pour {query}"

retrievers = RunnableParallel(
    vector=RunnableLambda(vector_search),
    keyword=RunnableLambda(keyword_search),
    graph=RunnableLambda(graph_search)
)

results = retrievers.invoke("LangChain LCEL")
print(results)

RunnableParallel renvoie un dict {nom_de_branche: résultat}.

RunnablePassthrough : conserver l’entrée

En RAG, le retriever veut la requête originale ; le générateur veut docs + requête :

from langchain_core.runnables import RunnablePassthrough

def retrieve(query: dict) -> str:
    return "Contenu document récupéré..."

chain = RunnableParallel(
    retrieved_docs=RunnableLambda(retrieve),
    original_query=RunnablePassthrough()
)

result = chain.invoke({"query": "Qu'est-ce que LCEL ?"})
print(result)

RunnablePassthrough repasse l’entrée telle quelle — crucial dans les pipes complexes.

RunnableLambda : transformation custom

from langchain_core.runnables import RunnableLambda

def format_output(result: dict) -> str:
    docs = result["retrieved_docs"]
    query = result["original_query"]["query"]
    return f"Références : {docs}\nQuestion : {query}\nRépondez à partir des références :"

chain = RunnableParallel(
    retrieved_docs=RunnableLambda(retrieve),
    original_query=RunnablePassthrough()
) | RunnableLambda(format_output)

formatted = chain.invoke({"query": "Qu'est-ce que LCEL ?"})
print(formatted)

RunnableLambda est la « colle universelle » du pipe.

Pipe RAG complet

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    ["LCEL est le langage d'expression de LangChain",
     "L'opérateur pipe sert à enchaîner les composants",
     "Runnable est l'abstraction centrale de LCEL"],
    embeddings
)
retriever = vectorstore.as_retriever()

rag_prompt = ChatPromptTemplate.from_template(
    """Répondez à partir des références suivantes.

Références :
{context}

Question : {question}

Réponse précise et détaillée :"""
)

def format_docs(docs) -> str:
    return "\n".join(doc.page_content for doc in docs)

rag_chain = (
    RunnableParallel(
        context=retriever | RunnableLambda(format_docs),
        question=RunnablePassthrough()
    )
    | rag_prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

answer = rag_chain.invoke("Qu'est-ce que LCEL ?")
print(answer)

Schéma :

{"question": "Qu'est-ce que LCEL ?"}
              ↓
    ┌─────────┴─────────┐
    ↓                   ↓
[retriever]        [Passthrough]
    ↓                   ↓
format_docs        question
    ↓                   ↓
    └─────────┬─────────┘
              ↓
        {"context": "...", "question": "..."}
              ↓
         [rag_prompt] → [model] → [parser]
              ↓
        "Contenu de la réponse..."

Cette implémentation rejoint l’optimisation RAG en pratique — reranking, multi-recall, etc. s’appliquent directement à cette structure LCEL.

Chapitre 5 : Migration des anciennes chaînes

Si votre projet utilise encore LLMChain, la migration est accessible. L’an dernier, j’ai migré un bot SAV e-commerce en deux jours. Quelques patterns courants.

LLMChain → syntaxe pipe

# Ancien (LLMChain)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

llm = OpenAI(temperature=0.7)
prompt = PromptTemplate(
    template="Question : {question}\nRéponse :",
    input_variables=["question"]
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(question="...")

# Nouveau (LCEL)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(temperature=0.7)
prompt = ChatPromptTemplate.from_template("Question : {question}\nRéponse :")
chain = prompt | model
result = chain.invoke({"question": "..."})

Points clés :

1. Modèle : OpenAI (Completion) → ChatOpenAI (Chat API).
2. Prompt : ChatPromptTemplate pour system message et multi-rôles.
3. Appel : run() → invoke(), retour Message — utilisez .content.

SequentialChain → RunnableParallel

L’ancien code enchaînait plusieurs étapes avec SequentialChain ; après migration, connectez directement avec le pipe :

# Ancien (SequentialChain)
from langchain.chains import SequentialChain, LLMChain

title_chain = LLMChain(
    llm=llm, prompt=title_prompt,
    output_key="title"
)

content_chain = LLMChain(
    llm=llm, prompt=content_prompt,
    output_key="content"
)

full_chain = SequentialChain(
    chains=[title_chain, content_chain],
    input_variables=["topic"],
    output_variables=["title", "content"]
)
result = full_chain({"topic": "Développement IA"})
print(result["title"], result["content"])

# Nouveau (LCEL)
from langchain_core.runnables import RunnableParallel

title_chain = title_prompt | model
content_chain = content_prompt | model

full_chain = RunnableParallel(
    title=title_chain,
    content=content_chain
)
result = full_chain.invoke({"topic": "Développement IA"})
print(result["title"].content, result["content"].content)

SequentialChain exécute en série — chaque chaîne attend la précédente. RunnableParallel parallélise et accélère. Pour du sériel (la deuxième étape dépend de la première), enchaînez avec le pipe :

chain = (
    title_prompt | model | StrOutputParser()
    | (lambda title: {"topic": topic, "title": title})
    | content_prompt | model
)

TransformChain → RunnableLambda

TransformChain insérait une logique custom ; migrez vers RunnableLambda :

# Ancien (TransformChain)
from langchain.chains import TransformChain

def transform_func(inputs: dict) -> dict:
    text = inputs["text"]
    processed = text.upper()
    return {"processed_text": processed}

transform_chain = TransformChain(
    input_variables=["text"],
    output_variables=["processed_text"],
    transform=transform_func
)

# Nouveau (RunnableLambda)
from langchain_core.runnables import RunnableLambda

transform_chain = RunnableLambda(transform_func)

RunnableLambda est plus souple — pas de déclaration explicite de variables, participation directe au pipe.

Pièges de migration

Quelques pièges que j’ai rencontrés :

Piège 1 : le type de retour a changé

LLMChain.run() renvoie une chaîne ; LCEL.invoke() renvoie un objet Message.

Vultr - Serveur cloud NVMe haute performance, 32 emplacements mondiaux, déploiement Docker en un clic

Voir les tarifs →Pub

result = chain.run(...)  # str
result = chain.invoke(...)  # AIMessage
text = result.content  # str

Solution : ajoutez StrOutputParser en fin de pipe.

chain = prompt | model | StrOutputParser()
result = chain.invoke(...)  # str directement

Piège 2 : migration des composants mémoire

Ancien code avec ConversationChain :

from langchain.chains import ConversationChain

chain = ConversationChain(llm=llm, memory=memory)

Nouveau code avec RunnableWithMessageHistory :

from langchain_core.runnables import RunnableWithMessageHistory

chain = prompt | model
chain_with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history=get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

Plus de paramètres — noms de champs explicites. Voir les appels d’outils par un Agent pour un système de dialogue complet.

Piège 3 : chemins d’import LangChain v0.3

Beaucoup de composants passent de langchain à langchain_core ou langchain_community :

# Anciens imports
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# Nouveaux imports
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory

L’IDE signale les erreurs — suivez les suggestions.

Cas de migration en production

Bot SAV e-commerce migré l’an dernier : ~300 lignes (LLMChain + SequentialChain + TransformChain), logique centrale après migration :

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)

intent_prompt = ChatPromptTemplate.from_template(
    """Analysez l'intention utilisateur, retournez une catégorie :
    - product_query (consultation produit)
    - order_status (suivi commande)
    - complaint (réclamation)
    - other (autre)

    Message : {message}
    Catégorie :"""
)

product_prompt = ChatPromptTemplate.from_template(
    "Consultation produit : {message}\nRecherchez dans le catalogue et répondez :"
)
order_prompt = ChatPromptTemplate.from_template(
    "Suivi commande : {message}\nInterrogez le statut et répondez :"
)

def route_by_intent(result):
    intent = result.content.strip().lower()
    if "product" in intent:
        return "product"
    elif "order" in intent:
        return "order"
    else:
        return "default"

intent_chain = intent_prompt | model | StrOutputParser() | RunnableLambda(route_by_intent)

full_chain = (
    {"message": RunnablePassthrough()}
    | RunnableParallel(
        intent=intent_chain,
        original=RunnablePassthrough()
    )
    # Routage par branche — RunnableBranch en production
)

async for chunk in full_chain.astream({"message": "Je veux voir la commande 12345"}):
    print(chunk, end="", flush=True)

~150 lignes après migration — moitié en moins. Streaming et async, autrefois développements à part, tiennent en une ligne.

Résumé

LCEL est l’architecture recommandée LangChain v0.3+. Pipe pour simplifier, Runnable pour unifier, streaming intégré pour l’UX.

Le défi n’est pas la syntaxe mais le changement de mindset : déclaration explicite vs flux implicite dans le pipe.

Migrez par étapes : d’abord les chaînes simples, puis la composition complexe, avec LangSmith pour déboguer les types.

Ensuite : gestion d’état avec LangGraph. LangGraph + LCEL : chaînes simples en LCEL, état complexe en LangGraph — combinaison mature aujourd’hui.

Navigation — série Développement IA en pratique

• Partie 1 : Claude API, de l’authentification au dialogue multi-tours
• Partie 2 : Prompt Engineering avancé en pratique
• Partie 8 : Optimisation RAG : équilibre précision et qualité de génération
• Partie 11 : Gestion d’état LangGraph — architecture Agent 2026
• Partie 13 : LangChain LCEL en pratique : de la chaîne traditionnelle à la réponse en streaming (cet article)
• Partie 13 : Appels d’outils Agent : faire appeler des API externes à l’IA