H.Otsuka Log

RAG徹底解説 ― フルスクラッチ実装で「なぜ動くのか」まで理解する

大塚隼斗更新: 38 分で読了

1. なぜRAGが必要なのか ―― LLMの知識問題

LLMは流暢に、自信満々に、そして頻繁に間違えます。先週起きた出来事を尋ねても答えられません。知らないからです。LLMの知識は学習が終わった時点で凍結されており、それ以降の世界を認識できません。

これはバグではなく、アーキテクチャ上の根本的な制約です。LLMは知識をパラメータ(学習で獲得した数十億の重み)として保持します。学習が終わればその知識は固定され、モデルは「自分が何を知らないか」を知りません。だから空白を、自信に満ちた作り話で埋めてしまう。これがハルシネーション(幻覚)です。単純な要約タスクなら発生率は1%程度に抑えられることもありますが、複雑なタスクでは50%を超えるという報告もあります。

この「知識問題」には3つの層があります。そして、素朴に思いつく解決策はどれもうまくいきません。

1.1 知識が学習時に凍結される

LLMが知っていることは、すべて学習時にスクレイピングされ、処理され、パラメータへと圧縮されたものです。カットオフ日以降、モデルは盲目になります。あなたの新製品リリースも、昨日のセキュリティインシデントも、今朝改定された人事ポリシーも知りません。カスタマーサポート・法務・社内検索・コンプライアンスのように正確さが要求される業務では、最新情報にアクセスできないモデルはそのままでは負債になりえます。

1.2 コンテキストウィンドウは答えにならない

「全部プロンプトに突っ込めばいいのでは?」という力技は魅力的です。コンテキストウィンドウは急速に拡大し、100万トークンを超えるモデルも登場しました。しかしこのアプローチには3つの致命的な問題があります。

  • コストが高い:トークン単位で課金されるため、毎クエリで知識ベース全体を送りつければ予算はすぐ尽きます。
  • ハードリミットがある:100万トークンでも、大企業の全ドキュメントやDB全体は入りきりません。
  • 文脈が増えるほど精度が下がる:これが最も見落とされがちな点です。Stanford・UC Berkeleyの研究(Liu et al., "Lost in the Middle")によると、LLMの性能はU字カーブを描きます。重要情報がプロンプトの先頭か末尾にあるときは高精度ですが、真ん中に埋もれると精度が急落します。20文書のマルチ文書QAで、鍵となる情報を中央に置くと、あるモデルでは精度が25%程度まで落ちたと報告されています。

つまり、コンテキストを増やすことは、モデルがそれを使ってくれることを保証しません。

1.3 ファインチューニングも答えにならない

ファインチューニングは、事前学習済みモデルを自前データでさらに学習させ、新しい知識や振る舞いを覚えさせる手法です。人を専門学校に通わせるようなイメージで、理論上はドメイン知識を教え込めます。

しかし実務では、解決するより多くの問題を生みます。GPU計算資源・機械学習の専門知識・入念に準備した学習データが必要で、完了までに数日〜数週間かかります。しかも結果は「スナップショット」にすぎません。元データが変われば、ファインチューン済みモデルはすぐ陳腐化します。事実の記憶(factual recall)はファインチューニングが苦手とする領域でもあります。

1.4 本当に必要なもの

必要なのは、モデル自体を変えずに、クエリごとに、適切な情報を、適切なタイミングで与えることです。それがRAGです。


2. RAGとは何か ―― オープンブック試験というメタファ

いちばん分かりやすい比喩はオープンブック試験です。

  • クローズドブック試験(=素のLLM):記憶している内容だけで答える。賢く推論もできるが、パラメータに入っている知識に縛られる。
  • オープンブック試験(=RAG):同じ推論能力を持つ学生が、答える前に関連ページを参照できる。

RAGの正式な定義は、2020年のPatrick Lewisらの論文(Facebook AI Research/University College London)に由来します。RAGモデルは、事前学習済み言語モデルのパラメトリックメモリ(parametric memory)と、外部知識インデックスを使うノンパラメトリックメモリ(non-parametric memory)を組み合わせます。後者にはニューラル検索器(neural retriever)でアクセスします。知識を更新したければ、モデルを再学習させる代わりに検索インデックスを差し替えるだけでよい ―― これがRAGの実務的な強さの核心です。

実装レベルでは、RAGは3ステップのループです。

要点は一つ。モデルを変えているのではなく、モデルに見せるものを変えている。この違いがRAGを強力かつ実用的にしています。加えて、どのソース文書を使ったかを提示できる(引用・帰属)ため、透明性という大きな利点も得られます。


3. アーキテクチャ全体像 ―― 2本のパイプライン

RAGシステムは、大きく2本のパイプラインから成ります。

  • オフライン・インジェストパイプライン:データを検索可能な状態に整える(事前処理)。
  • オンライン・検索/生成パイプライン:クエリに答える(推論時)。

重要な不変条件が一つあります。クエリと文書は同じベクトル空間に存在しなければならない。したがって、インジェスト時と検索時で同じ埋め込みモデルを使うことが必須です。

以降、この図の各ブロックを、フレームワークに頼らずPythonで実装していきます。


4. フルスクラッチ実装:環境準備

方針は「仕組みが見える最小構成」です。埋め込みとリランキングにはローカルで動く軽量モデルを使い、生成はローカルLLMランタイムの Ollama を使います(OpenAI互換APIでも代替可能。後述)。ベクトルストア・BM25・RRF・Agenticループはすべて自作します。

# Python 3.10+ を想定
python -m venv .venv
source .venv/bin/activate

# 埋め込み・リランキングモデル(sentence-transformers 経由でローカル実行)
pip install sentence-transformers numpy requests

# 日本語のスパース検索(BM25)用の形態素解析器(任意だが推奨)
pip install janome

# 生成用:Ollama をインストールし、モデルを取得しておく
#   https://ollama.com/download
#   ollama pull llama3.1
#   ollama pull qwen2.5   # 日本語ならこちらも良い
# 別ターミナルで `ollama serve` が起動している状態にしておく

以降のコードは、章ごとに完結した形で示しつつ、最後に全部を1ファイルに統合した rag.py を掲載します。まずは全ステップ共通で使う「教材コーパス」を用意します。実在の社内ドキュメントを模した、ごく短い日本語文書の集合です。

# corpus.py
# 教材用の小さな知識ベース。実運用ではPDF/DB/APIなどから読み込む。
DOCUMENTS = [
    {
        "id": "doc-hr-01",
        "title": "就業規則:勤務時間",
        "category": "HR",
        "text": (
            "当社の標準勤務時間は9時から18時までとし、休憩時間は12時から13時の1時間とする。"
            "フレックスタイム制度を利用する場合、コアタイムは11時から15時とする。"
            "コアタイム外の勤務については、各自の裁量で開始・終了時刻を調整してよい。"
        ),
    },
    {
        "id": "doc-hr-02",
        "title": "就業規則:休暇制度",
        "category": "HR",
        "text": (
            "年次有給休暇は入社半年後に10日付与され、以降勤続年数に応じて増加する。"
            "有給休暇の取得は原則として前日までに申請するものとするが、"
            "急な体調不良の場合は当日申請を認める。未消化の有給は翌年度に限り繰り越せる。"
        ),
    },
    {
        "id": "doc-it-01",
        "title": "情報システム:パスワードポリシー",
        "category": "IT",
        "text": (
            "社内システムのパスワードは12文字以上とし、英大文字・小文字・数字・記号を含めること。"
            "パスワードは90日ごとに変更する。過去3回分と同一のパスワードは再利用できない。"
            "パスワードを忘れた場合は、社内ポータルの『パスワード再設定』から本人確認のうえ再設定する。"
        ),
    },
    {
        "id": "doc-it-02",
        "title": "情報システム:VPN接続手順",
        "category": "IT",
        "text": (
            "社外から社内ネットワークへ接続するにはVPNクライアントを使用する。"
            "初回はポータルから証明書をダウンロードし、クライアントにインポートする。"
            "接続にはワンタイムパスワード(OTP)が必要で、認証アプリで生成した6桁のコードを入力する。"
        ),
    },
    {
        "id": "doc-fin-01",
        "title": "経費精算:申請ルール",
        "category": "Finance",
        "text": (
            "経費精算は月末締めで、翌月5営業日以内に申請する。"
            "1万円を超える支出には領収書の添付を必須とする。"
            "交通費は実費精算とし、定期区間を含む場合は差額のみを申請する。"
        ),
    },
    {
        "id": "doc-fin-02",
        "title": "経費精算:返金・キャンセルポリシー",
        "category": "Finance",
        "text": (
            "出張がキャンセルとなった場合、キャンセル料は経費として認められる。"
            "ただし、本人都合による直前キャンセルで発生した違約金は自己負担とする。"
            "誤って多く精算した金額は、翌月の給与から相殺する。"
        ),
    },
]

5. Step 1:埋め込みと類似度をゼロから理解する

RAGの心臓部は埋め込み(embedding) です。これはセマンティック検索(意味による検索)を可能にする仕組みで、これを理解せずにRAGは語れません。

テキスト埋め込みは、テキストを密なベクトル(floating-pointの配列。実運用では1,536次元や3,072次元がよく使われる)に変換します。肝は「意味を捉える(capture meaning)」という部分です。似た意図を持つ語や文は、たとえ使われている単語が違っても、ベクトル空間上で近くに配置されます。

たとえば「パスワードを再設定するには?」と「アカウントにログインできない」は、単語こそまったく違いますが、埋め込みに変換するとほぼ同じ向きのベクトルになります。この近さはコサイン類似度・内積・ユークリッド距離などで測ります。

まずは埋め込みと類似度を、素のコードで体感しましょう。

# step1_embedding.py
import numpy as np
from sentence_transformers import SentenceTransformer

# 多言語対応の軽量埋め込みモデル。日本語も扱える。
# e5系は "query: " / "passage: " のprefixを付けると精度が上がる設計だが、
# まずは仕組みを見るため素のテキストで確認する。
embedder = SentenceTransformer("intfloat/multilingual-e5-small")


def embed(texts: list[str]) -> np.ndarray:
    """テキスト(複数可)を正規化済みベクトルに変換する。
    normalize_embeddings=True により各ベクトルはL2ノルム1になるので、
    内積がそのままコサイン類似度になる。"""
    return embedder.encode(texts, normalize_embeddings=True)


def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    """正規化済みベクトル同士なら内積=コサイン類似度。
    正規化していない一般のベクトルにも対応できるよう、ここでは明示的に割る。"""
    denom = (np.linalg.norm(a) * np.linalg.norm(b))
    if denom == 0:
        return 0.0
    return float(np.dot(a, b) / denom)


if __name__ == "__main__":
    sentences = [
        "パスワードを再設定するには?",     # 0
        "アカウントにログインできない",       # 1
        "フランスの首都はどこですか?",       # 2
    ]
    vecs = embed(sentences)

    print("次元数:", vecs.shape)  # (3, 384) 程度
    print("類似度(0 vs 1)  意味が近い :", round(cosine_similarity(vecs[0], vecs[1]), 4))
    print("類似度(0 vs 2)  無関係     :", round(cosine_similarity(vecs[0], vecs[2]), 4))

実行すると、0 vs 1(パスワード再設定 と ログインできない)の類似度が、0 vs 2(無関係な首都の質問)より明らかに高くなります。キーワードが一致しなくても、意味が近ければ近いベクトルになる ―― これがセマンティック検索の土台です。RAGは「単語」ではなく「意味」で関連文書を見つけられるのです。

なぜ正規化するのか:ベクトルをL2正規化(長さ1に)しておくと、内積がそのままコサイン類似度になります。検索時に大量のベクトルとの類似度を計算するので、np.dot 一発で済むこの性質は実装を大きく単純化します。


6. Step 2:チャンキング ―― 最も効く前処理

文書をそのまま丸ごと埋め込むと、1つのベクトルに情報が詰め込まれすぎて「意味の解像度」が落ちます。逆に細かく刻みすぎると、文脈が切れて断片だけになります。そこでチャンキング(意味のある単位への分割)が必要です。RAGの品質はチャンクの品質に強く依存します ―― どれだけ賢い検索を組んでも、チャンクが支離滅裂な断片なら救えません。

まずは最も素朴な「文字数ベースのスライディングウィンドウ」を実装します。ポイントはオーバーラップです。関連する記述がチャンク境界にまたがって落ちてしまうのを防ぐため、隣り合うチャンクを一部重複させます。

# step2_chunking.py

def sliding_window_chunk(text: str, chunk_size: int = 300, overlap: int = 60) -> list[str]:
    """文字数ベースの単純なスライディングウィンドウ分割。
    chunk_size 文字ごとに切り出し、overlap 文字だけ重複させる。"""
    if chunk_size <= overlap:
        raise ValueError("chunk_size は overlap より大きくすること")
    chunks = []
    start = 0
    step = chunk_size - overlap
    while start < len(text):
        chunk = text[start:start + chunk_size]
        chunks.append(chunk)
        start += step
    return chunks

これでも動きますが、文の途中でぶつ切りにしてしまう欠点があります。実運用では区切り文字を優先順位付きで試す再帰的分割がよく使われます(LangChainの RecursiveCharacterTextSplitter がこの考え方)。段落 → 文 → 句読点 → 空白の順に、なるべく自然な境界で切ろうとし、それでもサイズを超える場合だけ強制分割します。素のPythonで書くと次のようになります。

# step2_chunking.py(続き)

def recursive_chunk(
    text: str,
    chunk_size: int = 300,
    overlap: int = 60,
    separators: tuple[str, ...] = ("\n\n", "\n", "。", "、", " ", ""),
) -> list[str]:
    """区切り文字を優先順位順に試す再帰的分割。
    まず先頭のセパレータで分け、各断片が chunk_size 以下ならそのまま、
    超える場合は次のセパレータで再帰的に分割する。
    最後に隣接断片を chunk_size を超えない範囲でまとめ直し、overlap を付与する。"""

    def _split(t: str, seps: tuple[str, ...]) -> list[str]:
        if len(t) <= chunk_size or not seps:
            return [t]
        sep, *rest = seps
        if sep == "":
            # これ以上分ける区切りがない → 文字数で強制分割
            return [t[i:i + chunk_size] for i in range(0, len(t), chunk_size)]
        parts = t.split(sep)
        # セパレータ自体は残して復元する(末尾を除く)
        pieces = [p + sep for p in parts[:-1]] + [parts[-1]]
        result = []
        for p in pieces:
            if len(p) <= chunk_size:
                result.append(p)
            else:
                result.extend(_split(p, tuple(rest)))
        return result

    # 1) 再帰分割で「chunk_size以下の小片」の列を得る
    pieces = _split(text, separators)

    # 2) 小片を chunk_size を超えない範囲で貪欲に連結
    merged: list[str] = []
    buf = ""
    for p in pieces:
        if len(buf) + len(p) <= chunk_size:
            buf += p
        else:
            if buf:
                merged.append(buf)
            buf = p
    if buf:
        merged.append(buf)

    # 3) 隣接チャンク間に overlap 文字の重複を付与
    if overlap <= 0 or len(merged) <= 1:
        return merged
    overlapped = [merged[0]]
    for i in range(1, len(merged)):
        tail = merged[i - 1][-overlap:]
        overlapped.append(tail + merged[i])
    return overlapped


if __name__ == "__main__":
    from corpus import DOCUMENTS
    sample = DOCUMENTS[2]["text"]  # パスワードポリシー
    for i, c in enumerate(recursive_chunk(sample, chunk_size=120, overlap=30)):
        print(f"[chunk {i}] ({len(c)}字) {c}")

チャンクサイズは測って決める:300文字が最適という保証はどこにもありません。契約書・論文・マニュアル・請求書では情報密度がまったく違います。本来はサイズとオーバーラップを変えながら、後述の評価(Step 14)で検索品質を測って決めるべきです。さらに一歩進めたセマンティックチャンキング(連続する文の埋め込み類似度が急落する箇所で区切る)という手法もあります。まずは固定長で動かし、評価基盤ができてから最適化するのが現実的です。


7. Step 3:自作ベクトルストア

埋め込んだチャンクを保存し、クエリベクトルとの類似度で上位K件を返す ―― これがベクトルストアの役割です。本番ではQdrant・Chroma・Milvus・pgvectorなどを使いますが、中身は「ベクトルを貯めて、内積で近いものを探す」だけです。numpyで自作してしまいましょう。

# step3_vectorstore.py
import numpy as np


class VectorStore:
    """最小限のインメモリ・ベクトルストア。
    正規化済みベクトルを行列として保持し、クエリとの内積で類似度検索する。"""

    def __init__(self):
        self.matrix: np.ndarray | None = None   # (N, dim) の正規化済みベクトル
        self.chunks: list[str] = []             # チャンク本文
        self.metadatas: list[dict] = []         # 出典・カテゴリなどのメタデータ

    def add(self, vectors: np.ndarray, chunks: list[str], metadatas: list[dict]) -> None:
        """ベクトルとチャンク・メタデータを追加する。"""
        assert len(chunks) == len(metadatas) == vectors.shape[0]
        if self.matrix is None:
            self.matrix = vectors
        else:
            self.matrix = np.vstack([self.matrix, vectors])
        self.chunks.extend(chunks)
        self.metadatas.extend(metadatas)

    def search(self, query_vec: np.ndarray, top_k: int = 5) -> list[dict]:
        """クエリベクトルに近いチャンクを上位 top_k 件返す。
        戻り値は {chunk, metadata, score} の辞書リスト。"""
        if self.matrix is None:
            return []
        # 正規化済み前提なら内積=コサイン類似度。全件との内積を一括計算。
        scores = self.matrix @ query_vec            # (N,)
        # 上位 top_k を取得(argpartition で部分ソートし、その中を降順ソート)
        k = min(top_k, len(scores))
        idx = np.argpartition(-scores, k - 1)[:k]
        idx = idx[np.argsort(-scores[idx])]
        return [
            {
                "chunk": self.chunks[i],
                "metadata": self.metadatas[i],
                "score": float(scores[i]),
            }
            for i in idx
        ]

argpartition を使う理由:全件を完全ソートすると O(N log N) ですが、上位K件だけ欲しいなら argpartition で O(N) に抑えられます。数千〜数万件なら全件内積でも十分高速です。数百万件規模になると、HNSWやIVFといった**近似最近傍探索(ANN)**インデックスが必要になり、そこが専用ベクトルDBの主な価値になります。ここでは仕組みが見えることを優先し、厳密な全探索にしています。


8. Step 4:インジェスト(オフライン処理)をつなぐ

Step 1〜3をつなぎ、「文書ロード → チャンキング → 埋め込み → 保存(+メタデータ付与)」というオフラインパイプラインを完成させます。メタデータ付与は後で効いてきます ―― 出典表示(引用)、カテゴリでの絞り込み、アクセス制御による情報漏洩防止に使えます。

# step4_ingest.py
from step1_embedding import embed
from step2_chunking import recursive_chunk
from step3_vectorstore import VectorStore
from corpus import DOCUMENTS


def build_index(documents: list[dict], chunk_size: int = 120, overlap: int = 30) -> VectorStore:
    """文書集合からベクトルインデックスを構築する。"""
    store = VectorStore()

    all_chunks: list[str] = []
    all_metas: list[dict] = []
    for doc in documents:
        chunks = recursive_chunk(doc["text"], chunk_size=chunk_size, overlap=overlap)
        for j, ch in enumerate(chunks):
            all_chunks.append(ch)
            all_metas.append({
                "doc_id": doc["id"],
                "title": doc["title"],
                "category": doc["category"],
                "chunk_index": j,
            })

    # まとめて埋め込む(バッチ化で高速)
    vectors = embed(all_chunks)
    store.add(vectors, all_chunks, all_metas)
    print(f"インデックス構築完了: {len(all_chunks)} チャンク")
    return store


if __name__ == "__main__":
    store = build_index(DOCUMENTS)
    # 動作確認:素朴な密検索
    from step1_embedding import embed as embed_query
    qvec = embed_query(["パスワードを忘れたときの対処法は?"])[0]
    for hit in store.search(qvec, top_k=3):
        print(round(hit["score"], 4), hit["metadata"]["title"], "→", hit["chunk"][:40])

「パスワードを忘れたときの対処法は?」というクエリに対し、原文には「忘れた」「対処法」という表現がそのままなくても、パスワード再設定のチャンクが上位に来るはずです。これが密検索(dense retrieval)です。


9. Step 5:生成 ―― プロンプト構築とLLM呼び出し

検索できたら、最後は生成です。ここでのプロンプトは3つの部品から組み立てます。

  1. システムプロンプト:LLMの振る舞いを指示(例:「あなたは社内規程に詳しいアシスタントです」)。
  2. 検索した文脈チャンク:知識ベースから取ってきた関連テキスト。
  3. ユーザーの元の質問:入力そのまま。

そして重要なのが、「与えられた文脈のみに基づいて答え、根拠がなければ『分からない』と答えよ」 という制約を明示することです。これがハルシネーションを抑え、引用可能性(トレーサビリティ)を担保します。

生成にはOllamaのローカルAPIを使います。

# step5_generate.py
import requests

OLLAMA_URL = "http://localhost:11434/api/chat"
GEN_MODEL = "qwen2.5"   # 日本語が得意なモデル。llama3.1 等でも可

SYSTEM_PROMPT = (
    "あなたは社内規程に詳しい日本語アシスタントです。"
    "以下の【文脈】に書かれている情報だけを根拠に、簡潔に答えてください。"
    "【文脈】に答えがない場合は、推測せずに「提供された資料には記載がありません」と答えてください。"
    "回答の末尾に、根拠にした文書のタイトルを『出典: ...』の形で示してください。"
)


def build_prompt(query: str, hits: list[dict]) -> str:
    """検索結果からLLMに渡す文脈付きプロンプトを構築する。"""
    context_blocks = []
    for i, h in enumerate(hits, start=1):
        title = h["metadata"]["title"]
        context_blocks.append(f"[{i}] (出典: {title}\n{h['chunk']}")
    context = "\n\n".join(context_blocks)
    return f"【文脈】\n{context}\n\n【質問】\n{query}"


def generate(query: str, hits: list[dict], model: str = GEN_MODEL) -> str:
    """Ollama のチャットAPIで回答を生成する。"""
    user_prompt = build_prompt(query, hits)
    resp = requests.post(
        OLLAMA_URL,
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_prompt},
            ],
            "stream": False,
            "options": {"temperature": 0.2},  # 事実回答なので低温に
        },
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()["message"]["content"]

OpenAI互換APIで代替する場合:Ollamaを使わない環境では、以下のように差し替えられます(pip install openai)。ロジックは同一です。

from openai import OpenAI
client = OpenAI()  # OPENAI_API_KEY を環境変数に設定

def generate(query, hits, model="gpt-4o-mini"):
    user_prompt = build_prompt(query, hits)
    resp = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.2,
    )
    return resp.choices[0].message.content

10. Step 6:これで動く最小RAG(Naive RAG)

ここまでの部品をつなげば、動くRAGが完成します。これは論文でいう Naive RAG(インデックス作成 → 検索 → 生成の素朴な3ステップ)です。

# step6_naive_rag.py
from step1_embedding import embed
from step4_ingest import build_index
from step5_generate import generate
from corpus import DOCUMENTS


class NaiveRAG:
    def __init__(self, documents: list[dict]):
        self.store = build_index(documents)

    def answer(self, query: str, top_k: int = 4) -> str:
        # 1) クエリ埋め込み(インジェストと同じモデルを使うのが必須条件)
        qvec = embed([query])[0]
        # 2) 類似度検索
        hits = self.store.search(qvec, top_k=top_k)
        # 3) 文脈付きで生成
        return generate(query, hits)


if __name__ == "__main__":
    rag = NaiveRAG(DOCUMENTS)
    for q in [
        "パスワードを忘れたらどうすればいい?",
        "有給休暇は入社してどれくらいで付与される?",
        "出張がキャンセルになったらキャンセル料は経費になる?",
        "社長の趣味は?",   # 資料に無い → 「記載がありません」と答えるべき
    ]:
        print("Q:", q)
        print("A:", rag.answer(q))
        print("-" * 60)

最後の「社長の趣味は?」に対して、システムプロンプトの制約が効いていれば「提供された資料には記載がありません」と返すはずです。知らないことを知らないと言えるようになった ―― これがRAGの最初のご利益です。

しかしNaive RAGには弱点があります。論文(Gao et al. のサーベイ)が指摘するように、低精度(無関係なチャンクが混ざる)・低再現率(関連チャンクを取りこぼす)という問題です。以降のステップで、この2つを潰していきます。


11. Step 7:ハイブリッド検索(BM25 + Dense + RRF)

密検索(Dense)は意味で拾えるのが強みですが、固有名詞・型番・略語・完全一致のキーワードには弱いことがあります。「OTP」「VPN」のような語は、意味ベクトルより「その文字列が含まれているか」で探したいことも多い。ここでスパース検索(Sparse) の出番です。

代表的なスパース検索がBM25です。単語の出現頻度(TF)と、その単語がどれだけ珍しいか(IDF)で重み付けし、キーワード一致をスコア化します。密と疎、両方の「いいとこ取り」をするのがハイブリッド検索で、両者のランキングをReciprocal Rank Fusion(RRF) で融合します。

まずBM25をゼロから実装します。日本語はスペースで区切られないので、形態素解析(janome)でトークン化します。

# step7_bm25.py
import math
from collections import Counter

try:
    from janome.tokenizer import Tokenizer
    _jp_tokenizer = Tokenizer()

    def tokenize(text: str) -> list[str]:
        """日本語を形態素に分割し、記号・空白を除いた表層形の列を返す。"""
        return [
            tok.surface
            for tok in _jp_tokenizer.tokenize(text)
            if tok.surface.strip() and tok.part_of_speech.split(",")[0] not in ("記号",)
        ]
except ImportError:
    # janome が無い場合は文字bigramで代替(日本語でも一応機能する)
    def tokenize(text: str) -> list[str]:
        t = "".join(text.split())
        return [t[i:i + 2] for i in range(len(t) - 1)] if len(t) >= 2 else [t]


class BM25:
    """Okapi BM25 のフルスクラッチ実装。"""

    def __init__(self, corpus_tokens: list[list[str]], k1: float = 1.5, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.corpus = corpus_tokens
        self.N = len(corpus_tokens)
        self.doc_len = [len(doc) for doc in corpus_tokens]
        self.avgdl = sum(self.doc_len) / self.N if self.N else 0.0

        # 各文書のTF(単語頻度)
        self.tf = [Counter(doc) for doc in corpus_tokens]

        # 各単語のDF(その語を含む文書数)→ IDF
        df = Counter()
        for doc in corpus_tokens:
            for term in set(doc):
                df[term] += 1
        self.idf = {
            term: math.log(1 + (self.N - n + 0.5) / (n + 0.5))
            for term, n in df.items()
        }

    def score(self, query_tokens: list[str], index: int) -> float:
        """クエリと文書 index の BM25 スコア。"""
        score = 0.0
        tf = self.tf[index]
        dl = self.doc_len[index]
        for term in query_tokens:
            if term not in tf:
                continue
            idf = self.idf.get(term, 0.0)
            freq = tf[term]
            denom = freq + self.k1 * (1 - self.b + self.b * dl / self.avgdl)
            score += idf * (freq * (self.k1 + 1)) / denom
        return score

    def search(self, query: str, top_k: int = 10) -> list[tuple[int, float]]:
        """(文書index, スコア) の上位 top_k を返す。"""
        q = tokenize(query)
        scored = [(i, self.score(q, i)) for i in range(self.N)]
        scored.sort(key=lambda x: x[1], reverse=True)
        return scored[:top_k]

次にRRFです。RRFは各検索手法の順位(rank) だけを使ってスコアを合成します。「順位が高いほど、そして複数の手法で共通して上位なほど、最終スコアが高い」という直感を、1 / (k + rank) の和で表現します。スコアのスケールが異なる密検索とBM25を、順位という共通通貨で公平に混ぜられるのが利点です。

# step7_rrf.py

def reciprocal_rank_fusion(rankings: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    """複数のランキング(各要素はID列。上位ほど前)をRRFで融合する。
    戻り値は (ID, 融合スコア) を降順に並べたリスト。"""
    fused: dict[str, float] = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking):  # rank は0始まり
            fused[doc_id] = fused.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
    return sorted(fused.items(), key=lambda x: x[1], reverse=True)

これらを組み込んだハイブリッド版のインデックスとリトリーバを作ります。チャンクにグローバル通し番号(chunk_id)を振り、密・疎それぞれのランキングをRRFで融合します。

# step7_hybrid.py
import numpy as np
from step1_embedding import embed
from step2_chunking import recursive_chunk
from step7_bm25 import BM25, tokenize
from step7_rrf import reciprocal_rank_fusion
from corpus import DOCUMENTS


class HybridRetriever:
    def __init__(self, documents: list[dict], chunk_size: int = 120, overlap: int = 30):
        self.chunks: list[str] = []
        self.metadatas: list[dict] = []

        for doc in documents:
            for j, ch in enumerate(recursive_chunk(doc["text"], chunk_size, overlap)):
                self.chunks.append(ch)
                self.metadatas.append({
                    "chunk_id": len(self.chunks) - 1,   # グローバル通し番号
                    "doc_id": doc["id"],
                    "title": doc["title"],
                    "category": doc["category"],
                    "chunk_index": j,
                })

        # 密検索用の行列
        self.matrix = embed(self.chunks)  # 正規化済み
        # 疎検索(BM25)用のトークン化コーパス
        self.bm25 = BM25([tokenize(c) for c in self.chunks])

    def _dense_rank(self, query: str, top_k: int) -> list[str]:
        qvec = embed([query])[0]
        scores = self.matrix @ qvec
        idx = np.argsort(-scores)[:top_k]
        return [str(i) for i in idx]  # chunk_id を文字列化してRRFに渡す

    def _sparse_rank(self, query: str, top_k: int) -> list[str]:
        return [str(i) for i, _ in self.bm25.search(query, top_k)]

    def search(self, query: str, top_k: int = 5, candidate_k: int = 20) -> list[dict]:
        """密・疎それぞれで candidate_k 件取り、RRFで融合して上位 top_k を返す。"""
        dense = self._dense_rank(query, candidate_k)
        sparse = self._sparse_rank(query, candidate_k)
        fused = reciprocal_rank_fusion([dense, sparse])
        results = []
        for chunk_id_str, score in fused[:top_k]:
            cid = int(chunk_id_str)
            results.append({
                "chunk": self.chunks[cid],
                "metadata": self.metadatas[cid],
                "score": score,
            })
        return results


if __name__ == "__main__":
    retriever = HybridRetriever(DOCUMENTS)
    for q in ["OTPって何に使う?", "有給の繰り越しはできる?"]:
        print("Q:", q)
        for h in retriever.search(q, top_k=3):
            print("  ", round(h["score"], 4), h["metadata"]["title"])
        print("-" * 50)

「OTP」のような略語は疎検索が確実に拾い、「有給の繰り越し」のような言い換えは密検索が拾います。RRFが両者を束ねることで、どちらの質問タイプにも強くなります。


12. Step 8:リランキング(Cross-Encoder)

密検索の類似度は「おおよその近さ」の近似です。埋め込みはクエリと文書を別々にエンコードして距離を測る(bi-encoder)ため、細かな関連性を取りこぼすことがあります。トピックは近いが実際には質問に答えていないチャンクが、上位に来てしまうこともある。文脈として5〜6件をLLMに渡すとき、1〜2件でも無関係が混ざると回答品質が薄まります。

そこで2段階検索にします。

  • Stage 1(広く網を張る):ハイブリッド検索で候補を多め(例:20件)に取る。高速。
  • Stage 2(絞り込む)Cross-Encoder で、クエリと各候補をペアで同時に入力し、「この文書は本当に質問に答えているか」を精密に採点。上位K件だけをLLMに渡す。

Cross-Encoderはクエリと文書を一緒に見るので、両者の関係性を理解できます。その代わり1ペアずつ処理するので遅い ―― だから全件には掛けず、高速な検索で候補を絞ってから使うのです。

# step8_rerank.py
from sentence_transformers import CrossEncoder

# 多言語対応のリランカー。日本語も扱える。
_reranker = CrossEncoder("jinaai/jina-reranker-v2-base-multilingual", trust_remote_code=True)
# 代替(英語中心・軽量): CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")


def rerank(query: str, hits: list[dict], top_n: int = 5) -> list[dict]:
    """候補チャンクをCross-Encoderで採点し直し、上位 top_n を返す。"""
    if not hits:
        return []
    pairs = [(query, h["chunk"]) for h in hits]
    scores = _reranker.predict(pairs)
    ranked = sorted(zip(hits, scores), key=lambda x: x[1], reverse=True)
    out = []
    for h, s in ranked[:top_n]:
        h = dict(h)
        h["rerank_score"] = float(s)
        out.append(h)
    return out

これでリトリーバは「ハイブリッドで20件 → リランクで上位5件」という、実運用に近い二段構えになりました。


13. Step 9:Agentic RAG ―― LLMに検索を委ねる

ここまでは「毎メッセージ必ず検索する」設計でした。しかし「こんにちは」「さっきの話、要約して」にまで検索を走らせるのは無駄で、ときに逆効果です(無関係な文脈がモデルを混乱させる)。

Agentic RAGは、検索するかどうか・何を検索するか・いつ十分かをLLM自身に判断させます。素のLLMは「ツール(関数)」の一覧を渡されると、function_call(この関数をこの引数で呼べ)か、テキスト回答(もう十分、直接答える)のどちらかを返せます。あなたのコードが関数を実行して結果を返し、LLMが次の一手を決める ―― このループが Agentic RAG の本質です。

ネイティブのツール呼び出しに対応しないローカルモデルでも動くよう、ここでは「LLMにJSONで意思決定を出力させ、それをパースしてループする」方式でエージェントループを自作します。会話履歴(これまでの検索結果を含む)がエージェントの記憶になります。

# step9_agentic_rag.py
import json
import re
import requests
from step7_hybrid import HybridRetriever
from step8_rerank import rerank
from corpus import DOCUMENTS

OLLAMA_URL = "http://localhost:11434/api/chat"
MODEL = "qwen2.5"

AGENT_SYSTEM = (
    "あなたは社内規程アシスタントです。ユーザーの質問に答えるために、"
    "必要なら文書検索ツールを使えます。次のいずれかのJSONのみを出力してください。\n"
    '  1) 検索したいとき: {"action": "search", "query": "検索クエリ"}\n'
    '  2) 十分な情報が集まり回答できるとき: {"action": "answer", "text": "最終回答"}\n'
    "回答は提供された検索結果のみを根拠にし、根拠が無ければその旨を述べてください。"
    "JSON以外のテキストは絶対に出力しないこと。"
)


def _chat(messages: list[dict]) -> str:
    resp = requests.post(
        OLLAMA_URL,
        json={"model": MODEL, "messages": messages, "stream": False,
              "options": {"temperature": 0.1}},
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()["message"]["content"]


def _parse_json(text: str) -> dict:
    """LLM出力から最初のJSONオブジェクトを取り出してパースする。"""
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if not match:
        return {"action": "answer", "text": text.strip()}
    try:
        return json.loads(match.group(0))
    except json.JSONDecodeError:
        return {"action": "answer", "text": text.strip()}


class AgenticRAG:
    def __init__(self, documents: list[dict]):
        self.retriever = HybridRetriever(documents)

    def _retrieve_tool(self, query: str, top_k: int = 4) -> str:
        """検索ツールの実体:ハイブリッド検索+リランク→文脈テキストを返す。"""
        candidates = self.retriever.search(query, top_k=20, candidate_k=20)
        hits = rerank(query, candidates, top_n=top_k)
        blocks = [f"(出典: {h['metadata']['title']}{h['chunk']}" for h in hits]
        return "\n".join(blocks) if blocks else "(該当なし)"

    def answer(self, question: str, max_steps: int = 5) -> str:
        messages = [
            {"role": "system", "content": AGENT_SYSTEM},
            {"role": "user", "content": question},
        ]
        for step in range(max_steps):
            raw = _chat(messages)
            decision = _parse_json(raw)

            if decision.get("action") == "search":
                query = decision.get("query", question)
                print(f"[step {step+1}] 検索: {query}")
                observation = self._retrieve_tool(query)
                # 検索結果を会話履歴に追加(これがエージェントの記憶になる)
                messages.append({"role": "assistant", "content": raw})
                messages.append({
                    "role": "user",
                    "content": f"検索結果:\n{observation}\n\n"
                               f"この情報で回答できるなら answer を、"
                               f"まだ足りなければ別のクエリで search を出力してください。",
                })
            else:
                print(f"[step {step+1}] 回答")
                return decision.get("text", raw)
        return "(最大ステップ数に到達しました)"


if __name__ == "__main__":
    agent = AgenticRAG(DOCUMENTS)
    for q in [
        "こんにちは",                                    # 検索不要 → 直接回答
        "パスワードは何文字以上必要で、何日ごとに変える?",   # 1回検索で回答
        "有給の申請ルールと、繰り越しの可否を両方教えて",     # 複数回検索が望ましい
    ]:
        print("Q:", q)
        print("A:", agent.answer(q))
        print("=" * 60)

このループにより、エージェントは「挨拶なら検索しない」「必要なら複数回クエリを変えて検索する」といった振る舞いを見せます。System Design One の記事が言う通り、Traditional RAGが「1ページだけ見て答える学生」なら、Agentic RAGは「複数のソースを確認し、怪しい箇所を読み直し、突き合わせてから書く学生」です。

実運用での発展余地:ここで自作したループは最小限です。プロダクション級では、(1) クエリ書き換え(「2つ目の点は?」のような会話的フォローアップを、検索に適した独立クエリへ変換)、(2) 反復検索(1回で足りなければ精緻化して再検索)、(3) 自己反省(生成した回答が文脈と矛盾していないか検証するノード)などを足します。LangGraphのようなグラフ構造のフレームワークは、これらを状態遷移として素直に書けるよう設計されています。


14. RAGの評価 ―― 「なんとなく良さそう」から脱却する

RAGは組み上げた瞬間から 「本当にうまく動いているのか分からない」 という壁にぶつかります。回答が悪いとき、原因が「チャンキング(関連情報が分断された)」「検索(正しいチャンクが候補に入っていない)」「リランキング(入っていたが下位に沈んだ)」「生成(文脈は正しいのにLLMが無視した)」のどれなのか、切り分けられないと改善は当てずっぽうになります。

そこで評価フレームワークが要ります。RAG専用の代表格が**RAGAS(Retrieval-Augmented Generation Assessment)**です。BLEUやROUGEのような表層的な類似度では、検索品質や事実の裏付けを測れません。RAGASは以下のような軸を、多くの場合LLMを審判(LLM-as-a-judge)として評価します。

  • Faithfulness(忠実性):生成された回答は、検索した文脈に忠実か。作り話をしていないか。
  • Answer Relevance(回答の関連性):回答は質問に対して的確で有用か。
  • Context Precision(文脈精度):検索した文書は関連性が高く、無駄が少ないか。
  • Context Recall(文脈再現率):回答に必要な情報を、検索が取りこぼしていないか。

概念を掴むため、Faithfulness をLLMで自作採点する最小実装を示します。回答を「主張(claim)」に分解し、各主張が文脈から支持されるかをLLMに判定させ、支持率をスコアとする、というRAGASの発想の縮約版です。

# step14_eval.py
import json
import re
import requests

OLLAMA_URL = "http://localhost:11434/api/chat"
JUDGE_MODEL = "qwen2.5"


def _judge(prompt: str) -> str:
    resp = requests.post(
        OLLAMA_URL,
        json={"model": JUDGE_MODEL,
              "messages": [{"role": "user", "content": prompt}],
              "stream": False, "options": {"temperature": 0.0}},
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()["message"]["content"]


def _extract_json(text: str, default):
    m = re.search(r"\{.*\}|\[.*\]", text, re.DOTALL)
    if not m:
        return default
    try:
        return json.loads(m.group(0))
    except json.JSONDecodeError:
        return default


def faithfulness(answer: str, context: str) -> float:
    """回答を主張に分解し、各主張が文脈で支持される割合を返す(0〜1)。"""
    # 1) 回答を独立した主張のリストに分解
    decompose = (
        "次の回答を、検証可能な独立した主張(事実の単位)に分解し、"
        'JSON配列 ["主張1", "主張2", ...] のみを出力してください。\n\n'
        f"回答:\n{answer}"
    )
    claims = _extract_json(_judge(decompose), default=[])
    if not claims:
        return 0.0

    # 2) 各主張が文脈から支持されるかを判定
    supported = 0
    for claim in claims:
        verify = (
            "以下の【文脈】だけを根拠に、【主張】が支持されるか判定してください。"
            '{"supported": true} または {"supported": false} のJSONのみ出力。\n\n'
            f"【文脈】\n{context}\n\n【主張】\n{claim}"
        )
        verdict = _extract_json(_judge(verify), default={"supported": False})
        if verdict.get("supported") is True:
            supported += 1

    return supported / len(claims)


def context_recall(question: str, ground_truth: str, context: str) -> float:
    """正解回答の各文が、検索文脈でカバーされているかの割合(0〜1)。"""
    sentences = [s for s in re.split(r"[\n]", ground_truth) if s.strip()]
    if not sentences:
        return 0.0
    covered = 0
    for s in sentences:
        verify = (
            "【文脈】に、【文】の内容を裏付ける情報が含まれるか判定してください。"
            '{"covered": true/false} のJSONのみ出力。\n\n'
            f"【文脈】\n{context}\n\n【文】\n{s}"
        )
        verdict = _extract_json(_judge(verify), default={"covered": False})
        if verdict.get("covered") is True:
            covered += 1
    return covered / len(sentences)


if __name__ == "__main__":
    ctx = ("社内システムのパスワードは12文字以上とし、英大文字・小文字・数字・記号を含める。"
           "パスワードは90日ごとに変更する。")
    ans_good = "パスワードは12文字以上必要で、90日ごとに変更します。"
    ans_bad = "パスワードは8文字以上で、毎年変更すればよいです。"  # 事実に反する

    print("faithfulness(good):", faithfulness(ans_good, ctx))  # 1.0 に近い
    print("faithfulness(bad) :", faithfulness(ans_bad, ctx))   # 低くなる

評価は最初から組み込む:理想は、文書ごとに数個の「質問と正解」ペアを用意し、チャンキング・検索・リランキング・生成の各段を独立に採点するパイプラインを、プロジェクト初期から持つことです。そうすれば、チャンクサイズやtop_kを「勘」ではなく「数値」でチューニングできます。実プロダクトでは、自作せず RAGAS(本家)や TruLens・ARES 等を使うのが近道です。


15. 発展:Advanced / Modular RAG の地図

Naive RAGを起点に、RAGは Advanced RAG → Modular RAG へと進化してきました。全体像を俯瞰しておくと、自分のシステムに何を足すべきか判断しやすくなります。

Advanced RAGは、検索の前後を最適化します。

  • 検索前(pre-retrieval):チャンク粒度の調整、インデックス構造の最適化、メタデータ付与、クエリ書き換え。
  • 検索(retrieval):埋め込みモデルのファインチューニング、ハイブリッド検索。
  • 検索後(post-retrieval):リランキング、プロンプト圧縮(ノイズ除去)、「Lost in the Middle」対策として関連文脈をプロンプトの端に再配置する。

Modular RAGは、検索・記憶・融合・ルーティング・予測などを差し替え可能なモジュールとして扱い、タスクに応じて組み替えます。Naive/Advanced はその特殊形と見なせます。代表的な発展テクニックには次のものがあります。

  • HyDE(Hypothetical Document Embeddings):クエリから「仮の回答」を生成し、それを埋め込んで検索する。質問と文書の表現ギャップを埋める。
  • Query Rewriting / Step-Back Prompting:曖昧なクエリを検索に適した形へ変換したり、抽象化してから推論する。
  • Iterative / Recursive / Adaptive Retrieval:多段推論のために検索を繰り返す。Self-RAG や FLARE は「いつ・何を検索するか」をモデル自身に適応的に決めさせる。
  • GraphRAG:エンティティ間の関係を知識グラフとして抽出し、単純な検索では答えられない「知識ベース全体にまたがる比較・要約」に対応する。

これらは本記事で作った土台(密+疎+RRF+リランク+エージェントループ)の上に、部品として足していけます。


16. RAG vs ファインチューニング vs ロングコンテキスト

最後に、RAGを「いつ使うべきか」を他の選択肢との比較で整理します。

観点RAGファインチューニングロングコンテキスト
得意なこと最新・固有の知識を与える振る舞い・文体・出力形式を変える少数の文書をその場で丸ごと処理
知識の更新インデックス差し替えのみ(即時)再学習が必要(数日〜数週間)毎回プロンプトに詰める
コストクエリごとに検索+少量トークン学習に高い初期コストトークン課金が膨らむ
出典の提示可能(引用・帰属)困難可能だが精度は文脈長に依存
弱点検索品質に依存事実の記憶が苦手・陳腐化中央の情報を見落とす(U字カーブ)

重要なのは、これらは排他ではなく補完関係だということです。ベストプラクティスは「ファインチューニングで文体・出力形式(例:特定のJSON構造)を教え、RAGで最新の知識を流し込む」という組み合わせです。ロングコンテキストとも相性がよく、「まずRAGで的確な20件に絞り、それをロングコンテキストで一括処理する」構成が有効です。「Lost in the Middle」の知見が示す通り、巨大な文脈に頼るより、小さく関連性の高いチャンクを渡すほうがうまくいきます。


17. まとめ

本記事では、RAGを理論と実装の両輪で辿りました。

  • LLMは知識が凍結され、コンテキスト増大でも精度が上がらず、ファインチューニングも陳腐化する。モデルを変えずに、適切な文脈をその場で与えるのがRAGでした。
  • 心臓部は埋め込みによる意味検索。チャンキングが品質を左右し、密+疎(BM25)+RRFで検索を頑健にし、Cross-Encoderリランキングで精度を高め、Agenticループで「いつ検索するか」までモデルに委ねられる。
  • そして何より、評価なきRAGは盲目のチューニングになる。Faithfulness・Answer Relevance・Context Precision/Recallを測る仕組みを最初から持つべきでした。

表面的には「テキストを取ってきてモデルに渡すだけ」に見えるRAGは、掘るほどに多くの可動部が絡み合い、それぞれが静かに壊れうることが分かります。だからこそ、フレームワークの抽象に頼る前に一度フルスクラッチで通してみると、「どの一行が、なぜ必要なのか」が腹落ちします。ここで作った最小実装を土台に、Advanced/Modularのテクニックを一つずつ足していってみてください。


18. 参考文献

解説記事・ブログ

チュートリアル・実装教材

動画

論文(一次資料)

ツール・ドキュメント