AI Engineering Academy · บทเรียน

สายงานแบบแยกแขนงและแบบขนาน

สร้างโครงสร้าง RunnableParallel และ RunnableBranch เพื่อเรียกใช้หลายสายงานพร้อมกัน หรือส่งอินพุตไปยังสายงานต่าง ๆ ตามเงื่อนไขแบบไดนามิก

สายงานแบบแยกแขนงและแบบขนาน เป็นบทเรียน AI Engineering Academy ฟรีบน CoddyKit นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน 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.

คำถามที่พบบ่อย

บทเรียน “สายงานแบบแยกแขนงและแบบขนาน” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “สายงานแบบแยกแขนงและแบบขนาน” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส AI Engineering Academy ให้อัปเกรดเป็น CoddyKit PRO คอร์ส AI Engineering Academy มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “สายงานแบบแยกแขนงและแบบขนาน”

สร้างโครงสร้าง RunnableParallel และ RunnableBranch เพื่อเรียกใช้หลายสายงานพร้อมกัน หรือส่งอินพุตไปยังสายงานต่าง ๆ ตามเงื่อนไขแบบไดนามิก คุณปฏิบัติ AI Engineering Academy ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน AI Engineering Academy หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน AI Engineering Academy บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน

บทเรียน “สายงานแบบแยกแขนงและแบบขนาน” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน AI Engineering Academy นี้ได้ไหม

ได้ บทเรียน AI Engineering Academy ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

← กลับไปที่ AI Engineering Academy