この手順で作るRAGの構成

RAG(Retrieval-Augmented Generation、検索拡張生成)は、文書を検索してからLLMに答えさせる仕組みです。質問に関係する文書を探し、その本文をプロンプトに入れて回答させます。LLMが学習していない社内情報にも、文書を根拠にして答えられます。

この記事では、外部のデータベースやフレームワークを使いません。Pythonスクリプト数本で動く最小構成を作ります。処理は2段階です。

  1. 索引づくり:テキストを短い断片(チャンク)に分けます。各チャンクを埋め込みベクトルに変換し、ファイルに保存します。埋め込みとは、文章の意味を数値の並び(ベクトル)で表したものです。
  2. 質問応答:質問も同じ方法でベクトルにします。NumPyで意味の近いチャンクを探し、その本文をプロンプトに入れてLLMに回答させます。

ベクトル検索はNumPyの行列積だけで行います。数千チャンク程度までなら、専用のベクトルデータベースがなくても実用的な速さで動きます。

前提:必要な環境、APIキー、費用の目安

必要なのは、Python 3.9以上と、コマンドを実行できるターミナルです。この記事のコードは、埋め込みと回答生成の両方にOpenAIのAPIを使います。1つのSDKとAPIキーで両方を済ませられるので、入門用の構成として選びました。APIキーは環境変数 OPENAI_API_KEY に設定済みとします。他社のAPIやローカルモデルに置き換える方法は後の節で扱います。

回答生成に使うモデル名は、コードに直接書かず環境変数 CHAT_MODEL で渡します。モデルの世代交代が速く、記事に書いた名前はすぐ古くなるからです。OpenAIのモデル一覧で、現行の軽量なモデルを確認してください。埋め込みモデルには、執筆時点で提供されている text-embedding-3-small を使います。

費用はごくわずかです。サンプル文書は合計500字ほどで、埋め込みに使うトークンは1回の索引づくりで数百程度です。トークンとは、LLMがテキストを処理する単位です。回答生成も1回あたり数千トークンに収まります。執筆時点の料金で軽量モデルを選べば、何度試しても数円程度です。料金は改定されるので、公式の料金ページで確認してください。なお、実際の社内文書を外部APIに送る場合は、事前に社内のルールを確認してください。

先に知っておきたい4つのつまずきどころ

最小構成のRAGで起きる失敗は、ほとんどが次の4つのどれかに当てはまります。手順の中でも該当する箇所で触れます。確認方法と戻し方は、後の節にまとめました。

1つ目はチャンクの大きさです。大きすぎると、1つのチャンクに複数の話題が混ざって検索の精度が落ちます。小さすぎると文脈が切れ、何についての記述か分からない断片が検索されます。

2つ目は文書側と質問側で埋め込みモデルが違うことです。モデルが違うとベクトル空間が別物になり、類似度の数値が意味を持たなくなります。次元数がたまたま同じだとエラーが出ません。それらしい誤った結果が返るので、気づきにくい失敗です。

3つ目は検索結果が空になる、または無関係になることです。ベクトル検索は、文書に答えがない質問でも「いちばん近いもの」を必ず返します。それをそのままLLMに渡すと、無関係な資料をもとにした回答が作られます。

4つ目はコンテキスト長の超過です。コンテキスト長とは、LLMが1回のリクエストで受け取れるトークン数の上限です。検索件数やチャンクの大きさを増やしすぎると、上限を超えてエラーになるか、費用と応答時間が増えます。

手順1:作業フォルダとライブラリを用意する

作業フォルダを作ります。その中に仮想環境(プロジェクト専用のPython環境)を作り、ライブラリを入れます。

mkdir mini-rag
cd mini-rag
python -m venv .venv
source .venv/bin/activate
pip install openai numpy
export CHAT_MODEL="公式ドキュメントで確認したモデル名"

Windowsのコマンドプロンプトでは、仮想環境の有効化に .venv\Scripts\activate、環境変数の設定に set CHAT_MODEL=モデル名 を使います。pip install が失敗したら、仮想環境が有効か確認してください。有効なら、プロンプトの先頭に (.venv) と表示されます。環境が壊れたときは、.venv フォルダを削除して最初からやり直せば元に戻ります。

手順2:社内メモ風のサンプル文書を作る

次のコードを make_docs.py として保存し、python make_docs.py で実行します。docs フォルダに、テキストファイルが3つ作られます。各ファイルの1行目はタイトルです。手元の文書を使う場合は、同じ形式(1行目がタイトル、UTF-8)のファイルを docs に置いてください。

from pathlib import Path

docs = {
    'expense.txt': (
        '経費精算の手順\n'
        '経費は発生した月の翌月5営業日までに精算システムで申請する。'
        '領収書は画像で添付し、原本は3か月保管する。'
        '1件5万円を超える支出は、事前に部門長の承認を得ること。'
        'タクシー代は22時以降の帰宅か、重い機材を運ぶ場合に限り認める。'
    ),
    'remote_work.txt': (
        'リモートワーク規程\n'
        '在宅勤務は週3日まで認める。前日17時までに勤怠システムで申請する。'
        '業務は会社貸与のPCで行い、私物PCでの業務は禁止する。'
        '公共のWi-Fiを使う場合は必ずVPNに接続する。'
        '通信費の補助として月3000円を給与と合わせて支給する。'
    ),
    'meeting_room.txt': (
        '会議室の予約ルール\n'
        '会議室はグループウェアの施設予約から予約する。'
        '予約は2週間先まで、1回あたり最大2時間とする。'
        '開始10分を過ぎても利用がない場合、予約は自動で取り消される。'
        '大会議室Aは役員会議を優先するため、総務部の承認が必要。'
    ),
}

Path('docs').mkdir(exist_ok=True)
for name, text in docs.items():
    Path('docs', name).write_text(text, encoding='utf-8')
    print('作成:', name, len(text), '文字')

手順3:共通部分と索引づくりのスクリプトを書く

まず、埋め込みを作る関数を common.py にまとめます。文書側と質問側で必ず同じ関数を通すためです。こうしておけば、埋め込みモデルの不一致を構造的に防げます。返すベクトルは長さ1に正規化します。正規化しておくと、内積がそのままコサイン類似度(向きの近さを表す-1〜1の値)になります。

import numpy as np
from openai import OpenAI

EMBED_MODEL = 'text-embedding-3-small'
client = OpenAI()  # OPENAI_API_KEY を環境変数から読む

def embed(texts):
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    vecs = np.array([d.embedding for d in resp.data], dtype=np.float32)
    return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)

次に build_index.py を書きます。本文を「。」で文に分け、上限の文字数に達するまで文を詰めてチャンクにします。文の途中で切らないので、意味が壊れにくくなります。各チャンクの先頭には、文書のタイトルを付けます。こうすると、2つ目以降のチャンクだけが検索されても、何の規程の話かが伝わります。

import json
from pathlib import Path
import numpy as np
from common import EMBED_MODEL, embed

CHUNK_SIZE = 100  # サンプルが短いため小さめ。実際の文書では300〜800字から試す

def split_text(text, max_chars=CHUNK_SIZE):
    sentences = [s + '。' for s in text.replace('\n', '。').split('。') if s.strip()]
    chunks, current = [], ''
    for s in sentences:
        if current and len(current) + len(s) > max_chars:
            chunks.append(current)
            current = ''
        current += s
    if current:
        chunks.append(current)
    return chunks

records = []
for path in sorted(Path('docs').glob('*.txt')):
    title, _, body = path.read_text(encoding='utf-8').partition('\n')
    for i, chunk in enumerate(split_text(body)):
        records.append({'source': path.name, 'chunk_id': i, 'text': f'【{title}】{chunk}'})

for r in records:
    print(len(r['text']), r['source'], r['text'][:30])

vectors = embed([r['text'] for r in records])
Path('index').mkdir(exist_ok=True)
np.save('index/vectors.npy', vectors)
meta = {'embed_model': EMBED_MODEL, 'records': records}
Path('index/meta.json').write_text(json.dumps(meta, ensure_ascii=False, indent=1), encoding='utf-8')
print('保存しました:', vectors.shape)

python build_index.py で実行します。チャンクごとの文字数と冒頭部分が表示されます。最後に (6, 1536) のような形が表示されます。これは「チャンク数 × ベクトルの次元数」で、次元数はモデルによって変わります。meta.json には、使った埋め込みモデル名も保存しています。後で、質問側のモデルと照合するためです。

手順4:NumPyで検索し、類似度を目で確認する

search.py を作ります。質問をベクトル化し、全チャンクとの内積を一度に計算して、上位を返します。読み込み時に、索引のモデル名と現在のモデル名を照合します。食い違っていたら、検索せずに止めます。

import json
import sys
from pathlib import Path
import numpy as np
from common import EMBED_MODEL, embed

TOP_K = 3
MIN_SCORE = 0.2  # 仮の値。下の確認手順で実際のスコアを見て調整する

meta = json.loads(Path('index/meta.json').read_text(encoding='utf-8'))
if meta['embed_model'] != EMBED_MODEL:
    sys.exit(f'埋め込みモデルが不一致です(索引: {meta["embed_model"]} / 現在: {EMBED_MODEL})。build_index.py を再実行してください')
vectors = np.load('index/vectors.npy')
records = meta['records']

def search(question, top_k=TOP_K, min_score=MIN_SCORE):
    q = embed([question])[0]
    scores = vectors @ q
    order = np.argsort(scores)[::-1][:top_k]
    return [(records[i], float(scores[i])) for i in order if scores[i] >= min_score]

if __name__ == '__main__':
    for r, s in search(sys.argv[1], min_score=-1):
        print(f'{s:.3f}  {r["source"]}#{r["chunk_id"]}  {r["text"][:40]}')

LLMに渡す前に、検索だけを試します。python search.py "タクシー代は使える?" を実行すると、expense.txt のチャンクが上位に来るはずです。続けて、文書にない質問も試します。たとえば python search.py "社員食堂のメニューは?" です。この場合も何かしらの結果は返りますが、スコアは低くなります。2つのスコアを見比べて、その間の値を MIN_SCORE に設定します。スコアの水準は埋め込みモデルごとに違うので、ほかの記事の閾値をそのまま使わないでください。

手順5:検索結果をプロンプトに入れてLLMに回答させる

最後に ask.py を作ります。検索結果が空なら、LLMを呼ばずに終了します。資料の合計文字数には上限を設け、コンテキスト長の超過を防ぎます。プロンプトでは「資料だけを根拠にする」「答えがなければそう言う」の2点を指示します。

import os
import sys
from common import client
from search import search

CHAT_MODEL = os.environ['CHAT_MODEL']
MAX_CONTEXT_CHARS = 3000

question = sys.argv[1]
hits = search(question)
if not hits:
    sys.exit('関連する資料が見つかりませんでした。質問を言い換えるか、MIN_SCORE を見直してください')

context = ''
for r, score in hits:
    block = f'[{r["source"]}]\n{r["text"]}\n\n'
    if len(context) + len(block) > MAX_CONTEXT_CHARS:
        break
    context += block

prompt = (
    '以下の社内資料だけを根拠に、質問に日本語で答えてください。\n'
    '資料に答えがなければ「資料に記載がありません」と答えてください。\n'
    '回答の最後に、根拠にしたファイル名を書いてください。\n\n'
    f'# 資料\n{context}# 質問\n{question}'
)

resp = client.chat.completions.create(
    model=CHAT_MODEL,
    messages=[{'role': 'user', 'content': prompt}],
)
print(resp.choices[0].message.content)

python ask.py "深夜に帰るときタクシーを使っていい?" を実行します。22時以降の帰宅なら認められるという趣旨の回答と、根拠の expense.txt が表示されれば完成です。CHAT_MODEL が未設定だと KeyError で止まります。その場合は、手順1の環境変数を設定し直してください。

失敗したときの確認点と戻し方

この構成では、索引(index フォルダ)は docs と設定値からいつでも作り直せます。設定を変える前に index を index_bak としてコピーしておけば、比較も復元も簡単です。

チャンクが大きすぎる・小さすぎる

build_index.py が表示するチャンクの文字数と、search.py が返すチャンクの中身を確認します。上位のチャンクに無関係な話題が混ざっていたら大きすぎるので、CHUNK_SIZE を下げます。単独では意味が通らない断片ばかりなら小さすぎるので、上げます。値を変えたら build_index.py を再実行します。戻すときは、元の値に戻して再実行するか、index_bak を戻します。

文書側と質問側で埋め込みモデルが違う

search.py の照合で止まった場合は、build_index.py を再実行すれば直ります。EMBED_MODEL を変えたときや、チャンクにタイトルを付けるなど前処理を変えたときも、必ず索引を作り直してください。質問側だけを変えると、エラーなしで検索精度が崩れます。

検索結果が空、または無関係になる

まず search.py でスコアを直接見ます。全部のスコアが低い場合は、次の順に疑います。チャンク数が0になっていないか(拡張子が .txt でない、文字コードがUTF-8でない)。質問と文書で言葉が大きくずれていないか(社内の略語など)。MIN_SCORE が高すぎないか。無関係なチャンクのスコアが高い場合は、チャンクが大きすぎる可能性があります。答えがない質問で「資料に記載がありません」と返るのは、正しい動作です。

コンテキスト長を超える

回答生成で、トークン上限を超えた旨のエラーが出たら、MAX_CONTEXT_CHARS か TOP_K を下げます。日本語の文字数とトークン数の比率はモデルによって違うので、文字数による上限は目安です。正確に管理したいときは、各社のトークン計算手段を使ってください。埋め込みAPIにも、1テキストあたりと1リクエストあたりの上限があります。文書が増えて索引づくりでエラーが出たら、embed() に渡すリストを数十〜数百件ずつに分けて呼び出します。

別のAPIやローカルモデルに置き換える

変更が必要な箇所は2つだけです。common.py の embed() と、ask.py の回答生成の呼び出しです。埋め込み側を変えたら、索引を必ず作り直してください。

構成向いている場合注意点
OpenAIで埋め込みと生成(本記事)1つのキーで手早く試したい文書が外部APIに送られる
Voyage AIの埋め込み+Anthropic Claudeで生成回答生成にClaudeを使いたいAnthropicは埋め込みAPIを提供しておらず、公式ドキュメントでVoyage AIなどを案内している。APIキーが2つ必要
Google Gemini APIで埋め込みと生成Googleのサービスに統一したいSDKと応答の形式が異なる
sentence-transformersなどのローカル埋め込み文書を外部に送りたくない、埋め込みの費用をなくしたい初回にモデルのダウンロードが必要。日本語に対応したモデルを選ぶ必要がある

動いた後に次に手を入れる順番

まず、自分の文書を docs に置き、答えが分かっている質問を10個ほど用意します。最初に search.py で、正解のチャンクが上位3件に入るかを確認します。入らなければ検索側の問題なので、チャンクの大きさや MIN_SCORE を調整します。入っているのに回答が誤っていれば生成側の問題なので、プロンプトかモデルを見直します。この2つを分けて確認すると、原因を早く絞り込めます。チャンク数が数万を超えたり、文書の頻繁な更新や属性での絞り込みが必要になったりした時点で、FAISSやChromaなどのベクトル検索ライブラリへの移行を検討してください。