RAGシステムをゼロから作る総まとめ！シリーズ完結と次のステップを解説

「RAGって名前はよく聞くけど、実際どういう仕組みなの？」「自分でも作れるの？」そんな疑問を持っている方に向けて、この記事ではRAG（Retrieval-Augmented Generation）をゼロから丁寧に解説します。

難しそうに聞こえますが、仕組みを理解してしまえば「なるほど、そういうことか！」となるはず。初心者の方でもわかるよう、図解イメージや実際のコードを交えながら説明していきますよ 🚀

Contents

🔍 RAGとは何か？まず概念を理解しよう
- 💡 なぜRAGが必要なの？
- 📚 試験でたとえると…
🏗️ RAGの全体的な仕組みを理解しよう
- ① データ準備フェーズ（インデックス構築）
- ② 質問応答フェーズ（RAG推論）
🔑 RAGを構成する3つのキーワード
🛠️ 実際にRAGシステムを作ってみよう
📁 ファイル構成のまとめ
⚡ RAGシステムの精度を上げるテクニック
🚀 さらに発展させるには？
❓ よくある質問
まとめ

🔍 RAGとは何か？まず概念を理解しよう

AI search database / Photo by Tima Miroshnichenko via Pexels

RAGとは Retrieval-Augmented Generation の略で、日本語にすると「検索で拡張した文章生成」です。

一言でいうと、「AIが回答する前に、関連する情報を検索してから答える仕組み」のことです。

💡 なぜRAGが必要なの？

ChatGPTのような大規模言語モデル（LLM）は、学習データをもとに回答を生成します。でも、次のような弱点があります。

❌ 学習データのカットオフ以降の情報を知らない（最新ニュースに対応できない）
❌ 自社の社内文書・独自データを参照できない
❌ ハルシネーション（もっともらしい嘘）が起きやすい

そこで登場するのがRAGです。質問に答える前に自分のデータベースから関連情報を検索し、その情報をLLMに渡して回答を生成させることで、これらの弱点を補えます。

📚 試験でたとえると…

試験のたとえが一番わかりやすいので使わせてください。

通常のLLM：教科書を暗記した状態で試験を受ける（記憶が古かったり間違っていることも）
RAG：試験中に教科書を開いて調べてから回答する（常に正確な情報をもとに答えられる）

RAGなら自社データや最新情報を「教科書」として使えるので、より正確で信頼性の高い回答が得られるわけです。

🏗️ RAGの全体的な仕組みを理解しよう

RAGシステムは大きく「データを準備する段階」と「質問に答える段階」の2つに分かれます。

① データ準備フェーズ（インデックス構築）

まず、参照したいドキュメント（PDF・テキスト・Webページなど）を事前に処理しておきます。

ドキュメント
    ↓ チャンキング（文書を適切なサイズに分割）
小さなテキストのかたまり（チャンク）
    ↓ エンベディング（テキストを数値ベクトルに変換）
数値ベクトル
    ↓ ベクトルDBに保存
ベクトルデータベース ✅

② 質問応答フェーズ（RAG推論）

ユーザーが質問してきたときの流れです。

ユーザーの質問
    ↓ エンベディング（質問もベクトルに変換）
質問ベクトル
    ↓ ベクトル類似度検索（似ている文書を探す）
関連チャンク（上位K件）
    ↓ プロンプトに組み込んでLLMに送信
最終的な回答 ✅

この「検索 → 生成」の流れがRAGの本質です。シンプルですよね？

🔑 RAGを構成する3つのキーワード

RAGを理解するうえで絶対に押さえておきたいキーワードが3つあります。

① エンベディング（Embedding）

テキストを数値のベクトルに変換する技術です。

たとえば「犬」というテキストは [0.12, -0.34, 0.87, ...] のような数百次元の数値配列に変換されます。この変換のポイントは、意味が似ているテキストは似たベクトルになること。

「犬」と「ワンちゃん」→ ベクトルが近い
「犬」と「自動車」→ ベクトルが遠い

この性質を利用して、「意味的に似ているドキュメントを検索する」ことができるわけです。

② ベクトルデータベース（Vector DB）

エンベディングされたベクトルを保存し、類似度検索を高速に行える専用データベースです。

代表的なものとしては：

pgvector：PostgreSQLの拡張機能。既存DBと統合しやすい
Chroma：Pythonから使いやすい軽量ベクトルDB
FAISS：Metaが開発した高速ベクトル検索ライブラリ
Pinecone：クラウド型のマネージドベクトルDB

この記事では、最も実用的で本番環境にも使いやすいpgvectorを使って解説します。

③ チャンキング（Chunking）

長いドキュメントをLLMに渡すには限界があります（トークン制限）。そのため、文書を適切なサイズの「チャンク（かたまり）」に分割する必要があります。

チャンクサイズは検索精度に直結する重要なパラメータです。

小さすぎる：文脈が失われて検索精度が下がる
大きすぎる：LLMのトークン制限に引っかかる・ノイズが増える
一般的な目安：500〜1000文字程度、前後に50〜100文字の重複（オーバーラップ）を設ける

🛠️ 実際にRAGシステムを作ってみよう

ここからは実際のPythonコードを使って、RAGシステムを段階的に構築していきます。

環境構築

まず必要なライブラリをインストールします。

pip install openai psycopg2-binary pgvector

また、PostgreSQLにpgvector拡張を有効化しておく必要があります。

-- PostgreSQLに接続して実行
CREATE EXTENSION IF NOT EXISTS vector;

ステップ1：データベースのセットアップ

ドキュメントとそのベクトルを保存するテーブルを作成します。

import psycopg2

def setup_database():
    """ドキュメント保存用テーブルを作成する"""
    conn = psycopg2.connect("postgresql://localhost/ragdb")
    cur = conn.cursor()

    # pgvector拡張の有効化
    cur.execute("CREATE EXTENSION IF NOT EXISTS vector")

    # ドキュメントテーブルの作成
    # embedding列はOpenAIのtext-embedding-3-smallが出力する1536次元
    cur.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id        SERIAL PRIMARY KEY,
            content   TEXT NOT NULL,
            metadata  JSONB,
            embedding VECTOR(1536)
        )
    """)

    conn.commit()
    conn.close()
    print("✅ データベースのセットアップ完了")

setup_database()

ステップ2：HNSWインデックスの作成

大量のベクトルを高速に検索するために、HNSWインデックスを作成します。

def create_index():
    """ベクトル検索用のHNSWインデックスを作成する"""
    conn = psycopg2.connect("postgresql://localhost/ragdb")
    cur = conn.cursor()

    # HNSWインデックスの作成
    # m=16: 各ノードの最大接続数（精度と速度のバランス）
    # ef_construction=64: インデックス構築時の探索幅
    cur.execute("""
        CREATE INDEX IF NOT EXISTS documents_embedding_idx
        ON documents
        USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 64)
    """)

    conn.commit()
    conn.close()
    print("✅ HNSWインデックスの作成完了")

create_index()

HNSWとは Hierarchical Navigable Small World の略で、グラフ構造を使った近傍探索アルゴリズムです。全件スキャンに比べて圧倒的に高速で、数百万件のベクトルでも実用的な速度で検索できます。

ステップ3：ドキュメントをエンベディングしてDBに登録

実際のドキュメントをチャンクに分割し、エンベディングしてDBに保存します。

import json
from openai import OpenAI

client = OpenAI()  # OPENAI_API_KEY環境変数から自動読み込み

def embed(text: str) -> list[float]:
    """テキストをOpenAIのAPIでベクトル化する"""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """テキストを重複付きで分割する"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():  # 空のチャンクは除外
            chunks.append(chunk)
        start += chunk_size - overlap  # オーバーラップ分だけ戻る
    return chunks

def ingest_document(text: str, metadata: dict = None):
    """ドキュメントをチャンキング→エンベディング→DB保存する"""
    chunks = chunk_text(text)
    print(f"📄 {len(chunks)}個のチャンクに分割しました")

    conn = psycopg2.connect("postgresql://localhost/ragdb")
    cur = conn.cursor()

    for i, chunk in enumerate(chunks):
        vec = embed(chunk)
        cur.execute(
            "INSERT INTO documents (content, metadata, embedding) VALUES (%s, %s, %s)",
            (chunk, json.dumps(metadata or {}), vec)
        )
        print(f"  ✅ チャンク {i+1}/{len(chunks)} を保存")

    conn.commit()
    conn.close()
    print("✅ ドキュメントの登録完了")

# 使い方
sample_text = """
Pythonは1991年にグイド・ヴァン・ロッサムによって作られたプログラミング言語です。
シンプルな文法と豊富なライブラリが特徴で、データサイエンス・Web開発・AI開発など
幅広い分野で使われています。Pythonの非同期処理にはasyncioライブラリを使います。
async/awaitキーワードを使って、I/O待ち時間を有効活用できます。
"""

ingest_document(sample_text, metadata={"source": "python_intro.txt"})

ステップ4：ベクトル類似度検索の実装

登録したドキュメントの中から、質問に近いものを検索します。

def search(query: str, top_k: int = 3) -> list[tuple]:
    """クエリに意味的に近いドキュメントチャンクを検索する"""
    # 質問もベクトルに変換
    query_vec = embed(query)

    conn = psycopg2.connect("postgresql://localhost/ragdb")
    cur = conn.cursor()

    # pgvectorの<->演算子でコサイン距離を計算して近い順に取得
    # 1 - 距離 = コサイン類似度（1に近いほど似ている）
    cur.execute("""
        SELECT
            content,
            metadata,
            1 - (embedding <-> %s::vector) AS similarity
        FROM documents
        ORDER BY embedding <-> %s::vector
        LIMIT %s
    """, (query_vec, query_vec, top_k))

    results = cur.fetchall()
    conn.close()
    return results

# 検索テスト
results = search("Pythonで非同期処理をするには？")
for content, metadata, similarity in results:
    print(f"類似度: {similarity:.3f}")
    print(f"内容: {content[:100]}...")
    print(f"ソース: {metadata}")
    print("---")

ここでのポイントを整理します 👇

<-> 演算子：pgvectorのコサイン距離演算子。SQLだけで類似検索が実現できます
類似度 = 1 – 距離：距離が小さいほど似ているので、1から引いて「似ている度合い」に変換しています
HNSWインデックスが効く：先ほど作ったインデックスのおかげで、大量データでも高速に検索できます

ステップ5：検索結果をLLMに渡してRAG応答を生成

いよいよRAGの核心部分です。検索で得られた関連情報をLLMへのプロンプトに組み込みます。

def rag_answer(question: str, top_k: int = 3) -> str:
    """RAGを使って質問に回答する"""
    # 1. 関連ドキュメントを検索
    search_results = search(question, top_k=top_k)

    if not search_results:
        return "関連する情報が見つかりませんでした。"

    # 2. 検索結果をコンテキストとして整形
    context_parts = []
    for i, (content, metadata, similarity) in enumerate(search_results):
        context_parts.append(f"【参考情報 {i+1}】\n{content}")

    context = "\n\n".join(context_parts)

    # 3. プロンプトを構築
    prompt = f"""以下の参考情報をもとに、質問に答えてください。
参考情報にない内容については「提供された情報には含まれていません」と答えてください。

=== 参考情報 ===
{context}

=== 質問 ===
{question}

=== 回答 ==="""

    # 4. LLMに送信して回答を生成
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "あなたは親切なアシスタントです。提供された参考情報のみをもとに回答してください。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.1  # 低めにすると事実に忠実な回答になる
    )

    return response.choices[0].message.content

# RAGテスト
answer = rag_answer("Pythonの非同期処理について教えてください")
print(answer)

このコードの流れを整理するとこうなります 👇

質問をベクトル化して類似ドキュメントを検索
検索結果を「参考情報」としてテキストに整形
「参考情報＋質問」をまとめたプロンプトを作成
LLMに送信して回答を得る

プロンプトで「参考情報にないことは答えるな」と指示するのがポイントです。これによりハルシネーションを抑制できます。

📁 ファイル構成のまとめ

ここまで作ったコードを整理すると、プロジェクト構成はこうなります。

rag_project/
├── 01_setup_db.py       # テーブル＋拡張機能のセットアップ
├── 02_create_index.py   # HNSWインデックスの作成
├── 03_ingest.py         # ドキュメントのチャンキング→DB保存
├── 04_search.py         # ベクトル類似度検索
├── 05_rag.py            # 検索＋LLM生成のRAGメイン処理
└── requirements.txt     # 必要なライブラリ一覧

各ファイルの役割が明確に分かれているので、後から改修や機能追加もしやすい構成です。

⚡ RAGシステムの精度を上げるテクニック

基本のRAGが動いたら、次は精度向上を目指しましょう。よく使われるテクニックを紹介します。

① チャンクサイズの最適化

チャンクサイズはケースバイケースです。

技術ドキュメント：300〜500文字（細かく正確な情報を拾いたい）
一般的な記事・マニュアル：500〜1000文字
書籍・長文：1000〜1500文字（文脈を保持したい）

実際には複数のサイズで試して、回答品質を比較するのがベストです。

② ハイブリッド検索

ベクトル検索（意味的類似度）と全文検索（キーワードマッチ）を組み合わせる手法です。

ベクトル検索が得意：「非同期処理の方法」→「asyncioの使い方」のような意味的な検索
全文検索が得意：固有名詞・型番・コードなど完全一致が重要な検索

両者のスコアを組み合わせる RRF（Reciprocal Rank Fusion） という手法が特によく使われます。

③ 再ランキング（Reranker）

初回の類似度検索で上位20件を取得し、専用の再ランキングモデルでさらに精度よく並び替える手法です。

# 概念コード（Cohereの再ランキングAPIを使う例）
# pip install cohere

import cohere

co = cohere.Client("YOUR_API_KEY")

def rerank(query: str, documents: list[str], top_n: int = 3):
    """検索結果を再ランキングで精度向上させる"""
    results = co.rerank(
        query=query,
        documents=documents,
        top_n=top_n,
        model="rerank-multilingual-v3.0"  # 日本語対応モデル
    )
    return results.results

④ メタデータフィルタリング

ドキュメントにメタデータ（日付・カテゴリ・ソースなど）を付けておくと、検索範囲を絞り込めます。

def search_with_filter(query: str, source_filter: str, top_k: int = 3):
    """特定のソースのドキュメントだけを検索する"""
    query_vec = embed(query)

    conn = psycopg2.connect("postgresql://localhost/ragdb")
    cur = conn.cursor()

    # metadataのJSONBフィールドでフィルタリング
    cur.execute("""
        SELECT content, 1 - (embedding <-> %s::vector) AS similarity
        FROM documents
        WHERE metadata ->> 'source' = %s
        ORDER BY embedding <-> %s::vector
        LIMIT %s
    """, (query_vec, source_filter, query_vec, top_k))

    return cur.fetchall()

🚀 さらに発展させるには？

基本のRAGシステムが動くようになったら、次のステップとして以下の発展が考えられます。

アプリ化する

Streamlit：Pythonだけでチャット画面のWebアプリを作れる。プロトタイプに最適
FastAPI：RAGをAPIエンドポイントとして公開できる。本番サービス化に向いている

PDFやWebページを取り込む

PyMuPDF（fitz）：PDFからテキストを抽出するライブラリ
BeautifulSoup / Scrapy：WebページのHTMLからテキストを抽出
LangChain / LlamaIndex：これらの処理をまとめて担ってくれるフレームワーク

エンベディングモデルを変える

OpenAI text-embedding-3-large：精度を最優先したいとき（コスト増）
sentence-transformers：ローカルで動く無料のエンベディングモデル
multilingual-e5-large：日本語を含む多言語に強いモデル

❓ よくある質問

Q. LangChainを使わずに作る意味はあるの？

あります！LangChainなどのフレームワークは便利ですが、「何が起きているかわからない」ブラックボックスになりがちです。今回のようにゼロから作ると、各コンポーネントの役割を完全に理解できるので、トラブル対応や独自カスタマイズがしやすくなります。仕組みを理解してからフレームワークを使うのが理想的です。