0Pricing
AI Engineering Academy · Lezione

Chain ramificate e parallele

Costruirà strutture RunnableParallel e RunnableBranch per eseguire più chain simultaneamente o indirizzare l'input verso chain diverse in base a condizioni dinamiche.

Chain ramificate e parallele è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.

Parti di questa lezione non sono ancora state tradotte e vengono mostrate in inglese.

Why Parallel and Branching Matter

Real-world LLM pipelines often need to do multiple things at once or route requests differently based on content. Parallel chains run multiple branches simultaneously, reducing latency when tasks are independent. Branching chains route input to different specialized chains based on dynamic conditions. LCEL supports both patterns natively with RunnableParallel and RunnableBranch.

RunnableParallel Basics

RunnableParallel takes a dictionary where each key maps to a Runnable. When invoked, it runs all branches concurrently and returns a dictionary with each key containing the result of its branch. This is ideal when you want to generate multiple outputs from the same input — for example, generating a summary and extracting keywords simultaneously.

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'])

Parallel with Dict Shorthand

LCEL provides a convenient shorthand: passing a plain dict as a step in a pipe chain automatically wraps it in RunnableParallel. This makes parallel branches feel natural and eliminates the explicit class instantiation. The dict keys become the result keys, and the values are the branches that run concurrently.

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'])

Understanding RunnableBranch

RunnableBranch routes input to different chains based on a condition. You provide a list of (condition, runnable) pairs and a default runnable. The branch evaluates conditions in order and runs the first matching branch. This enables intent-based routing — sending customer service queries to a support chain and technical questions to a documentation chain.

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

Semantic Routing with LLM Classification

A more flexible routing pattern uses a classifier LLM call to determine which branch to use. The router first calls a small model to classify the intent of the input, then uses the classification to route to the appropriate specialist chain. This handles nuanced cases that keyword matching misses, at the cost of one extra API call.

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)
)

Parallel RAG: Multiple Retrievers

In advanced RAG systems, you might retrieve from multiple data sources simultaneously and merge the results. RunnableParallel lets you query a product database, a FAQ store, and a documentation index at the same time. A merging step then combines the top results before passing context to the LLM, giving the model a richer information base.

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
)

Conditional Chains with itemgetter

When you need to select a specific key from a parallel result or pass only part of the context to the next step, Python's operator.itemgetter works as a lightweight Runnable selector. This is useful after a parallel step when different branches produce different keys and you need to extract only the relevant one for the next stage of processing.

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

Measuring Parallel Speedup

The key benefit of RunnableParallel is wall-clock time reduction. Three sequential LLM calls taking 2 seconds each would take 6 seconds total. In parallel they complete in approximately 2 seconds — the time of the slowest branch. However, parallel calls multiply your token usage at once, so watch rate limits. Use max_concurrency in batch() or set per-key rate limits if needed.

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')

Async Parallel Chains

For truly non-blocking parallel execution in async applications, use RunnableParallel.ainvoke(). Under the hood, LCEL uses asyncio.gather() to run the branches concurrently on the event loop. This is especially important in FastAPI services where each request handler is a coroutine — using the async interface prevents event loop blocking even with multiple simultaneous LLM calls.

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'])

Nesting Parallel and Sequential Chains

Complex pipelines often mix sequential and parallel steps. You can nest RunnableParallel inside a sequential pipe and vice versa. For example: classify the intent (sequential), then run retrieval and context formatting in parallel, then generate the final response (sequential). LangChain evaluates the nesting correctly, making sophisticated pipelines readable without callback spaghetti.

# 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
)

Error Handling in Branches

When one branch of a RunnableParallel fails, the entire parallel invocation raises an exception by default. Use .with_fallbacks() on individual branch runnables to handle branch-level failures gracefully. For RunnableBranch, wrap each branch in a try-except within a RunnableLambda or use a default fallback chain to handle routing errors.

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

Quick Check

Test your understanding of branching and parallel chains in LCEL.

Lesson Recap

In this lesson you learned: RunnableParallel runs multiple chains concurrently and returns a dict of results, reducing latency for independent LLM calls, RunnableBranch routes input to different specialist chains based on conditions or LLM classification, and nesting parallel and sequential steps lets you build sophisticated multi-path pipelines that remain readable. Next up we explore streaming output in LangChain.

Domande Frequenti

La lezione «Chain ramificate e parallele» è gratuita?

Sì — il testo completo di «Chain ramificate e parallele» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.

Cosa imparerò in «Chain ramificate e parallele»?

Costruirà strutture RunnableParallel e RunnableBranch per eseguire più chain simultaneamente o indirizzare l'input verso chain diverse in base a condizioni dinamiche. Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Engineering Academy?

Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.

Quanto tempo richiede la lezione «Chain ramificate e parallele»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?

Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Architettura di LangChain e astrazioni fondamentali
  2. Creare chain con LCEL
  3. Chain ramificate e parallele
  4. Streaming dell'output in LangChain
← Torna a AI Engineering Academy