第3章 LCEL

1. LCELとは

LCEL（LangChain Expression Language）は、LangChainでチェーンを構築するための宣言的な記法である。従来のLegacyChain（LLMChain、SequentialChainなど）に代わり、より柔軟で直感的なチェーン構築を可能にする。LangChain v0.1以降、LCELが推奨される標準的な方法となっている。

1.1 設計思想

LCELの設計思想は以下の3点に集約される。

Table 1. LCELの設計原則

LCELを使用することで、プロンプト、モデル、出力パーサーなどの各コンポーネントを自由に組み合わせてチェーンを構築できる。また、構築したチェーンは自動的にストリーミングと非同期処理に対応する。

1.2 パイプ演算子

LCELの最も特徴的な要素は、パイプ演算子（|）によるチェーンの構築である。Unix/Linuxのパイプと同様に、前の処理の出力が次の処理の入力となる。

Fig 1. パイプ演算子によるチェーン構築

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 各コンポーネントの定義
prompt = ChatPromptTemplate.from_template(
    "{topic}について3つのポイントで説明してください。"
)
model = ChatOpenAI(model="gpt-4o", temperature=0)
parser = StrOutputParser()

# パイプ演算子でチェーンを構築
chain = prompt | model | parser

# チェーンの実行
result = chain.invoke({"topic": "機械学習"})
print(result)

このコードでは、prompt → model → parser という順序で処理が流れる。promptは辞書を受け取りChatPromptValueを出力、modelはChatPromptValueを受け取りAIMessageを出力、parserはAIMessageを受け取り文字列を出力する。

Fig 2. チェーンのデータフロー

{"topic": "機械学習"}
    ↓ prompt
ChatPromptValue(messages=[HumanMessage(...)])
    ↓ model
AIMessage(content="...")
    ↓ parser
"機械学習について3つのポイント..."（文字列）

2. Runnable

Runnableは、LCELの基盤となるインターフェースである。LangChainのほぼすべてのコンポーネント（プロンプト、モデル、パーサー、リトリーバーなど）はRunnableを実装しており、統一された方法で呼び出し・組み合わせができる。

2.1 RunnableSequence

RunnableSequenceは、複数のRunnableを順番に実行するクラスである。パイプ演算子（|）で連結すると、自動的にRunnableSequenceが生成される。

Fig 3. RunnableSequenceの明示的な構築

from langchain_core.runnables import RunnableSequence

# パイプ演算子を使う方法（推奨）
chain1 = prompt | model | parser

# RunnableSequenceを直接使う方法（同等）
chain2 = RunnableSequence(first=prompt, middle=[model], last=parser)

# どちらも同じ結果
result1 = chain1.invoke({"topic": "Python"})
result2 = chain2.invoke({"topic": "Python"})

通常はパイプ演算子を使用するが、プログラムで動的にチェーンを構築する場合はRunnableSequenceを直接使用することもある。

2.2 RunnableParallel

RunnableParallelは、複数のRunnableを並列に実行し、結果を辞書として返すクラスである。複数の処理を同時に実行したい場合や、入力データを複数の経路に分岐させたい場合に使用する。

Fig 4. RunnableParallelの使用例

from langchain_core.runnables import RunnableParallel

# 2つの異なるプロンプトを並列実行
summary_prompt = ChatPromptTemplate.from_template(
    "{text}を50文字で要約してください。"
)
keyword_prompt = ChatPromptTemplate.from_template(
    "{text}からキーワードを3つ抽出してください。"
)

# 並列チェーンの構築
parallel_chain = RunnableParallel(
    summary=summary_prompt | model | parser,
    keywords=keyword_prompt | model | parser
)

# 実行（両方の処理が並列に実行される）
result = parallel_chain.invoke({
    "text": "LangChainはLLMアプリケーション開発のためのフレームワークです..."
})

print(result["summary"])   # 要約
print(result["keywords"])  # キーワード

辞書リテラルを使用して、より簡潔に記述することもできる。

Fig 5. 辞書による並列チェーン

# 辞書リテラルで記述（同等）
parallel_chain = {
    "summary": summary_prompt | model | parser,
    "keywords": keyword_prompt | model | parser
}

2.3 RunnableLambda

RunnableLambdaは、任意のPython関数をRunnableとしてラップするクラスである。LCELのチェーンに独自の処理を組み込む際に使用する。

Fig 6. RunnableLambdaの使用例

from langchain_core.runnables import RunnableLambda

# カスタム処理をRunnableにラップ
def format_output(text: str) -> str:
    """出力を整形する"""
    lines = text.strip().split("\n")
    return "\n".join(f"• {line}" for line in lines if line)

formatter = RunnableLambda(format_output)

# チェーンに組み込み
chain = prompt | model | parser | formatter

result = chain.invoke({"topic": "デザインパターン"})

デコレータを使用して、より簡潔に定義することもできる。

Fig 7. @chainデコレータの使用

from langchain_core.runnables import chain

@chain
def process_and_format(input_dict: dict) -> str:
    """入力を処理して整形する"""
    topic = input_dict["topic"]
    # 何らかの処理
    return f"処理結果: {topic}"

# そのままチェーンに組み込める
full_chain = process_and_format | model | parser

3. 実践パターン

LCELを使用した実践的なパターンをいくつか紹介する。

Fig 8. 条件分岐（RunnableBranch）

from langchain_core.runnables import RunnableBranch

# 入力に基づいて異なるチェーンを実行
branch = RunnableBranch(
    # (条件関数, 実行するチェーン) のタプルのリスト
    (lambda x: x["type"] == "summary", summary_chain),
    (lambda x: x["type"] == "translate", translate_chain),
    # デフォルト（条件に一致しない場合）
    default_chain
)

result = branch.invoke({"type": "summary", "text": "..."})

Fig 9. 値の受け渡し（RunnablePassthrough）

from langchain_core.runnables import RunnablePassthrough

# 入力をそのまま出力に渡しつつ、追加処理も行う
chain = RunnablePassthrough.assign(
    # 元の入力に "processed" キーを追加
    processed=lambda x: x["text"].upper()
) | prompt | model | parser

# 入力: {"text": "hello"}
# assign後: {"text": "hello", "processed": "HELLO"}

Fig 10. 変数のバインド（bind）

# モデルのパラメータを固定
model_with_params = model.bind(
    temperature=0,
    max_tokens=500
)

# ツールをバインド
model_with_tools = model.bind_tools([search_tool, calculator_tool])

chain = prompt | model_with_tools | parser

4. デバッグと最適化

LCELチェーンのデバッグと最適化について解説する。

Fig 11. チェーンの可視化

# チェーンの構造を確認
chain = prompt | model | parser
print(chain)
# RunnableSequence(first=ChatPromptTemplate(...), middle=[...], last=...)

# 入力/出力スキーマの確認
print(chain.input_schema.schema())
print(chain.output_schema.schema())

Fig 12. with_configによる設定

# 実行時の設定
result = chain.invoke(
    {"topic": "Python"},
    config={
        "callbacks": [callback_handler],  # コールバック
        "tags": ["production"],           # タグ付け
        "metadata": {"user_id": "123"}    # メタデータ
    }
)

Fig 13. ストリーミング出力

# ストリーミングでトークン単位に出力
for chunk in chain.stream({"topic": "AI"}):
    print(chunk, end="", flush=True)

Fig 14. バッチ処理

# 複数の入力を並列処理
inputs = [
    {"topic": "Python"},
    {"topic": "JavaScript"},
    {"topic": "Go"}
]

# max_concurrencyで並列数を制御
results = chain.batch(inputs, config={"max_concurrency": 3})

LangSmith（後述）と組み合わせることで、チェーンの各ステップの入出力、レイテンシ、トークン使用量などを詳細にトレースできる。本番環境でのデバッグや最適化に必須のツールである。