▣ アプリ開発/2026-04-20上級

Gemma 4 で RAG パイプラインを構築する — ローカルLLMで自分だけの検索AIを作る

Gemma 4をベースにしたRAG（Retrieval-Augmented Generation）パイプラインを、ChromaDB＋LangChainで構築する完全実装ガイド。ローカル実行からAPI活用まで、実際に動くコードで全工程を解説します。

Gemma 4²¹ RAG⁵ ChromaDB LangChain³ ローカルLLM²⁰ ベクトル検索 Python¹³

✦ プレミアム記事

「自分のドキュメントに基づいて回答するAI」を作りたいと思ったことはないでしょうか。社内ドキュメントの検索、アプリのFAQボット、コードベースへの質問応答など、応用範囲は広い。

RAG（Retrieval-Augmented Generation）は、このニーズに応える実践的なアーキテクチャです。LLMの知識だけに頼るのではなく、都度関連ドキュメントを検索してから回答を生成することで、根拠のある回答と情報の最新性を両立します。

ここで扱うのはGemma 4をローカルで実行してRAGパイプラインを構築する完全な実装を、コードと一緒に解説します。クラウドAPIに依存しない、手元で完結するシステムを作ることが目標です。

RAG の仕組みをひと言で

検索エンジンと生成AIを組み合わせたアーキテクチャです。

ユーザーの質問 → 関連ドキュメントを検索 → 検索結果をコンテキストとしてLLMに渡す → LLMがコンテキストを参照しながら回答を生成

これにより、LLMが学習データとして持っていない情報についても、ドキュメントを提供すれば答えられるようになります。

環境構築

必要なコンポーネントを準備します。

# Python ライブラリのインストール
pip install langchain langchain-community chromadb ollama sentence-transformers
 
# Ollama のインストール（ローカルLLM実行環境）
# macOS: brew install ollama
# Linux: curl -fsSL https://ollama.ai/install.sh | sh
 
# Gemma 4 モデルのダウンロード
ollama pull gemma4
 
# 埋め込みモデルのダウンロード（軽量で高精度）
ollama pull nomic-embed-text

Ollama はローカルマシンでLLMを実行するためのランタイムです。Gemma 4 をダウンロードすれば、APIキーなしでモデルを使えます。M1/M2 Mac なら8Bモデルでも快適に動作します。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦Gemma 4をOllamaでローカル実行してRAGに組み込む手順を、ChromaDBとLangChainの実装コード付きで習得できる

✦ドキュメント分割・埋め込み・検索・生成の4ステップを全て実装し、本番品質のRAGシステムを動かせるようになる

✦よくある詰まりポイント（埋め込み次元不一致・チャンク戦略の失敗・コンテキスト長超過）と対処法が分かる

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

Step 1: ドキュメントの読み込みと分割

RAGの第一ステップは、ドキュメントをLLMが扱いやすいサイズに分割することです。

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import DirectoryLoader, TextLoader, PyPDFLoader
from langchain.schema import Document
from pathlib import Path
import logging
 
logger = logging.getLogger(__name__)
 
def load_documents(source_dir: str) -> list[Document]:
    """
    ディレクトリ内のテキスト・PDFファイルを読み込む
    
    Args:
        source_dir: ドキュメントが格納されたディレクトリパス
    
    Returns:
        Document オブジェクトのリスト
    """
    documents = []
    source_path = Path(source_dir)
    
    if not source_path.exists():
        raise FileNotFoundError(f"ディレクトリが見つかりません: {source_dir}")
    
    # テキストファイルの読み込み
    txt_files = list(source_path.glob("**/*.txt"))
    for txt_file in txt_files:
        try:
            loader = TextLoader(str(txt_file), encoding="utf-8")
            documents.extend(loader.load())
        except Exception as e:
            logger.warning(f"テキスト読み込みエラー {txt_file}: {e}")
    
    # PDFファイルの読み込み
    pdf_files = list(source_path.glob("**/*.pdf"))
    for pdf_file in pdf_files:
        try:
            loader = PyPDFLoader(str(pdf_file))
            documents.extend(loader.load())
        except Exception as e:
            logger.warning(f"PDF読み込みエラー {pdf_file}: {e}")
    
    logger.info(f"{len(documents)} 件のドキュメントを読み込みました")
    return documents
 
def split_documents(documents: list[Document]) -> list[Document]:
    """
    ドキュメントをチャンクに分割する
    
    チャンクサイズとオーバーラップは、コンテンツの種類によって調整が必要:
    - 技術文書: chunk_size=1000, overlap=200
    - Q&A形式: chunk_size=500, overlap=100
    - 長い説明文: chunk_size=2000, overlap=400
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,        # 1チャンクの最大文字数
        chunk_overlap=200,      # チャンク間のオーバーラップ（文脈の連続性を保つ）
        length_function=len,
        separators=["\n\n", "\n", "。", ".", " ", ""]  # 日本語対応
    )
    
    chunks = splitter.split_documents(documents)
    logger.info(f"{len(chunks)} チャンクに分割しました")
    return chunks

チャンク戦略の落とし穴: チャンクサイズを大きくすれば文脈が保たれますが、検索の精度が落ちます。小さくすれば検索精度は上がりますが、回答に必要な情報が分断されます。コンテンツの性質に合わせた調整が重要です。

Step 2: ベクトル埋め込みとChromaDBへの保存

ドキュメントを意味的に検索できるよう、テキストをベクトル（数値の配列）に変換してデータベースに格納します。

from langchain.embeddings import OllamaEmbeddings
from langchain.vectorstores import Chroma
import chromadb
from pathlib import Path
 
def create_vector_store(
    chunks: list[Document],
    persist_dir: str = "./chroma_db",
    embedding_model: str = "nomic-embed-text"
) -> Chroma:
    """
    チャンクをベクトル化してChromaDBに保存する
    
    Args:
        chunks: 分割されたドキュメントのリスト
        persist_dir: ChromaDBの永続化ディレクトリ
        embedding_model: 使用する埋め込みモデル名
    
    Returns:
        Chroma ベクトルストアオブジェクト
    """
    
    # 埋め込みモデルの設定（Ollama経由でローカル実行）
    embeddings = OllamaEmbeddings(
        model=embedding_model,
        base_url="http://localhost:11434"  # Ollamaのデフォルトポート
    )
    
    # 既存DBがあれば削除して再作成（インデックス更新時）
    persist_path = Path(persist_dir)
    
    print(f"ベクトル化中... {len(chunks)} チャンク")
    
    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=str(persist_path),
        collection_name="my_documents"
    )
    
    vector_store.persist()
    print(f"ChromaDB に保存しました: {persist_dir}")
    
    return vector_store
 
def load_existing_vector_store(
    persist_dir: str = "./chroma_db",
    embedding_model: str = "nomic-embed-text"
) -> Chroma:
    """既存のChromaDBを読み込む（再インデックス不要）"""
    
    embeddings = OllamaEmbeddings(
        model=embedding_model,
        base_url="http://localhost:11434"
    )
    
    return Chroma(
        persist_directory=persist_dir,
        embedding_function=embeddings,
        collection_name="my_documents"
    )

埋め込みモデルの選択: nomic-embed-text はサイズが小さく（137MB）、英語・日本語ともに実用的な精度が出ます。多言語ドキュメントを扱う場合は multilingual-e5-large も選択肢です。ただしサイズが大きくなります。

Step 3: 検索とコンテキスト構築

ユーザーの質問に対して、最も関連性の高いチャンクを取得します。

from langchain.schema import Document
 
def retrieve_relevant_chunks(
    query: str,
    vector_store: Chroma,
    top_k: int = 4,
    score_threshold: float = 0.7
) -> list[Document]:
    """
    クエリに関連するドキュメントチャンクを検索する
    
    Args:
        query: ユーザーの質問
        vector_store: ChromaDB ベクトルストア
        top_k: 取得するチャンク数
        score_threshold: 最低スコア閾値（低いほど関連性が低い）
    
    Returns:
        関連チャンクのリスト（スコアが高い順）
    """
    
    # スコア付き類似度検索
    results_with_scores = vector_store.similarity_search_with_score(
        query=query,
        k=top_k
    )
    
    # スコア閾値でフィルタリング（ChromaDB のスコアは距離なので低いほど良い）
    filtered_results = [
        doc for doc, score in results_with_scores
        if score <= (1.0 - score_threshold)  # 距離を類似度スコアに変換
    ]
    
    if not filtered_results:
        # 閾値を超えるものがなければ、上位1件は必ず返す
        filtered_results = [results_with_scores[0][0]] if results_with_scores else []
    
    return filtered_results
 
def build_context(chunks: list[Document], max_context_chars: int = 3000) -> str:
    """
    検索されたチャンクからコンテキスト文字列を構築する
    
    Args:
        chunks: 取得したドキュメントチャンク
        max_context_chars: Gemma 4 に渡すコンテキストの最大文字数
    
    Returns:
        フォーマットされたコンテキスト文字列
    """
    context_parts = []
    total_chars = 0
    
    for i, chunk in enumerate(chunks, 1):
        chunk_text = f"[参考資料 {i}]\n{chunk.page_content}\n"
        
        if total_chars + len(chunk_text) > max_context_chars:
            # コンテキスト上限を超える場合は途中で切る
            remaining = max_context_chars - total_chars
            if remaining > 100:  # 意味のある長さが残っている場合のみ追加
                context_parts.append(chunk_text[:remaining] + "...[省略]")
            break
        
        context_parts.append(chunk_text)
        total_chars += len(chunk_text)
    
    return "\n".join(context_parts)

Step 4: Gemma 4 での回答生成

検索したコンテキストをGemma 4に渡して回答を生成します。

import ollama
from typing import Generator
 
def generate_answer(
    query: str,
    context: str,
    model: str = "gemma4",
    stream: bool = True
) -> str | Generator:
    """
    Gemma 4 でコンテキストを参照した回答を生成する
    
    Args:
        query: ユーザーの質問
        context: 検索されたドキュメントコンテキスト
        model: 使用するモデル名
        stream: ストリーミング出力を使用するか
    
    Returns:
        生成された回答テキスト（stream=Trueの場合はジェネレータ）
    """
    
    system_prompt = """あなたは提供された資料に基づいて質問に答えるアシスタントです。
 
重要なルール:
1. 回答は必ず提供された参考資料に基づくこと
2. 資料に記載がない情報は「この資料には記載がありません」と明示する
3. 回答の根拠となった参考資料の番号を示す
4. 日本語で、分かりやすく丁寧に回答する"""
    
    user_message = f"""以下の参考資料を参照して質問に答えてください。
 
参考資料:
{context}
 
質問: {query}"""
    
    if stream:
        # ストリーミングで逐次出力
        response_stream = ollama.chat(
            model=model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_message}
            ],
            stream=True
        )
        return response_stream
    else:
        # 一括生成
        response = ollama.chat(
            model=model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_message}
            ]
        )
        return response["message"]["content"]

パイプラインの統合と実行

4つのステップを組み合わせて、使えるRAGシステムにします。

import sys
 
def run_rag_pipeline(
    query: str,
    vector_store: Chroma,
    verbose: bool = False
) -> str:
    """
    完全なRAGパイプラインを実行する
    
    Args:
        query: ユーザーの質問
        vector_store: 構築済みのChromaDB
        verbose: デバッグ情報を出力するか
    
    Returns:
        生成された回答
    """
    
    # Step 1: 関連チャンクを検索
    chunks = retrieve_relevant_chunks(query, vector_store)
    
    if not chunks:
        return "関連する資料が見つかりませんでした。質問を変えて試してみてください。"
    
    if verbose:
        print(f"検索: {len(chunks)} 件のチャンクを取得")
        for i, chunk in enumerate(chunks):
            print(f"  [{i+1}] {chunk.page_content[:100]}...")
    
    # Step 2: コンテキストを構築
    context = build_context(chunks)
    
    # Step 3: 回答を生成（ストリーミング）
    print("回答: ", end="", flush=True)
    
    full_response = ""
    for chunk_data in generate_answer(query, context, stream=True):
        token = chunk_data["message"]["content"]
        print(token, end="", flush=True)
        full_response += token
    
    print()  # 改行
    return full_response
 
def main():
    """メイン実行フロー"""
    
    # ドキュメントディレクトリ（初回のみインデックス構築）
    docs_dir = "./documents"
    db_dir = "./chroma_db"
    
    import os
    if not os.path.exists(db_dir):
        print("初回: ドキュメントをインデックス化中...")
        
        # ドキュメントの読み込みと分割
        docs = load_documents(docs_dir)
        chunks = split_documents(docs)
        
        # ベクトルストアの構築
        vector_store = create_vector_store(chunks, db_dir)
        print("インデックス構築完了")
    else:
        print("既存インデックスを読み込み中...")
        vector_store = load_existing_vector_store(db_dir)
    
    # 対話ループ
    print("\nRAGシステム起動完了。質問を入力してください（'quit' で終了）\n")
    
    while True:
        query = input("質問: ").strip()
        
        if not query:
            continue
        if query.lower() in ["quit", "exit", "終了"]:
            print("終了します。")
            break
        
        answer = run_rag_pipeline(query, vector_store, verbose=False)
        print()
 
if __name__ == "__main__":
    main()

実際に動かしてみた結果

./documents/ に自社のFAQドキュメント（テキスト形式、約20ファイル、計15万文字）を入れてテストしました。

インデックス構築時間: 約8分（初回のみ）回答生成時間: 平均15〜25秒

精度については、ドキュメントに明確に記載されている情報への質問では9割以上の正解率でした。複数のドキュメントをまたぐ質問（「AとBの違いは？」など）では精度が下がりますが、それでも出典を示してくれるので確認がしやすいです。

よくある詰まりポイントと対処法

埋め込み次元の不一致エラー: 既存のChromaDBと異なる埋め込みモデルで再インデックスしようとすると発生します。chroma_db ディレクトリを削除して再作成してください。

Gemma 4 がコンテキストを無視する問題: システムプロンプトの指示が弱い場合に起きます。「資料に記載がない情報は答えないこと」という指示を強く書くことで改善されます。

コンテキスト長超過によるエラー: build_context の max_context_chars を小さくするか、取得チャンク数(top_k)を減らします。Gemma 4 の有効なコンテキスト窓は設定によりますが、日本語では3,000〜5,000文字程度を目安にするとよいです。

日本語の検索精度が低い問題: nomic-embed-text は多少の日本語対応がありますが、純日本語コンテンツには cl-nagoya/sup-simcse-ja-base などの日本語特化モデルの方が精度が出ることがあります。HuggingFace から取得して HuggingFaceEmbeddings で使用します。

個人開発者の視点から（実体験メモ）

Antigravity API との組み合わせ

ローカル実行ではなく、クラウドのGemma 4 API（Antigravity API）を使う場合は、ollama.chat の部分を google.generativeai に置き換えます。

import google.generativeai as genai
 
genai.configure(api_key="YOUR_API_KEY")
 
def generate_answer_cloud(query: str, context: str) -> str:
    """Antigravity API（クラウド）でGemma 4を使う"""
    model = genai.GenerativeModel(
        model_name="gemma-4-it",  # instruction-tuned版
        system_instruction="""提供された参考資料に基づいて質問に答えてください。
資料にない情報は「資料に記載がありません」と明示してください。"""
    )
    
    response = model.generate_content(
        f"参考資料:\n{context}\n\n質問: {query}"
    )
    return response.text