AI Engineering Academy · 강의

분기 및 병렬 체인

RunnableParallel 및 RunnableBranch 구성 요소를 구축해 여러 체인을 동시에 실행하거나 동적인 조건에 따라 입력을 서로 다른 체인으로 전달합니다.

분기 및 병렬 체인은(는) CoddyKit의 무료 AI Engineering Academy 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Engineering Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Engineering Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

이 강의의 일부는 아직 번역되지 않았으며 영어로 표시됩니다.

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.

자주 묻는 질문

“분기 및 병렬 체인” 강의는 무료인가요?

네 — “분기 및 병렬 체인” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Engineering Academy 강의 전체를 잠금 해제할 수 있습니다. AI Engineering Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

“분기 및 병렬 체인”에서 뭘 배우나요?

RunnableParallel 및 RunnableBranch 구성 요소를 구축해 여러 체인을 동시에 실행하거나 동적인 조건에 따라 입력을 서로 다른 체인으로 전달합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Engineering Academy을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Engineering Academy을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Engineering Academy은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.

“분기 및 병렬 체인” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Engineering Academy 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Engineering Academy 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Engineering Academy(으)로 돌아가기