EzoAI Course Catalog
Course 12 / Layer 2 -- 技術特化

RAG実践 LangChain / LangGraph 完全ガイド

社内ドキュメントをAIが検索・回答するRAGシステムを構築します。LangChainの基礎からベクトルDB、チャンク戦略、検索精度チューニング、LangGraphによるマルチステップRAG、LangSmithでの監視まで。LangGraphまでカバーする日本語コースは他にありません。

中級-上級約12時間(720分)9 Sections + 2 Reviewハンズオン比率 66%前提: Python基礎

目次

  1. RAGアーキテクチャ -- 仕組みと設計パターン 50min
  2. LangChain基礎 -- Chain/Template/Parser 60min
  3. ベクトルDB -- Chroma/FAISS/Pinecone 60min
  4. 復習A -- 基本RAGパイプライン構築 30min
  5. ドキュメントローダーとチャンク戦略 60min
  6. 検索精度チューニング 65min
  7. 復習B -- 高精度RAG構築 30min
  8. LangGraph入門 -- ステートマシンRAG 70min
  9. LangSmith -- デバッグ/評価/監視 55min
  10. 高度なRAGパターン -- Self-RAG/Adaptive RAG 60min
  11. 総合ハンズオン: 社内ドキュメント検索システム 140min
Section 01 -- 50min(講義30 + ハンズオン20)

RAGアーキテクチャ -- 仕組みと設計パターン

RAGパイプラインの全体像

RAGとは何か

Retrieval-Augmented Generation(検索拡張生成)。LLM単体では学習データに含まれない社内文書や最新情報を参照できませんが、RAGは外部データソースから関連情報を検索し、その結果をプロンプトに挿入してからLLMに回答を生成させます。知識の鮮度と正確性を飛躍的に改善する手法です。

仕組みを一言でまとめると「質問が来たら、まず関連文書を探し、見つけた文書を添えてLLMに渡す」だけ。シンプルですが、検索精度とチャンク戦略次第で回答品質が大きく変わります。

%%{init:{'theme':'dark','themeVariables':{'primaryColor':'#00A5BF','primaryBorderColor':'#007A8F','primaryTextColor':'#e8e8e8','lineColor':'#00A5BF','secondaryColor':'#1a1a1a','background':'#141414','mainBkg':'#1a1a1a','nodeBorder':'#00A5BF'}}}%%
graph LR
  subgraph Indexing["Indexing Phase"]
    A["ドキュメント読込"] --> B["チャンク分割"]
    B --> C["Embedding
(ベクトル変換)"]
    C --> D["ベクトルDB
に格納"]
  end
  subgraph Query["Query Phase"]
    E["ユーザーの質問"] --> F["質問を
Embedding"]
    F --> G["類似検索
(Top-K)"]
    G --> H["関連チャンク
を取得"]
  end
  subgraph Generation["Generation Phase"]
    H --> I["プロンプト構築
(質問+文脈)"]
    I --> J["LLM回答生成"]
  end
RAGの3フェーズ: Indexing(事前準備) → Retrieval(検索) → Generation(回答生成)

RAGの3ステップを理解する

1. Indexing -- ドキュメントの準備

PDF、CSV、Webページなどのドキュメントをチャンク(断片)に分割し、各チャンクをEmbeddingモデルでベクトル化してDBに格納します。この前処理の質がRAG全体の精度を左右するため、チャンクサイズやoverlap設定は慎重に決める必要があります。

2. Retrieval -- 類似検索

ユーザーの質問も同じEmbeddingモデルでベクトル化し、ベクトルDB内で類似度の高いチャンクをTop-K件取得します。コサイン類似度やユークリッド距離が代表的な類似度指標です。

3. Generation -- 回答生成

取得したチャンクを「参考情報」としてプロンプトに挿入し、LLMに回答を生成させます。プロンプトには「以下の情報のみを参照して回答してください」と制約を加えることで、ハルシネーションを抑制できます。

Naive RAG vs Advanced RAG vs Modular RAG

手法特徴精度実装難度
Naive RAG単純な検索+生成。チャンク分割→ベクトル検索→LLM呼出し中低
Advanced RAGRe-ranking、Hybrid Search、Query Transform等で検索精度を改善高中
Modular RAG検索/生成/評価をモジュール化しグラフで制御。LangGraphと相性がよい最高高
Tips: RAGが有効な場面と不要な場面の判断基準
RAGを導入する前に「本当に検索が必要か」を判断する基準を持っておくと、不要なインフラコストを避けられます。
場面RAGの要否理由
社内データに基づく質問応答有効LLMの学習データに含まれない社内固有の情報を参照する必要がある
最新ニュースに基づく回答有効LLMの知識カットオフ以降の情報が必要
法令や規約の正確な引用有効一言一句正確に引用する必要があり、ハルシネーション防止に必須
一般的なプログラミング質問不要LLMの学習データに十分含まれており、検索結果がノイズになる
ブレインストーミング・創作不要正確な事実参照より自由な発想が求められる場面
翻訳・要約(入力テキストが与えられる場合)不要処理対象のテキストが入力として渡されるため外部検索が不要
判断に迷ったら「LLMが学習時に見ていない情報か?」と自問する。Yesなら検索が要る。Noなら検索はノイズになります。

理解度チェック: Section 01

Q1. RAGの3つのフェーズを正しい順序で並べたものはどれですか?

正解: B。まずドキュメントを前処理(Indexing)し、質問に対して類似検索(Retrieval)を行い、取得した情報をもとにLLMが回答を生成(Generation)します。

Q2. Advanced RAGがNaive RAGより精度が高い主な理由は?

正解: B。検索フェーズにRe-ranking、Hybrid Search、Query Transformなどの改善手法を追加することで、LLMに渡す文脈の質が向上します。
ハンズオン: 最小RAGパイプラインを動かす 20min
目標: OpenAI + Chromaで最小構成のRAGを体験し、検索と回答生成の流れを確認する

パート1: 環境構築(5min)

pip install langchain langchain-openai langchain-chroma chromadb

パート2: 最小RAG実装(15min)

import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser os.environ["OPENAI_API_KEY"] = "sk-..." # ご自身のAPIキーを設定 # --- 1. Indexing: ドキュメントをベクトルDBに格納 --- documents = [ Document(page_content="当社の有給休暇は入社6ヶ月後に10日付与されます。勤続年数に応じて最大20日まで増加します。"), Document(page_content="リモートワークは週3日まで可能です。申請はHRシステムから行ってください。事前に上長の承認が必要です。"), Document(page_content="出張経費の精算は出張後2週間以内に経費精算システムで申請してください。領収書の添付が必須です。"), Document(page_content="社内研修は四半期ごとに開催されます。参加希望者はLMSから申し込んでください。受講費用は会社負担です。"), Document(page_content="育児休業は子が1歳になるまで取得可能です。延長申請により最長2歳まで延長できます。"), ] embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents(documents, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) # --- 2. Retrieval + Generation: 質問に回答 --- llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) prompt = ChatPromptTemplate.from_template(""" 以下の参考情報のみを使って質問に回答してください。 参考情報に含まれない内容には「情報が見つかりませんでした」と回答してください。 参考情報: {context} 質問: {question} """) def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) # --- 3. 実行 --- questions = [ "リモートワークは週何日まで可能ですか?", "有給休暇は何日もらえますか?", "社員食堂の営業時間は?", # ドキュメントにない質問 ] for q in questions: print(f"Q: {q}") print(f"A: {chain.invoke(q)}") print("-" * 40)
期待される出力

最初の2問はドキュメントから正確に回答されます。3問目の「社員食堂の営業時間」はドキュメントに情報がないため、「情報が見つかりませんでした」と返答されるのが理想的な動作です。プロンプトの制約指示がハルシネーション抑制に効いていることを確認してください。

参考リンク

Section 02 -- 60min(講義25 + ハンズオン35)

LangChain基礎 -- Chain/Template/Parser

LangChainはLLMアプリケーション構築のデファクトフレームワークです。v0.3では設計が大きく整理され、langchain-coreを軸にプロバイダー別パッケージへ分離されました。

LangChain v0.3 のパッケージ構成

パッケージ役割インストール
langchain-core基盤クラス: Runnable, PromptTemplate, OutputParserpip install langchain-core
langchain-openaiOpenAI統合: ChatOpenAI, OpenAIEmbeddingspip install langchain-openai
langchain-communityコミュニティ統合: 各種ローダー、ベクトルDBpip install langchain-community
langchain高レベルChain(legacy含む)pip install langchain
langgraphステートマシン/グラフ実行(Sec06で詳説)pip install langgraph

主要コンポーネント

ChatModel

LLMとの対話を抽象化。ChatOpenAI、ChatAnthropic等のプロバイダー実装を差し替え可能。temperatureやmax_tokensを設定します。

PromptTemplate

変数を埋め込めるテンプレート。ChatPromptTemplateはsystem/human/aiメッセージを組み立てます。再利用性が高く、プロンプト管理が容易に。

OutputParser

LLMの出力を構造化データに変換。StrOutputParser(文字列)、JsonOutputParser(JSON)、PydanticOutputParser(型付きオブジェクト)。

LCEL (LangChain Expression Language)

パイプ演算子(|)でコンポーネントを連結。prompt | llm | parserのように宣言的にChainを組みます。非同期、ストリーミング、バッチ処理を自動サポート。

LCEL -- パイプ演算子でChainを組む

LCELはUnixパイプに似た考え方です。各コンポーネントはRunnableインタフェースを実装しており、前のコンポーネントの出力が次の入力になります。

%%{init:{'theme':'dark','themeVariables':{'primaryColor':'#00A5BF','primaryBorderColor':'#007A8F','primaryTextColor':'#e8e8e8','lineColor':'#00A5BF','secondaryColor':'#1a1a1a','background':'#141414','mainBkg':'#1a1a1a','nodeBorder':'#00A5BF'}}}%%
graph LR
  A["PromptTemplate
{topic}を埋める"] -->|"|"| B["ChatModel
LLMに送信"]
  B -->|"|"| C["OutputParser
文字列に変換"]
  style A fill:#1a1a1a,stroke:#00A5BF
  style B fill:#1a1a1a,stroke:#00A5BF
  style C fill:#1a1a1a,stroke:#00A5BF
LCEL: パイプ演算子で3つのコンポーネントを連結
ハンズオン: 3種のChainを構築する 35min
目標: SimpleChain、SequentialChain、RouterChainの3パターンをLCELで実装する

Chain 1: SimpleChain(10min)

最も基本的なパターン。テンプレートにトピックを埋めてLLMに送信し、文字列で受け取ります。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) # SimpleChain: テンプレート → LLM → 文字列 prompt = ChatPromptTemplate.from_template( "{topic}について、3つのポイントで100文字以内に要約してください。" ) chain = prompt | llm | StrOutputParser() result = chain.invoke({"topic": "RAGアーキテクチャ"}) print(result) # バッチ実行も可能 results = chain.batch([ {"topic": "ベクトルデータベース"}, {"topic": "プロンプトエンジニアリング"}, ]) for r in results: print(r) print("-" * 40)

Chain 2: SequentialChain(12min)

前のChainの出力を次のChainの入力にする連鎖パターン。要約→翻訳のように段階処理に使います。

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) # Chain1: 技術用語を日本語で解説 explain_prompt = ChatPromptTemplate.from_template( "{term}を初心者向けに日本語で3行以内に解説してください。" ) explain_chain = explain_prompt | llm | StrOutputParser() # Chain2: 解説文からクイズを作成 quiz_prompt = ChatPromptTemplate.from_template( "以下の解説文をもとに、4択クイズを1問作成してください。\n\n解説文:\n{explanation}" ) quiz_chain = quiz_prompt | llm | StrOutputParser() # Sequential: explain → quiz sequential_chain = ( {"explanation": explain_chain, "term": RunnablePassthrough()} | quiz_chain ) result = sequential_chain.invoke({"term": "Embedding"}) print(result)

Chain 3: RouterChain(13min)

質問の内容に応じて異なるChainにルーティングします。「技術質問」と「業務質問」で回答スタイルを切り替える例です。

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 分類Chain: 質問のカテゴリを判定 classify_prompt = ChatPromptTemplate.from_template( "以下の質問を「technical」か「business」に分類してください。1単語のみ回答。\n\n質問: {question}" ) classify_chain = classify_prompt | llm | StrOutputParser() # 技術質問用Chain tech_prompt = ChatPromptTemplate.from_template( "あなたはシニアエンジニアです。技術的に正確に回答してください。\n\n質問: {question}" ) tech_chain = tech_prompt | llm | StrOutputParser() # 業務質問用Chain biz_prompt = ChatPromptTemplate.from_template( "あなたはビジネスコンサルタントです。実務的な視点で回答してください。\n\n質問: {question}" ) biz_chain = biz_prompt | llm | StrOutputParser() # Router: 分類結果に応じてChainを切替 def route(info): category = info["category"].strip().lower() if "technical" in category: return tech_chain.invoke({"question": info["question"]}) return biz_chain.invoke({"question": info["question"]}) router_chain = ( {"category": classify_chain, "question": lambda x: x["question"]} | RunnableLambda(route) ) # テスト questions = [ {"question": "LangChainのRunnableインタフェースの設計思想は?"}, {"question": "RAG導入でROIを最大化するには?"}, ] for q in questions: print(f"Q: {q['question']}") print(f"A: {router_chain.invoke(q)}") print("-" * 40)
補足: LCELの内部動作

パイプ演算子(|)はPythonの__or__メソッドをオーバーロードしています。chain = a | b | cと書くと、内部的にはRunnableSequence([a, b, c])が生成されます。各要素のinvoke/ainvoke/stream/batchが統一インタフェースで呼び出される仕組みです。

参考リンク

自走チャレンジ
テーマ: 講師がSimpleChainを作ったのと同じ構造で、以下の要件のChainを自力で構築してください。「入力テキストを受け取り → 要約 → 要約の品質を5段階でスコアリング → スコアが3以下なら再要約」のSequentialChain。
条件: LCELのパイプ演算子(|)で記述すること。条件分岐にはRunnableBranchまたはif文を使用。
ここで動画を一度止めて、7分間取り組んでください

ヒント: スコアリングLLMの出力をパースして数値に変換する部分がポイントです。OutputParserでint型に変換するか、プロンプトで「スコアのみを数字1文字で出力してください」と制約をかけると安定します。

講師の解答例を見る
from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 要約Chain summarize_prompt = ChatPromptTemplate.from_template( "以下のテキストを3文以内で要約してください: {text}" ) summarize_chain = summarize_prompt | llm | StrOutputParser() # スコアリングChain score_prompt = ChatPromptTemplate.from_template( "以下の要約の品質を1-5の数字のみで評価してください。" "評価基準: 正確性、簡潔さ、情報の網羅性。 要約: {summary} スコア:" ) score_chain = score_prompt | llm | StrOutputParser() # 再要約Chain resummarize_prompt = ChatPromptTemplate.from_template( "以下の要約は品質が低いと判定されました。" "元のテキストからより正確で簡潔な要約を作成してください: " "元テキスト: {text} 前回の要約: {summary}" ) resummarize_chain = resummarize_prompt | llm | StrOutputParser() # 統合実行 def summarize_with_retry(text: str) -> dict: summary = summarize_chain.invoke({"text": text}) score_str = score_chain.invoke({"summary": summary}) score = int(score_str.strip()[0]) if score <= 3: summary = resummarize_chain.invoke({ "text": text, "summary": summary }) score_str = score_chain.invoke({"summary": summary}) score = int(score_str.strip()[0]) return {"summary": summary, "score": score} result = summarize_with_retry("ここに長いテキストを入力...") print(f"要約: {result['summary']}") print(f"スコア: {result['score']}")

解説ポイント: LLMの出力をプログラムで条件分岐に使う場合、出力フォーマットの安定性が命です。「数字のみで出力」と指定しても説明文が混じることがあるので、strip()[0]で先頭1文字を取るなどの防御的パースが必要になります。

Section 03 -- 60min(講義25 + ハンズオン35)

ベクトルDB -- Chroma / FAISS / Pinecone

RAGの「検索」を支えるのがベクトルデータベースです。テキストを数百〜数千次元のベクトルに変換し、類似度で検索します。SQLのLIKE検索とは根本的に異なり、「意味的に近い」文書を見つけられるのが最大の強みです。

Embeddingの仕組み

Embeddingモデルはテキストを固定長のベクトル(数値の配列)に変換します。意味が近い文は近いベクトル空間上に配置される性質を持ちます。「犬」と「猫」のベクトルは近く、「犬」と「経済学」は遠くなります。

%%{init:{'theme':'dark','themeVariables':{'primaryColor':'#00A5BF','primaryBorderColor':'#007A8F','primaryTextColor':'#e8e8e8','lineColor':'#00A5BF','secondaryColor':'#1a1a1a','background':'#141414','mainBkg':'#1a1a1a','nodeBorder':'#00A5BF'}}}%%
graph LR
  A["テキスト
有給休暇は10日"] --> B["Embedding
Model"]
  B --> C["ベクトル
[0.12, -0.34, 0.56, ...]
1536次元"]
  D["テキスト
休みは何日?"] --> E["Embedding
Model"]
  E --> F["ベクトル
[0.11, -0.31, 0.58, ...]
1536次元"]
  C --> G["コサイン類似度
= 0.92(高い)"]
  F --> G
意味が近い文は近いベクトルに変換される

3つのベクトルDBを比較する

項目ChromaFAISSPinecone
提供形態OSS / ローカルMeta製OSS / ローカルクラウドマネージド
セットアップpip install chromadbpip install faiss-cpupip install pinecone-client
スケーラビリティ中(数万件向け)高(数百万件対応)最高(自動スケール)
検索速度(100K件)数十ms数ms数十ms(ネットワーク込み)
永続化ローカルファイルローカルファイルクラウド(自動)
適用場面プロトタイプ、小規模RAG大規模・高速検索本番運用、チーム利用

Embeddingモデルの選び方

Embeddingモデルの選択はRAGの検索精度に直結します。コスト、精度、日本語対応の3軸で代表的なモデルを比較します。

モデル提供元次元数日本語精度コスト(100万トークン)特徴
text-embedding-3-smallOpenAI1536良好$0.02コスパ最良。ほとんどのRAGでこれを選ぶのが正解
text-embedding-3-largeOpenAI3072良好$0.13精度重視。ストレージ2倍・コスト6.5倍で精度向上は数%程度
embed-multilingual-v3.0Cohere1024優秀$0.10多言語対応が強み。日本語+英語の混在ドキュメントに最適
BGE-M3BAAI(OSS)1024優秀無料(セルフホスト)オープンソース。ローカルで動かせるため機密データ向き
multilingual-e5-largeMicrosoft(OSS)1024優秀無料(セルフホスト)HuggingFace上で利用可能。日本語ベンチマークで高スコア
Tips: 選定の判断基準
API利用が許される環境ならtext-embedding-3-smallが安牌。機密データを扱う場合はBGE-M3やmultilingual-e5-largeをローカルで動かす選択肢が現実的です。一度選んだモデルを途中で変更すると、全ドキュメントのre-embeddingが必要になるため、最初の選定は慎重に。
ハンズオン: 3つのDBに格納して検索精度を比較 35min
目標: 同じドキュメントをChroma、FAISS、(模擬)Pineconeに格納し、検索速度と結果の違いを比較する

パート1: 共通データの準備(5min)

import time from langchain_openai import OpenAIEmbeddings from langchain_core.documents import Document embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 社内マニュアル風ドキュメント(10件) docs = [ Document(page_content="当社の勤務時間はフレックスタイム制で、コアタイムは10:00-15:00です。", metadata={"source": "就業規則"}), Document(page_content="有給休暇は入社6ヶ月後に10日付与されます。勤続年数に応じて最大20日まで増加します。", metadata={"source": "就業規則"}), Document(page_content="リモートワークは週3日まで可能です。利用には上長の事前承認が必要です。", metadata={"source": "勤怠ルール"}), Document(page_content="出張経費は出張後2週間以内に経費精算システムで申請してください。", metadata={"source": "経費規程"}), Document(page_content="社内研修は四半期ごとに開催されます。LMSから申し込んでください。", metadata={"source": "人材育成"}), Document(page_content="育児休業は子が1歳になるまで取得可能です。延長申請で最長2歳まで。", metadata={"source": "育児支援"}), Document(page_content="社内Slackの利用ガイドラインに従い、機密情報の投稿は禁止されています。", metadata={"source": "情報セキュリティ"}), Document(page_content="PCの持ち出しには情報システム部への申請が必要です。VPN接続を必須とします。", metadata={"source": "情報セキュリティ"}), Document(page_content="年末調整の書類は毎年11月中旬までに人事部へ提出してください。", metadata={"source": "税務"}), Document(page_content="健康診断は年1回、会社指定の医療機関で受診してください。費用は会社負担です。", metadata={"source": "福利厚生"}), ] query = "テレワークは何日までできますか?"

パート2: Chroma(10min)

from langchain_chroma import Chroma start = time.time() chroma_db = Chroma.from_documents(docs, embeddings, collection_name="test") chroma_results = chroma_db.similarity_search_with_score(query, k=3) chroma_time = time.time() - start print(f"[Chroma] 検索時間: {chroma_time:.3f}s") for doc, score in chroma_results: print(f" score={score:.4f} | {doc.page_content[:50]}...")

パート3: FAISS(10min)

from langchain_community.vectorstores import FAISS start = time.time() faiss_db = FAISS.from_documents(docs, embeddings) faiss_results = faiss_db.similarity_search_with_score(query, k=3) faiss_time = time.time() - start print(f"[FAISS] 検索時間: {faiss_time:.3f}s") for doc, score in faiss_results: print(f" score={score:.4f} | {doc.page_content[:50]}...") # FAISSはローカル保存/読み込み可能 faiss_db.save_local("faiss_index") # loaded = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)

パート4: 結果比較(10min)

print("=" * 60) print("検索速度比較:") print(f" Chroma: {chroma_time:.3f}s") print(f" FAISS : {faiss_time:.3f}s") print() print("Top-1 結果比較:") print(f" Chroma: {chroma_results[0][0].page_content[:60]}") print(f" FAISS : {faiss_results[0][0].page_content[:60]}") print() print("考察:") print("- 両方とも「リモートワーク」の文書をTop-1に返すはず") print("- 「テレワーク」という単語は文書にないが、意味的類似性で検索できている") print("- FAISSは純粋な検索速度が速い(初期化を除く)") print("- Chromaはメタデータフィルタリングが使いやすい")
補足: Pineconeを使う場合

Pineconeはクラウドサービスのため、APIキーの取得とインデックス作成が必要です。pinecone.io で無料アカウントを作成し、langchain-pineconeパッケージを使います。本番環境でスケーラビリティが必要な場合に検討してください。

参考リンク

Review Hands-on A -- 30min

復習A: 基本RAGパイプライン構築

Sec01-03で学んだRAGの基本構造を1本のスクリプトにまとめます。ドキュメント読込からChroma格納、LLMによる回答生成まで通しで実装してください。

課題: 業務マニュアルRAGの構築 30min
成果物: 5問のテスト質問に正確に回答できるRAGスクリプト

要件

  1. 以下のDLテンプレート「サンプルドキュメント.txt」をドキュメントソースとして使用してください
  2. Chromaにベクトル格納し、retrieverを構築してください
  3. LCELでRAG Chainを組み、5問に回答してください

テスト質問(5問)

  1. 有給休暇の付与日数は?
  2. リモートワークの上限日数と必要な手続きは?
  3. 出張経費の申請期限は?
  4. 育児休業は何歳まで延長できますか?
  5. PCの社外持ち出しに必要なことは?
# 復習A: 基本RAGパイプライン # 以下を埋めて完成させてください import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser os.environ["OPENAI_API_KEY"] = "sk-..." # 1. ドキュメントを定義(サンプルドキュメント.txtの内容をDocumentに変換) documents = [ # TODO: Document(page_content="...", metadata={"source": "..."}) ] # 2. ベクトルDB構築 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents(documents, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 3. RAG Chain構築(LCEL) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) prompt = ChatPromptTemplate.from_template(""" 以下の参考情報のみを使って質問に回答してください。 参考情報に含まれない内容には「情報が見つかりませんでした」と回答してください。 参考情報: {context} 質問: {question} """) def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) # 4. テスト実行 questions = [ "有給休暇の付与日数は?", "リモートワークの上限日数と必要な手続きは?", "出張経費の申請期限は?", "育児休業は何歳まで延長できますか?", "PCの社外持ち出しに必要なことは?", ] for q in questions: print(f"Q: {q}") print(f"A: {chain.invoke(q)}") print("-" * 50)
Section 04 -- 60min(講義25 + ハンズオン35)

ドキュメントローダーとチャンク戦略

RAGの品質はドキュメントの前処理で8割決まると言っても過言ではありません。どの形式をどう読み込み、どの粒度で分割するか。この設計を雑にすると、どれだけ高性能なLLMを使っても回答精度は頭打ちになります。

ドキュメントローダー一覧

形式ローダーパッケージ備考
PDFPyPDFLoaderlangchain-communityページ単位で分割。OCR不要のテキストPDF向け
CSVCSVLoaderlangchain-community行単位でDocumentに変換
ExcelUnstructuredExcelLoaderlangchain-communityunstructuredパッケージが必要
WebWebBaseLoaderlangchain-communityBeautifulSoupでHTML→テキスト
NotionNotionDirectoryLoaderlangchain-communityエクスポートしたMarkdownを読込
GitHubGitLoaderlangchain-communityリポジトリのコードを読込

チャンク戦略の4種

1. 固定サイズ分割

文字数で機械的に分割。実装が最も簡単だが、文の途中で切れるリスクがあります。CharacterTextSplitterで実現。

2. 再帰的分割(推奨)

段落→文→単語の順で再帰的に分割。文脈を保ちやすく、多くのケースでベストプラクティスです。RecursiveCharacterTextSplitterがデフォルト選択肢。

3. セマンティック分割

Embeddingの類似度が変化する箇所で分割。意味の区切りで切れるため精度が高いが、Embedding計算コストが発生します。SemanticChunkerで実現。

4. 構造ベース分割

Markdown見出し、HTMLタグ、PDFのページ区切りなど、ドキュメントの構造を利用。MarkdownHeaderTextSplitter等。

チャンクサイズとoverlapの選び方

チャンクサイズはトレードオフです。小さすぎると1チャンクに十分な文脈が含まれず、大きすぎるとノイズが増えて検索精度が落ちます。

パラメータ推奨値考慮点
chunk_size500〜1000文字Q&Aなら短め(300-500)、技術文書なら長め(800-1200)
chunk_overlapchunk_sizeの10〜20%前後のチャンクの意味的接続を維持する

オーバーラップ(重複)の効果

チャンク間のオーバーラップは「前後のチャンクの文脈を繋ぐ接着剤」です。設定値による影響を具体例で示します。

例えば以下の3文を含むドキュメントを、chunk_size=100で分割する場合を考えます。

文A(50文字): 「リモートワークは週3日まで利用可能です。」
文B(50文字): 「利用にはHRシステムから事前申請が必要です。」
文C(50文字): 「申請は上長の承認後、翌営業日に有効になります。」

overlap設定Chunk 1の内容Chunk 2の内容「申請方法は?」の検索精度
0%(重複なし)文A + 文B文C のみ低 -- 文Bと文Cの関連が切れるため、Chunk 2だけでは申請→承認の流れが不明
10%(10文字)文A + 文B文Bの末尾10文字 + 文C中 -- 「事前申請が必要」の断片がChunk 2にも含まれ、文脈が部分的に繋がる
20%(20文字)文A + 文B文Bの後半20文字 + 文C高 -- 「HRシステムから事前申請が必要」がChunk 2にも含まれ、申請→承認の一連の流れが検索で引ける

実務での推奨値はchunk_sizeの10〜20%。20%を超えるとストレージの無駄が増え、検索速度にも影響します。FAQのような短い独立した項目の場合は0%でも問題ありません。

注意: チャンクサイズのアンチパターン
chunk_sizeを2000以上にすると、1チャンクに複数の話題が混在し、検索精度が著しく低下します。「1チャンク = 1つの意味単位」が原則です。
ハンズオン: 4種のチャンク戦略で分割して比較 35min
目標: PDF読込後に4種のチャンク戦略で分割し、検索精度の違いを確認する

パート1: ドキュメント読込+チャンク分割(20min)

from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader from langchain_text_splitters import ( CharacterTextSplitter, RecursiveCharacterTextSplitter, ) from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings # --- サンプルテキスト(PDF読込の代わり) --- sample_text = """ 第1章 有給休暇制度 当社の有給休暇は、入社6ヶ月経過後に10日が付与されます。 勤続年数に応じて付与日数は増加し、6年6ヶ月以上の勤続で最大20日となります。 未消化の有給休暇は翌年度に限り繰り越し可能です。 半日単位での取得も可能で、事前に上長の承認を得てください。 第2章 リモートワーク制度 リモートワークは週3日まで利用可能です。 コアタイム(10:00-15:00)中はオンラインで連絡が取れる状態を維持してください。 利用にあたっては、HRシステムから事前申請し、上長の承認を得る必要があります。 セキュリティの観点から、VPN接続を必須とし、公共Wi-Fiでの業務は禁止します。 第3章 出張経費精算 出張経費は出張完了後2週間以内に経費精算システムで申請してください。 申請には領収書の原本(またはスキャン画像)の添付が必須です。 日当は国内出張5,000円、海外出張10,000円です。 宿泊費の上限は国内15,000円、海外30,000円(税込)です。 上限を超える場合は事前に部長承認が必要です。 """ # 1. 固定サイズ分割 fixed_splitter = CharacterTextSplitter( separator="\n", chunk_size=200, chunk_overlap=30, ) fixed_chunks = fixed_splitter.split_text(sample_text) # 2. 再帰的分割(推奨) recursive_splitter = RecursiveCharacterTextSplitter( chunk_size=200, chunk_overlap=30, separators=["\n\n", "\n", "。", "、", " ", ""], ) recursive_chunks = recursive_splitter.split_text(sample_text) # 3. セマンティック分割 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") semantic_splitter = SemanticChunker(embeddings) semantic_chunks = [doc.page_content for doc in semantic_splitter.create_documents([sample_text])] # 結果比較 for name, chunks in [("固定サイズ", fixed_chunks), ("再帰的", recursive_chunks), ("セマンティック", semantic_chunks)]: print(f"\n{'='*50}") print(f"[{name}] チャンク数: {len(chunks)}") for i, chunk in enumerate(chunks): print(f" Chunk {i+1} ({len(chunk)}文字): {chunk[:60]}...")

パート2: 検索精度比較(15min)

from langchain_chroma import Chroma from langchain_core.documents import Document query = "出張の日当はいくらですか?" # 各チャンク戦略でベクトルDBを構築し検索 for name, chunks in [("固定サイズ", fixed_chunks), ("再帰的", recursive_chunks), ("セマンティック", semantic_chunks)]: docs = [Document(page_content=c) for c in chunks] db = Chroma.from_documents(docs, embeddings, collection_name=name.replace(" ", "_")) results = db.similarity_search_with_score(query, k=2) print(f"\n[{name}]") for doc, score in results: print(f" score={score:.4f} | {doc.page_content[:80]}...") print("\n考察:") print("- 再帰的分割は章ごとにきれいに分かれやすく、検索精度が安定する") print("- 固定サイズは文の途中で切れると検索精度が低下する") print("- セマンティック分割は意味の区切りで分割できるが、計算コストが高い")
補足: 構造ベース分割(Markdown)
from langchain_text_splitters import MarkdownHeaderTextSplitter markdown_text = """ # 有給休暇制度 入社6ヶ月後に10日付与。最大20日。 ## 取得方法 半日単位で取得可能。上長承認が必要。 # リモートワーク制度 週3日まで利用可能。 ## セキュリティ要件 VPN必須。公共Wi-Fi禁止。 """ headers = [("#", "Chapter"), ("##", "Section")] splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers) chunks = splitter.split_text(markdown_text) for chunk in chunks: print(f"metadata={chunk.metadata}") print(f"content={chunk.page_content[:60]}") print("-" * 40)

参考リンク

自走チャレンジ
テーマ: サンプルドキュメント(DL済み)を使い、4種のチャンク戦略(固定サイズ/再帰的/セマンティック/構造ベース)のうち2つを自分で選んで実装し、同じ質問で検索精度を比較してください。
条件: 同一ドキュメント・同一質問で比較すること。検索結果のTop-3チャンクの「関連度」を自分で1-5で評価し、平均スコアを算出してください。
ここで動画を一度止めて、10分間取り組んでください

ヒント: 比較しやすい組み合わせは「固定サイズ vs 再帰的」(初学者向け)または「再帰的 vs セマンティック」(挑戦したい方向け)です。テスト質問は「複数セクションの情報を統合しないと答えられない質問」が差が出やすいです。

講師の解答例を見る
from langchain_text_splitters import ( CharacterTextSplitter, RecursiveCharacterTextSplitter, ) from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma embeddings = OpenAIEmbeddings() # ドキュメント読み込み(共通) with open("sample_doc.txt", "r") as f: text = f.read() # 戦略A: 固定サイズ splitter_fixed = CharacterTextSplitter( chunk_size=500, chunk_overlap=50, separator="" ) chunks_fixed = splitter_fixed.create_documents([text]) db_fixed = Chroma.from_documents(chunks_fixed, embeddings, collection_name="fixed") # 戦略B: 再帰的 splitter_recursive = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=[" ", " ", "。", " ", ""] ) chunks_recursive = splitter_recursive.create_documents([text]) db_recursive = Chroma.from_documents(chunks_recursive, embeddings, collection_name="recursive") # 比較テスト query = "返品ポリシーの条件と手続きの流れを教えてください" print("=== 固定サイズ ===") for doc in db_fixed.similarity_search(query, k=3): print(f" {doc.page_content[:80]}...") print(" === 再帰的 ===") for doc in db_recursive.similarity_search(query, k=3): print(f" {doc.page_content[:80]}...") # 典型的な結果: # 固定サイズ: 文の途中で切れたチャンクが混入(関連度平均3.0) # 再帰的: 段落単位で切れるため文脈が保持される(関連度平均4.3)

解説ポイント: 再帰的チャンカーが多くのケースで固定サイズを上回るのは、日本語の文章構造(段落→文→句)に沿って分割するためです。ただしMarkdownやHTML等の構造化文書では、構造ベースのチャンカーがさらに優位になります。

Section 05 -- 65min(講義25 + ハンズオン40)

検索精度チューニング

RAGの回答品質はRetrievalフェーズの精度に強く依存します。良い文書を引けなければ、どれほど優れたLLMでもまともな回答は返せません。ここでは検索精度を引き上げる5つの手法を実装します。

5つのチューニング手法

手法概要効果
Top-K調整取得するチャンク数を調整(k=3〜5が一般的)適切なkで情報不足/ノイズ過多を防止
Score閾値類似度がn以上のチャンクだけ採用無関係な低スコア文書を除外
Hybrid Searchベクトル検索 + キーワード検索(BM25)の組み合わせ意味検索とキーワード一致の両方を活用
Re-ranking検索結果をCross-EncoderやCohere Rerankで再順位付けTop-K内の順序を最適化
Multi-Query1つの質問を複数表現に変換して検索表現のゆれに対するカバー範囲を拡大

Hybrid Search

ベクトル検索は「意味」で探すのが得意ですが、固有名詞や型番のようなキーワード一致は苦手です。BM25(キーワードベースの検索)と組み合わせることで、両方の強みを活かせます。

%%{init:{'theme':'dark','themeVariables':{'primaryColor':'#00A5BF','primaryBorderColor':'#007A8F','primaryTextColor':'#e8e8e8','lineColor':'#00A5BF','secondaryColor':'#1a1a1a','background':'#141414','mainBkg':'#1a1a1a','nodeBorder':'#00A5BF'}}}%%
graph TD
  Q["ユーザーの質問"] --> V["ベクトル検索
(意味的類似性)"]
  Q --> K["BM25検索
(キーワード一致)"]
  V --> M["結果マージ
(重み付け統合)"]
  K --> M
  M --> R["Re-ranking
(再順位付け)"]
  R --> F["最終結果
(Top-K)"]
Hybrid Search: ベクトル検索とBM25を統合し、Re-rankingで仕上げる

Multi-Query Retriever

「テレワークの上限日数は?」という質問を「リモートワークは週何日まで?」「在宅勤務の制限は?」など複数の表現に変換してから検索します。同じ意味の異なる表現をカバーでき、検索のRecall(再現率)が向上します。

ハンズオン: Hybrid Search + Re-ranking の実装 40min
目標: BM25とベクトル検索を統合するHybrid Searchを実装し、Re-rankingで精度を向上させる

パート1: BM25 + ベクトル検索のHybrid(20min)

from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_chroma import Chroma from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever from langchain_core.documents import Document embeddings = OpenAIEmbeddings(model="text-embedding-3-small") docs = [ Document(page_content="当社の有給休暇は入社6ヶ月後に10日付与されます。勤続年数に応じて最大20日。", metadata={"id": 1}), Document(page_content="リモートワークは週3日まで可能です。HRシステムから事前申請し上長承認を得てください。", metadata={"id": 2}), Document(page_content="出張経費は出張後2週間以内に経費精算システムで申請。領収書の添付が必須です。", metadata={"id": 3}), Document(page_content="育児休業は子が1歳になるまで取得可能。延長申請により最長2歳まで延長できます。", metadata={"id": 4}), Document(page_content="PCの持ち出しには情報システム部への申請が必要。VPN接続を必須とします。", metadata={"id": 5}), Document(page_content="社内Slackの利用ガイドラインに従い、機密情報の投稿は禁止されています。", metadata={"id": 6}), Document(page_content="健康診断は年1回、会社指定の医療機関で受診してください。費用は会社負担。", metadata={"id": 7}), Document(page_content="年末調整の書類は毎年11月中旬までに人事部へ提出してください。", metadata={"id": 8}), ] # ベクトル検索用Retriever vectorstore = Chroma.from_documents(docs, embeddings) vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # BM25(キーワード検索)用Retriever bm25_retriever = BM25Retriever.from_documents(docs) bm25_retriever.k = 3 # Hybrid: 両方を統合(重みを調整可能) ensemble_retriever = EnsembleRetriever( retrievers=[vector_retriever, bm25_retriever], weights=[0.6, 0.4], # ベクトル60% + BM25 40% ) # 比較テスト query = "テレワークの申請方法を教えて" print("=== ベクトル検索のみ ===") for doc in vector_retriever.invoke(query): print(f" [{doc.metadata['id']}] {doc.page_content[:50]}...") print("\n=== BM25のみ ===") for doc in bm25_retriever.invoke(query): print(f" [{doc.metadata['id']}] {doc.page_content[:50]}...") print("\n=== Hybrid Search ===") for doc in ensemble_retriever.invoke(query): print(f" [{doc.metadata['id']}] {doc.page_content[:50]}...")

パート2: Multi-Query Retriever(10min)

from langchain.retrievers.multi_query import MultiQueryRetriever llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3) multi_retriever = MultiQueryRetriever.from_llm( retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), llm=llm, ) import logging logging.basicConfig(level=logging.INFO) query = "在宅勤務のルールは?" results = multi_retriever.invoke(query) print(f"\n=== Multi-Query結果({len(results)}件) ===") for doc in results: print(f" [{doc.metadata['id']}] {doc.page_content[:60]}...")

パート3: Score閾値フィルタリング(10min)

# Score閾値でフィルタリング results_with_score = vectorstore.similarity_search_with_score( "社員食堂のメニューは?", # ドキュメントにない質問 k=3, ) THRESHOLD = 0.5 # Chromaのスコアは距離(小さいほど類似) print("=== Score閾値フィルタリング ===") for doc, score in results_with_score: status = "採用" if score < THRESHOLD else "除外" print(f" score={score:.4f} [{status}] {doc.page_content[:50]}...") print() print("考察:") print("- ドキュメントにない質問でもTop-K件は必ず返される") print("- Score閾値を設定することで無関係な結果を除外できる") print("- 閾値はドメインに応じて調整が必要(0.3〜0.7が一般的)")

理解度チェック: Section 05

Q3. Hybrid Searchが単独のベクトル検索より優れるケースは?

正解: B。ベクトル検索は意味的類似性は得意ですが、固有名詞や型番のような正確なキーワード一致は苦手です。BM25との組み合わせでカバーできます。

参考リンク

Review Hands-on B -- 30min

復習B: 高精度RAGの構築

Sec04-05で学んだチャンク戦略とチューニング手法を組み合わせ、復習Aの基本RAGを高精度版に改良してください。

課題: 高精度RAGパイプライン 30min
成果物: RecursiveCharacterTextSplitter + Hybrid Search + Score閾値を組み合わせた高精度RAG

改良ポイント

  1. チャンク分割をRecursiveCharacterTextSplitter(chunk_size=300, overlap=50)に変更
  2. BM25 + ベクトル検索のEnsembleRetrieverに差し替え
  3. Score閾値を設定し、無関係な結果を除外するロジックを追加
  4. 復習Aと同じ5問で回答精度を比較
Section 06 -- 70min(講義30 + ハンズオン40)

LangGraph入門 -- ステートマシンRAG

LangGraphはLangChain上に構築されたステートマシンフレームワークです。条件分岐、ループ、ツール呼出し、マルチステップ推論をグラフ(有向グラフ)として表現できます。「検索して、結果を評価し、不十分なら再検索する」といった反復的なRAGパターンはLangGraphが最も得意とする領域です。

なぜLangGraphが必要か

LCELのパイプラインは直線的です。A → B → Cと一方向に流れます。「Cの結果が不十分ならBに戻る」「条件によってDかEに分岐する」といった制御フローはLCELだけでは表現が困難。LangGraphはこの問題をステートマシンで解決します。

4つの基本概念

State

グラフ全体で共有するデータ構造。TypedDictで定義し、各ノードがStateを読み書きします。「質問」「検索結果」「回答」「評価スコア」などを保持。

Node

処理の単位。Stateを受け取り、加工して返す関数。「検索ノード」「生成ノード」「評価ノード」など。

Edge

ノード間の遷移。add_edgeで無条件に遷移。前のノード完了後に必ず次のノードが実行されます。

Conditional Edge

条件付き遷移。add_conditional_edgesで、Stateの内容に応じて次のノードを選択。Self-RAGのループ制御に使います。

Stateの定義方法 -- TypedDictの実践パターン

StateはPythonのTypedDictで定義します。各フィールドがグラフ全体の「共有メモリ」として機能し、ノードはStateを受け取って部分的に更新を返します。

from typing import TypedDict, List, Annotated from langgraph.graph import add_messages # メッセージ蓄積用のreducer from langchain_core.messages import BaseMessage # 基本パターン: RAG用State class RAGState(TypedDict): question: str # ユーザーの質問(不変) context: str # 検索で取得した文脈(retrieveノードが更新) answer: str # 生成された回答(generateノードが更新) is_quality_ok: bool # 品質チェック結果(evaluateノードが更新) retry_count: int # リトライ回数(evaluateノードが更新) # 発展パターン: チャット履歴付きState class ChatRAGState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] # メッセージ蓄積 context: str route: str # "rag" or "direct" の分岐用 # ノードの実装ではStateの一部だけを返せばOK def retrieve(state: RAGState) -> dict: # question を読み取り、context だけを更新して返す docs = retriever.invoke(state["question"]) return {"context": "\n".join(d.page_content for d in docs)} # ← answer, is_quality_ok, retry_count は変更しない

Annotatedとreducer(add_messages等)を使うと、ノードが返した値を「上書き」ではなく「追記」にできます。チャット履歴のように蓄積が必要なフィールドで活用してください。

%%{init:{'theme':'dark','themeVariables':{'primaryColor':'#00A5BF','primaryBorderColor':'#007A8F','primaryTextColor':'#e8e8e8','lineColor':'#00A5BF','secondaryColor':'#1a1a1a','background':'#141414','mainBkg':'#1a1a1a','nodeBorder':'#00A5BF'}}}%%
graph TD
  S["START"] --> R["retrieve
ベクトル検索"]
  R --> G["generate
回答生成"]
  G --> E["evaluate
品質チェック"]
  E -->|"品質OK"| END["END"]
  E -->|"品質NG
(retry < 3)"| R
  E -->|"retry >= 3"| END
Self-RAG: 回答品質が不十分なら再検索するループ構造
ハンズオン: Self-RAGをLangGraphで実装 40min
目標: 検索 → 回答生成 → 品質チェック → 必要なら再検索、のループをLangGraphで構築する

パート1: State定義とノード実装(25min)

import os from typing import TypedDict, List, Annotated from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langgraph.graph import StateGraph, END import operator os.environ["OPENAI_API_KEY"] = "sk-..." # --- ドキュメント準備 --- docs = [ Document(page_content="当社の有給休暇は入社6ヶ月後に10日付与。勤続年数に応じて最大20日まで増加。未消化分は翌年度に限り繰越可能。"), Document(page_content="リモートワークは週3日まで可能。コアタイム10:00-15:00はオンライン必須。HRシステムで事前申請し上長承認を得ること。"), Document(page_content="出張経費は出張後2週間以内に申請。日当は国内5,000円、海外10,000円。宿泊費上限は国内15,000円、海外30,000円。"), Document(page_content="育児休業は子が1歳まで取得可能。延長申請で最長2歳まで。復帰後は短時間勤務制度(子が3歳まで)を利用可能。"), Document(page_content="健康診断は年1回、会社指定医療機関で受診。35歳以上は人間ドック受診可(会社負担上限3万円)。"), ] embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents(docs, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # --- State定義 --- class RAGState(TypedDict): question: str context: str answer: str is_quality_ok: bool retry_count: int # --- Node: 検索 --- def retrieve(state: RAGState) -> dict: docs = retriever.invoke(state["question"]) context = "\n\n".join(doc.page_content for doc in docs) return {"context": context} # --- Node: 回答生成 --- def generate(state: RAGState) -> dict: prompt = ChatPromptTemplate.from_template(""" 以下の参考情報のみを使って質問に回答してください。 具体的な数値や条件を含めて回答してください。 参考情報に含まれない場合は「情報が見つかりませんでした」と回答してください。 参考情報: {context} 質問: {question} """) chain = prompt | llm | StrOutputParser() answer = chain.invoke({"context": state["context"], "question": state["question"]}) return {"answer": answer} # --- Node: 品質評価 --- def evaluate(state: RAGState) -> dict: eval_prompt = ChatPromptTemplate.from_template(""" 以下の質問と回答を評価してください。 回答が質問に対して具体的かつ正確に答えているなら「OK」、不十分なら「NG」と1単語で回答してください。 質問: {question} 回答: {answer} """) chain = eval_prompt | llm | StrOutputParser() result = chain.invoke({"question": state["question"], "answer": state["answer"]}) is_ok = "ok" in result.strip().lower() return {"is_quality_ok": is_ok, "retry_count": state.get("retry_count", 0) + 1} # --- 条件分岐 --- def should_retry(state: RAGState) -> str: if state["is_quality_ok"]: return "end" if state.get("retry_count", 0) >= 3: return "end" return "retry"

パート2: グラフ構築と実行(15min)

# --- グラフ構築 --- workflow = StateGraph(RAGState) # ノード追加 workflow.add_node("retrieve", retrieve) workflow.add_node("generate", generate) workflow.add_node("evaluate", evaluate) # エッジ追加 workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "generate") workflow.add_edge("generate", "evaluate") # 条件分岐エッジ workflow.add_conditional_edges( "evaluate", should_retry, { "retry": "retrieve", # 品質NGなら再検索 "end": END, # 品質OKまたはリトライ上限 }, ) # コンパイル app = workflow.compile() # --- 実行 --- questions = [ "有給休暇は最大何日もらえますか?", "海外出張の日当と宿泊費の上限は?", "育児休業後に利用できる制度は?", ] for q in questions: print(f"\nQ: {q}") result = app.invoke({ "question": q, "context": "", "answer": "", "is_quality_ok": False, "retry_count": 0, }) print(f"A: {result['answer']}") print(f" (リトライ回数: {result['retry_count']}, 品質OK: {result['is_quality_ok']})") print("-" * 50)
グラフの可視化

LangGraphのグラフはMermaid形式で出力可能です。

# グラフをMermaid形式で出力 print(app.get_graph().draw_mermaid())

出力されたMermaidコードをMermaid Live Editorに貼り付けると、グラフを視覚的に確認できます。

参考リンク

自走チャレンジ
テーマ: 講師がSelf-RAGを構築したのと同じフレームワークで、以下の要件のグラフを自力で設計してください。「質問 → 関連度チェック → 関連度高ならRAG回答 → 関連度低ならWeb検索フォールバック → 回答生成」
条件: まずノードとエッジを紙(orテキスト)で設計してからコードを書くこと。最低4ノード+1条件分岐エッジを含めること。
ここで動画を一度止めて、10分間取り組んでください

ヒント: まず「ノード一覧」「各ノードの入出力」「条件分岐の判定基準」の3つをテキストで書き出してください。コードはその後です。LangGraphはグラフ設計が8割、コーディングが2割です。

講師の解答例を見る
from langgraph.graph import StateGraph, END from typing import TypedDict, Literal class GraphState(TypedDict): question: str context: str relevance: str # "high" or "low" answer: str source: str # "rag" or "web" # ノード1: ベクトル検索 def retrieve(state: GraphState) -> GraphState: docs = vectorstore.similarity_search(state["question"], k=3) context = " ".join([d.page_content for d in docs]) return {**state, "context": context} # ノード2: 関連度チェック def check_relevance(state: GraphState) -> GraphState: prompt = f"""以下の質問とコンテキストの関連度を判定してください。 質問: {state['question']} コンテキスト: {state['context'][:500]} 関連度が高ければ"high"、低ければ"low"のみを出力してください。""" result = llm.invoke(prompt).content.strip().lower() relevance = "high" if "high" in result else "low" return {**state, "relevance": relevance} # ノード3: RAG回答生成 def rag_answer(state: GraphState) -> GraphState: prompt = f"コンテキストに基づいて回答: {state['context']} 質問: {state['question']}" answer = llm.invoke(prompt).content return {**state, "answer": answer, "source": "rag"} # ノード4: Web検索フォールバック def web_search(state: GraphState) -> GraphState: # TavilyやSerpAPIで検索(簡略化) search_result = f"[Web検索結果] {state['question']}に関する情報..." prompt = f"以下の検索結果に基づいて回答: {search_result} 質問: {state['question']}" answer = llm.invoke(prompt).content return {**state, "answer": answer, "source": "web"} # 条件分岐 def route_by_relevance(state: GraphState) -> Literal["rag_answer", "web_search"]: return "rag_answer" if state["relevance"] == "high" else "web_search" # グラフ構築 graph = StateGraph(GraphState) graph.add_node("retrieve", retrieve) graph.add_node("check_relevance", check_relevance) graph.add_node("rag_answer", rag_answer) graph.add_node("web_search", web_search) graph.set_entry_point("retrieve") graph.add_edge("retrieve", "check_relevance") graph.add_conditional_edges("check_relevance", route_by_relevance) graph.add_edge("rag_answer", END) graph.add_edge("web_search", END) app = graph.compile()

解説ポイント: LangGraphの条件分岐(add_conditional_edges)は、ノードの出力をもとに次のノードを選択する仕組みです。判定関数の戻り値がノード名と一致している必要があるので、typoに注意してください。

Section 07 -- 55min(講義25 + ハンズオン30)

LangSmith -- デバッグ / 評価 / 監視

LangSmithはLangChainの実行をトレース・評価・監視するプラットフォームです。RAGパイプラインの各ステップでどんな入出力が流れたかをリアルタイムに可視化でき、「なぜ変な回答が返ったか」の原因特定に威力を発揮します。

LangSmithの3つの機能

1. トレース

各ステップ(検索、プロンプト構築、LLM呼出し、パース)の入出力をタイムスタンプ付きで記録。遅いステップやエラー箇所を即座に特定できます。

2. 評価

テストデータセットを作成し、RAGの回答精度を定量測定。正解率、F1スコア、LLM-as-Judgeによる評価を自動化できます。

3. 監視

プロダクション環境のトレースをダッシュボードで監視。レイテンシ、エラー率、コストの推移をリアルタイムで追跡します。

セットアップ

# 環境変数の設定(LangSmith接続) export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY="lsv2_..." # LangSmithのAPIキー export LANGCHAIN_PROJECT="c12-rag-demo" # プロジェクト名
Tips: LangSmithの無料枠
smith.langchain.comでアカウント作成すると、月5,000トレースまで無料で利用可能です。ハンズオンの学習用途には十分な量です。
ハンズオン: トレース確認と評価パイプライン 30min
目標: LangSmithでRAGパイプラインのトレースを確認し、テストデータセットで回答精度を定量評価する

パート1: トレースを有効化してRAGを実行(10min)

import os # LangSmith接続 os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "lsv2_..." # ご自身のAPIキー os.environ["LANGCHAIN_PROJECT"] = "c12-rag-demo" # Sec01で作成したRAG Chainをそのまま実行 # chain.invoke() を呼ぶだけで自動的にトレースされる from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser docs = [ Document(page_content="当社の有給休暇は入社6ヶ月後に10日付与。最大20日。"), Document(page_content="リモートワークは週3日まで可能。上長承認が必要。"), Document(page_content="出張経費は2週間以内に申請。領収書添付必須。"), ] embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents(docs, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) prompt = ChatPromptTemplate.from_template(""" 参考情報のみを使って回答してください。 参考情報: {context} 質問: {question} """) def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) # 実行(自動的にLangSmithにトレースが送信される) result = chain.invoke("有給は何日ですか?") print(result) print() print("LangSmith (https://smith.langchain.com/) にアクセスして") print("c12-rag-demo プロジェクトのトレースを確認してください")

パート2: 評価データセットの作成と自動評価(20min)

from langsmith import Client from langsmith.evaluation import evaluate client = Client() # 評価データセットの作成 dataset_name = "c12-rag-eval" # データセットが既に存在する場合は削除(デモ用) try: existing = client.read_dataset(dataset_name=dataset_name) client.delete_dataset(dataset_id=existing.id) except: pass dataset = client.create_dataset(dataset_name=dataset_name) # テストケースを追加 test_cases = [ {"input": "有給休暇は何日もらえますか?", "expected": "入社6ヶ月後に10日。最大20日。"}, {"input": "リモートワークの上限は?", "expected": "週3日まで。上長承認が必要。"}, {"input": "出張経費の申請期限は?", "expected": "出張後2週間以内。"}, {"input": "社員食堂の場所は?", "expected": "情報が見つかりませんでした"}, ] for tc in test_cases: client.create_example( inputs={"question": tc["input"]}, outputs={"answer": tc["expected"]}, dataset_id=dataset.id, ) # 評価実行 def predict(inputs: dict) -> dict: return {"answer": chain.invoke(inputs["question"])} # LLM-as-Judge評価 from langsmith.evaluation import LangChainStringEvaluator evaluator = LangChainStringEvaluator( "qa", config={"llm": ChatOpenAI(model="gpt-4o-mini", temperature=0)}, prepare_data=lambda run, example: { "prediction": run.outputs["answer"], "reference": example.outputs["answer"], "input": example.inputs["question"], }, ) results = evaluate( predict, data=dataset_name, evaluators=[evaluator], experiment_prefix="rag-v1", ) print("評価完了。LangSmithで結果を確認してください。")
LangSmith画面の見方

LangSmithにログインし、プロジェクト一覧から「c12-rag-demo」を選択すると、実行したトレースが表示されます。各トレースをクリックすると、Retriever → PromptTemplate → ChatOpenAI → StrOutputParser の各ステップの入出力、所要時間、トークン数が確認できます。

Datasets & Testingタブから「c12-rag-eval」を開くと、各テストケースの評価結果(Pass/Fail)が一覧で確認できます。

参考リンク

Section 08 -- 60min(講義25 + ハンズオン35)

高度なRAGパターン -- Self-RAG / Adaptive RAG / Corrective RAG

Sec06で実装したSelf-RAGの発展形として、Adaptive RAGとCorrective RAGを学びます。いずれもLangGraphのステートマシンで実現する設計パターンで、質問の難易度や検索結果の質に応じて動的にパイプラインを制御します。

3つのパターン比較

パターン判断対象分岐ロジック適用場面
Self-RAG回答の品質回答生成後に自己評価。不十分なら再検索回答精度を確実に担保したい場合
Adaptive RAG質問の複雑さ質問をルーティング。簡単→直接回答、複雑→RAG実行不要な検索コストを削減したい場合
Corrective RAG検索結果の関連度検索結果を評価。無関係→Web検索にフォールバック社内DBにない質問にも対応したい場合
%%{init:{'theme':'dark','themeVariables':{'primaryColor':'#00A5BF','primaryBorderColor':'#007A8F','primaryTextColor':'#e8e8e8','lineColor':'#00A5BF','secondaryColor':'#1a1a1a','background':'#141414','mainBkg':'#1a1a1a','nodeBorder':'#00A5BF'}}}%%
graph TD
  S["START"] --> CL["classify
質問を分類"]
  CL -->|"simple"| DIR["direct_answer
LLM直接回答"]
  CL -->|"complex"| RET["retrieve
ベクトル検索"]
  RET --> CHK["check_relevance
関連度チェック"]
  CHK -->|"relevant"| GEN["generate
回答生成"]
  CHK -->|"irrelevant"| WEB["web_search
Web検索"]
  WEB --> GEN
  GEN --> EVAL["evaluate
品質評価"]
  EVAL -->|"OK"| END["END"]
  EVAL -->|"NG"| RET
  DIR --> END
Adaptive + Corrective + Self-RAG の統合パターン
ハンズオン: Adaptive RAGの実装 35min
目標: 質問の複雑さを判定し、簡単な質問はLLM直接回答、複雑な質問はRAGパイプラインで処理するAdaptive RAGを構築する
import os from typing import TypedDict, Literal from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langgraph.graph import StateGraph, END os.environ["OPENAI_API_KEY"] = "sk-..." # --- ドキュメント準備 --- docs = [ Document(page_content="当社の有給休暇は入社6ヶ月後に10日付与。勤続年数に応じて最大20日。"), Document(page_content="リモートワークは週3日まで。コアタイム10:00-15:00はオンライン必須。"), Document(page_content="出張日当は国内5,000円、海外10,000円。宿泊費上限は国内15,000円、海外30,000円。"), Document(page_content="育児休業は子が1歳まで。延長で最長2歳。復帰後は短時間勤務(子3歳まで)可能。"), ] embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents(docs, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # --- State --- class AdaptiveState(TypedDict): question: str route: str context: str answer: str # --- Node: 質問分類 --- def classify(state: AdaptiveState) -> dict: prompt = ChatPromptTemplate.from_template(""" 以下の質問を分類してください。 - 社内規則やルールに関する質問 → "rag" - 一般的な知識やPythonの使い方など → "direct" 1単語(ragまたはdirect)のみ回答してください。 質問: {question} """) chain = prompt | llm | StrOutputParser() route = chain.invoke({"question": state["question"]}).strip().lower() route = "rag" if "rag" in route else "direct" return {"route": route} # --- Node: RAG検索 --- def retrieve(state: AdaptiveState) -> dict: docs = retriever.invoke(state["question"]) context = "\n\n".join(doc.page_content for doc in docs) return {"context": context} # --- Node: RAG回答 --- def rag_generate(state: AdaptiveState) -> dict: prompt = ChatPromptTemplate.from_template(""" 参考情報のみを使って回答してください。 参考情報: {context} 質問: {question} """) chain = prompt | llm | StrOutputParser() answer = chain.invoke({"context": state["context"], "question": state["question"]}) return {"answer": answer} # --- Node: 直接回答 --- def direct_answer(state: AdaptiveState) -> dict: prompt = ChatPromptTemplate.from_template( "以下の質問に簡潔に回答してください。\n\n質問: {question}" ) chain = prompt | llm | StrOutputParser() answer = chain.invoke({"question": state["question"]}) return {"answer": answer} # --- ルーティング --- def route_question(state: AdaptiveState) -> str: return state["route"] # --- グラフ構築 --- workflow = StateGraph(AdaptiveState) workflow.add_node("classify", classify) workflow.add_node("retrieve", retrieve) workflow.add_node("rag_generate", rag_generate) workflow.add_node("direct_answer", direct_answer) workflow.set_entry_point("classify") workflow.add_conditional_edges( "classify", route_question, {"rag": "retrieve", "direct": "direct_answer"}, ) workflow.add_edge("retrieve", "rag_generate") workflow.add_edge("rag_generate", END) workflow.add_edge("direct_answer", END) app = workflow.compile() # --- テスト --- questions = [ "有給休暇は何日取れますか?", # → RAG "Pythonでリスト内包表記の書き方は?", # → Direct "海外出張の日当はいくら?", # → RAG "LangChainとは何ですか?", # → Direct ] for q in questions: result = app.invoke({ "question": q, "route": "", "context": "", "answer": "", }) print(f"Q: {q}") print(f"Route: {result['route']}") print(f"A: {result['answer']}") print("-" * 50)
深掘り: Corrective RAGの仕組みと実装

Corrective RAGの核心は「検索結果の関連度を評価し、低ければ別のソースにフォールバックする」という判断ロジックです。通常のRAGは検索結果をそのままLLMに渡しますが、Corrective RAGは「この検索結果は本当に質問と関連があるか?」を間に挟みます。

処理の流れは以下の4ステップです。

  1. ユーザーの質問をベクトルDBで検索(通常のRetrieval)
  2. 取得したチャンクをLLMに渡し、質問との関連度を「relevant / irrelevant / ambiguous」で判定
  3. relevantなら通常通り回答生成。irrelevantならWeb検索(Tavily等)にフォールバック。ambiguousなら両方の結果を統合
  4. 最終的な回答を生成
# Corrective RAGの関連度チェックノード def check_relevance(state: dict) -> dict: """検索結果が質問に関連しているか判定""" prompt = ChatPromptTemplate.from_template(""" 以下の参考情報が質問に関連しているか判定してください。 1単語で回答: relevant / irrelevant / ambiguous 質問: {question} 参考情報: {context} """) chain = prompt | llm | StrOutputParser() relevance = chain.invoke({ "question": state["question"], "context": state["context"], }).strip().lower() if "irrelevant" in relevance: return {"relevance": "irrelevant"} elif "ambiguous" in relevance: return {"relevance": "ambiguous"} return {"relevance": "relevant"} # Web検索フォールバックノード def web_search(state: dict) -> dict: """ベクトルDB検索で関連文書が見つからない場合のフォールバック""" from langchain_community.tools.tavily_search import TavilySearchResults search = TavilySearchResults(max_results=3) results = search.invoke(state["question"]) context = "\n\n".join(r["content"] for r in results) return {"context": context} # 条件分岐: relevanceに応じてルーティング def route_by_relevance(state: dict) -> str: if state["relevance"] == "relevant": return "generate" elif state["relevance"] == "irrelevant": return "web_search" else: # ambiguous return "web_search" # 安全側に倒す

Web検索にはTavily Search API(langchain-communityのTavilySearchResults)が最も統合しやすい選択肢です。Brave Search APIやGoogle Custom Search APIも代替として使えます。Corrective RAGの肝は「判定精度」にあるため、check_relevanceノードのプロンプトを丁寧に作り込む価値があります。

理解度チェック: Section 08

Q4. Adaptive RAGの最大のメリットは?

正解: B。簡単な質問にはLLM直接回答、複雑な質問にはRAGを使うことで、レイテンシとAPIコストを最適化しつつ回答品質を維持できます。

参考リンク

自走チャレンジ
テーマ: 講師のAdaptive RAGを参考に、以下の改善を自力で実装してください。質問の複雑さを3段階(simple/medium/complex)に分類し、simpleはベクトル検索のみ、mediumはHybrid Search、complexはMulti-Query+Re-rankingで検索する条件分岐。
条件: 分類ノード + 3つの検索ノード + 回答生成ノードの最低5ノード構成。テスト質問を各レベル1つずつ用意して動作確認すること。
ここで動画を一度止めて、10分間取り組んでください

ヒント: 質問の複雑さ分類は、LLMに「この質問を回答するのに必要な情報源の数」で判定させると安定します。1つなら simple、2-3なら medium、4つ以上なら complex。

講師の解答例を見る
from langgraph.graph import StateGraph, END from typing import TypedDict, Literal class AdaptiveState(TypedDict): question: str complexity: str # simple / medium / complex context: str answer: str # 複雑さ分類ノード def classify_complexity(state: AdaptiveState) -> AdaptiveState: prompt = f"""以下の質問の複雑さを判定してください。 - simple: 1つの事実で回答できる(例: "Xの定義は?") - medium: 2-3の情報を統合する必要がある(例: "XとYの違いは?") - complex: 4つ以上の情報源や多角的分析が必要(例: "Xの導入効果を評価してください") 質問: {state['question']} complexityのみを出力(simple/medium/complex):""" result = llm.invoke(prompt).content.strip().lower() for level in ["simple", "medium", "complex"]: if level in result: return {**state, "complexity": level} return {**state, "complexity": "medium"} # simple: ベクトル検索のみ def search_simple(state: AdaptiveState) -> AdaptiveState: docs = vectorstore.similarity_search(state["question"], k=3) context = " ".join([d.page_content for d in docs]) return {**state, "context": context} # medium: Hybrid Search(ベクトル + BM25) def search_medium(state: AdaptiveState) -> AdaptiveState: from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever bm25 = BM25Retriever.from_documents(documents, k=3) vector_ret = vectorstore.as_retriever(search_kwargs={"k": 3}) ensemble = EnsembleRetriever( retrievers=[bm25, vector_ret], weights=[0.4, 0.6] ) docs = ensemble.invoke(state["question"]) context = " ".join([d.page_content for d in docs[:5]]) return {**state, "context": context} # complex: Multi-Query + Re-ranking def search_complex(state: AdaptiveState) -> AdaptiveState: # Multi-Query: 元の質問から3つの派生クエリを生成 mq_prompt = f"以下の質問を3つの異なる視点から言い換えてください。1行1クエリで出力: {state['question']}" queries = llm.invoke(mq_prompt).content.strip().split(" ") queries = [q.strip() for q in queries if q.strip()][:3] queries.append(state["question"]) # 全クエリで検索し結果を統合 all_docs = [] for q in queries: all_docs.extend(vectorstore.similarity_search(q, k=3)) # 重複除去 seen = set() unique_docs = [] for d in all_docs: key = d.page_content[:100] if key not in seen: seen.add(key) unique_docs.append(d) # Re-ranking(LLMベース簡易版) rerank_prompt = f"""質問: {state['question']} 以下の文書を関連度順に並べ替え、上位3件の番号を出力してください: """ + " ".join([f"[{i}] {d.page_content[:150]}" for i, d in enumerate(unique_docs)]) ranking = llm.invoke(rerank_prompt).content context = " ".join([d.page_content for d in unique_docs[:5]]) return {**state, "context": context} # 回答生成ノード def generate_answer(state: AdaptiveState) -> AdaptiveState: prompt = f"コンテキスト: {state['context']} 質問: {state['question']} 回答:" answer = llm.invoke(prompt).content return {**state, "answer": answer} # ルーティング def route_by_complexity(state) -> Literal["search_simple", "search_medium", "search_complex"]: return f"search_{state['complexity']}" # グラフ構築 graph = StateGraph(AdaptiveState) graph.add_node("classify", classify_complexity) graph.add_node("search_simple", search_simple) graph.add_node("search_medium", search_medium) graph.add_node("search_complex", search_complex) graph.add_node("generate", generate_answer) graph.set_entry_point("classify") graph.add_conditional_edges("classify", route_by_complexity) graph.add_edge("search_simple", "generate") graph.add_edge("search_medium", "generate") graph.add_edge("search_complex", "generate") graph.add_edge("generate", END) app = graph.compile() # テスト for q in [ "RAGとは何ですか?", # simple "RAGとFine-tuningの使い分けは?", # medium "社内文書検索にRAGを導入する際の設計判断ポイントを網羅的に教えてください" # complex ]: result = app.invoke({"question": q, "complexity": "", "context": "", "answer": ""}) print(f"Q: {q}") print(f" complexity: {result['complexity']}") print(f" A: {result['answer'][:100]}...") print("-" * 50)

解説ポイント: Adaptive RAGの本質は「全ての質問を同じパイプラインで処理しない」ことです。simple質問にMulti-Query+Re-rankingを使うとレイテンシとコストが無駄になり、complex質問に単純検索だけだと回答品質が落ちます。質問の性質に応じてパイプラインを切り替えることで、コストと品質のバランスを最適化できます。

Section 09 -- 140min(講義10 + ハンズオン130)

総合ハンズオン: 社内ドキュメント検索システム

ここまで学んだ全技術を統合し、社内ドキュメント(PDF/テキスト)を対象としたRAG検索システムを構築します。10ステップで段階的に完成させていきます。

前提条件
OpenAI APIキーが有効であること。LangSmithアカウントがセットアップ済みであること。Python 3.10以上を推奨します。
社内ドキュメントRAG検索システム構築 130min
成果物: PDF/テキストから検索・回答できるRAGシステム。LangGraphによるマルチステップ処理、LangSmithによる評価付き。

Step 1: 要件定義(10min)

対象ドキュメント、想定質問、求める回答品質を整理してください。DLテンプレートの「RAG設計シート」を使います。

Step 2: ドキュメント読込 + チャンク設定(15min)

from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.documents import Document # サンプルドキュメント(DLテンプレートの内容を使用) raw_docs = [ Document(page_content="""就業規則 第3章 有給休暇 当社の有給休暇は、入社6ヶ月経過後に10日が付与されます。 勤続年数に応じて付与日数は増加し、6年6ヶ月以上で最大20日。 未消化分は翌年度に限り繰越可能。半日単位の取得も可能。 時間単位の有給休暇は年5日分(40時間)まで取得可能です。 事前に上長の承認を得てください。急な体調不良の場合は事後申請可。""", metadata={"source": "就業規則", "chapter": "有給休暇"}), Document(page_content="""就業規則 第5章 リモートワーク リモートワークは週3日まで利用可能です。 コアタイム(10:00-15:00)中はオンラインで連絡が取れる状態を維持してください。 利用にあたってはHRシステムから事前申請し、上長の承認を得ること。 セキュリティの観点からVPN接続を必須とし、公共Wi-Fiでの業務は禁止。 自宅以外(カフェ、コワーキング等)での作業は部長承認が必要です。""", metadata={"source": "就業規則", "chapter": "リモートワーク"}), Document(page_content="""経費規程 第2章 出張経費 出張経費は出張完了後2週間以内に経費精算システムで申請してください。 申請には領収書原本(またはスキャン画像)の添付が必須です。 日当は国内出張5,000円、海外出張10,000円。 宿泊費上限は国内15,000円、海外30,000円(税込)。 上限超過の場合は事前に部長承認が必要です。 交通費は原則として最も経済的な経路を選択してください。""", metadata={"source": "経費規程", "chapter": "出張経費"}), Document(page_content="""情報セキュリティポリシー 第1章 社内情報の取り扱いは機密レベルに応じて3段階に分類されます。 レベル1(公開可): 会社概要、採用情報など。 レベル2(社外秘): 社内マニュアル、組織図、売上データなど。 レベル3(極秘): 個人情報、未公開の経営計画、契約書など。 レベル2以上の情報をAIツールに入力する場合は、情報システム部の承認が必要です。 無料版AIツールへのレベル2以上の情報入力は禁止されています。""", metadata={"source": "情報セキュリティ", "chapter": "機密分類"}), Document(page_content="""福利厚生制度 育児休業は子が1歳になるまで取得可能。延長申請により最長2歳まで。 復帰後は短時間勤務制度(子が3歳になるまで)を利用可能。 健康診断は年1回、会社指定の医療機関で受診。費用は会社負担。 35歳以上は人間ドック受診可(会社負担上限3万円)。 カフェテリアプラン(年間5万円分)で書籍購入、資格取得費用等に充当可能。""", metadata={"source": "福利厚生", "chapter": "各種制度"}), ] # RecursiveCharacterTextSplitterで分割 splitter = RecursiveCharacterTextSplitter( chunk_size=300, chunk_overlap=50, separators=["\n\n", "\n", "。", "、", " ", ""], ) chunks = splitter.split_documents(raw_docs) print(f"チャンク数: {len(chunks)}") for i, chunk in enumerate(chunks): print(f" [{i}] ({len(chunk.page_content)}文字) {chunk.metadata} | {chunk.page_content[:40]}...")

Step 3: ベクトルDB構築(10min)

from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents( chunks, embeddings, collection_name="company_docs", persist_directory="./chroma_db", ) print(f"ベクトルDB構築完了。チャンク数: {vectorstore._collection.count()}")

Step 4: 基本RAGパイプライン構築(15min)

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) rag_prompt = ChatPromptTemplate.from_template(""" あなたは社内規則に詳しいアシスタントです。 以下の参考情報のみを使って質問に回答してください。 具体的な数値や条件を含めてください。 参考情報に含まれない場合は「社内規則に該当する情報が見つかりませんでした」と回答してください。 回答の最後に、参照した情報源(source)を明記してください。 参考情報: {context} 質問: {question} """) def format_docs(docs): return "\n\n".join( f"[{doc.metadata.get('source', '不明')}] {doc.page_content}" for doc in docs ) basic_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | rag_prompt | llm | StrOutputParser() ) # テスト print(basic_chain.invoke("有給休暇は何日もらえますか?"))

Step 5: Hybrid Search + Re-ranking 追加(15min)

from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever # BM25 Retriever bm25_retriever = BM25Retriever.from_documents(chunks) bm25_retriever.k = 3 # Vector Retriever vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # Ensemble(Hybrid) hybrid_retriever = EnsembleRetriever( retrievers=[vector_retriever, bm25_retriever], weights=[0.6, 0.4], ) # Hybrid版Chain hybrid_chain = ( {"context": hybrid_retriever | format_docs, "question": RunnablePassthrough()} | rag_prompt | llm | StrOutputParser() ) # 比較テスト q = "VPN接続のルールは?" print("=== Basic RAG ===") print(basic_chain.invoke(q)) print("\n=== Hybrid RAG ===") print(hybrid_chain.invoke(q))

Step 6: LangGraphでマルチステップRAG化(20min)

from typing import TypedDict from langgraph.graph import StateGraph, END class SystemState(TypedDict): question: str route: str context: str answer: str is_quality_ok: bool retry_count: int def classify_question(state: SystemState) -> dict: prompt = ChatPromptTemplate.from_template( "質問を分類: 社内規則→'rag'、一般知識→'direct'。1単語のみ。\n質問: {question}" ) chain = prompt | llm | StrOutputParser() route = chain.invoke({"question": state["question"]}).strip().lower() return {"route": "rag" if "rag" in route else "direct"} def retrieve_docs(state: SystemState) -> dict: docs = hybrid_retriever.invoke(state["question"]) return {"context": format_docs(docs)} def generate_answer(state: SystemState) -> dict: chain = rag_prompt | llm | StrOutputParser() answer = chain.invoke({"context": state["context"], "question": state["question"]}) return {"answer": answer} def direct_answer(state: SystemState) -> dict: prompt = ChatPromptTemplate.from_template("簡潔に回答: {question}") chain = prompt | llm | StrOutputParser() return {"answer": chain.invoke({"question": state["question"]})} def evaluate_answer(state: SystemState) -> dict: eval_prompt = ChatPromptTemplate.from_template( "回答が具体的で正確なら'OK'、不十分なら'NG'。1単語。\n質問:{question}\n回答:{answer}" ) chain = eval_prompt | llm | StrOutputParser() result = chain.invoke({"question": state["question"], "answer": state["answer"]}) return {"is_quality_ok": "ok" in result.lower(), "retry_count": state.get("retry_count", 0) + 1} def route_fn(state: SystemState) -> str: return state["route"] def retry_fn(state: SystemState) -> str: if state["is_quality_ok"] or state.get("retry_count", 0) >= 2: return "end" return "retry" # グラフ構築 wf = StateGraph(SystemState) wf.add_node("classify", classify_question) wf.add_node("retrieve", retrieve_docs) wf.add_node("generate", generate_answer) wf.add_node("direct", direct_answer) wf.add_node("evaluate", evaluate_answer) wf.set_entry_point("classify") wf.add_conditional_edges("classify", route_fn, {"rag": "retrieve", "direct": "direct"}) wf.add_edge("retrieve", "generate") wf.add_edge("generate", "evaluate") wf.add_conditional_edges("evaluate", retry_fn, {"retry": "retrieve", "end": END}) wf.add_edge("direct", END) app = wf.compile() # テスト test_qs = [ "出張の日当と宿泊費上限を教えて", "AIツールに社内情報を入力する際のルールは?", "Pythonのリスト内包表記の書き方は?", ] for q in test_qs: r = app.invoke({"question": q, "route": "", "context": "", "answer": "", "is_quality_ok": False, "retry_count": 0}) print(f"Q: {q}") print(f"Route: {r['route']} | Retry: {r['retry_count']}") print(f"A: {r['answer']}") print("=" * 60)

Step 7: LangSmithでトレース + 評価設定(15min)

import os os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "lsv2_..." os.environ["LANGCHAIN_PROJECT"] = "c12-final-system" # 上記のappを再実行するだけで自動トレース for q in test_qs: r = app.invoke({"question": q, "route": "", "context": "", "answer": "", "is_quality_ok": False, "retry_count": 0}) print(f"Q: {q} → A: {r['answer'][:50]}...") print("\nLangSmith (https://smith.langchain.com/) で") print("c12-final-system プロジェクトのトレースを確認してください")

Step 8: テストケース10問で精度測定(15min)

test_cases = [ {"q": "有給休暇は何日もらえますか?", "expected": "10日(入社6ヶ月後)。最大20日。"}, {"q": "リモートワークは週何日まで?", "expected": "週3日まで"}, {"q": "出張の日当は?", "expected": "国内5,000円、海外10,000円"}, {"q": "宿泊費の上限は?", "expected": "国内15,000円、海外30,000円"}, {"q": "育児休業は何歳まで延長できる?", "expected": "最長2歳まで"}, {"q": "カフェテリアプランの金額は?", "expected": "年間5万円分"}, {"q": "AIツールに入力してはいけない情報は?", "expected": "レベル2以上(社外秘、極秘)の情報"}, {"q": "健康診断で人間ドックを受けられるのは?", "expected": "35歳以上。会社負担上限3万円"}, {"q": "時間単位の有給は年何時間?", "expected": "年5日分(40時間)"}, {"q": "社員食堂のメニューは?", "expected": "情報なし"}, ] correct = 0 for tc in test_cases: r = app.invoke({"question": tc["q"], "route": "", "context": "", "answer": "", "is_quality_ok": False, "retry_count": 0}) # 簡易判定: 期待キーワードが含まれているか keywords = [kw for kw in tc["expected"].split("、") if len(kw) > 1] hit = any(kw in r["answer"] for kw in keywords) if keywords else "見つかりません" in r["answer"] correct += 1 if hit else 0 status = "PASS" if hit else "FAIL" print(f"[{status}] Q: {tc['q']}") print(f" Expected: {tc['expected']}") print(f" Got: {r['answer'][:80]}...") print() print(f"精度: {correct}/{len(test_cases)} ({correct/len(test_cases)*100:.0f}%)")

Step 9: 精度改善(15min)

精度が目標に達しない場合、以下のパラメータを調整してStep 8を再実行してください。

Step 10: ドキュメント整備(10min)

完成したシステムの設計をDLテンプレートの「RAG設計シート」に記録してください。チャンク戦略、検索設定、評価結果を記載しておくと、後日のチューニングが効率的に進みます。

最終チェック: Section 09

Q5. RAGシステムの精度改善で最も効果が高いのは?

正解: B。RAGの回答品質はRetrievalフェーズの精度に最も強く依存します。適切なチャンクサイズ、overlap、検索手法の選択が、LLMの変更よりも効果的です。
Downloads

DLテンプレート

各ハンズオンで使用するテンプレートとサンプルファイルです。

RAG設計シート.md requirements.txt サンプルドキュメント.txt テストケース集.md LangSmith評価シート.md