LangChain

Mark as read

LangChain 技術概要 — LLMアプリケーション開発フレームワークの全容

1. はじめに

1.1 LangChainとは

LangChain は、大規模言語モデル（LLM）を活用したアプリケーションの開発を効率化するためのオープンソースフレームワークである。2022年10月にHarrison Chase氏によって開発が開始され、Python版およびJavaScript/TypeScript版が提供されている。

LangChainの核心的な価値は、LLMとの対話を「構成可能なコンポーネント」として抽象化し、複雑なAIアプリケーションを宣言的かつモジュラーに構築できる点にある。単なるAPIラッパーではなく、プロンプトエンジニアリング、外部データの取り込み、エージェント的推論、メモリ管理といった横断的な関心事を統一的に扱うアーキテクチャを提供する。

1.2 LangChainエコシステムの全体像

LangChainは単一のライブラリではなく、以下のコンポーネントから成るエコシステムとして構成されている。

コンポーネント	役割
langchain-core	基本的な抽象化（Runnable, BaseMessage, PromptTemplate等）を定義するコアパッケージ
langchain	チェーン、エージェント、検索戦略等の高レベルコンポーネント
langchain-community	サードパーティ統合のコレクション
langchain-{provider}	特定プロバイダー向けパッケージ（langchain-openai, langchain-anthropic等）
LangGraph	ステートフルなマルチアクターアプリケーション構築のためのオーケストレーションフレームワーク
LangSmith	LLMアプリケーションのデバッグ、テスト、監視、評価のためのプラットフォーム
LangServe	LangChainチェーンをREST APIとしてデプロイするためのライブラリ

┌─────────────────────────────────────────────────────────┐
│                    LangSmith (監視・評価)                  │
├─────────────────────────────────────────────────────────┤
│  LangServe (API化)  │  LangGraph (オーケストレーション)     │
├─────────────────────────────────────────────────────────┤
│                  langchain (高レベルAPI)                   │
├─────────────────────────────────────────────────────────┤
│  langchain-openai │ langchain-anthropic │ langchain-xxx  │
├─────────────────────────────────────────────────────────┤
│               langchain-core (基本抽象化)                  │
└─────────────────────────────────────────────────────────┘

1.3 インストールと基本セットアップ

# コアパッケージのインストール
pip install langchain

# プロバイダーパッケージのインストール（例：OpenAI）
pip install langchain-openai

# その他よく使われるパッケージ
pip install langchain-anthropic    # Anthropic Claude
pip install langchain-google-genai # Google Gemini
pip install langchain-community    # コミュニティ統合
pip install langgraph              # グラフベースオーケストレーション
pip install langsmith              # 監視・評価

環境変数の設定:

# LLMプロバイダーのAPIキー
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."

# LangSmith（オプション・推奨）
export LANGSMITH_API_KEY="ls__..."
export LANGSMITH_TRACING="true"
export LANGSMITH_PROJECT="my-project"

最小限の動作確認:

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# モデルの初期化
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 単純な呼び出し
response = llm.invoke([HumanMessage(content="LangChainとは何ですか？")])
print(response.content)

2. コアアーキテクチャ — langchain-core

2.1 Runnableインターフェース

LangChainのアーキテクチャの中心にあるのが Runnable プロトコルである。すべてのLangChainコンポーネント（LLM、プロンプトテンプレート、出力パーサー、リトリーバー等）は Runnable インターフェースを実装しており、統一的なAPIで操作できる。

Runnableが提供する標準メソッド

メソッド	説明	用途
`invoke(input)`	単一入力に対する同期実行	基本的な実行
`ainvoke(input)`	単一入力に対する非同期実行	非同期コンテキスト
`batch(inputs)`	複数入力に対するバッチ実行	大量データ処理
`abatch(inputs)`	非同期バッチ実行	非同期大量データ処理
`stream(input)`	ストリーミング出力	リアルタイム応答
`astream(input)`	非同期ストリーミング	非同期リアルタイム応答
`astream_events(input)`	イベントベースストリーミング	中間ステップの監視

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
    ("system", "あなたは{role}として回答してください。"),
    ("human", "{question}")
])

# invoke — 同期実行
chain = prompt | llm
result = chain.invoke({"role": "専門家", "question": "量子コンピュータとは？"})

# stream — ストリーミング
for chunk in chain.stream({"role": "専門家", "question": "量子コンピュータとは？"}):
    print(chunk.content, end="", flush=True)

# batch — バッチ実行
results = chain.batch([
    {"role": "専門家", "question": "AIとは？"},
    {"role": "初心者向け", "question": "AIとは？"},
])

# async — 非同期実行
import asyncio
async def main():
    result = await chain.ainvoke({"role": "専門家", "question": "量子コンピュータとは？"})
    print(result.content)

asyncio.run(main())

2.2 LCEL（LangChain Expression Language）

LCEL は、Runnableコンポーネントを宣言的に連結するためのDSL（ドメイン固有言語）である。Pythonの | （パイプ）演算子を使って、データフローを直感的に記述できる。

LCELの基本的なパイプライン構築

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

# コンポーネントの定義
prompt = ChatPromptTemplate.from_template(
    "以下のトピックについて3つの要点を挙げてください: {topic}"
)
model = ChatOpenAI(model="gpt-4o", temperature=0.7)
output_parser = StrOutputParser()

# LCELによるチェーン構築（パイプ演算子）
chain = prompt | model | output_parser

# 実行
result = chain.invoke({"topic": "再生可能エネルギー"})
print(result)

LCELの高度な合成パターン

from langchain_core.runnables import (
    RunnableParallel,
    RunnablePassthrough,
    RunnableLambda,
    RunnableBranch,
)

# 1. 並列実行 — RunnableParallel
parallel_chain = RunnableParallel(
    summary=prompt_summary | model | StrOutputParser(),
    keywords=prompt_keywords | model | StrOutputParser(),
    sentiment=prompt_sentiment | model | StrOutputParser(),
)
# 入力を3つのチェーンに同時に渡して並列処理
results = parallel_chain.invoke({"text": "..."})
# results = {"summary": "...", "keywords": "...", "sentiment": "..."}

# 2. パススルー — RunnablePassthrough
chain = RunnableParallel(
    context=retriever,                    # 検索結果を取得
    question=RunnablePassthrough(),       # 入力をそのまま渡す
) | prompt | model | StrOutputParser()

# 3. カスタムロジック — RunnableLambda
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

chain = (
    {"context": retriever | RunnableLambda(format_docs), 
     "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 4. 条件分岐 — RunnableBranch
branch_chain = RunnableBranch(
    (lambda x: "技術" in x["topic"], technical_chain),
    (lambda x: "ビジネス" in x["topic"], business_chain),
    default_chain,  # デフォルトのチェーン
)

# 5. フォールバック — with_fallbacks
reliable_model = ChatOpenAI(model="gpt-4o").with_fallbacks(
    [ChatOpenAI(model="gpt-4o-mini"), ChatAnthropic(model="claude-sonnet-4-20250514")]
)

# 6. リトライ — with_retry
retry_model = ChatOpenAI(model="gpt-4o").with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

2.3 メッセージシステム

LangChainは、LLMとの対話を構造化されたメッセージオブジェクトとして表現する。

from langchain_core.messages import (
    SystemMessage,
    HumanMessage,
    AIMessage,
    ToolMessage,
    FunctionMessage,
)

messages = [
    SystemMessage(content="あなたは親切なアシスタントです。"),
    HumanMessage(content="Pythonの特徴を教えてください。"),
    AIMessage(content="Pythonは読みやすい構文を持つ汎用プログラミング言語です..."),
    HumanMessage(content="具体例を挙げてください。"),
]

# マルチモーダルメッセージ（画像付き）
from langchain_core.messages import HumanMessage

multimodal_message = HumanMessage(
    content=[
        {"type": "text", "text": "この画像について説明してください。"},
        {
            "type": "image_url",
            "image_url": {"url": "data:image/png;base64,{base64_data}"},
        },
    ]
)

3. プロンプトテンプレート

3.1 基本的なプロンプトテンプレート

プロンプトテンプレートは、LLMに渡すプロンプトをパラメータ化し、再利用可能にするための仕組みである。

from langchain_core.prompts import (
    PromptTemplate,
    ChatPromptTemplate,
    MessagesPlaceholder,
    FewShotPromptTemplate,
    FewShotChatMessagePromptTemplate,
)

# 1. 基本的な文字列テンプレート
template = PromptTemplate.from_template(
    "以下の文章を{language}に翻訳してください:\n\n{text}"
)
result = template.invoke({"language": "英語", "text": "今日はいい天気です。"})

# 2. ChatPromptTemplate（推奨）
chat_template = ChatPromptTemplate.from_messages([
    ("system", "あなたは{expertise}の専門家です。簡潔に回答してください。"),
    ("human", "{question}"),
])

# 3. MessagesPlaceholder（会話履歴の挿入）
chat_with_history = ChatPromptTemplate.from_messages([
    ("system", "あなたは親切なアシスタントです。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
])

from langchain_core.messages import HumanMessage, AIMessage
result = chat_with_history.invoke({
    "chat_history": [
        HumanMessage(content="私の名前はタロウです。"),
        AIMessage(content="こんにちは、タロウさん！"),
    ],
    "input": "私の名前を覚えていますか？",
})

3.2 Few-Shotプロンプト

# Few-Shot例の定義
examples = [
    {"input": "嬉しい", "output": "positive"},
    {"input": "悲しい", "output": "negative"},
    {"input": "普通", "output": "neutral"},
]

# Few-Shot用テンプレート
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

# 最終的なプロンプト
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "以下の例に従って、テキストの感情を分類してください。"),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | llm | StrOutputParser()
result = chain.invoke({"input": "今日は最高の日だ！"})

3.3 動的Few-Shot選択

from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 大量の例からセマンティック検索で最適な例を選択
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples=all_examples,           # 大量の例
    embeddings=OpenAIEmbeddings(),
    vectorstore_cls=Chroma,
    k=3,                             # 選択する例の数
)

dynamic_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    example_selector=example_selector,  # 静的examplesの代わり
)

4. モデル統合

4.1 Chat Models

LangChainは、すべての主要なLLMプロバイダーに対して統一インターフェースを提供する。

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

# OpenAI
openai_llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    max_tokens=4096,
    timeout=30,
    max_retries=2,
)

# Anthropic
anthropic_llm = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    temperature=0,
    max_tokens=4096,
)

# Google
google_llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-pro",
    temperature=0,
)

# すべて同じインターフェースで使用可能
for llm in [openai_llm, anthropic_llm, google_llm]:
    result = llm.invoke("LangChainとは何ですか？")
    print(f"{llm.__class__.__name__}: {result.content[:50]}...")

4.2 構造化出力（Structured Output）

from pydantic import BaseModel, Field
from typing import List, Optional

# Pydanticモデルで出力スキーマを定義
class MovieReview(BaseModel):
    """映画レビューの構造化データ"""
    title: str = Field(description="映画のタイトル")
    rating: int = Field(description="1-10のスコア", ge=1, le=10)
    genres: List[str] = Field(description="ジャンルのリスト")
    summary: str = Field(description="レビューの要約")
    recommended: bool = Field(description="推奨するかどうか")

# with_structured_output で型安全な出力を得る
structured_llm = llm.with_structured_output(MovieReview)

review = structured_llm.invoke(
    "映画『インセプション』についてレビューしてください。"
)
# review は MovieReview 型のオブジェクト
print(f"タイトル: {review.title}")
print(f"スコア: {review.rating}/10")
print(f"ジャンル: {', '.join(review.genres)}")
print(f"推奨: {'はい' if review.recommended else 'いいえ'}")

4.3 ツール呼び出し（Tool Calling）

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """指定した都市の天気を取得する。"""
    # 実際にはAPIを呼び出す
    return f"{city}の天気: 晴れ、気温25度"

@tool
def calculate(expression: str) -> str:
    """数学的な計算を実行する。"""
    return str(eval(expression))

# ツールをモデルにバインド
llm_with_tools = llm.bind_tools([get_weather, calculate])

# モデルが必要に応じてツールを呼び出す
response = llm_with_tools.invoke("東京の天気と、15 * 23の計算結果を教えてください。")

# ツール呼び出し情報へのアクセス
for tool_call in response.tool_calls:
    print(f"ツール: {tool_call['name']}")
    print(f"引数: {tool_call['args']}")

5. 出力パーサー

5.1 主要な出力パーサー

from langchain_core.output_parsers import (
    StrOutputParser,
    JsonOutputParser,
    PydanticOutputParser,
    CommaSeparatedListOutputParser,
    XMLOutputParser,
)

# 1. 文字列パーサー（最も基本的）
str_parser = StrOutputParser()

# 2. JSONパーサー
json_parser = JsonOutputParser()
chain = prompt | llm | json_parser
# 出力: {"key": "value", ...}

# 3. Pydanticパーサー
class Recipe(BaseModel):
    name: str
    ingredients: List[str]
    steps: List[str]
    prep_time_minutes: int

pydantic_parser = PydanticOutputParser(pydantic_object=Recipe)

prompt_with_format = ChatPromptTemplate.from_messages([
    ("system", "レシピをJSON形式で出力してください。\n{format_instructions}"),
    ("human", "{dish}のレシピを教えてください。"),
])

chain = (
    prompt_with_format.partial(
        format_instructions=pydantic_parser.get_format_instructions()
    )
    | llm
    | pydantic_parser
)

recipe = chain.invoke({"dish": "カレーライス"})
# recipe は Recipe 型

# 4. カンマ区切りリストパーサー
list_parser = CommaSeparatedListOutputParser()
chain = prompt | llm | list_parser
# 出力: ["item1", "item2", "item3"]

# 5. ストリーミング対応JSONパーサー
from langchain_core.output_parsers import JsonOutputParser

streaming_chain = prompt | llm | JsonOutputParser()
async for chunk in streaming_chain.astream({"topic": "AI"}):
    print(chunk)  # 部分的なJSON結果が逐次出力される

6. ドキュメント処理

6.1 Document Loaders（ドキュメントローダー）

LangChainは100以上のドキュメントローダーを提供し、多様なデータソースからの読み込みをサポートする。

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    CSVLoader,
    UnstructuredHTMLLoader,
    WebBaseLoader,
    GitLoader,
    DirectoryLoader,
    JSONLoader,
    NotionDBLoader,
    ConfluenceLoader,
    S3FileLoader,
)

# PDF読み込み
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()
# 各ページがDocumentオブジェクトとして返される
for doc in pdf_docs:
    print(f"ページ {doc.metadata['page']}: {doc.page_content[:100]}...")

# Web ページ読み込み
web_loader = WebBaseLoader(
    web_paths=["https://example.com/article1", "https://example.com/article2"],
)
web_docs = web_loader.load()

# ディレクトリ内の全ファイルを再帰的に読み込み
dir_loader = DirectoryLoader(
    "docs/",
    glob="**/*.md",
    loader_cls=TextLoader,
    show_progress=True,
)
all_docs = dir_loader.load()

# JSON読み込み（jq式でフィールド指定）
json_loader = JSONLoader(
    file_path="data.json",
    jq_schema=".messages[].content",
    text_content=False,
)
json_docs = json_loader.load()

# Git リポジトリ読み込み
git_loader = GitLoader(
    clone_url="https://github.com/owner/repo",
    repo_path="./temp_repo",
    branch="main",
    file_filter=lambda f: f.endswith(".py"),
)
git_docs = git_loader.load()

6.2 Text Splitters（テキスト分割）

ドキュメントをLLMのコンテキストウィンドウに収まるチャンクに分割する。

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter,
    MarkdownHeaderTextSplitter,
    HTMLHeaderTextSplitter,
    RecursiveJsonSplitter,
    Language,
)

# 1. 再帰的文字分割（最も汎用的・推奨）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,           # チャンクの最大文字数
    chunk_overlap=200,         # チャンク間のオーバーラップ
    length_function=len,
    separators=["\n\n", "\n", "。", "、", " ", ""],  # 日本語向けセパレータ
)

chunks = text_splitter.split_documents(documents)

# 2. Markdownヘッダーベース分割
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header1"),
        ("##", "Header2"),
        ("###", "Header3"),
    ]
)
md_chunks = md_splitter.split_text(markdown_text)

# 3. プログラミング言語対応分割
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=2000,
    chunk_overlap=200,
)
python_chunks = python_splitter.split_documents(python_docs)

# 4. トークンベース分割
token_splitter = TokenTextSplitter(
    chunk_size=500,       # トークン数
    chunk_overlap=50,
    model_name="gpt-4o",
)
token_chunks = token_splitter.split_documents(documents)

7. Embedding（埋め込み）とVector Stores（ベクトルストア）

7.1 Embeddingモデル

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

# OpenAI Embeddings
openai_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=1536,
)

# HuggingFace Embeddings（ローカル実行）
hf_embeddings = HuggingFaceEmbeddings(
    model_name="intfloat/multilingual-e5-large",
    model_kwargs={"device": "cpu"},
    encode_kwargs={"normalize_embeddings": True},
)

# 使用例
text = "LangChainはLLMアプリケーション開発フレームワークです。"
embedding_vector = openai_embeddings.embed_query(text)
print(f"次元数: {len(embedding_vector)}")  # 1536

# 複数テキストのバッチ埋め込み
texts = ["テキスト1", "テキスト2", "テキスト3"]
vectors = openai_embeddings.embed_documents(texts)

7.2 Vector Stores

from langchain_community.vectorstores import (
    Chroma,
    FAISS,
    Pinecone,
    Qdrant,
    Weaviate,
    Milvus,
    PGVector,
)

# 1. Chroma（ローカル開発に最適）
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=openai_embeddings,
    persist_directory="./chroma_db",
    collection_name="my_collection",
)

# 2. FAISS（高速な類似検索）
vectorstore = FAISS.from_documents(
    documents=chunks,
    embedding=openai_embeddings,
)
# ローカル保存・読み込み
vectorstore.save_local("faiss_index")
loaded_vectorstore = FAISS.load_local(
    "faiss_index", 
    openai_embeddings,
    allow_dangerous_deserialization=True,
)

# 3. Qdrant（本番環境向け）
from langchain_qdrant import QdrantVectorStore
vectorstore = QdrantVectorStore.from_documents(
    documents=chunks,
    embedding=openai_embeddings,
    url="http://localhost:6333",
    collection_name="my_collection",
)

# 類似検索
results = vectorstore.similarity_search(
    "LangChainの特徴は？",
    k=5,
)
for doc in results:
    print(f"スコア: {doc.metadata.get('score', 'N/A')}")
    print(f"内容: {doc.page_content[:100]}...")

# スコア付き類似検索
results_with_scores = vectorstore.similarity_search_with_score(
    "LangChainの特徴は？",
    k=5,
)
for doc, score in results_with_scores:
    print(f"スコア: {score:.4f} — {doc.page_content[:50]}...")

# フィルタ付き検索
filtered_results = vectorstore.similarity_search(
    "LangChainの特徴は？",
    k=5,
    filter={"source": "official_docs"},
)

# MMR（Maximal Marginal Relevance）検索 — 多様性を考慮
mmr_results = vectorstore.max_marginal_relevance_search(
    "LangChainの特徴は？",
    k=5,
    fetch_k=20,
    lambda_mult=0.5,  # 0=多様性重視、1=関連性重視
)

7.3 Retriever（リトリーバー）

# Vector Storeからリトリーバーを生成
retriever = vectorstore.as_retriever(
    search_type="similarity",     # "similarity", "mmr", "similarity_score_threshold"
    search_kwargs={
        "k": 5,
        "score_threshold": 0.7,   # similarity_score_threshold時のみ
    },
)

# リトリーバーの使用
docs = retriever.invoke("LangChainの特徴は？")

# 高度なリトリーバー
from langchain.retrievers import (
    ContextualCompressionRetriever,
    MultiQueryRetriever,
    EnsembleRetriever,
    ParentDocumentRetriever,
    SelfQueryRetriever,
)

# マルチクエリリトリーバー — 1つの質問から複数の検索クエリを生成
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm,
)

# アンサンブルリトリーバー — 複数のリトリーバーを組み合わせ
from langchain_community.retrievers import BM25Retriever

bm25_retriever = BM25Retriever.from_documents(documents, k=5)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vectorstore.as_retriever()],
    weights=[0.4, 0.6],  # BM25とベクトル検索の重み
)

# 圧縮リトリーバー — 検索結果を要約・フィルタリング
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(),
)

# セルフクエリリトリーバー — 自然言語からメタデータフィルタを自動生成
from langchain.chains.query_constructor.base import AttributeInfo

metadata_field_info = [
    AttributeInfo(name="source", description="ドキュメントのソース", type="string"),
    AttributeInfo(name="year", description="公開年", type="integer"),
]

self_query_retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    document_contents="技術ドキュメント",
    metadata_field_info=metadata_field_info,
)

8. RAG（Retrieval-Augmented Generation）

8.1 基本的なRAGパイプライン

RAGは、外部知識ベースから関連情報を検索し、それをコンテキストとしてLLMに提供することで、より正確で根拠のある回答を生成するパターンである。

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

# Step 1: ドキュメントの読み込み
loader = PyPDFLoader("technical_document.pdf")
documents = loader.load()

# Step 2: テキスト分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
)
chunks = text_splitter.split_documents(documents)

# Step 3: ベクトルストアの構築
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./rag_db",
)

# Step 4: リトリーバーの作成
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20},
)

# Step 5: プロンプトテンプレート
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """以下のコンテキストに基づいて質問に回答してください。
コンテキストに記載されていない情報については「この情報はコンテキストに含まれていません」と回答してください。

コンテキスト:
{context}"""),
    ("human", "{question}"),
])

# Step 6: ドキュメントフォーマット関数
def format_docs(docs):
    return "\n\n---\n\n".join(
        f"[出典: {doc.metadata.get('source', '不明')} / "
        f"ページ: {doc.metadata.get('page', 'N/A')}]\n{doc.page_content}"
        for doc in docs
    )

# Step 7: RAGチェーンの構築
rag_chain = (
    {
        "context": retriever | RunnableLambda(format_docs),
        "question": RunnablePassthrough(),
    }
    | rag_prompt
    | ChatOpenAI(model="gpt-4o", temperature=0)
    | StrOutputParser()
)

# 実行
answer = rag_chain.invoke("このドキュメントの主要な結論は何ですか？")
print(answer)

8.2 会話型RAG（Conversational RAG）

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain, create_history_aware_retriever

# 会話履歴を考慮した検索クエリ生成
contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", "会話履歴と最新の質問から、独立した検索クエリを生成してください。"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

history_aware_retriever = create_history_aware_retriever(
    llm=ChatOpenAI(model="gpt-4o"),
    retriever=retriever,
    prompt=contextualize_prompt,
)

# 回答生成プロンプト
answer_prompt = ChatPromptTemplate.from_messages([
    ("system", "コンテキストに基づいて質問に回答してください。\n\n{context}"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

# チェーンの構築
question_answer_chain = create_stuff_documents_chain(
    llm=ChatOpenAI(model="gpt-4o"),
    prompt=answer_prompt,
)

conversational_rag_chain = create_retrieval_chain(
    history_aware_retriever,
    question_answer_chain,
)

# 会話の実行
chat_history = []

response1 = conversational_rag_chain.invoke({
    "input": "LangChainのメリットは何ですか？",
    "chat_history": chat_history,
})
chat_history.extend([
    HumanMessage(content="LangChainのメリットは何ですか？"),
    AIMessage(content=response1["answer"]),
])

response2 = conversational_rag_chain.invoke({
    "input": "それをもっと具体的に説明してください。",  # 「それ」は前の回答を参照
    "chat_history": chat_history,
})

9. エージェント

9.1 エージェントの概念

エージェントは、LLMを推論エンジンとして使用し、ツールを呼び出しながら自律的にタスクを遂行するコンポーネントである。従来のチェーンが事前定義されたステップを順番に実行するのに対し、エージェントはLLMが動的に次のアクションを決定する。

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# ツールの定義
@tool
def search_web(query: str) -> str:
    """Webを検索して最新情報を取得する。"""
    # 実際にはTavily, Serper等のAPIを使用
    return f"検索結果: {query}に関する情報..."

@tool
def get_current_time() -> str:
    """現在の日時を取得する。"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate_math(expression: str) -> str:
    """数学的な計算を実行する。安全な数式のみ受け付ける。"""
    import ast
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"計算エラー: {e}"

# エージェントプロンプト
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", """あなたは有能なアシスタントです。
ユーザーの質問に答えるためにツールを活用してください。
複数のツールを組み合わせて使うこともできます。"""),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

# エージェントの作成
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_web, get_current_time, calculate_math]
agent = create_tool_calling_agent(llm, tools, agent_prompt)

# AgentExecutor でラップ
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # ステップの詳細表示
    max_iterations=10,      # 最大イテレーション数
    handle_parsing_errors=True,
    return_intermediate_steps=True,
)

# 実行
result = agent_executor.invoke({
    "input": "現在の日時と、円周率の100乗を計算してください。",
    "chat_history": [],
})

print(result["output"])
for step in result["intermediate_steps"]:
    print(f"ツール: {step[0].tool}, 入力: {step[0].tool_input}")
    print(f"結果: {step[1]}")

9.2 LangChain Agentの最新API（create_react_agent）

from langchain.agents import create_react_agent
from langchain import hub

# LangChain Hub からプロンプトを取得
react_prompt = hub.pull("hwchase17/react")

# ReActエージェントの作成
react_agent = create_react_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=tools,
    prompt=react_prompt,
)

executor = AgentExecutor(agent=react_agent, tools=tools, verbose=True)
result = executor.invoke({"input": "東京の天気を調べて"})

9.3 LangGraphベースのエージェント（推奨）

最新のLangChainでは、エージェントの構築にLangGraphの使用が推奨されている。

from langgraph.prebuilt import create_react_agent as create_langgraph_agent

# LangGraphベースのReActエージェント
langgraph_agent = create_langgraph_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=tools,
    prompt="あなたは有能なアシスタントです。ツールを活用して回答してください。",
)

# 実行
result = langgraph_agent.invoke({
    "messages": [("human", "東京の天気と現在時刻を教えてください")]
})

# ストリーミング実行
for chunk in langgraph_agent.stream({
    "messages": [("human", "東京の天気と現在時刻を教えてください")]
}):
    print(chunk)

10. メモリとチャット履歴

10.1 メモリの種類

from langchain.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    ConversationSummaryBufferMemory,
    ConversationTokenBufferMemory,
)

# 1. バッファメモリ — すべての会話を保持
buffer_memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
)

# 2. ウィンドウメモリ — 直近N回の会話のみ保持
window_memory = ConversationBufferWindowMemory(
    k=10,                    # 直近10回
    memory_key="chat_history",
    return_messages=True,
)

# 3. 要約メモリ — 過去の会話を要約して保持
summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    memory_key="chat_history",
    return_messages=True,
)

# 4. 要約バッファメモリ — 直近は全文、古いものは要約
summary_buffer_memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    max_token_limit=2000,
    memory_key="chat_history",
    return_messages=True,
)

10.2 チャット履歴の永続化

from langchain_community.chat_message_histories import (
    RedisChatMessageHistory,
    SQLChatMessageHistory,
    FileChatMessageHistory,
)

# Redis による永続化
redis_history = RedisChatMessageHistory(
    session_id="user_123_session_456",
    url="redis://localhost:6379/0",
    ttl=3600,  # 1時間で期限切れ
)

# SQLite による永続化
sql_history = SQLChatMessageHistory(
    session_id="user_123",
    connection="sqlite:///chat_history.db",
)

# ファイルによる永続化
file_history = FileChatMessageHistory(
    file_path="chat_history.json",
)

# メッセージの追加と取得
redis_history.add_user_message("こんにちは")
redis_history.add_ai_message("こんにちは！何かお手伝いできることはありますか？")
messages = redis_history.messages

11. コールバックとトレーシング

11.1 コールバックハンドラ

from langchain_core.callbacks import (
    BaseCallbackHandler,
    StdOutCallbackHandler,
    AsyncCallbackHandler,
)
from langchain_core.outputs import LLMResult

class CustomCallbackHandler(BaseCallbackHandler):
    """カスタムコールバックハンドラ"""

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"🚀 LLM開始: {serialized.get('name', 'unknown')}")
        print(f"   プロンプト数: {len(prompts)}")

    def on_llm_end(self, response: LLMResult, **kwargs):
        print(f"✅ LLM完了")
        if response.llm_output:
            usage = response.llm_output.get("token_usage", {})
            print(f"   トークン使用量: {usage}")

    def on_llm_error(self, error, **kwargs):
        print(f"❌ LLMエラー: {error}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"🔗 チェーン開始: {serialized.get('name', 'unknown')}")

    def on_chain_end(self, outputs, **kwargs):
        print(f"🔗 チェーン完了")

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"🔧 ツール開始: {serialized.get('name', 'unknown')}")
        print(f"   入力: {input_str}")

    def on_tool_end(self, output, **kwargs):
        print(f"🔧 ツール完了: {output[:100]}...")

    def on_retriever_start(self, serialized, query, **kwargs):
        print(f"🔍 検索開始: {query}")

    def on_retriever_end(self, documents, **kwargs):
        print(f"🔍 検索完了: {len(documents)}件")

# 使用方法
handler = CustomCallbackHandler()

# 方法1: コンストラクタで指定
llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])

# 方法2: 実行時に指定
result = chain.invoke({"input": "質問"}, config={"callbacks": [handler]})

11.2 LangSmithによるトレーシング

import os

# LangSmithの設定
os.environ["LANGSMITH_API_KEY"] = "ls__..."
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-rag-app"

# 設定するだけで自動的にトレースが収集される
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"topic": "AI"})
# → LangSmith UIでトレースを確認可能

# カスタムランの記録
from langsmith import traceable

@traceable(name="custom_processing")
def process_data(data):
    """カスタム処理ロジック"""
    result = chain.invoke(data)
    return result

12. LangServe — APIとしてのデプロイ

12.1 基本的なLangServeセットアップ

# server.py
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

app = FastAPI(
    title="LangChain API Server",
    version="1.0",
    description="LangChainチェーンをREST APIとして提供",
)

# チェーンの定義
translate_chain = (
    ChatPromptTemplate.from_template(
        "以下のテキストを{target_language}に翻訳してください:\n\n{text}"
    )
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)

summarize_chain = (
    ChatPromptTemplate.from_template(
        "以下のテキストを{max_words}語以内で要約してください:\n\n{text}"
    )
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)

# ルートの追加
add_routes(app, translate_chain, path="/translate")
add_routes(app, summarize_chain, path="/summarize")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# サーバー起動
python server.py

# エンドポイント:
# POST http://localhost:8000/translate/invoke    — 同期実行
# POST http://localhost:8000/translate/stream     — ストリーミング
# POST http://localhost:8000/translate/batch      — バッチ実行
# GET  http://localhost:8000/translate/playground  — Playground UI

12.2 クライアントからの呼び出し

from langserve import RemoteRunnable

# リモートチェーンをローカルのRunnableのように使用
remote_chain = RemoteRunnable("http://localhost:8000/translate")

# 同期呼び出し
result = remote_chain.invoke({
    "text": "Hello, World!",
    "target_language": "日本語",
})

# ストリーミング
for chunk in remote_chain.stream({
    "text": "Hello, World!",
    "target_language": "日本語",
}):
    print(chunk, end="", flush=True)

# バッチ
results = remote_chain.batch([
    {"text": "Hello", "target_language": "日本語"},
    {"text": "Goodbye", "target_language": "日本語"},
])

13. 実践的なユースケースと設計パターン

13.1 マルチモーダルRAG

from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
import base64

def encode_image(image_path: str) -> str:
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

# 画像を理解するRAGシステム
multimodal_llm = ChatOpenAI(model="gpt-4o")

def analyze_image_with_context(image_path: str, question: str, context_docs: list):
    image_data = encode_image(image_path)
    context = "\n".join(doc.page_content for doc in context_docs)

    message = HumanMessage(
        content=[
            {"type": "text", "text": f"コンテキスト:\n{context}\n\n質問: {question}"},
            {
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{image_data}"},
            },
        ]
    )

    response = multimodal_llm.invoke([message])
    return response.content

13.2 SQL質問応答

from langchain_community.utilities import SQLDatabase
from langchain.chains import create_sql_query_chain
from langchain_community.tools.sql_database.tool import QuerySQLDataBaseTool

# データベース接続
db = SQLDatabase.from_uri(
    "sqlite:///sample.db",
    include_tables=["users", "orders", "products"],
    sample_rows_in_table_info=3,
)

# SQLクエリ生成チェーン
query_chain = create_sql_query_chain(
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
    db=db,
)

# クエリ実行ツール
execute_query = QuerySQLDataBaseTool(db=db)

# 質問 → SQL生成 → 実行 → 回答のパイプライン
answer_prompt = ChatPromptTemplate.from_messages([
    ("system", """SQLクエリの結果に基づいてユーザーの質問に回答してください。
質問: {question}
SQLクエリ: {query}
クエリ結果: {result}"""),
    ("human", "上記に基づいて日本語で回答してください。"),
])

full_chain = (
    RunnablePassthrough.assign(query=query_chain).assign(
        result=lambda x: execute_query.invoke(x["query"])
    )
    | answer_prompt
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)

answer = full_chain.invoke({"question": "先月の売上トップ5の商品は何ですか？"})

13.3 カスタムツールの高度な実装

from langchain_core.tools import StructuredTool, BaseTool
from pydantic import BaseModel, Field
from typing import Optional, Type
import aiohttp

# 方法1: @tool デコレータ（最もシンプル）
@tool
def search_database(query: str, limit: int = 10) -> str:
    """データベースを検索する。
    
    Args:
        query: 検索クエリ
        limit: 結果の最大数
    """
    # 実装
    return f"検索結果: {query} (最大{limit}件)"

# 方法2: StructuredTool（入力スキーマを明示的に定義）
class SearchInput(BaseModel):
    query: str = Field(description="検索クエリ")
    category: str = Field(description="検索カテゴリ", default="all")
    limit: int = Field(description="結果の最大数", default=10, ge=1, le=100)

def search_func(query: str, category: str = "all", limit: int = 10) -> str:
    return f"検索: {query} カテゴリ: {category} 上限: {limit}"

structured_tool = StructuredTool.from_function(
    func=search_func,
    name="advanced_search",
    description="詳細検索を実行する",
    args_schema=SearchInput,
)

# 方法3: BaseTool継承（最も柔軟）
class AsyncWebSearchTool(BaseTool):
    name: str = "async_web_search"
    description: str = "非同期でWeb検索を実行する"
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str, category: str = "all", limit: int = 10) -> str:
        """同期実行"""
        import requests
        # 実装
        return f"同期検索結果: {query}"

    async def _arun(self, query: str, category: str = "all", limit: int = 10) -> str:
        """非同期実行"""
        async with aiohttp.ClientSession() as session:
            # 実装
            return f"非同期検索結果: {query}"

14. パフォーマンスとベストプラクティス

14.1 パフォーマンス最適化

# 1. キャッシュの活用
from langchain_community.cache import InMemoryCache, SQLiteCache
from langchain_core.globals import set_llm_cache

# インメモリキャッシュ
set_llm_cache(InMemoryCache())

# SQLiteキャッシュ（永続化）
set_llm_cache(SQLiteCache(database_path=".langchain_cache.db"))

# 2. バッチ処理の活用
results = chain.batch(
    inputs=[{"q": q} for q in questions],
    config={"max_concurrency": 5},  # 並列度制御
)

# 3. ストリーミングの活用
async for event in chain.astream_events(
    {"input": "質問"},
    version="v2",
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

# 4. Rate Limiting
from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=1,
    check_every_n_seconds=0.1,
    max_bucket_size=10,
)

llm = ChatOpenAI(model="gpt-4o", rate_limiter=rate_limiter)

14.2 エラーハンドリング

# フォールバック戦略
from langchain_core.runnables import RunnableWithFallbacks

primary = ChatOpenAI(model="gpt-4o")
fallback1 = ChatOpenAI(model="gpt-4o-mini")
fallback2 = ChatAnthropic(model="claude-sonnet-4-20250514")

robust_llm = primary.with_fallbacks([fallback1, fallback2])

# リトライ設定
retry_llm = ChatOpenAI(model="gpt-4o").with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_if_exception_type=(TimeoutError, ConnectionError),
)

14.3 セキュリティのベストプラクティス

# 1. プロンプトインジェクション対策
from langchain_core.prompts import ChatPromptTemplate

# 入力をシステムメッセージから分離
secure_prompt = ChatPromptTemplate.from_messages([
    ("system", """あなたはカスタマーサポートBotです。
以下のルールに従ってください:
- 製品に関する質問にのみ回答する
- 個人情報を要求しない
- 不適切な内容には応答しない
"""),
    ("human", "{user_input}"),  # ユーザー入力はhumanメッセージに限定
])

# 2. APIキーの安全な管理
from langchain_core.utils import get_from_env

api_key = get_from_env("OPENAI_API_KEY", "openai_api_key")

# 3. SQLインジェクション対策（SQL質問応答の場合）
db = SQLDatabase.from_uri(
    "sqlite:///sample.db",
    include_tables=["users", "products"],  # テーブルを制限
    custom_table_info={
        "users": "ユーザーテーブル（id, name, emailのみ）"
    },
)

15. LangChainの最新動向とロードマップ

15.1 アーキテクチャの進化

LangChainは2024年以降、以下の方向に進化している。

LangGraphへの統合強化: 複雑なエージェントワークフローはLangGraphで構築することが推奨されるようになった。LangChainの AgentExecutor は引き続き利用可能だが、より高度な制御が必要な場合はLangGraphが推奨される。
Deep Agents: LangChainエコシステムの最上位レイヤーとして、最小限の設定で強力なエージェントを構築できる「Deep Agents」が導入された。
パッケージの分離: langchain-community から個別プロバイダーパッケージ（langchain-openai, langchain-anthropic 等）への移行が進んでいる。
標準化されたツールインターフェース: ツール呼び出しのインターフェースがLLMプロバイダー間で標準化され、bind_tools() と with_structured_output() が推奨APIとなっている。

15.2 エコシステムの階層構造

┌─────────────────────────────────────────────────────────┐
│          Deep Agents（バッテリー同梱・最小設定）           │
│   → 最も簡単にエージェントを構築                          │
├─────────────────────────────────────────────────────────┤
│        LangChain Agents（中程度のカスタマイズ）            │
│   → create_react_agent, ツールバインディング               │
├─────────────────────────────────────────────────────────┤
│          LangGraph（低レベルオーケストレーション）          │
│   → ステートマシン、条件分岐、人間介入                    │
├─────────────────────────────────────────────────────────┤
│            langchain-core（基本抽象化）                    │
│   → Runnable, LCEL, Messages, Tools                     │
└─────────────────────────────────────────────────────────┘

15.3 まとめ

LangChainは、LLMアプリケーション開発における事実上の標準フレームワークとしての地位を確立している。その主要な価値は以下の通りである。

価値	説明
モデル非依存	単一のコードでOpenAI、Anthropic、Google等のモデルを切り替え可能
構成可能性	LCELによる宣言的なパイプライン構築
豊富な統合	100以上のドキュメントローダー、30以上のベクトルストア、20以上のLLMプロバイダー
プロダクションレディ	LangSmithによる監視、LangServeによるAPI化、LangGraphによるオーケストレーション
アクティブな開発	急速に進化するLLM技術への迅速な対応

LangChainを使用することで、プロンプトエンジニアリング、RAG、エージェント、マルチモーダルアプリケーションといった複雑なAIシステムを、モジュラーかつ保守可能な形で構築できる。特に、LangGraphとの連携により、従来のシーケンシャルなチェーンでは実現困難だったステートフルで動的なワークフローの構築が可能となっている。