0Pricing
AI Engineering Academy · Leçon

Chaînes conditionnelles et parallèles

Construisez des structures RunnableParallel et RunnableBranch pour exécuter plusieurs chaînes simultanément ou orienter l’entrée vers différentes chaînes selon des conditions dynamiques.

Chaînes conditionnelles et parallèles est une leçon AI Engineering Academy gratuite sur CoddyKit. Ceci est la leçon 3 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Engineering Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Engineering Academy comprend 4 leçons au total.

Pourquoi le parallélisme et les embranchements sont importants

Les pipelines LLM du monde réel doivent souvent effectuer plusieurs tâches simultanément ou router les requêtes différemment selon leur contenu. Les chaînes parallèles exécutent plusieurs branches simultanément, ce qui réduit la latence lorsque les tâches sont indépendantes. Les chaînes à embranchements orientent l’entrée vers différentes chaînes spécialisées selon des conditions dynamiques. LCEL prend nativement en charge ces deux modèles avec RunnableParallel et RunnableBranch.

Principes de base de RunnableParallel

RunnableParallel prend un dictionnaire dans lequel chaque clé correspond à un Runnable. Lors de l’appel, il exécute toutes les branches simultanément et renvoie un dictionnaire dont chaque clé contient le résultat de sa branche. C’est idéal lorsque vous souhaitez générer plusieurs sorties à partir de la même entrée, par exemple produire simultanément un résumé et extraire des mots-clés.

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

model = ChatOpenAI(model='gpt-4o-mini')
parser = StrOutputParser()

parallel = RunnableParallel(
    summary=(
        ChatPromptTemplate.from_template('Summarize: {text}') | model | parser
    ),
    keywords=(
        ChatPromptTemplate.from_template('Extract keywords from: {text}') | model | parser
    )
)

result = parallel.invoke({'text': 'Long document text here...'})
print(result['summary'])
print(result['keywords'])

Parallélisme avec la notation abrégée des dictionnaires

LCEL propose une notation abrégée pratique : le fait de transmettre un dict simple comme étape d’une chaîne pipeline l’enveloppe automatiquement dans RunnableParallel. Les branches parallèles deviennent ainsi naturelles et vous évitez l’instanciation explicite de la classe. Les clés du dictionnaire deviennent les clés du résultat et les valeurs sont les branches exécutées simultanément.

from langchain_core.runnables import RunnablePassthrough

# Dict shorthand creates RunnableParallel automatically
chain = (
    RunnablePassthrough.assign(
        sentiment=(
            ChatPromptTemplate.from_template('Sentiment of: {review}')
            | model | parser
        ),
        aspects=(
            ChatPromptTemplate.from_template('List aspects mentioned in: {review}')
            | model | parser
        )
    )
)

result = chain.invoke({'review': 'Great battery but poor camera quality.'})
print(result['sentiment'])
print(result['aspects'])

Comprendre RunnableBranch

RunnableBranch oriente l’entrée vers différentes chaînes en fonction d’une condition. Vous fournissez une liste de paires (condition, runnable) ainsi qu’un runnable par défaut. La branche évalue les conditions dans l’ordre et exécute la première branche correspondante. Cela permet un routage fondé sur l’intention : envoyer les demandes au service client vers une chaîne d’assistance et les questions techniques vers une chaîne de documentation.

from langchain_core.runnables import RunnableBranch

technical_chain = (
    ChatPromptTemplate.from_template('Technical answer: {query}') | model | parser
)
general_chain = (
    ChatPromptTemplate.from_template('General answer: {query}') | model | parser
)

branch = RunnableBranch(
    (lambda x: 'error' in x['query'].lower() or 'bug' in x['query'].lower(),
     technical_chain),
    general_chain  # default branch
)

result = branch.invoke({'query': 'I got a TypeError in my code'})
# Routes to technical_chain because 'error' is in the query

Routage sémantique avec classification par LLM

Un modèle de routage plus flexible utilise un appel de classification par un LLM pour déterminer la branche à utiliser. Le routeur appelle d’abord un petit modèle afin de classer l’intention de l’entrée, puis utilise cette classification pour l’orienter vers la chaîne spécialisée appropriée. Cette approche gère les cas nuancés que la recherche par mots-clés ne détecte pas, au prix d’un appel d’API supplémentaire.

from langchain_core.output_parsers import StrOutputParser

# Step 1: classify intent
classify_prompt = ChatPromptTemplate.from_template(
    'Classify this query as exactly one of: billing, technical, general.\nQuery: {query}'
)
classifier = classify_prompt | model | StrOutputParser()

# Step 2: route based on classification
def route(classification_result: dict):
    topic = classification_result['topic'].strip().lower()
    if topic == 'billing':
        return billing_chain
    elif topic == 'technical':
        return technical_chain
    return general_chain

full_chain = (
    RunnablePassthrough.assign(topic=lambda x: classifier.invoke(x))
    | RunnableLambda(route)
)

RAG parallèle : plusieurs récupérateurs

Dans les systèmes RAG avancés, vous pouvez récupérer simultanément des données provenant de plusieurs sources et fusionner les résultats. RunnableParallel vous permet d’interroger en même temps une base de données de produits, une base de FAQ et un index de documentation. Une étape de fusion combine ensuite les meilleurs résultats avant de transmettre le contexte au LLM, ce qui fournit au modèle une base d’informations plus riche.

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

# Assume these retrievers are already set up
faq_retriever = faq_vectorstore.as_retriever(search_kwargs={'k': 3})
doc_retriever = doc_vectorstore.as_retriever(search_kwargs={'k': 3})

retrieval = RunnableParallel(
    faq_results=faq_retriever,
    doc_results=doc_retriever
)

def merge_docs(retrieved: dict) -> str:
    all_docs = retrieved['faq_results'] + retrieved['doc_results']
    return '\n\n'.join(d.page_content for d in all_docs)

pipeline = (
    retrieval
    | RunnableLambda(merge_docs)
    | ChatPromptTemplate.from_template('Context: {context}\nAnswer: {question}')
    | model | parser
)

Chaînes conditionnelles avec itemgetter

Lorsque vous devez sélectionner une clé spécifique dans un résultat parallèle ou ne transmettre qu’une partie du contexte à l’étape suivante, operator.itemgetter de Python fonctionne comme un sélecteur Runnable léger. Cela est utile après une étape parallèle, lorsque différentes branches produisent des clés différentes et que vous devez extraire uniquement celle qui est pertinente pour l’étape de traitement suivante.

from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough

# After a parallel step, extract just the summary
chain = (
    RunnableParallel(
        summary=summary_chain,
        sentiment=sentiment_chain
    )
    | itemgetter('summary')  # pass only the summary onward
    | translate_chain
)

# itemgetter works because dict.__getitem__ is a valid transform

Mesurer l’accélération du traitement parallèle

Le principal avantage de RunnableParallel est la réduction du temps réel écoulé. Trois appels séquentiels à un LLM prenant chacun 2 secondes nécessiteraient 6 secondes au total. En parallèle, ils se terminent en environ 2 secondes, soit le temps d’exécution de la branche la plus lente. Toutefois, les appels parallèles multiplient votre consommation de jetons au même moment : surveillez donc les limites de débit. Utilisez max_concurrency dans batch() ou définissez des limites de débit par clé si nécessaire.

import time

# Measure sequential time
start = time.time()
result1 = chain_a.invoke(input_data)
result2 = chain_b.invoke(input_data)
result3 = chain_c.invoke(input_data)
seq_time = time.time() - start
print(f'Sequential: {seq_time:.2f}s')

# Measure parallel time
start = time.time()
results = RunnableParallel(a=chain_a, b=chain_b, c=chain_c).invoke(input_data)
par_time = time.time() - start
print(f'Parallel: {par_time:.2f}s')
print(f'Speedup: {seq_time/par_time:.1f}x')

Chaînes parallèles asynchrones

Pour une exécution parallèle véritablement non bloquante dans les applications asynchrones, utilisez RunnableParallel.ainvoke(). En interne, LCEL utilise asyncio.gather() pour exécuter les branches simultanément dans la boucle d’événements. Cela est particulièrement important dans les services FastAPI, où chaque gestionnaire de requête est une coroutine : l’utilisation de l’interface asynchrone empêche le blocage de la boucle d’événements, même lorsque plusieurs appels à un LLM sont effectués simultanément.

import asyncio

async def analyze_document(text: str) -> dict:
    parallel = RunnableParallel(
        summary=summary_chain,
        keywords=keyword_chain,
        sentiment=sentiment_chain
    )
    # All three chains run concurrently with asyncio.gather internally
    result = await parallel.ainvoke({'text': text})
    return result

# In FastAPI:
from fastapi import FastAPI
app = FastAPI()

@app.post('/analyze')
async def analyze(request: dict):
    return await analyze_document(request['text'])

Imbrication de chaînes parallèles et séquentielles

Les pipelines complexes combinent souvent des étapes séquentielles et parallèles. Vous pouvez imbriquer RunnableParallel dans un pipeline séquentiel, et inversement. Par exemple : classer l’intention (séquentiellement), puis exécuter en parallèle la recherche et la mise en forme du contexte, avant de générer la réponse finale (séquentiellement). LangChain évalue correctement l’imbrication, ce qui permet de conserver des pipelines sophistiqués lisibles sans enchevêtrement de fonctions de rappel.

# Full pipeline: classify → parallel retrieval → generate
pipeline = (
    RunnablePassthrough.assign(
        intent=classify_chain  # sequential: classify first
    )
    | RunnablePassthrough.assign(
        context=RunnableParallel(  # parallel: retrieve from both sources
            faq=faq_retriever,
            docs=doc_retriever
        )
    )
    | format_context_chain  # sequential: format merged context
    | generate_answer_chain  # sequential: call LLM
)

Gestion des erreurs dans les branches

Lorsqu’une branche de RunnableParallel échoue, l’invocation parallèle entière lève une exception par défaut. Utilisez .with_fallbacks() sur les objets Runnable de chaque branche afin de gérer correctement les échecs au niveau des branches. Pour RunnableBranch, entourez chaque branche d’un bloc try-except dans un RunnableLambda, ou utilisez une chaîne de secours par défaut pour gérer les erreurs de routage.

from langchain_core.runnables import RunnableParallel

# Wrap each branch with a fallback
safe_summary = summary_chain.with_fallbacks([
    RunnableLambda(lambda x: 'Summary unavailable')
])
safe_sentiment = sentiment_chain.with_fallbacks([
    RunnableLambda(lambda x: 'Sentiment unavailable')
])

robust_parallel = RunnableParallel(
    summary=safe_summary,
    sentiment=safe_sentiment
)

# Now one branch failing won't kill the entire parallel call

Vérification rapide

Vérifiez votre compréhension des branchements et des chaînes parallèles dans LCEL.

Récapitulatif de la leçon

Dans cette leçon, vous avez appris que RunnableParallel exécute plusieurs chaînes simultanément et renvoie un dictionnaire de résultats, ce qui réduit la latence des appels indépendants à un LLM ; que RunnableBranch achemine l’entrée vers différentes chaînes spécialisées selon des conditions ou une classification effectuée par un LLM ; et que l’imbrication d’étapes parallèles et séquentielles permet de créer des pipelines sophistiqués à plusieurs chemins tout en restant lisibles. Nous allons maintenant explorer la sortie en continu dans LangChain.

Questions Fréquemment Posées

La leçon « Chaînes conditionnelles et parallèles » est-elle gratuite ?

Oui — le texte complet de « Chaînes conditionnelles et parallèles » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Engineering Academy, passe à CoddyKit PRO. Le cours AI Engineering Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Chaînes conditionnelles et parallèles » ?

Construisez des structures RunnableParallel et RunnableBranch pour exécuter plusieurs chaînes simultanément ou orienter l’entrée vers différentes chaînes selon des conditions dynamiques. Tu pratiques AI Engineering Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Engineering Academy ?

Aucune expérience préalable n'est requise. AI Engineering Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 3 sur 4.

Combien de temps prend la leçon « Chaînes conditionnelles et parallèles » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Engineering Academy ?

Oui. Chaque leçon AI Engineering Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Architecture de LangChain et abstractions fondamentales
  2. Construire des chaînes avec LCEL
  3. Chaînes conditionnelles et parallèles
  4. Diffuser la sortie avec LangChain
← Retour à AI Engineering Academy