【2026年最新】LangChain LCEL完全ガイド｜Pythonで作るAIチェーン導入5ステップ

「LangChainを使いたいのに、どこから始めればいいかわからない」「LLMChainが非推奨と言われたが、代替の書き方が理解できない」——こうした悩みを持つ開発者は2026年現在も多い。

本記事では、LangChain Expression Language（LCEL）を使った現代的なチェーン構築を、5ステップのハンズオン形式で解説する。LLMChainなどのレガシー構文は使わず、2026年時点の公式推奨アーキテクチャのみで解説する。コードは全てコピーペーストで動作する。

LangChain LCELとは何か — 旧アーキテクチャとの違い

LangChain Expression Language（LCEL）は、2023年8月にLangChainが発表した新しいチェーン構築インターフェースだ。2026年現在、LLMChain・SequentialChainなどのレガシークラスはlangchain-classicパッケージに移行され、公式の新規開発対象から除外されている。現在も動きはするが、新規プロジェクトで使うべきではない。

LCELの核心は「パイプ演算子（|）による宣言的チェーン構築」だ。

旧来コードとLCELの比較

# 旧来（LLMChain — 非推奨・langchain-classicに移行）
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

llm = OpenAI()
prompt = PromptTemplate(input_variables=["product"], template="{product}のキャッチコピーを考えてください。")
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run("スマートスピーカー")

# LCEL（現代的な書き方 — 公式推奨）
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_template("{product}のキャッチコピーを考えてください。")
parser = StrOutputParser()

# パイプ演算子でチェーンを構築
chain = prompt | llm | parser
result = chain.invoke({"product": "スマートスピーカー"})
print(result)

行数はほぼ同じだが、LCELには以下の決定的なメリットがある。

• ストリーミングが自動対応：chain.stream() を呼ぶだけで逐次出力が可能
• 非同期が無料：chain.ainvoke() / chain.astream() が即座に使える
• バッチ処理が簡単：chain.batch([...]) でリストを一括処理
• LangSmithトレースが自動連携：デバッグが格段に楽になる
• 並列処理が宣言的に書ける：RunnableParallelで複数ブランチを同時実行

環境構築 — インストールから動作確認まで

LangChainはv1.0（2025年10月正式リリース）以降、パッケージが役割ごとに分割されている。最小構成で始めるには以下の3パッケージで十分だ。

# 基本パッケージ（最小構成）
pip install langchain langchain-core langchain-openai

# RAG・ベクトルストアも使う場合
pip install langchain langchain-core langchain-openai langchain-community chromadb

# LangGraph（エージェント・ステート管理）も使う場合
pip install langchain langchain-core langchain-openai langgraph

# Python 3.10以上が必要（v1.0でPython 3.9サポート終了）
python --version  # Python 3.10.x 以上を確認

次にAPIキーを設定する。

import os
os.environ["OPENAI_API_KEY"] = "sk-..."  # 本番では.envファイルから読み込む

# python-dotenv推奨（pip install python-dotenv）
from dotenv import load_dotenv
load_dotenv()  # .envファイルを読み込む

5ステップ実装フロー — LCELハンズオン

以下の5ステップで「プロンプト→LLM→出力パーサー」の基本チェーンから、並列処理・ストリーミング・RAGまで段階的に習得する。

1. Step 1：基本チェーンを作る（Runnable、パイプ演算子の理解）
2. Step 2：ストリーミング出力を実装する（UX向上・リアルタイム応答）
3. Step 3：RunnablePassthroughで変数を引き回す（複雑なパイプラインの基礎）
4. Step 4：RunnableParallelで並列処理する（処理の高速化）
5. Step 5：RAGチェーンを構築する（実用的なユースケース）

Step 1：基本チェーンを作る

LCELの最小単位は「プロンプトテンプレート → LLM → 出力パーサー」の3要素だ。それぞれがRunnableインターフェースを実装しており、パイプ演算子（|）で接続できる。

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

# 1. 各コンポーネントを定義
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは技術記事を書くプロのライターです。"),
    ("human", "{topic}について、エンジニア向けに200字で説明してください。"),
])

parser = StrOutputParser()

# 2. パイプ演算子でチェーンを組み立てる
chain = prompt | model | parser

# 3. 実行
result = chain.invoke({"topic": "LangChain LCEL"})
print(result)
# → 「LangChain Expression Language（LCEL）は...」のような出力が得られる

ここで重要なのが「全ての部品がRunnableである」という点だ。Runnableは以下の統一インターフェースを持つ。

• runnable.invoke(input)：同期・単一入力
• runnable.stream(input)：同期・ストリーミング
• runnable.batch([input1, input2, ...])：同期・バッチ
• await runnable.ainvoke(input)：非同期・単一入力
• runnable.astream(input)：非同期ストリーミング

Step 2：ストリーミング出力を実装する

チャットUIや長文生成では、全体の完了を待たずに逐次表示（ストリーミング）が必要だ。LCELではチェーン定義を変えずに .stream() に切り替えるだけでよい。

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

model = ChatOpenAI(model="gpt-4o-mini", streaming=True)
prompt = ChatPromptTemplate.from_template("Pythonで{task}する方法を詳しく説明してください。")
chain = prompt | model | StrOutputParser()

# ストリーミング実行 — チェーン定義はinvokeと全く同じ
print("生成中...\n")
for chunk in chain.stream({"task": "CSVファイルを読み込んで集計"}):
    print(chunk, end="", flush=True)
print("\n\n完了")

# 非同期ストリーミング（FastAPI等での利用）
import asyncio

async def stream_async():
    async for chunk in chain.astream({"task": "APIレスポンスをキャッシュ"}):
        print(chunk, end="", flush=True)

asyncio.run(stream_async())

Step 3：RunnablePassthroughで変数を引き回す

チェーンが長くなると「前のステップの入力値を後のステップでも使いたい」ケースが出てくる。そこで使うのが RunnablePassthrough だ。

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

model = ChatOpenAI(model="gpt-4o-mini")

# Step1：質問からキーワードを抽出するチェーン
keyword_prompt = ChatPromptTemplate.from_template(
    "以下の質問から、検索に使うキーワードを3つ抽出してください。質問: {question}"
)
keyword_chain = keyword_prompt | model | StrOutputParser()

# Step2：元の質問とキーワードを両方使って回答するチェーン
answer_prompt = ChatPromptTemplate.from_template(
    "質問: {question}\n検索キーワード: {keywords}\n上記の質問に詳しく答えてください。"
)
answer_chain = answer_prompt | model | StrOutputParser()

# RunnablePassthroughで元の質問を保持しながら、キーワードを追加する
full_chain = (
    {"keywords": keyword_chain, "question": RunnablePassthrough()}
    | answer_chain
)

result = full_chain.invoke("LangChainのLCELはどんな場合に使うべきですか？")
print(result)

Step 4：RunnableParallelで並列処理する

複数の処理を独立して実行できる場合は、RunnableParallelで並列化することでレスポンスタイムを大幅に短縮できる。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")

# 2つのチェーンを並列定義
summary_chain = (
    ChatPromptTemplate.from_template("{topic}を3行で要約してください。")
    | model
    | StrOutputParser()
)

usecase_chain = (
    ChatPromptTemplate.from_template("{topic}の主な活用事例を3つ箇条書きで挙げてください。")
    | model
    | StrOutputParser()
)

# RunnableParallelで同時実行（逐次実行より約50%高速）
parallel_chain = RunnableParallel(
    summary=summary_chain,
    usecases=usecase_chain,
    topic=RunnablePassthrough()  # 元のtopicもそのまま保持
)

result = parallel_chain.invoke({"topic": "LangChain LCEL"})
print("=== 要約 ===")
print(result["summary"])
print("\n=== 活用事例 ===")
print(result["usecases"])

Step 5：RAGチェーンを構築する

LCELの最も代表的なユースケースが RAG（Retrieval-Augmented Generation）チェーンだ。ドキュメントを検索して、その内容をコンテキストとしてLLMに渡すパターンを実装する。

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader

# 1. ドキュメントを読み込んでチャンク分割
loader = TextLoader("./company_docs.txt", encoding="utf-8")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = splitter.split_documents(docs)

# 2. ベクトルストアを作成
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(splits, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 3. RAGプロンプトを定義
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "以下のコンテキストのみを使って質問に答えてください。コンテキスト: {context}"),
    ("human", "{question}"),
])

# 4. LCELでRAGチェーンを組み立て
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# 5. 実行
answer = rag_chain.invoke("弊社の返金ポリシーはどうなっていますか？")
print(answer)

# ストリーミングも同様に使える
for chunk in rag_chain.stream("弊社のサポート時間を教えてください"):
    print(chunk, end="", flush=True)

LangChain vs LangGraph vs CrewAI vs Mastra — 選択基準マトリクス

2026年時点でPythonのAIエージェント/LLMフレームワーク選択は以下の4択に絞られることが多い。それぞれの特性を正確に理解することが重要だ。

観点	LangChain LCEL	LangGraph	CrewAI	Mastra
得意な用途	チェーン・RAG・シンプルなパイプライン	複雑なエージェント・ステート管理・ループ	マルチエージェント・役割分担	TypeScript製・フロントエンド統合
言語	Python	Python	Python	TypeScript
学習コスト	低〜中	高（グラフ設計が必要）	低（役割・タスクで直感的）	中（TS知識必要）
ストリーミング	ネイティブ対応	対応	部分対応	対応
Human-in-the-loop	手動実装	ネイティブ対応	部分対応	対応
ツール統合数	750以上	LangChain共有	多数	50以上
GitHub Stars（2026）	97,000以上	10,000以上	44,000以上	10,000以上
本番実績	Klarna、Uber等	LinkedIn等	Fortune 500 60%採用	スタートアップ中心
向いている開発者	Pythonバックエンド・データサイエンティスト	複雑なエージェント開発者	プロトタイプ重視・ビジネスユーザー	フロントエンド・フルスタック

使い分けの判断基準

LangChain LCELを選ぶべき場面:

• RAGシステム、チャットボット、テキスト変換パイプラインを作る
• 既存のLangChainコードベース（LLMChain等）をモダン化したい
• ストリーミング・非同期処理が必要
• OpenAI以外の多様なLLMプロバイダーを使いたい（750以上のツール統合）

LangGraphへ移行するタイミング:

• エージェントのループ・条件分岐・ステート永続化が必要になった時
• Human-in-the-loopのワークフローが必要な時
• LCELでは表現できない複雑な制御フローが出てきた時

重要なのは、LangChainとLangGraphは競合しないという点だ。LangChainがチェーン構築の基礎、LangGraphがその上に乗るステート管理エンジンという関係で、多くのプロジェクトは両方を組み合わせて使う。

LCELの高度な活用パターン

OutputParserで構造化データを取得する

LLMの出力を常にフリーテキストで受け取るのではなく、JSON・PydanticモデルなどとしてパースするとPythonで扱いやすくなる。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List

# Pydanticモデルで出力スキーマを定義
class ArticleAnalysis(BaseModel):
    title: str = Field(description="記事タイトル")
    keywords: List[str] = Field(description="主要キーワード3〜5個")
    difficulty: str = Field(description="難易度: beginner/intermediate/advanced")
    summary: str = Field(description="100字以内の要約")

model = ChatOpenAI(model="gpt-4o-mini")

# with_structured_output()を使う方法（推奨）
structured_model = model.with_structured_output(ArticleAnalysis)

prompt = ChatPromptTemplate.from_template(
    "以下の記事を分析してください。\n\n記事: {article}"
)

chain = prompt | structured_model

result = chain.invoke({
    "article": "LangChain LCELはパイプ演算子を使ったチェーン構築ライブラリです..."
})

print(f"タイトル: {result.title}")
print(f"キーワード: {result.keywords}")
print(f"難易度: {result.difficulty}")

RunnableLambdaでカスタム処理を挟む

既製のRunnableクラスだけではカバーできない独自処理は、RunnableLambdaで任意のPython関数をチェーンに組み込める。

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

# 文字数カウントをチェーンに挟む例
def count_chars(text: str) -> dict:
    return {"text": text, "char_count": len(text)}

def format_with_stats(data: dict) -> str:
    return f"文字数: {data['char_count']}\n\n{data['text']}"

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("{topic}について説明してください。")

chain = (
    prompt
    | model
    | StrOutputParser()
    | RunnableLambda(count_chars)  # カスタム関数をRunnableとしてチェーンに組み込む
    | RunnableLambda(format_with_stats)
)

result = chain.invoke({"topic": "量子コンピュータ"})
print(result)
# → 「文字数: 312\n\n量子コンピュータとは...」のような出力

フォールバック（エラー回復）を設定する

本番環境では特定のモデルがAPIエラーを返すケースがある。LCELの .with_fallbacks() でバックアップモデルを設定しておくと自動的にフォールバックする。

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

# プライマリ：GPT-4o
primary_model = ChatOpenAI(model="gpt-4o")

# フォールバック：Claude 3.5 Sonnet
fallback_model = ChatAnthropic(model="claude-3-5-sonnet-20241022")

# フォールバック付きモデルを作成
model_with_fallback = primary_model.with_fallbacks([fallback_model])

prompt = ChatPromptTemplate.from_template("{question}に答えてください。")
chain = prompt | model_with_fallback | StrOutputParser()

# GPT-4oがエラーになった場合、自動的にClaudeで再試行する
result = chain.invoke({"question": "AIエージェントの定義とは？"})
print(result)

LangSmithによるデバッグとモニタリング

LCELのもう一つの強力な特徴が、LangSmithとの自動統合だ。環境変数を数行設定するだけで、全てのチェーン実行がLangSmithのダッシュボードに記録される。

import os

# LangSmith設定（LangSmithアカウントが必要）
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."  # LangSmithのAPIキー
os.environ["LANGCHAIN_PROJECT"] = "my-project"  # プロジェクト名（任意）

# 以降のLCEL実行は全て自動でLangSmithに記録される
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chain = (
    ChatPromptTemplate.from_template("{topic}を説明してください。")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

result = chain.invoke({"topic": "ベクトルデータベース"})
# → https://smith.langchain.com で実行トレースを確認できる

LangSmithでは以下を可視化できる。

• 各コンポーネントの入出力と処理時間
• トークン消費量とコスト
• エラーが発生したステップの特定
• プロンプトの比較A/Bテスト

初心者が必ずハマる3つの失敗パターン

失敗パターン1：LLMChainを2026年に使い続ける

症状：既存のチュートリアルを参考にしたら「DeprecationWarning」が大量に出てきた。または古い書き方で書いたコードが動くが、将来的なバージョンアップで動かなくなるリスクがある。

根本原因：LLMChain・SequentialChain・SimpleSequentialChainなどのレガシークラスが、LangChain v1.0以降は `langchain-classic` パッケージに移行されており、メンテナンス対象外になっている。

解決策：新規コードでは必ずLCELを使う。既存コードは段階的に移行する。移行は一般に以下のパターンで対応できる。

# 移行前（非推奨）
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run({"topic": "AI"})

# 移行後（LCEL）
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"topic": "AI"})

失敗パターン2：RunnablePassthroughを使わず変数をチェーン間で渡せない

症状：チェーンを複数つなげると「前のステップの入力が後のステップで消えてしまう」「KeyErrorが出る」。

根本原因：LCELのパイプ演算子（|）は、前のステップの出力のみを次のステップに渡す。元の入力値を保持したい場合は明示的に指定しなければならない。

解決策：辞書形式（{"key1": chain1, "key2": RunnablePassthrough()}）で複数の値を束ねる。

# NGパターン（前の入力が消える）
chain = step1_chain | step2_chain  # step2にはstep1の出力しか渡らない

# OKパターン（元の入力を保持）
from langchain_core.runnables import RunnablePassthrough

chain = (
    {"step1_result": step1_chain, "original_input": RunnablePassthrough()}
    | step2_chain
)

失敗パターン3：非同期処理で sync/async を混在させる

症状：FastAPIやDjangoで使おうとすると「RuntimeError: This event loop is already running」が出る。または `asyncio.run()` が入れ子になってクラッシュする。

根本原因：LCELの同期メソッド（.invoke()）を非同期コンテキスト（asyncioイベントループ内）で呼ぶと競合する。

解決策：FastAPI等の非同期フレームワーク内では 必ず await chain.ainvoke() を使う。

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

app = FastAPI()
chain = (
    ChatPromptTemplate.from_template("{question}に答えてください。")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# NG：FastAPI内でsyncのinvoke()を呼ぶ
@app.get("/ask-sync")
def ask_sync(question: str):
    return chain.invoke({"question": question})  # デッドロックの可能性あり

# OK：asyncコンテキストではainvokeを使う
@app.get("/ask")
async def ask(question: str):
    result = await chain.ainvoke({"question": question})  # 非同期で安全
    return {"answer": result}

# ストリーミングAPIの例
from fastapi.responses import StreamingResponse

@app.get("/ask-stream")
async def ask_stream(question: str):
    async def generate():
        async for chunk in chain.astream({"question": question}):
            yield chunk
    return StreamingResponse(generate(), media_type="text/plain")

LangChainを実務で使う5つのユースケース

1. 社内ドキュメントQAボット（RAG）

社内規定・製品マニュアル・FAQをベクトルストアに格納し、社員が自然言語で質問できるシステム。RAGチェーン（Step 5で解説）を基盤に、Slack Bot や社内ポータルと統合するケースが最も多い。

詳細な実装については、ベクトルデータベース比較ガイドも参照してほしい。

2. コンテンツ自動生成パイプライン

キーワードから記事タイトル案を複数生成→ユーザーが選択→本文生成→SEOメタデータ生成、というような多段パイプラインをLCELで構築する。RunnableParallelで並列化することで処理時間を最小化できる。

3. データ変換・バッチ処理

CSVデータの各行にLLMを適用して感情分析・カテゴリ分類・翻訳を行う処理は、chain.batch([...]) で簡潔に書ける。1,000件のデータを10並列で処理するといった用途に向いている。

4. マルチモーダル処理チェーン

画像・PDF・テキストが混在するドキュメントを処理する場合、LangChainの Document Loader群（PDF・Web・S3等）を活用すると前処理コードを大幅に削減できる。

5. LangGraphへの移行準備

シンプルなチェーンから始めてLCELに慣れた後、状態管理や条件分岐が必要になった時点でLangGraphに段階移行する、というロードマップが推奨される。LCELのコンポーネントはそのままLangGraphのノードとして再利用できる。

マルチエージェントの役割分担型システムが必要になった場合は CrewAIガイド も参照してほしい。TypeScriptスタックであれば Mastra完全ガイド が参考になる。

LangChain v1.0以降の変更点サマリー

2025年10月に正式リリースされたLangChain v1.0は、以下の点で大きな変更が入っている。2023〜2024年の古いチュートリアルとの差分を把握しておくことが重要だ。

• Python 3.10以上が必須（3.9サポート終了）
• LLMChain・SequentialChain等がlangchain-classicに移行（新規利用非推奨）
• パッケージ分割の明確化：langchain-core（コアインターフェース）/ langchain（高レベルAPI）/ langchain-community（サードパーティ統合）の3層構造
• LangGraph v1.0の同時リリース：エージェント・ステート管理はLangGraphが正式な推奨パス
• LangSmith連携の強化：トレーシングのAPI仕様が安定化

古いコードを発見したら「import from langchain.chains」という形のインポートが非推奨のサインだ。from langchain_core.* または from langchain_community.* 形式に移行することを検討する。

まとめ — LangChain LCELを使うべき理由

LangChain LCELは「とりあえずLLMを使いたい」段階を超えて、本番品質のLLMアプリケーションを構築するための現代的な基盤だ。本記事のポイントを整理する。

• LLMChainは使わない：2026年時点では非推奨。LCELのパイプ演算子に移行する
• 5ステップで習得：基本チェーン → ストリーミング → Passthrough → Parallel → RAG
• ストリーミング・非同期は無料でついてくる：追加実装なしで.stream()/.ainvoke()が使える
• LangGraphは競合でなく補完：複雑なエージェントが必要になった時点でLangGraph化する
• 3つの失敗パターンを先に把握する：LLMChain使用・変数の消失・sync/async混在

まず pip install langchain langchain-core langchain-openai でインストールし、Step 1の基本チェーンを手元で動かすことから始めてほしい。LCELの感覚は、書けば書くほど直感的に身につく。

---

参考・公式リソース

• LangChain LCEL 公式ドキュメント
• LangChain GitHub（97,000以上のスター）
• LangSmith 公式ドキュメント
• LangGraph 公式ドキュメント
• LangChain Python Tutorial 2026（PyCharm Blog）

関連記事（自動）

• ▸ AIエージェント監視ツール比較｜OTel・LangSmith
• ▸ AIエージェント実装ロードマップ｜RAG・実行・運用の必須技術【2026】
• ▸ ReAct・Reflexion他｜AIエージェント推論パターン【2026】
• ▸ Claude Code Skills ガイド｜Skill自作7パターン+配布【2026】
• ▸ ストリーミング応答 実装ガイド｜SSE・トークン逐次表示【2026】