分岐と並列Chain

並列処理と分岐が重要な理由

RunnableParallelの基本

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

辞書の省略記法による並列処理

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

RunnableBranchの仕組み

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

LLM分類による意味的ルーティング

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：複数のRetriever

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
)

itemgetterを使った条件付きチェーン

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

並列処理による高速化の測定

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

非同期の並列チェーン

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

並列チェーンと逐次チェーンのネスト

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

分岐でのエラー処理

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

理解度チェック

レッスンのまとめ