0Pricing
AI Engineering Academy · Ders

Dallanma ve Paralel Zincirler

Birden çok zinciri eş zamanlı çalıştırmak veya girdiyi dinamik koşullara göre farklı zincirlere yönlendirmek için RunnableParallel ve RunnableBranch yapılarını oluşturun.

Dallanma ve Paralel Zincirler, CoddyKit'te ücretsiz bir AI Engineering Academy dersidir. Bu, 4 dersinin 3. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, AI Engineering Academy öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. AI Engineering Academy kursu toplamda 4 dersten oluşur.

Bu dersin bazı bölümleri henüz çevrilmemiş olup İngilizce olarak gösterilmektedir.

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.

Sıkça Sorulan Sorular

“Dallanma ve Paralel Zincirler” dersi ücretsiz mi?

Evet — “Dallanma ve Paralel Zincirler” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve AI Engineering Academy kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. AI Engineering Academy kursu toplamda 4 dersten oluşur.

“Dallanma ve Paralel Zincirler” dersinde ne öğreneceğim?

Birden çok zinciri eş zamanlı çalıştırmak veya girdiyi dinamik koşullara göre farklı zincirlere yönlendirmek için RunnableParallel ve RunnableBranch yapılarını oluşturun. AI Engineering Academy ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.

AI Engineering Academy öğrenmeye başlamak için deneyim gerekli mi?

Önceden deneyim gerekmez. CoddyKit'te AI Engineering Academy, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 3. dersidir.

“Dallanma ve Paralel Zincirler” dersi ne kadar sürer?

Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.

Bu AI Engineering Academy dersinde kod yazıp çalıştırabilir miyim?

Evet. Her AI Engineering Academy dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.

Bu kursun tüm dersleri

  1. LangChain Mimarisi ve Temel Soyutlamalar
  2. LCEL ile Zincirler Oluşturma
  3. Dallanma ve Paralel Zincirler
  4. LangChain'de Akışlı Çıktı
← AI Engineering Academy Sayfasına Dön