6. LCELによる宣言的パイプライン

これまでの章では、LLMとのやり取りを調整するために命令型コードを書いてきました。プロンプトを作成し、モデルを呼び出し、レスポンスを解析する、という流れです。これは機能しますが、AIアプリケーションがより複雑になるにつれて、このアプローチは冗長になり、保守が困難になります。深くネストされた関数呼び出し、各ステップでの手動エラー処理、全体的なデータフローの理解の難しさに直面することになります。

LangChain Expression Language(LCEL)は、何を実行したいかを宣言することで、この問題を解決します。どのように実行するかではなく、何を実行したいかを宣言するのです。順次関数を呼び出す手続き型コードを書く代わりに、Unixパイプラインのように読めるシンプルなパイプ演算子(|)を使用してコンポーネントを組み合わせます。その結果、AIシステムを通るデータの流れを明確に表現する、よりクリーンで保守しやすいコードが得られます。

この章では、線形ワークフローを構築するためのLCELを紹介します。線形ワークフローとは、データが分岐やループなしに開始から終了まで流れる一連の操作です。LCELをいつ使用するか、パイプラインをどのように組み合わせるか、そして同期的にもストリーミング出力でも実行する方法について説明します。

6.1) なぜLCELなのか?

LCELが解決する問題

具体的な例から始めましょう。次のような処理が必要なカスタマーサポートアシスタントを構築しているとします:

ユーザーの質問を受け取る
システム指示とユーザーメッセージを含むプロンプトにフォーマットする
LLMに送信する
レスポンスを解析してテキストコンテンツのみを抽出する

LCELを使わずに命令型で書くと次のようになります:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # ステップ1: 質問を使ってプロンプトをフォーマットする
    messages = prompt.invoke({"question": question})
    
    # ステップ2: LLMを呼び出す
    response = llm.invoke(messages)
    
    # ステップ3: 出力を解析してテキストコンテンツを抽出する
    result = parser.invoke(response)
    
    return result
 
# 使用例
result = answer_question("How do I reset my password?")
print(result)

出力:

パスワードをリセットするには、次の手順に従ってください:
1. ログインページに移動します
2. 「パスワードを忘れた場合」をクリックします
3. メールアドレスを入力します
4. リセットリンクが記載されたメールを確認します
5. リンクをクリックして新しいパスワードを作成します

これは機能しますが、次のような問題があります:

冗長性: すべてのステップで明示的な変数の割り当てと関数呼び出しが必要です。実際のロジック(フォーマット → 呼び出し → 解析)がボイラープレートに埋もれています。

硬直的な構造: ステップを追加したい場合(質問の検証やレスポンスのログ記録など)、関数の途中にコードを挿入する必要があり、複雑さが増します。

組み込みストリーミングなし: トークンが到着するたびにストリーミングするには、llm.stream()を使用して非同期イテレーションを手動で処理するように関数全体を書き直す必要があります。

不明瞭なデータフロー: コードを読んでも、これがシンプルなパイプラインであることがすぐにはわかりません。変数の割り当てを追跡してフローを理解する必要があります。

では、LCEL版を見てみましょう:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# コンポーネントを定義する(前と同じ)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# パイプ演算子を使ってチェーンに組み合わせる
chain = prompt | llm | parser
 
# 使用例
result = chain.invoke({"question": "How do I reset my password?"})
print(result)

出力:

パスワードをリセットするには、次の手順に従ってください:
1. ログインページに移動します
2. 「パスワードを忘れた場合」をクリックします
3. メールアドレスを入力します
4. リセットリンクが記載されたメールを確認します
5. リンクをクリックして新しいパスワードを作成します

出力は同じですが、コードは劇的に異なります:

宣言的: chain = prompt | llm | parserが1行で全体のフローを表現しています。左から右に読みます: プロンプト → LLM → パーサー。

組み合わせ可能: 各コンポーネント(prompt、llm、parser)は独立しており、再利用可能です。パイプラインを書き直すことなくコンポーネントを交換できます。

ストリーミング組み込み: パイプライン定義を変更せずにchain.invoke()とchain.stream()を切り替えられます。

明確な意図: |演算子により、データフローが一目でわかります。

LCELが適合する場所: 線形ワークフロー

LCELは線形ワークフロー向けに設計されています。線形ワークフローとは、データが分岐やループなしに一方向に開始から終了まで流れるシーケンスです。

この線形パターンは多くのAIアプリケーションをカバーします。ドキュメントQ&Aシステムを考えてみましょう: 質問を受け取る → 関連ドキュメントを取得する → プロンプトをフォーマットする → LLMに送信する → 回答を解析する。各ステップは、あるステップの出力が次のステップの入力になる明確なシーケンスです。

しかし、ワークフローが次のようなことをする必要がある場合はどうでしょうか:

質問に基づいてLLMにどのツールを呼び出すか決定させる
ツールを呼び出し、結果を観察してから次に何をするか決定する
異なるアプローチで失敗した操作を再試行する

これらのシナリオにはループと条件分岐が必要です。LCELでは処理できないものです。これがLangGraphが存在する理由です(第15章で紹介します)。

2つのアプローチの視覚的な比較は次のとおりです:

LCELは、各ステップが前のステップの出力を処理する単純なパイプラインに最適です。データは一方向にのみ流れます。

LangGraphは、意思決定とループが必要な場合に使用します。エージェントは行動し、結果を観察し、再び考えることができます。

この章では、LCELに焦点を当てます。なぜ最初に線形パイプラインをマスターするのでしょうか?

基礎: パイプ演算子(|)はLangChainのコア構文です。これを理解すると、他のすべてが簡単になります。
LangGraphの前提条件: LangGraphエージェントは、ノード内でLCELチェーンを広範囲に使用します
実世界のパターン: 複雑なエージェントは、LCELチェーンを組み合わせることで構築されます

この章の残りの部分では、パイプ演算子(|)を使用してこれらの線形パイプラインを構築する方法を示します。

6.2) |演算子によるパイプラインの組み合わせ

LCELパイプの仕組み

LCELのパイプ演算子(|)は、Runnableインターフェースのおかげで機能します。

Runnableとは何か?

RunnableはLangChainの標準インターフェースです。コンポーネントがRunnableインターフェースを実装すると、|演算子を使用して他のコンポーネントとチェーン化できます。

すべてのRunnableは次のメソッドを提供します:

.invoke(input) - 一度実行して完全な結果を取得する
.stream(input) - 実行してLLMがトークンを生成するたびに各単語を受け取る
.batch(inputs) - 異なる入力で複数回実行してすべての結果を取得する

この章では、.invoke()と.stream()に焦点を当てます(.batch()については、必要に応じて後で説明します)。

なぜ|が機能するのか?

Runnableクラスが、Pythonの演算子オーバーロードを使用して|演算子を実装しているためです。prompt | llmと書くと、2つのコンポーネントを順次実行する新しいRunnableが作成されます。

ほとんどのLangChainコンポーネントはRunnableです

これが、非常に多くの異なるコンポーネントをチェーン化できる理由です:

ChatPromptTemplateはRunnableです
ChatOpenAIはRunnableです
StrOutputParserはRunnableです
|で作成したカスタムチェーンも、それ自体がRunnableです!

これは、より単純なパイプラインを組み合わせて複雑なパイプラインを構築できることを意味します。

コンポーネントの接続: 入力/出力型

|でコンポーネントを接続する場合、あるコンポーネントの出力型が次のコンポーネントの入力型と一致することを確認する必要があります。

主要なコンポーネントのシグネチャ:

コンポーネント	入力型	出力型
`ChatPromptTemplate`	`dict`	`list[BaseMessage]`
`ChatOpenAI` (LLM)	`list[BaseMessage]`	`AIMessage`
`StrOutputParser`	`AIMessage`	`str`

フローの例:

python

chain = prompt | llm | parser

各コンポーネントを通過する際のデータ型の変換は次のとおりです:

prompt: dictを入力として受け取り、list[BaseMessage]に変換します
llm: list[BaseMessage]を入力として受け取り、AIMessageに変換します
parser: AIMessageを入力として受け取り、strに変換します

実際の動作を見てみましょう:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "What is 2+2?"})
print(result)  # "2+2は4です。"

各ステップで何が起こるか:

ステップ	入力	コンポーネント	出力
1	`- dict -` `{"question": "What is 2+2?"}`	→ prompt →	`- list[BaseMessage] -` `[SystemMessage(...), HumanMessage(...)]`
2	`- list[BaseMessage] -` `[SystemMessage(...), HumanMessage(...)]`	→ llm →	`- AIMessage -` `AIMessage(content="2+2は4です。")`
3	`- AIMessage -` `AIMessage(content="2+2は4です。")`	→ parser →	`- str -` `"2+2は4です。"`

型が一致しない場合はどうなるか?

互換性のないコンポーネントを接続しようとすると、エラーが発生します:

python

# エラー: これは機能しません
chain = llm | prompt  # LLMはAIMessageを出力しますが、promptは入力としてdictを必要とします

エラーメッセージには、次のコンポーネントが期待する入力型と実際に受け取った入力型が表示されます。

6.3) チェーンの実行: .invoke()と.stream()

パイプラインの実行

チェーンを組み合わせたら、.invoke()メソッドを使用して実行します。これは、パイプラインを実行する同期的な方法です。結果を返す前に、完全なレスポンスを待ちます。

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# 同期呼び出し
result = chain.invoke({"question": "What is 2+2?"})
print(result)

出力:

2+2は4です。

.invoke()メソッドは単純です: 最初のコンポーネントが期待する入力パラメータを渡すと、最後のコンポーネントが生成した出力が返されます。

同じチェーンからのトークンのストリーミング

.invoke()はシンプルですが、ユーザー向けアプリケーションには制限があります。完全なレスポンスが完了するまで、ユーザーには何も表示されません。長いレスポンス(10〜20秒)の場合、これはユーザーエクスペリエンスが悪くなります。

ストリーミングは、完全なレスポンスを待つのではなく、生成されたトークンをすぐに表示します。これは、ChatGPTで見られるタイピング効果です。

ストリーミングの動作を見てみましょう:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# 到着したチャンクをストリーミングする
for chunk in chain.stream({"question": "Explain what Python is in one sentence"}):
    print(chunk, end="", flush=True)

出力(リアルタイムで、トークンごとに表示されます):

Pythonは、シンプルさと可読性で知られる汎用性の高い高水準プログラミング言語で、Web開発、データサイエンス、自動化で広く使用されています。

チェーン定義は上記の.invoke()の例と同じであることに注意してください。再構築する必要はありませんでした。.invoke()の代わりに.stream()を呼び出しただけです。

.stream()メソッドは、LLMがトークンを生成する間、結果をチャンクとして返します。print(chunk, end="", flush=True)は、各チャンクを画面にすぐに表示し、ライブタイピング効果を作成します。

.invoke()と.stream()をいつ使用するか

.invoke()を使用する場合:

最終結果のみが必要な場合(分析、翻訳、分類)
レスポンスが短く、待ち時間が問題にならない場合
次のステップに進む前に完全な出力が必要な場合

.stream()を使用する場合:

ユーザーが進捗を確認する必要がある場合(チャット、コンテンツ生成)
レスポンスが長く、待ち時間が目立つ場合
リアルタイムフィードバックが重要なUIを構築している場合

どちらのメソッドも同じチェーンで機能します。チェーンを一度定義してから、ニーズに基づいて実行モードを選択します。

この章では、LangChainの宣言的パイプライン構文であるLCELを学習しました:

パイプラインの構築: パイプ演算子を使用します: prompt | llm | parser
柔軟な実行: 完全な結果には.invoke()を、リアルタイム出力には.stream()を使用します
型安全性: コンポーネントをチェーン化する際に、出力型を入力型に一致させます

同じチェーンが両方の実行モードで機能します。一度定義すれば、どこでも使用できます。

次へ: 第7章では、Pydanticを使用した構造化出力について説明し、LLMレスポンスから検証済みのJSONデータを抽出できるようにします。