専門家一覧
LangChain Expression Language (LCEL) を活用した複雑なAIチェーンの設計

LangChain LCELで描くAIアーキテクチャ：手続き型から宣言型へ、UNIX思想で組む堅牢なチェーン設計

この記事は急速に進化する技術について解説しています。最新情報は公式ドキュメントをご確認ください。

約10分で読めます
文字サイズ:
LangChain LCELで描くAIアーキテクチャ：手続き型から宣言型へ、UNIX思想で組む堅牢なチェーン設計
目次

この記事の要点

  • 手続き型から宣言型へのAIチェーン設計移行
  • UNIX思想に基づく堅牢なチェーン構築
  • RAGや動的ルーティングの実装パターン

LangChain Expression Language（LCEL）は、AI開発におけるパラダイムシフトと言えるでしょう。手続き型（How）から宣言型（What）への思考の転換を促し、UNIXパイプの思想を受け継ぐこの記述法をマスターすれば、AIコードはシンプルかつ堅牢なものに進化します。ビジネスの現場で「まず動くものを作る」高速プロトタイピングを実現するためにも、この技術の本質を理解することは非常に重要です。

1. なぜ今、LCEL（LangChain Expression Language）なのか

従来のLangChain（Chainクラスベースの古いスタイル）は、オブジェクト指向的なアプローチが強く、カスタマイズの際には独自のクラス継承や複雑なラッパーが必要になるケースが多々ありました。これは単純なチャットボット程度なら問題ありませんが、GraphRAG（グラフ構造を用いた検索拡張生成）やマルチモーダルデータを取り扱う現代の高度なRAGシステムにおいては、構造的な限界に直面します。

「手続き型」Chainの限界とスパゲッティコードのリスク

従来の LLMChain や SequentialChain といったクラスは、処理の流れを隠蔽（ブラックボックス化）する傾向がありました。「何が入力され、内部でどう変換され、何が出力されているのか」を正確に追跡するには、ライブラリのソースコード深部まで潜る必要が生じます。

特に、近年のAIアプリケーションで求められる以下の要件を満たそうとすると、手続き型のコードは急速に複雑化し、いわゆる「スパゲッティコード」に陥るリスクが高まります。

  • 複雑なルーティング: クエリの内容に応じて、検索対象のデータソース（ベクトルDB、グラフDB、Web検索など）を動的に切り替える。
  • マルチモーダル対応: テキストだけでなく、画像や図表を含むドキュメントを統合的に処理する。
  • 非同期・並列処理: 複数の検索や生成タスクを同時に走らせ、レイテンシを最小化する。

UNIXパイプの思想を受け継ぐ「宣言的」記述のメリット

LCELは、UNIXのパイプライン処理（input | process1 | process2 | output）に強くインスパイアされた設計思想を持っています。これは「どのように処理を実行するか（How）」ではなく、「データがどう流れるか（What）」を定義する宣言的なアプローチです。

複雑化するAIアーキテクチャにおいて、LCELを採用するメリットは明確です：

  1. 可読性と透明性: 処理の流れが左から右へと直感的に記述でき、データフローが一目瞭然になります。
  2. プロダクション品質の標準装備: 定義したチェーンは、追加のコードなしでストリーミング（.stream()）、非同期処理（.ainvoke()）、バッチ処理（.batch()）を自動的にサポートします。
  3. コンポーザビリティ（構成可能性）: 小さな部品（Runnable）を組み合わせることで、GraphRAGのような複雑なパイプラインも、管理可能な単位で構築できます。

Runnableプロトコル：共通インターフェースの理解

LCELの根幹にあるのが「Runnableプロトコル」です。これは、すべてのLangChainコンポーネント（LLM、プロンプト、パーサー、リトリーバー、そしてカスタム関数など）が実装している共通のインターフェースです。

最新のLangChainエコシステムでは、すべてのRunnableが以下の標準メソッドを提供しています：

  • invoke: 単一の入力に対する呼び出し
  • batch: 入力リストに対する並列呼び出し（効率的な処理に不可欠）
  • stream: チャンクごとのストリーミング出力（UX向上に必須）

この共通規格により、テキスト生成モデルであれ、画像解析モデルであれ、あるいは検索リランカーであれ、あらゆるコンポーネントを統一的な「パイプ」で繋ぐことが可能になります。これが、変化の激しいAI技術トレンドに追従し、仮説を即座に形にして検証できる柔軟なアーキテクチャの基盤となるのです。

2. LCELの基本文法と「パイプ」の魔術

ここでは、Pythonのビット演算子である |（パイプ）が、LangChainの世界でどのようにデータを運ぶのかを見ていきます。UNIXのパイプライン処理と同様に、あるプロセスの出力を次のプロセスの入力としてシームレスに接続する設計思想がここにあります。

基本の「|」演算子：Prompt + LLM + OutputParser

最も基本的な構成は、プロンプトに入力を流し込み、それをLLM（大規模言語モデル）に渡し、最後に文字列として出力するパターンです。

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

# 1. モデルの準備
# ※利用可能な最新モデルを指定してください（例: "ChatGPT"など）
model = ChatOpenAI(model="ChatGPT")

# 2. プロンプトの定義
prompt = ChatPromptTemplate.from_template(
    "{topic}について、簡潔に3行で説明してください。"
)

# 3. 出力パーサーの定義（AIMessageオブジェクトを文字列に変換）
output_parser = StrOutputParser()

# 4. LCELによるチェーンの構築
# データの流れ: 辞書型入力 -> プロンプト -> LLM -> 文字列出力
chain = prompt | model | output_parser

# 5. 実行
# 入力データ: {"topic": "量子コンピュータ"}
result = chain.invoke({"topic": "量子コンピュータ"})

print(result)
# 出力例:
# 量子コンピュータは、量子力学の重ね合わせや量子もつれを利用して計算を行う次世代のコンピュータです。
# 従来のコンピュータでは解読困難な問題を高速に処理できる可能性を秘めています。
# 創薬、材料探索、暗号解読など、多岐にわたる分野での応用が期待されています。

この | 演算子は、左側のコンポーネント（Runnable）の出力を、右側のコンポーネントの入力として自動的に渡す役割を果たします。手続き型のコードを書くことなく、宣言的にデータフローを定義できる点が大きな利点です。

RunnablePassthroughとitemgetter：データの受け渡し

開発現場でよく直面するのが、「複数の引数をプロンプトに渡したい」あるいは「入力データをそのまま次のステップにも渡したい」というケースです。ここで活躍するのが RunnablePassthrough と itemgetter です。

例えば、RAG（検索拡張生成）の初期段階を考えてみましょう。入力された質問を使って検索を行い、その結果と元の質問の両方をプロンプトに埋め込む必要があります。

from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough

# 仮のリトリーバー関数（本来はVectorStoreなどを使用）
def fake_retriever(query: str):
    return f"【検索結果】{query}に関する社内ドキュメント..."

# 複雑なプロンプト
template = """
以下の文脈に基づいて質問に答えてください。

文脈:
{context}

質問:
{question}
"""
prompt = ChatPromptTemplate.from_template(template)

# RAGチェーンの構築
rag_chain = (
    # 入力された辞書（例: {"question": "..."}）を受け取り並列処理
    {
        # "context"キーには、"question"の値を使って検索した結果を入れる
        "context": itemgetter("question") | fake_retriever,
        # "question"キーには、入力の"question"をそのままパスする
        "question": itemgetter("question")
    }
    | prompt
    | model
    | StrOutputParser()
)

# 実行
print(rag_chain.invoke({"question": "有給休暇の申請方法は？"}))

ここで重要なのは、最初の辞書定義 { ... } 自体が RunnableParallel として機能している点です。これにより、入力データから必要な情報を抽出し、加工して次のプロンプトへ渡すフローを記述できます。データ構造の変換をチェーンの中に組み込むことで、コードの見通しが劇的に良くなります。

型安全性と入力スキーマの定義

LCELはPydanticと深く統合されています。チェーンの入力スキーマを定義することで、型安全性を確保し、LangServeなどでAPIとして公開する際のバリデーションを自動化できます。

from langchain_core.pydantic_v1 import BaseModel, Field

# 入力スキーマの定義
class TopicInput(BaseModel):
    topic: str = Field(description="解説してほしいトピック")

# チェーンに型情報を付与
chain_with_type = chain.with_types(input_type=TopicInput)

# これにより、LangServeなどでAPI化する際に自動的にOpenAPIドキュメントが生成されます

このように型情報を付与することで、開発時の入力ミスを防ぐだけでなく、運用時の信頼性も向上します。大規模な業務システムにおいて、データの整合性を保つための重要なプラクティスです。

3. 実践パターンA：RunnableParallelによる並列処理の最適化

2. LCELの基本文法と「パイプ」の魔術 - Section Image

システム全体を捉えると、AIアプリケーションのボトルネックは「LLMの待ち時間」や「外部APIのレイテンシ」にある場合があります。これらを直列（Sequential）に処理するのは、ユーザー体験を損なう要因です。

LCELの RunnableParallel を使えば、依存関係のないタスクを並列に実行し、全体の処理時間を短縮できます。ビジネスへの最短距離を描く上で、パフォーマンスの最適化は欠かせません。

直列処理のボトルネックと並列化の必要性

例えば、「製品の長所」と「短所」を別々に分析して統合するタスクを考えます。これを順番に行うと、(長所生成時間) + (短所生成時間) がかかります。並列化すれば、より遅い方の処理時間だけで済みます。

コンテキスト取得と質問生成を同時に走らせるRAG

より高度なRAGのシナリオでは、「検索（Retrieval）」と同時に「クエリの言い換え（Query Expansion）」や「関連トピックの生成」を行いたい場合があります。

from langchain_core.runnables import RunnableParallel

# 長所分析用チェーン
pros_chain = (
    ChatPromptTemplate.from_template("{product}の長所を3つ挙げて")
    | model
    | StrOutputParser()
)

# 短所分析用チェーン
cons_chain = (
    ChatPromptTemplate.from_template("{product}の短所を3つ挙げて")
    | model
    | StrOutputParser()
)

# 並列実行チェーンの定義
analysis_chain = RunnableParallel(
    pros=pros_chain,
    cons=cons_chain
)

# 実行
# 入力 {"product": "スマートフォン"} が両方のチェーンに同時に渡される
result = analysis_chain.invoke({"product": "最新の折りたたみスマホ"})

print(f"--- 長所 ---\n{result['pros']}")
print(f"--- 短所 ---\n{result['cons']}")

実行時間の短縮効果とコード実装

RunnableParallel（または辞書による暗黙的な定義）を使用すると、Pythonの asyncio やスレッドプールが活用され、効率的にタスクが処理されます。特に外部検索APIやデータベースアクセスを含む処理において、この設計パターンはパフォーマンス向上の鍵となります。

4. 実践パターンB：RunnableBranchによる動的ルーティング

ユーザーの入力は予測不可能です。ある時は「コードを書いて」と言い、ある時は「詩を書いて」と言います。これら全てを単一のプロンプトで処理しようとすると、プロンプトが肥大化し、精度が落ちる可能性があります。

そこで必要なのが、入力内容に応じて処理ルートを切り替える「動的ルーティング」です。従来はPythonの if-else 文で書いていましたが、LCELでは RunnableBranch を使ってチェーンの中にロジックを組み込めます。

ユーザーの意図に応じたチェーンの切り替え

ここでは、入力トピックに応じて「技術解説モード」と「クリエイティブモード」を切り替える例を見てみましょう。

from langchain_core.runnables import RunnableBranch

# 1. 分類用チェーン（ルーター）
# ユーザーの入力を 'tech', 'creative', 'other' のいずれかに分類
classification_chain = (
    ChatPromptTemplate.from_template(
        "以下の入力を 'tech', 'creative', 'other' のいずれかに分類して。出力は分類ラベルのみ。\n入力: {input}"
    )
    | model
    | StrOutputParser()
)

# 2. 各専用チェーンの定義
tech_chain = (
    ChatPromptTemplate.from_template("エンジニア向けに技術的に解説して: {input}")
    | model
    | StrOutputParser()
)

creative_chain = (
    ChatPromptTemplate.from_template("詩的な表現で物語風に書いて: {input}")
    | model
    | StrOutputParser()
)

other_chain = (
    ChatPromptTemplate.from_template("普通に答えて: {input}")
    | model
    | StrOutputParser()
)

# 3. ブランチ（分岐）の定義
# (条件, 実行するチェーン) のタプルリストと、デフォルトチェーンを指定
branch = RunnableBranch(
    (lambda x: "tech" in x["topic"].lower(), tech_chain),
    (lambda x: "creative" in x["topic"].lower(), creative_chain),
    other_chain  # デフォルト
)

# ※注: 上記はlambdaで簡易判定していますが、実運用では前段のclassification_chainの結果を使います。
# より実用的なパイプライン構成:

full_chain = (
    {
        "topic": classification_chain,
        "input": itemgetter("input")
    }
    | branch
)

# 実行例
# full_chain.invoke({"input": "再帰ニューラルネットワークの仕組み"}) -> tech_chainへ
# full_chain.invoke({"input": "雨の日の静かな森"}) -> creative_chainへ

分類タスクと実行タスクの分離

このパターンのポイントは、「判断するAI」と「実行するAI」を分けることです。単一のLLMにすべてをやらせるのではなく、軽量なモデルで分類を行い、専門的なプロンプトを持つチェーンへ振り分けることで、コスト削減と精度向上を両立できる可能性があります。経営者視点で見ても、リソースの最適化は非常に重要です。

5. 複雑なチェーンのデバッグと可観測性

4. 実践パターンB：RunnableBranchによる動的ルーティング - Section Image

LCELでチェーンを組んでいると、「パイプの中でデータがどう変形したか見えない」という問題に直面することがあります。宣言的記述の特性によるものです。しかし、LangChainには可観測性（Observability）ツールが用意されています。

chain.get_graph().print_ascii()による構造可視化

まず、構築したチェーンがどのような構造になっているかを確認しましょう。LCELオブジェクトはグラフ構造を持っています。

# チェーンの構造をアスキーアートで表示
chain.get_graph().print_ascii()

これを実行すると、コンソールにデータの流れが図示されます。分岐や並列処理が意図通りに接続されているかを視覚的にチェックできます。

LangSmithを活用したトレースの可視化

LangSmithを利用することで、チェーン実行のトレース（追跡）を記録できます。

  • 各ステップの入出力データ
  • 実行にかかった時間（レイテンシ）
  • 消費したトークン数
  • 発生したエラーの詳細

これらがWeb UI上で可視化されるため、「どのステップで意図しないデータ変換が起きたか」を特定できます。複雑なLCELチェーンを運用する場合に役立ちます。

ユニットテストの書きやすさとLCEL

LCELは関数型プログラミングの性質を持つため、テストが容易です。各Runnableは独立して動くため、個別のコンポーネントごとにユニットテストを書くことができます。

まとめ

短所分析用チェーン - Section Image 3

LCELは、LangChainを「便利ツール」から「開発フレームワーク」へと進化させるアップデートです。

  • 宣言的記述: | 演算子でデータの流れを明確にする。
  • 並列処理: RunnableParallel でパフォーマンスを最大化する。
  • 動的制御: RunnableBranch で柔軟なロジックを組む。
  • 可観測性: グラフ可視化とLangSmithでブラックボックスを排除する。

手続き型のコードから脱却し、宣言的なパイプラインを設計することは、最初は少し頭の切り替えが必要かもしれません。しかし、慣れてしまえば、その見通しの良さと拡張性を実感できるはずです。皆さんもぜひ、この新しいパラダイムを取り入れ、次世代のAIエージェント開発に挑戦してみてください。何か疑問があれば、ぜひ議論を深めていきましょう。

LangChain LCELで描くAIアーキテクチャ：手続き型から宣言型へ、UNIX思想で組む堅牢なチェーン設計 - Conclusion Image

コメント

コメントは1週間で消えます
0 / 150
コメントを読み込み中...