取り組みの背景 — RAG の概念を整理しつつ、なぜ今必要なのか
LLM(大規模言語モデル)は汎用的な知識を持っていますが、自社固有の情報——社内ドキュメント、API仕様書、ナレッジベース、顧客対応履歴——には対応できません。ファインチューニングは高コストで、データが更新されるたびに再学習が必要です。
RAG(Retrieval-Augmented Generation) はこの課題を解決するアーキテクチャです。ユーザーの質問に関連するドキュメントをベクトル検索で取得し、その情報をコンテキストとして LLM に渡すことで、正確で最新の回答を生成します。
ここで扱うのはAntigravity の AI 支援を最大限に活用しながら、RAG パイプラインをゼロから構築する方法を解説します。対象読者は、Python の基礎知識があり、LLM アプリケーションの開発に取り組みたいエンジニアです。なお、プロンプトの書き方に不安がある方は先にAntigravity プロンプトエンジニアリング上級ガイドを読んでおくと、RAG のプロンプト設計がスムーズになります。
RAG アーキテクチャの全体設計
RAG システムは大きく3つのフェーズで構成されます。
インジェストフェーズ(データ取り込み)
ドキュメントを読み込み、適切なサイズのチャンクに分割し、Embedding モデルでベクトル化してデータベースに格納します。
# Antigravity で生成した RAG インジェストパイプライン
# document_ingestor.py
import os
from pathlib import Path
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
DirectoryLoader,
TextLoader,
PyPDFLoader,
UnstructuredMarkdownLoader,
)
from langchain_google_genai import GoogleGenerativeAIEmbeddings
import chromadb
from chromadb.config import Settings
# --- 設定 ---
CHROMA_PERSIST_DIR = "./chroma_db"
COLLECTION_NAME = "company_knowledge"
CHUNK_SIZE = 800 # トークン数ではなく文字数
CHUNK_OVERLAP = 200 # チャンク間のオーバーラップ
def load_documents(source_dir: str) -> list:
"""複数形式のドキュメントを一括読み込み"""
loaders = {
"**/*.txt": TextLoader,
"**/*.md": UnstructuredMarkdownLoader,
"**/*.pdf": PyPDFLoader,
}
documents = []
for glob_pattern, loader_cls in loaders.items():
loader = DirectoryLoader(
source_dir,
glob=glob_pattern,
loader_cls=loader_cls,
show_progress=True,
)
documents.extend(loader.load())
return documents
def chunk_documents(documents: list) -> list:
"""ドキュメントを意味的に適切なサイズに分割"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
separators=["\n## ", "\n### ", "\n\n", "\n", "。", ".", " "],
length_function=len,
)
return splitter.split_documents(documents)
def create_embeddings_and_store(chunks: list):
"""Embedding を生成し ChromaDB に格納"""
# Gemini の Embedding モデルを使用
embeddings = GoogleGenerativeAIEmbeddings(
model="models/text-embedding-004",
google_api_key=os.getenv("GOOGLE_API_KEY"),
)
client = chromadb.PersistentClient(
path=CHROMA_PERSIST_DIR,
settings=Settings(anonymized_telemetry=False),
)
# コレクションを作成(既存の場合はリセット)
collection = client.get_or_create_collection(
name=COLLECTION_NAME,
metadata={"hnsw:space": "cosine"}, # コサイン類似度
)
# バッチ処理で効率的に格納
batch_size = 100
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
texts = [chunk.page_content for chunk in batch]
metadatas = [chunk.metadata for chunk in batch]
ids = [f"doc_{i + j}" for j in range(len(batch))]
# Embedding 生成
vectors = embeddings.embed_documents(texts)
collection.add(
ids=ids,
documents=texts,
embeddings=vectors,
metadatas=metadatas,
)
print(f" 格納完了: {i + len(batch)}/{len(chunks)} チャンク")
print(f"✅ 全 {len(chunks)} チャンクを ChromaDB に格納しました")
# --- 実行 ---
if __name__ == "__main__":
docs = load_documents("./knowledge_base")
print(f"📄 {len(docs)} ドキュメントを読み込みました")
chunks = chunk_documents(docs)
print(f"✂️ {len(chunks)} チャンクに分割しました")
create_embeddings_and_store(chunks)
期待される出力:
📄 47 ドキュメントを読み込みました
✂️ 312 チャンクに分割しました
格納完了: 100/312 チャンク
格納完了: 200/312 チャンク
格納完了: 300/312 チャンク
格納完了: 312/312 チャンク
✅ 全 312 チャンクを ChromaDB に格納しました
検索フェーズ(リトリーバル)
ユーザーの質問をベクトル化し、類似度の高いチャンクを検索します。ここでの精度が RAG 全体の品質を左右します。
生成フェーズ(ジェネレーション)
検索結果をプロンプトに組み込み、LLM が回答を生成します。
チャンク戦略の設計 — RAG 品質の80%はここで決まる
RAG の品質を最も大きく左右するのは、ドキュメントをどうチャンクに分割するかです。Antigravity のエージェントに「チャンク戦略を最適化して」と依頼すると、以下のような高度な分割ロジックを提案してくれます。
セマンティックチャンキング
単純な文字数分割ではなく、文章の意味的なまとまりを保った分割を行います。
# semantic_chunker.py — 意味的なまとまりを保つ高度なチャンク分割
from langchain_experimental.text_splitter import SemanticChunker
from langchain_google_genai import GoogleGenerativeAIEmbeddings
import os
def create_semantic_chunks(documents: list) -> list:
"""Embedding ベースのセマンティックチャンキング"""
embeddings = GoogleGenerativeAIEmbeddings(
model="models/text-embedding-004",
google_api_key=os.getenv("GOOGLE_API_KEY"),
)
# 文の意味的な距離が閾値を超えたところで分割
chunker = SemanticChunker(
embeddings,
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=90, # 上位10%の距離で分割
)
chunks = []
for doc in documents:
split = chunker.create_documents(
[doc.page_content],
metadatas=[doc.metadata],
)
chunks.extend(split)
return chunks
# --- 親子チャンク戦略(Parent-Child Chunking)---
class ParentChildChunker:
"""
大きな「親チャンク」で文脈を保持し、
小さな「子チャンク」で検索精度を上げる二段構え
"""
def __init__(self, parent_size=2000, child_size=400, overlap=100):
self.parent_splitter = RecursiveCharacterTextSplitter(
chunk_size=parent_size,
chunk_overlap=0,
)
self.child_splitter = RecursiveCharacterTextSplitter(
chunk_size=child_size,
chunk_overlap=overlap,
)
def split(self, documents: list) -> tuple:
"""親チャンクと子チャンクを同時に生成"""
parent_chunks = []
child_chunks = []
for doc in documents:
parents = self.parent_splitter.split_documents([doc])
for idx, parent in enumerate(parents):
parent.metadata["parent_id"] = f"{doc.metadata.get('source', 'unknown')}_{idx}"
parent_chunks.append(parent)
# 親チャンクをさらに小さく分割
children = self.child_splitter.split_documents([parent])
for child in children:
child.metadata["parent_id"] = parent.metadata["parent_id"]
child_chunks.append(child)
return parent_chunks, child_chunks
# 期待される動作:
# - 子チャンクで精密な検索を行い
# - ヒットした子チャンクの親チャンクをコンテキストとして LLM に渡す
# → 検索精度と文脈の豊かさを両立
チャンクサイズの選び方
チャンクサイズは一律に決められるものではなく、データの性質によって最適解が異なります。
Antigravity のエージェントに各パターンのベンチマークコードを生成させ、実データで比較するのが最も確実です。一般的な目安として、技術ドキュメント(API リファレンス等)は 400〜600 文字の小さめのチャンクが有効です。コードの文脈は短い範囲で完結することが多いためです。一方、ナラティブなドキュメント(社内Wiki、議事録等)は 800〜1200 文字のやや大きめのチャンクが適しています。文脈が長い文章では、短すぎるチャンクだと意味が失われます。法務・契約文書のように厳密性が求められる場合は、1500〜2000 文字の大きなチャンクを使い、条項全体の文脈を保持する点が肝心です。
リトリーバルの高度化 — ハイブリッド検索とリランキング
ベクトル検索だけでは不十分なケースがあります。キーワードの完全一致が重要な場面(エラーコード検索、商品型番の特定など)では、BM25 などのキーワード検索と組み合わせたハイブリッド検索が効果的です。
# hybrid_retriever.py — ベクトル検索 + BM25 のハイブリッド検索
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma
from langchain_google_genai import GoogleGenerativeAIEmbeddings
import os
def create_hybrid_retriever(chunks: list, k: int = 5):
"""ベクトル検索と BM25 を組み合わせたハイブリッドリトリーバー"""
embeddings = GoogleGenerativeAIEmbeddings(
model="models/text-embedding-004",
google_api_key=os.getenv("GOOGLE_API_KEY"),
)
# 1. ベクトル検索リトリーバー
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
collection_name="hybrid_search",
)
vector_retriever = vectorstore.as_retriever(
search_type="mmr", # Maximal Marginal Relevance
search_kwargs={"k": k, "fetch_k": k * 3},
)
# 2. BM25 キーワード検索リトリーバー
bm25_retriever = BM25Retriever.from_documents(
chunks,
k=k,
)
# 3. アンサンブル(重み付け統合)
ensemble = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4], # ベクトル検索を少し重視
)
return ensemble
# --- リランキング(Cohere Reranker を使用)---
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
def add_reranking(base_retriever, top_n: int = 3):
"""検索結果をリランカーで再スコアリング"""
reranker = CohereRerank(
model="rerank-v3.5",
cohere_api_key=os.getenv("COHERE_API_KEY"),
top_n=top_n,
)
return ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=base_retriever,
)
# 使用例:
# retriever = create_hybrid_retriever(chunks, k=10)
# reranked_retriever = add_reranking(retriever, top_n=3)
# results = reranked_retriever.invoke("Stripe Webhook の署名検証方法は?")
#
# 期待される動作:
# → BM25 が「Stripe」「Webhook」「署名検証」のキーワードマッチで候補を取得
# → ベクトル検索が意味的に近い文書も取得
# → リランカーが上位3件に絞り込み、最も関連性の高い文書を返す
MMR(Maximal Marginal Relevance)による多様性確保
ベクトル検索で search_type="mmr" を指定すると、類似度だけでなく結果の多様性も考慮されます。同じような内容のチャンクばかりが返されることを防ぎ、LLM に多角的な情報を提供できます。
生成フェーズ — プロンプト設計とガードレール
検索で取得したコンテキストを LLM に適切に渡すプロンプト設計が、回答品質の鍵を握ります。
# rag_chain.py — RAG チェーンの構築
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
def build_rag_chain(retriever):
"""検索結果を元に回答を生成する RAG チェーン"""
# --- プロンプトテンプレート ---
template = ChatPromptTemplate.from_messages([
("system", """あなたは社内ナレッジに基づいて正確に回答するアシスタントです。
以下のルールを厳守してください:
1. 提供されたコンテキストの情報のみに基づいて回答する
2. コンテキストに情報がない場合は「この情報は社内ナレッジベースに見つかりませんでした」と明示する
3. 推測や外部知識を混ぜない
4. 回答の根拠となったドキュメントのソースを明記する
5. 技術的な内容にはコード例を含める
コンテキスト:
{context}"""),
("human", "{question}"),
])
# --- LLM ---
llm = ChatGoogleGenerativeAI(
model="gemini-2.5-pro",
temperature=0.1, # 事実に基づく回答のため低めに設定
max_output_tokens=4096,
)
# --- チェーン構築 ---
def format_docs(docs):
"""検索結果を整形してコンテキスト文字列にする"""
formatted = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "不明")
formatted.append(
f"[ソース {i}: {source}]\n{doc.page_content}"
)
return "\n\n---\n\n".join(formatted)
chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
}
| template
| llm
| StrOutputParser()
)
return chain
# --- ストリーミング対応版 ---
def query_with_streaming(chain, question: str):
"""ストリーミングで回答を表示"""
print(f"\n📝 質問: {question}\n")
print("=" * 60)
for chunk in chain.stream(question):
print(chunk, end="", flush=True)
print("\n" + "=" * 60)
# 使用例:
# chain = build_rag_chain(reranked_retriever)
# query_with_streaming(chain, "新規プロジェクトのセットアップ手順を教えてください")
#
# 期待される出力:
# 📝 質問: 新規プロジェクトのセットアップ手順を教えてください
# ============================================================
# 社内ナレッジベースに基づくと、新規プロジェクトのセットアップは
# 以下の手順で行います:
# 1. GitHub でリポジトリを作成し...
# [ソース: docs/setup-guide.md]
# ============================================================
評価とモニタリング — RAG の品質を定量化する
RAG システムは「動いている」だけでは不十分です。回答品質を定量的に測定し、継続的に改善する仕組みが必要です。
# rag_evaluator.py — RAG パイプラインの評価フレームワーク
from dataclasses import dataclass
from langchain_google_genai import ChatGoogleGenerativeAI
import json
@dataclass
class EvalResult:
question: str
expected: str
actual: str
faithfulness: float # コンテキストに忠実か(0-1)
relevance: float # 質問に関連しているか(0-1)
completeness: float # 回答が十分か(0-1)
class RAGEvaluator:
"""LLM-as-a-Judge パターンで RAG 回答を評価"""
def __init__(self):
self.judge = ChatGoogleGenerativeAI(
model="gemini-2.5-pro",
temperature=0,
)
def evaluate_faithfulness(
self, context: str, answer: str
) -> float:
"""回答がコンテキストに忠実かを評価"""
prompt = f"""以下の回答が、提供されたコンテキストの情報のみに基づいているか評価してください。
コンテキスト:
{context}
回答:
{answer}
0.0(完全にコンテキスト外)〜 1.0(完全にコンテキストに忠実)のスコアを、
JSON形式 {{"score": 0.X, "reason": "理由"}} で返してください。"""
response = self.judge.invoke(prompt)
result = json.loads(response.content)
return result["score"]
def evaluate_relevance(
self, question: str, answer: str
) -> float:
"""回答が質問に関連しているかを評価"""
prompt = f"""以下の回答が質問に対して適切に回答しているか評価してください。
質問: {question}
回答: {answer}
0.0(無関係)〜 1.0(完全に的確)のスコアを、
JSON形式 {{"score": 0.X, "reason": "理由"}} で返してください。"""
response = self.judge.invoke(prompt)
result = json.loads(response.content)
return result["score"]
def run_eval_suite(
self, chain, retriever, test_cases: list[dict]
) -> list[EvalResult]:
"""テストケース一覧で評価を実行"""
results = []
for case in test_cases:
# 検索結果を取得
docs = retriever.invoke(case["question"])
context = "\n".join(d.page_content for d in docs)
# 回答を生成
answer = chain.invoke(case["question"])
# 評価
faithfulness = self.evaluate_faithfulness(context, answer)
relevance = self.evaluate_relevance(case["question"], answer)
results.append(EvalResult(
question=case["question"],
expected=case.get("expected", ""),
actual=answer,
faithfulness=faithfulness,
relevance=relevance,
completeness=0.0, # 別途評価
))
# サマリー出力
avg_faith = sum(r.faithfulness for r in results) / len(results)
avg_rel = sum(r.relevance for r in results) / len(results)
print(f"\n📊 評価結果サマリー({len(results)} ケース)")
print(f" 忠実性(Faithfulness): {avg_faith:.2f}")
print(f" 関連性(Relevance): {avg_rel:.2f}")
return results
# テストケース例:
# test_cases = [
# {"question": "デプロイ手順は?", "expected": "wrangler deploy を実行..."},
# {"question": "API キーの発行方法は?", "expected": "管理画面から..."},
# ]
# evaluator = RAGEvaluator()
# results = evaluator.run_eval_suite(chain, retriever, test_cases)
本番運用のベストプラクティス
RAG システムを本番環境で安定運用するためのポイントをまとめます。
インクリメンタルインジェスト: ドキュメントが更新されたら差分だけをベクトルDBに反映します。ファイルのハッシュ値を記録し、変更があったファイルのみ再 Embedding を実行する設計にすると、コストと処理時間を大幅に削減できます。
キャッシュ戦略: 同じ質問に対する回答をキャッシュすることで、レイテンシとAPI費用を削減します。質問のベクトル類似度が閾値(例: 0.95)以上の場合にキャッシュヒットとみなす「セマンティックキャッシュ」が効果的です。
フォールバック設計: ベクトル検索の結果が低スコア(例: コサイン類似度 0.3 未満)の場合、「関連情報が見つかりませんでした」と正直に回答する方が、ハルシネーションを含む回答より遥かに信頼性が高くなります。
監視メトリクス: 本番環境では以下の指標を常時モニタリングすべきです。検索レイテンシ(P95 が 500ms 以下を目標)、回答生成レイテンシ(P95 が 3秒以下を目標)、検索結果の平均類似度スコア(低下傾向はデータの陳腐化を示唆)、ユーザーフィードバック率(👍/👎 の比率)。
個人開発者の視点から(実体験メモ)
ここまでの要点
RAG は「LLM を自社データで賢くする」最も実践的なアプローチです。ここで扱うのはAntigravity の AI 支援を活用しながら、インジェスト → 検索 → 生成 → 評価の全工程を実装しました。
複数の LLM を使い分けたい場合は、Antigravity マルチモデル完全攻略ガイドも参考になります。また、Embedding やモデルの軽量化に興味がある方は AI モデル量子化入門もあわせてご覧ください。
重要なポイントを振り返ります。チャンク戦略が RAG 品質の8割を決めるため、セマンティックチャンキングや親子チャンク戦略を検討すること。ハイブリッド検索(ベクトル + BM25)とリランキングで検索精度を大幅に向上させられること。LLM-as-a-Judge パターンで回答品質を定量的に測定し、継続的に改善すること。
RAG の設計パターンをさらに深く