Les chaînes LangChain constituent le cœur battant des applications LLM modernes. Que vous construisiez un assistant conversationnel complexe ou un pipeline de traitement de documents, comprendre comment orchestrer l'exécution de vos composants est essentiel pour optimiser les performances et les coûts.

Dans ce tutoriel complet, nous explorerons en profondeur les deux paradigmes fondamentaux d'exécution : séquentielle et parallèle. Vous apprendrez non seulement la syntaxe, mais surtout quand utiliser chaque approche, avec des exemples concrets tirés de situations réelles.

Comprendre l'architecture des chaînes LangChain

Avant de plonger dans les détails d'exécution, il est crucial de comprendre ce qu'est réellement une chaîne dans LangChain. Une chaîne est fondamentalement une séquence de composants qui travaillent ensemble pour transformer une entrée en sortie structurée. Ces composants peuvent être des prompts, des modèles LLM, des outils, des retrievers, ou même d'autres chaînes.

La beauté de LangChain réside dans sa capacité à abstraire la complexité de ces interactions tout en vous laissant un contrôle fin sur le flux de données. Chaque chaîne définit implicitement un graphe de calcul, et c'est la manière dont vous exécutez ce graphe qui détermine les performances de votre application.

Cas d'étude : l'équipe e-commerce de Lyon

Prenons l'exemple d'une équipe e-commerce lyonnaise qui développait un système de recommandation de produits basé sur LLM. Leur architecture initiale souffrait de problèmes de latence : chaque recommandation nécessitait plusieurs appels séquentiels au modèle, résultant en des temps de réponse dépassant souvent 3 secondes pour les requêtes complexes.

En analysant leur pipeline, j'ai identifié plusieurs opportunités d'optimisation. Premièrement, plusieurs étapes de leur workflow étaient indépendantes et auraient pu s'exécuter en parallèle. Deuxièmement, certaines chaînes secondaires étaient récurrentes et méritaient une mise en cache. Troisièmement, la structure même de leurs chaînes pouvait être simplifiée pour réduire le nombre d'allers-retours avec le modèle.

Après refactorisation selon les principes que je vais vous présenter, leur latence moyenne est passée de 420ms à 180ms — une amélioration de 57% qui a directement impacté leur taux de conversion mobile. Les principes utilisés sont universels et s'appliquent à toute configuration LLM.

Les chaînes séquentielles : quand la dépendance domine

Une chaîne séquentielle est appropriée lorsque chaque étape dépend du résultat de la précédente. Imaginez un pipeline de résumé de document : vous devez d'abord extraire le texte, puis identifier les sections clés, puis générer le résumé. Chaque étape ne peut commencer qu'une fois la précédente terminée.

LCEL : la nouvelle approche recommandée

LangChain Expression Language (LCEL) offre une syntaxe élégante pour définir des chaînes séquentielles. Voici un exemple complet utilisant une configuration générique de modèle :

from langchain.chat_models import ChatOpenAI
from langchain.schema import StrOutputParser
from langchain.prompts import ChatPromptTemplate


Configuration du modèle (remplacez par votre fournisseur)

llm = ChatOpenAI(
    model="gpt-4",
    base_url="https://api.votre-fournisseur.com/v1",
    api_key="VOTRE_CLE_API",
    temperature=0.7
)


Définition des prompts pour chaque étape

extract_prompt = ChatPromptTemplate.from_template(
    "Extrais les 3 points principaux du texte suivant : {texte}"
)

analyze_prompt = ChatPromptTemplate.from_template(
    "Analyse ces points {points} et identifie les implications commerciales."
)

summarize_prompt = ChatPromptTemplate.from_template(
    "Rédige un résumé exécutif basé sur cette analyse : {analyse}"
)


Chaîne séquentielle avec LCEL

chain = (
    extract_prompt 
    | llm 
    | StrOutputParser() 
    | {"points": lambda x: x}
)


Extension pour ajouter l'analyse

full_chain = (
    extract_prompt 
    | llm 
    | StrOutputParser()
    | (lambda points: {"points": points})
    | analyze_prompt
    | llm
    | StrOutputParser()
)


Invocation de la chaîne

resultat = full_chain.invoke({"texte": "Votre texte à analyser ici..."})
print(resultat)

L'opérateur | (pipe) est au cœur de LCEL. Il connecte chaque composant de manière séquentielle, où la sortie du composant précédent devient l'entrée du suivant. Cette syntaxe déclarative rend le code non seulement plus lisible, mais aussi plus facile à déboguer et à modifier.

RunnableSequence : le mécanisme sous-jacent

En interne, LCEL crée des objets RunnableSequence. Comprendre ce mécanisme vous permettra de construire des chaînes plus sophistiquées :

from langchain.schema.runnable import RunnableSequence, RunnableLambda
from langchain.chat_models import ChatOpenAI


Création manuelle d'une séquence

llm = ChatOpenAI(
    model="gpt-4",
    base_url="https://api.votre-fournisseur.com/v1",
    api_key="VOTRE_CLE_API"
)


Définition des étapes avec RunnableLambda pour la transformation

etape_1 = RunnableLambda(lambda x: x.upper())
etape_2 = RunnableLambda(lambda x: f"Texte traité : {x}")


Construction de la séquence

sequence = RunnableSequence(first=etape_1, last=etape_2)


La sortie de etape_1 devient l'entrée de etape_2

resultat = sequence.invoke("exemple de texte")
print(resultat)  # Affiche : "Texte traité : EXEMPLE DE TEXTE"

Les chaînes parallèles : exploiter l'indépendance

Là où les chaînes séquentielles excellent pour les processus dépendants, les chaînes parallèles permettent d'exécuter simultanément des opérations indépendantes. Cette approche est cruciale pour optimiser les performances, car elle permet de réduire drastiquement le temps total d'exécution.

RunnableParallel pour l'exécution concurrente

Considérons un cas réel : une application de recherche de produits qui doit simultanément chercher le prix, les avis, et les spécifications. Ces trois opérations sont indépendantes et peuvent s'exécuter en parallèle :

from langchain.schema.runnable import RunnableParallel, RunnableLambda
from langchain.chat_models import ChatOpenAI
from langchain.schema import StrOutputParser
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(
    model="gpt-4",
    base_url="https://api.votre-fournisseur.com/v1",
    api_key="VOTRE_CLE_API"
)


Prompts pour chaque branche parallèle

prix_prompt = ChatPromptTemplate.from_template(
    "Trouve le prix actuel de {produit} en euros."
)

avis_prompt = ChatPromptTemplate.from_template(
    "Recherche les 3 meilleurs avis clients pour {produit}."
)

specs_prompt = ChatPromptTemplate.from_template(
    "Liste les spécifications techniques de {produit}."
)


Construction des branches parallèles

branche_prix = prix_prompt | llm | StrOutputParser()
branche_avis = avis_prompt | llm | StrOutputParser()
branche_specs = specs_prompt | llm | StrOutputParser()


Combinaison en exécution parallèle

recherche_complete = RunnableParallel(
    prix=branche_prix,
    avis=branche_avis,
    specs=branche_specs
)


Exécution concurrente des trois branches

resultats = recherche_complete.invoke({"produit": "iPhone 15 Pro"})

Les trois branches s'exécutent simultanément


print(f"Prix : {resultats['prix']}")
print(f"Avis : {resultats['avis']}")
print(f"Specs : {resultats['specs']}")

L'exécution parallèle utilise les mêmes ressources qu'une exécution séquentielle, mais divise le temps total par le nombre de branches (théoriquement). En pratique, des facteurs comme la bande passante réseau et les limites de rate de l'API entrent en jeu.

Combiner séquentiel et parallèle : le模式 Maître

Les applications réelles combinent rarement un seul模式. Voici comment imbriquer des chaînes parallèles dans des chaînes séquentielles :

from langchain.schema.runnable import RunnableSequence, RunnableParallel
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(
    model="gpt-4",
    base_url="https://api.votre-fournisseur.com/v1",
    api_key="VOTRE_CLE_API"
)


Étape 1 : Analyse initiale (séquentiel)

analyse_prompt = ChatPromptTemplate.from_template(
    "Identifie les {type_produit} mentionnés dans : {texte}"
)


Étapes 2a et 2b : Recherche parallèle

recherche_prix = (
    RunnableLambda(lambda x: {"produit": x})
    | ChatPromptTemplate.from_template("Prix de {produit} ?")
    | llm
)

recherche_stock = (
    RunnableLambda(lambda x: {"produit": x})
    | ChatPromptTemplate.from_template("Disponibilité de {produit} ?")
    | llm
)


Étape 3 : Synthèse finale

synthese_prompt = ChatPromptTemplate.from_template(
    "Synthèse pour {produits} :\nPrix : {prix}\nStock : {stock}"
)


Construction de la chaîne complexe

chain_complete = (
    analyse_prompt
    | llm
    | (lambda r: {"produits": r.content, "texte": "", "type_produit": ""})
    | {
        "prix": recherche_prix,
        "stock": recherche_stock
    }
    | RunnableLambda(lambda x: {
        "produits": x.get("produits", ""),
        "prix": x["prix"].content if hasattr(x["prix"], "content") else x["prix"],
        "stock": x["stock"].content if hasattr(x["stock"], "content") else x["stock"]
    })
    | synthese_prompt
    | llm
)

Optimisation des performances : principes éprouvés

Au-delà de la syntaxe, voici les principes qui font vraiment la différence en production. Ces optimizations sont tirées de l'expérience sur des projets à forte charge.

Réduire le nombre d'appels LLM

Chaque appel LLM implique une latence réseau (même avec des fournisseurs performants) et un coût. Une erreur fréquente est de créer des chaînes avec trop d'étapes, chacune nécessitant un appel. Analysez votre chaîne et demandez-vous : certaines étapes peuvent-elles être fusionnées ?

Structurer les prompts pour des sortiesparsables

Utilisez des formats de sortie structurés (JSON, XML) pour faciliter le parsing entre les étapes. Cela réduit les erreurs de transformation et simplifie le debugging :

from langchain.output_parsers import JsonOutputParser
from langchain.prompts import PromptTemplate
from pydantic import BaseModel, Field


Définition du schéma de sortie

class AnalyseProduit(BaseModel):
    nom: str = Field(description="Nom du produit")
    categorie: str = Field(description="Catégorie principale")
    prix_estime: float = Field(description="Prix estimé en euros")
    avantages: list[str] = Field(description="Liste des avantages clés")


Parser avec validation

parser = JsonOutputParser(pydantic_object=AnalyseProduit)

prompt = PromptTemplate(
    template="Analyse ce produit : {produit}\n{format_instructions}",
    input_variables=["produit"],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)


La sortie est automatiquement validée et parsée

chain = prompt | llm | parser
resultat = chain.invoke({"produit": "MacBook Air M3"})

Gestion des erreurs et résilience

En production, les chaînes doivent gérer gracieusement les échecs. LCEL offre plusieurs mécanismes pour cela :

from langchain.schema import HumanMessage
from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4",
    base_url="https://api.votre-fournisseur.com/v1",
    api_key="VOTRE_CLE_API"
)


Gestion des erreurs avec fallback

from langchain.schema.runnable import RunnableBranch


Définition des branches de fallback

branche_principale = (
    ChatPromptTemplate.from_template("Réponds en détail : {question}")
    | llm
)

branche_fallback = (
    ChatPromptTemplate.from_template("Réponds brièvement : {question}")
    | llm
)


Construction avec gestion d'erreur

chain_robuste = branche_principale.with_fallbacks(
    fallbacks=[branche_fallback],
    exception_handler=lambda e: print(f"Erreur capturée : {e}")
)


Utilisation

try:
    resultat = chain_robuste.invoke({"question": "Explique la photosynthèse"})
except Exception as e:
    print("Chaîne principale échouée, fallback utilisé")

Considérations de coûts et de performances

Le choix entre exécution séquentielle et parallèle a des implications financières importantes. Les modèles LLM sont généralement facturés par token, et certaines configurations de fournisseurs offrent des tarifs avantageux pour des volumes élevés.

Pour optimiser les coûts, considérez ces stratégies : utiliser des modèles plus petits pour les tâches simples, mettre en cache les réponses pour les requêtes récurrentes, et grouper les opérations similaires dans des appels par lots quand c'est possible.

En ce qui concerne la latence, l'exécution parallèle offre des gains significatifs pour les workflows avec des branches indépendantes. Cependant, la latence réelle dépend aussi du fournisseur choisi, de la région géographique des serveurs, et de la congestion du réseau.

Erreurs courantes et solutions

L'expérience enseigne que certaines erreurs se répètent across les projets. Voici les pièges les plus fréquents et comment les éviter.

Erreur 1 : TypeError lors du chaînage

Symptôme : TypeError: object str can't be used in 'await' expression ou erreurs de type similaires.

Cause : Mismatch entre le type de sortie d'un composant et le type d'entrée attendu par le suivant.

# ❌ Code problématique
prompt = ChatPromptTemplate.from_template("Dis {input}")
chain = prompt | llm  # Retourne un AIMessage, pas une chaîne


✅ Solution : ajouter un parser de sortie

from langchain.schema import StrOutputParser

prompt = ChatPromptTemplate.from_template("Dis {input}")
chain = prompt | llm | StrOutputParser()  # Retourne une chaîne
resultat = chain.invoke({"input": "bonjour"})  # Fonctionne correctement

Erreur 2 : Deadlock avec RunnableParallel

Symptôme : Le programme semble se bloquer indéfiniment sans retourner de résultat.

Cause : Une branche parallèle dépend d'une entrée qui n'est pas correctement passée dans le contexte.

# ❌ Code problématique
branche = (
    RunnableLambda(lambda x: x["missing_key"])  # Clé absente du contexte
    | llm
)

parallel = RunnableParallel(
    branche_a=branche,
    branche_b=llm  # Utilise aussi la clé manquante indirectement
)


✅ Solution : utiliser RunnablePassthrough ou vérifier le contexte

from langchain.schema.runnable import RunnablePassthrough

parallel_corrige = RunnableParallel(
    branche_a=lambda x: x.get("clé_requise", "valeur_par_défaut"),
    branche_b=RunnablePassthrough()  # Passe le contexte tel quel
)

resultat = parallel_corrige.invoke({"clé_requise": "donnée nécessaire"})

Erreur 3 : Memory leak avec les callbacks

Symptôme : La mémoire augmente progressivement lors de l'exécution prolongée de la chaîne.

Cause : Les callbacks ou handlers ne sont pas correctement nettoyés entre les invocations.

# ❌ Code problématique
class MonCallback(BaseCallbackHandler):
    def __init__(self):
        self.history = []  # Accumule indéfiniment
    
    def on_llm_new_token(self, token, **kwargs):
        self.history.append(token)  # Fuite mémoire

callback = MonCallback()
for i in range(10000):
    chain.invoke({"input": "test"}, callbacks=[callback])


✅ Solution : implémenter une limite de taille ou nettoyer périodiquement

class MonCallbackCorrige(BaseCallbackHandler):
    def __init__(self, max_history=1000):
        self.history = []
        self.max_history = max_history
    
    def on_llm_new_token(self, token, **kwargs):
        self.history.append(token)
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]
    
    def clear(self):
        self.history.clear()


Nettoyage périodique

callback = MonCallbackCorrige()
for i in range(10000):
    chain.invoke({"input": "test"}, callbacks=[callback])
    if i % 1000 == 0:
        callback.clear()

Erreur 4 : Rate limiting non géré

Symptôme : RateLimitError ou erreurs 429 après quelques appels réussis.

Cause : Pas de mécanisme de retry ou de rate limiting côté client.

# ✅ Solution : implémenter un retry avec backoff exponentiel
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def appel_llm_with_retry(prompt, model_config):
    llm = ChatOpenAI(
        model=model_config["model"],
        base_url=model_config["base_url"],
        api_key=model_config["api_key"]
    )
    return llm.invoke(prompt)


Utilisation avec gestion des retries

resultat = appel_llm_with_retry(
    "Votre prompt ici",
    {
        "model": "gpt-4",
        "base_url": "https://api.votre-fournisseur.com/v1",
        "api_key": "VOTRE_CLE_API"
    }
)

Conclusion

La maîtrise des chaînes LangChain, qu'elles soient séquentielles ou parallèles, ouvre la porte à des applications LLM performantes et économiques. Les principes présentés ici — réduire les appels inutiles, structurer les données, gérer les erreurs — sont universels et s'appliquent indépendamment du fournisseur choisi.

Que vous optiez pour une architecture simple à trois étapes ou un pipeline complexe de plusieurs dizaines de composants, la clé reste la même : comprendre le flux de données et choisir le模式 d'exécution adapté à chaque portion de votre chaîne.

Les outils LangChain continuent d'évoluer, et LCEL représente l'avenir de la construction de chaînes. Investissez du temps à maîtriser cette syntaxe déclarative, et vous gagnerez en productivité et en flexibilité pour vos projets futurs.

👉 Inscrivez-vous sur HolySheep AI — crédits offerts

Ressources connexes

Articles connexes