実践チュートリアル：SQLite3 FTS5 + Chroma によるローカル BM25+RAG パイプライン構築

1. チュートリアルの目的と全体アーキテクチャ

本チュートリアルでは、外部の検索サーバー（Elasticsearch や OpenSearch など）を立ち上げることなく、完全ローカル環境で動作するプロダクション品質のハイブリッド検索（BM25 + 密ベクトル）および RAG（検索拡張生成）システムをゼロから構築します。

第5章から第8章で解説した要素技術を結集し、以下の技術スタックでパイプラインを統合します。

flowchart TD
    UserQuery["ユーザー質問"] --> Tokenizer["日本語形態素解析 (Janome)"]
    Tokenizer --> FTS5["SQLite3 FTS5 (BM25 疎検索)"]
    UserQuery --> Embed["Ollama bge-m3 (埋め込み)"]
    Embed --> ChromaDB["Chroma (密ベクトル検索)"]

    FTS5 --> Ensemble["EnsembleRetriever (RRF 順位融合)"]
    ChromaDB --> Ensemble

    Ensemble --> FlashRank["FlashRank (Cross-Encoder リランカー)"]
    FlashRank --> Prompt["プロンプトテンプレート注入"]
    Prompt --> Gemma["Ollama gemma4:12b (回答生成)"]
    Gemma --> Answer["最終回答"]

---

2. 環境準備と事前セットアップ

Ollama モデルの取得

ローカル LLM 実行環境である Ollama で、埋め込みモデル bge-m3 と生成モデル gemma4:12b を事前にダウンロードします。

ollama pull bge-m3
ollama pull gemma4:12b

Python パッケージのインストール

必要なライブラリを一括でインストールします。SQLite3 は Python 3 の標準ライブラリ（sqlite3）に含まれるため、追加の DB サーバーインストールは不要です。

pip install langchain langchain-core langchain-community langchain-chroma langchain-ollama flashrank janome

---

3. Step 1: SQLite3 FTS5 による BM25 検索リトリーバーの実装

SQLite3 に組み込まれている FTS5（Full-Text Search 5）仮想テーブルを利用します。 FTS5 の組み込み関数 bm25() は、文書内の単語頻度と転置インデックスに基づき、C 言語レベルで高速に BM25 スコアを計算します。

日本語テキストは空白区切りがないため、Janome で形態素解析を行い、分かち書きしたトークン列を FTS5 の tokens カラムに保存することで、正確なキーワード検索と BM25 スコアリングを実現します。

SQLite3 FTS5 カスタムリトリーバーのコード

LangChain の BaseRetriever を継承し、LangChain の各種コンポーネント（EnsembleRetriever や LCEL パイプライン）と直接結合できるカスタムリトリーバーを作成します。

import sqlite3
from typing import Any
from janome.tokenizer import Tokenizer
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from pydantic import Field


class SQLiteFTS5Retriever(BaseRetriever):
    """SQLite3 FTS5 をバックエンドとする BM25 カスタムリトリーバー."""

    db_path: str = Field(description="SQLite データベースファイルのパス")
    table_name: str = Field(
        default="docs_fts", description="FTS5 仮想テーブル名"
    )
    k: int = Field(default=4, description="取得件数")
    _tokenizer: Any = None

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        self._tokenizer = Tokenizer()

    def _tokenize(self, text: str) -> list[str]:
        """形態素解析を行い、名詞・動詞・形容詞の基本形を空白区切りトークンとして抽出."""
        tokens = []
        for token in self._tokenizer.tokenize(text):
            pos = token.part_of_speech.split(",")[0]
            if pos in ["名詞", "動詞", "形容詞"]:
                tokens.append(token.base_form)
        return tokens

    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun | None = None,
    ) -> list[Document]:
        tokens = self._tokenize(query)
        if not tokens:
            return []

        # FTS5 の MATCH 構文用クエリ（OR 検索で関連文書を広くヒット）
        fts_query = " OR ".join(f'"{t}"' for t in tokens)

        conn = sqlite3.connect(self.db_path)
        cur = conn.cursor()

        # FTS5 の bm25() 関数は値が小さい（負に大きい）ほど関連度が高い
        sql = f"""
            SELECT doc_id, content, metadata_json, bm25({self.table_name}) AS rank
            FROM {self.table_name}
            WHERE {self.table_name} MATCH ?
            ORDER BY rank ASC
            LIMIT ?
        """

        cur.execute(sql, (fts_query, self.k))
        rows = cur.fetchall()
        conn.close()

        import json

        documents = []
        for doc_id, content, metadata_json, rank in rows:
            meta = json.loads(metadata_json) if metadata_json else {}
            meta["doc_id"] = doc_id
            meta["bm25_score"] = float(rank)
            documents.append(Document(page_content=content, metadata=meta))

        return documents

---

4. Step 2: データ準備と FTS5・Chroma へのインデックス登録

検証用の社内ナレッジドキュメント群を用意し、SQLite3 FTS5 仮想テーブルと Chroma ベクトルストアの両方に登録します。

import json
import sqlite3
from pathlib import Path
from janome.tokenizer import Tokenizer
from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_ollama import OllamaEmbeddings

# 1. サンプルナレッジベース文書
raw_documents = [
    Document(
        page_content="有給休暇は勤続半年後に10日付与されます。取得申請は希望日の3日前までに社内ポータルから提出してください。",
        metadata={"id": "HR-001", "category": "人事", "title": "有給休暇規定"},
    ),
    Document(
        page_content="夏季休暇は毎年7月1日から9月30日までの期間中に、最大3日間（有給扱いで連続または分割可）取得できます。",
        metadata={"id": "HR-002", "category": "人事", "title": "夏季特別休暇"},
    ),
    Document(
        page_content="社内ネットワークで通信エラー ERR-502 が発生した場合は、ゲートウェイのタイムアウトが原因です。卓上ルータの電源を再投入し、改善しない場合は内線8811の情シスへ連絡してください。",
        metadata={"id": "IT-001", "category": "情シス", "title": "ネットワーク障害対応"},
    ),
    Document(
        page_content="業務で使用する社内PCには EDR セキュリティソフトの常時稼働が義務付けられており、月次の OS セキュリティパッチ適用は必須です。",
        metadata={"id": "SEC-001", "category": "セキュリティ", "title": "端末セキュリティ方針"},
    ),
    Document(
        page_content="慶弔休暇として、本人の結婚時には5日、配偶者の出産時には3日の特別休暇を取得できます。事後2週間以内に証明書類を提出してください。",
        metadata={"id": "HR-003", "category": "人事", "title": "慶弔休暇規程"},
    ),
]

DB_DIR = Path("./tutorial_data")
DB_DIR.mkdir(parents=True, exist_ok=True)
SQLITE_DB_PATH = str(DB_DIR / "fts5_knowledge.db")
CHROMA_PERSIST_DIR = str(DB_DIR / "chroma_index")

# 2. SQLite3 FTS5 テーブルの初期化とデータ登録
tokenizer = Tokenizer()


def tokenize_text(text: str) -> str:
    tokens = []
    for token in tokenizer.tokenize(text):
        pos = token.part_of_speech.split(",")[0]
        if pos in ["名詞", "動詞", "形容詞"]:
            tokens.append(token.base_form)
    return " ".join(tokens)


conn = sqlite3.connect(SQLITE_DB_PATH)
cur = conn.cursor()

# FTS5 仮想テーブルの作成
cur.execute("""
    CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
        doc_id UNINDEXED,
        content,
        metadata_json UNINDEXED,
        tokens,
        tokenize = 'unicode61'
    );
""")
cur.execute("DELETE FROM docs_fts;")  # 既存データのクリア

for doc in raw_documents:
    doc_id = doc.metadata.get("id", "")
    content = doc.page_content
    meta_json = json.dumps(doc.metadata, ensure_ascii=False)
    token_str = tokenize_text(content)
    cur.execute(
        "INSERT INTO docs_fts (doc_id, content, metadata_json, tokens) VALUES (?, ?, ?, ?)",
        (doc_id, content, meta_json, token_str),
    )

conn.commit()
conn.close()
print("✓ SQLite3 FTS5 インデックス作成完了")

# 3. Chroma ベクトルストアの作成とデータ登録（bge-m3）
ollama_embeddings = OllamaEmbeddings(model="bge-m3")

chroma_vectorstore = Chroma.from_documents(
    documents=raw_documents,
    embedding=ollama_embeddings,
    persist_directory=CHROMA_PERSIST_DIR,
)
print("✓ Chroma ベクトルインデックス作成完了")

---

5. Step 3: ハイブリッド検索とスタンドアローン FlashRank によるリランキング

FTS5 の BM25 疎検索と Chroma の密ベクトル検索を EnsembleRetriever で結合し、取得した候補文書を FlashRank で再スコアリング（リランク）します。

Important

FlashRank のスタンドアローン化と langchain-community の非推奨化について

従来、LangChain では langchain_community.document_compressors.FlashrankRerank が提供されていましたが、LangChain エコシステムのモノリスパッケージ再編に伴い langchain-community 側の実装は非推奨（deprecated）となっています。 現在は、公式のスタンドアローンパッケージ flashrank（PrithivirajDamodaran/FlashRank）を直接使用し、LangChain の BaseDocumentCompressor を継承した軽量ラッパーを定義することが推奨されます。

スタンドアローン版 flashrank は ONNX Runtime をベースにしており、PyTorch や CUDA などの重量な依存なしで、CPU 上でミリ秒オーダーの超高速リランキング（Listwise / Pairwise）を実行できます。

FlashRank カスタムコンプレッサーの実装とハイブリッド結合

from typing import Any, Sequence
from flashrank import Ranker, RerankRequest
from langchain.retrievers import ContextualCompressionRetriever, EnsembleRetriever
from langchain_core.callbacks import Callbacks
from langchain_core.documents import Document
from langchain_core.documents.compressor import BaseDocumentCompressor
from pydantic import Field


class StandaloneFlashRankCompressor(BaseDocumentCompressor):
    """スタンドアローン版 flashrank を直接利用する LangChain 互換リランカー."""

    model_name: str = Field(
        default="ms-marco-MiniLM-L-12-v2", description="FlashRank モデル名"
    )
    top_n: int = Field(default=2, description="取得する上位件数")
    cache_dir: str | None = Field(default=None, description="モデルキャッシュ先")
    _ranker: Any = None

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        # スタンドアローン flashrank の Ranker を初期化
        # 推奨モデル:
        # - ms-marco-MiniLM-L-12-v2: ~34MB, 精度と速度のベストバランス（デフォルト）
        # - ms-marco-MultiBERT-L-12: ~150MB, 多言語対応（100言語以上）
        # - ms-marco-TinyBERT-L-2-v2: ~4MB, 超軽量・最速
        self._ranker = Ranker(
            model_name=self.model_name,
            cache_dir=self.cache_dir,
        )

    def compress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Callbacks | None = None,
    ) -> Sequence[Document]:
        if not documents:
            return []

        # flashrank の RerankRequest 形式に入力ドキュメントを変換
        passages = [
            {"id": i, "text": doc.page_content, "meta": doc.metadata}
            for i, doc in enumerate(documents)
        ]
        rerank_request = RerankRequest(query=query, passages=passages)
        results = self._ranker.rerank(rerank_request)

        # 上位 top_n 件を抽出し、スコアを metadata に付与して Document を再構築
        top_results = results[: self.top_n]
        return [
            Document(
                page_content=r["text"],
                metadata={**r["meta"], "rerank_score": float(r["score"])},
            )
            for r in top_results
        ]


# 1. 2つのリトリーバーを準備（第1段階：広めに上位4件ずつ取得）
fts5_retriever = SQLiteFTS5Retriever(
    db_path=SQLITE_DB_PATH,
    table_name="docs_fts",
    k=4,
)
chroma_retriever = chroma_vectorstore.as_retriever(search_kwargs={"k": 4})

# 2. EnsembleRetriever による RRF（逆順位融合）ハイブリッド検索
ensemble_retriever = EnsembleRetriever(
    retrievers=[fts5_retriever, chroma_retriever],
    weights=[0.5, 0.5],
)

# 3. スタンドアローン FlashRank によるリランキング（第2段階：上位2件に厳選）
compressor = StandaloneFlashRankCompressor(
    model_name="ms-marco-MiniLM-L-12-v2",
    top_n=2,
)
hybrid_rerank_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever,
)

---

6. Step 4: Ollama（gemma4:12b）による推論と LCEL パイプライン結合

リランキングされた文脈をプロンプトに注入し、ローカル LLM である gemma4:12b に渡して正確な回答を生成する LCEL（LangChain Expression Language）チェーンを構築します。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_ollama import ChatOllama

# 1. プロンプトテンプレートの定義
prompt = ChatPromptTemplate.from_template("""あなたは親切で正確な社内アシスタントです。
以下の【提供ドキュメント】の内容のみに基づいて、質問に明確に回答してください。
ドキュメントに記載がない内容については無理に答えず、「提供された情報からは回答できません」と答えてください。

【提供ドキュメント】
{context}

【質問】
{question}

【回答】:""")


# 2. 検索ドキュメントの文字列整形関数
def format_docs(docs: list[Document]) -> str:
    formatted = []
    for d in docs:
        doc_id = d.metadata.get("id", "N/A")
        title = d.metadata.get("title", "")
        formatted.append(f"[{doc_id}: {title}]\n{d.page_content}")
    return "\n\n".join(formatted)


# 3. 推論 LLM の初期化
llm = ChatOllama(model="gemma4:12b", temperature=0)

# 4. LCEL パイプラインの結合
rag_chain = (
    {
        "context": hybrid_rerank_retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | llm
    | StrOutputParser()
)

# 5. 実行と動作検証
test_queries = [
    "夏休みは最大で何日休めますか？",
    "通信障害で ERR-502 が出たときはどうすればいいですか？",
]

for q in test_queries:
    print(f"\n==========================================")
    print(f"質問: {q}")
    answer = rag_chain.invoke(q)
    print(f"回答:\n{answer}")

実行結果の検証

==========================================
質問: 夏休みは最大で何日休めますか？
回答:
夏季休暇は、毎年7月1日から9月30日までの期間中に、最大3日間（有給扱いで連続または分割可）取得することができます。

==========================================
質問: 通信障害で ERR-502 が出たときはどうすればいいですか？
回答:
通信エラー ERR-502 が発生した場合は、ゲートウェイのタイムアウトが原因です。
まずは卓上ルータの電源を再投入してください。それでも改善しない場合は、内線8811の情シスへ連絡してください。

SQLite3 FTS5 の BM25 により「ERR-502」などの特定エラーコードが完全一致で確実にヒットし、Chroma の密ベクトルにより「夏休み」と「夏季休暇」の意味的類似性が捉えられ、FlashRank で上位に絞り込まれた上で gemma4:12b が簡潔かつハルシネーションなく回答していることが確認できます。

---

7. ステップアップ：日本語特化モデル（ruri-v3-310m & ruri-v3-reranker-310m）への切り替え

なぜ日本語特化モデルを採用するのか？

本チュートリアルの基本構成で採用した bge-m3 および FlashRank は優れた多言語汎用性能を持ちますが、日本語の社内文書、契約書、技術仕様書、公的文書などのドメインに特化して極限の検索精度を追求する場合、東北大学・名古屋大学等の研究グループが開発した ruri シリーズへの切り替えが極めて効果的です。

• cl-nagoya/ruri-v3-310m: 日本語テキスト埋め込みベンチマーク（JMTEB）において多言語モデル（multilingual-e5 や bge-m3）を上回るトップクラスの検索精度を記録した、約 3.1 億パラメータの日本語特化埋め込みモデルです。
• cl-nagoya/ruri-v3-reranker-310m: 同様に日本語の質問と文書ペアの関連度判定に特化してファインチューニングされた Cross-Encoder リランカーモデルです。微妙な語彙の言い回しや助詞の違いを高精度に識別します。

切り替えに必要な追加パッケージ

Hugging Face Hub 上の PyTorch / transformers / sentence-transformers モデルを直接ローカル実行するために、以下のライブラリを追加します。

pip install sentence-transformers transformers torch

1. 埋め込みモデルを ruri-v3-310m に差し替える

langchain_community.embeddings.HuggingFaceEmbeddings（または langchain_huggingface.HuggingFaceEmbeddings）を利用して、Chroma の埋め込み関数を bge-m3 から ruri-v3-310m に変更します。

Note

ruri シリーズは、検索クエリと文書（インデックス対象）で異なるプレフィックスを推奨する場合があります。 通常の検索用途では、クエリの先頭に クエリ: を付与することで検索精度が向上します。

from langchain_community.embeddings import HuggingFaceEmbeddings

# 日本語特化埋め込みモデルの初期化
ruri_embeddings = HuggingFaceEmbeddings(
    model_name="cl-nagoya/ruri-v3-310m",
    model_kwargs={"device": "cpu"},  # GPU 利用時は "cuda"
    encode_kwargs={"normalize_embeddings": True},
)

# Chroma ベクトルストアの作成（ruri-v3-310m を使用）
ruri_chroma_vectorstore = Chroma.from_documents(
    documents=raw_documents,
    embedding=ruri_embeddings,
    persist_directory=str(DB_DIR / "chroma_ruri_index"),
)

ruri_vector_retriever = ruri_chroma_vectorstore.as_retriever(
    search_kwargs={"k": 4}
)

2. リランカーを ruri-v3-reranker-310m に差し替える

LangChain の CrossEncoderReranker と HuggingFaceCrossEncoder を使用して、FlashrankRerank を ruri-v3-reranker-310m に置き換えます。

from langchain.retrievers import ContextualCompressionRetriever, EnsembleRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 1. 日本語特化 Cross-Encoder リランカーの準備
ruri_cross_encoder = HuggingFaceCrossEncoder(
    model_name="cl-nagoya/ruri-v3-reranker-310m",
    model_kwargs={"device": "cpu"},  # GPU 利用時は "cuda"
)

# 2. リランカーコンプレッサーの作成（上位2件に厳選）
ruri_compressor = CrossEncoderReranker(model=ruri_cross_encoder, top_n=2)

# 3. ハイブリッドリトリーバー（SQLite3 FTS5 + ruri Chroma）の統合
ruri_ensemble_retriever = EnsembleRetriever(
    retrievers=[fts5_retriever, ruri_vector_retriever],
    weights=[0.5, 0.5],
)

# 4. 2段階圧縮リトリーバーの構築
ruri_hybrid_rerank_retriever = ContextualCompressionRetriever(
    base_compressor=ruri_compressor,
    base_retriever=ruri_ensemble_retriever,
)

3. 日本語特化パイプラインの実行

構築した ruri_hybrid_rerank_retriever を、先ほど定義した LCEL パイプラインの context に渡すだけで、完全な日本語特化型ローカル RAG パイプラインとして稼働します。

japanese_specialized_rag_chain = (
    {
        "context": ruri_hybrid_rerank_retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | llm
    | StrOutputParser()
)

result = japanese_specialized_rag_chain.invoke("慶弔休暇は結婚のときに何日休めますか？")
print(result)

---

8. まとめと実運用の勘所

本チュートリアルでは、以下の設計パターンを実現しました。

1. ゼロ外部デーモン & 低リソース: SQLite3 FTS5 と Chroma の組み合わせにより、Elasticsearch 等の重量なサーバープロセスなしで、Python プロセス完結の安定した永続インデックスを運用できます。
2. 疎密ハイブリッド検索（BM25 + Dense）: 固有名詞やエラーコード（ERR-502）は SQLite3 FTS5 が確実に捉え、自然言語の類語や言い換え表現は bge-m3（または ruri-v3-310m）がカバーします。
3. 2段階検索による高精度化: EnsembleRetriever で幅広く集めた候補を、FlashRank（または ruri-v3-reranker-310m）で厳選してコンテキスト長を節約し、gemma4:12b の回答精度を最大化します。
4. ドメインに応じた柔軟なモデル選定: 多言語汎用ユースケースでは軽量な bge-m3 + FlashRank、高精度な日本語業務特化ユースケースでは ruri-v3-310m + ruri-v3-reranker-310m をシームレスに切り替えて活用できます。