HANDS-ON / 55 MIN
RAGの構成要素 ── 5つのパーツを順に体験する
ファイル: 04_rag_components.py
EXERCISE
スクリプトを実行する
python 04_rag_components.py を実行
Document Loader
→
Text Splitter
→
Embedding
→
Vector Store
→
Retriever
Step 1 ── Document Loader でPDFを読み込む
# Step 1: Document Loader
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("data/sample.pdf")
pages = loader.load()
print(f"読み込んだページ数: {len(pages)}")
print(pages[0].page_content[:200])
PDFの内容がテキストとして取得される。RAGの第一歩は、対象文書をシステムに取り込むこと。
Step 2 ── Text Splitter で文書を分割する
# Step 2: Text Splitter
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
)
chunks = splitter.split_documents(pages)
chunk_size=500
500文字ごとに区切る。LLMに渡せるテキスト長には上限があるため、文書全体ではなく適切なサイズに分割する。
chunk_overlap=50
前後50文字を重複させる。文脈が途切れるのを防ぐ。
Step 3 ── Embedding でテキストをベクトルに変換する
# Step 3: Embedding
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
text = "RAGとは検索拡張生成のことです"
vector = embeddings.embed_query(text)
print(f"ベクトルの次元数: {len(vector)}")
「意味が近いテキスト同士は数値的にも近くなる」という性質がある。これが類似検索の基盤。
# 2つの文の類似度を比較
text_a = "RAGとは検索拡張生成です"
text_b = "Retrieval-Augmented Generationの略です"
text_c = "今日の天気は晴れです"
# コサイン類似度: 1に近いほど意味が近い
# RAG同士 → 0.4〜0.5前後（意味が近いので高い）
# RAGと天気 → 0.1〜0.2前後（関係ない話題なので低い）
Step 4 ── Vector Store に文書を格納する
# Step 4: Vector Store
from langchain_community.vectorstores import FAISS
vectorstore = FAISS.from_documents(chunks, embeddings)
print(f"Vector Storeへの格納完了。格納チャンク数: {len(chunks)}")
FAISSはMeta社が開発した、ローカルで動作するVector Store。本番環境ではPinecone、Weaviateなどのクラウドサービスを使うこともある。
Step 5 ── Retriever で検索する
# Step 5: Retriever
retriever = vectorstore.as_retriever(
search_kwargs={"k": 3} # 上位3件を返す
)
question = "PCが起動しない場合はどうすればいいですか"
results = retriever.invoke(question)
for i, doc in enumerate(results):
print(f"--- 結果{i+1} ---")
print(doc.page_content[:200])
質問の「意味」に近い文書が、大量のチャンクの中から見つかる。これがRAGの核心。
EXERCISE
自分で試してみよう
- chunk_size を 200 に変えてチャンク数の変化を観察する
- Step 3の文を自由に変えて類似度の変化を確認する
- Step 5の質問を変えて検索結果の違いを見る
この操作がどう活きるか
最終演習では、このStep 1〜5が retrieve_node という1つの関数に集約される。
# 最終演習でのretrieve_node
def retrieve_node(state):
question = state["question"]
documents = retriever.invoke(question)
return {"documents": documents}