【FastAPIで作る社内ナレッジ検索(RAG) 第3回】質問に近い社内文書を探す(ベクトル検索)を実装する

2026年9月30日2026年10月1日

社内ナレッジ検索(RAG)の心臓部は、「質問に近い文書の断片を、正しく見つけてくること」です。前回の第2回では、pgvector入りのPostgreSQLに文書の断片を保存しました。第3回では、質問と断片を「数値の並び」に変える埋め込みの部品を作り、質問に近い断片を近い順に取り出すベクトル検索を実装します。

「社内文書をAIに検索させるとき、『言葉が違っても意味が近い文書』を見つけてくれるって本当? キーワード検索と何が違うの?」

先に結論をお伝えします。ベクトル検索は、文章を数値の並び(ベクトル)にして「向きの近さ」で探す仕組みで、PostgreSQLとpgvectorだけで実装できます。ただし、近さの質は「文章を数値にする埋め込みモデル」の出来でほぼ決まります。この回では外部サービスなしで動く簡易版を作り、本物のモデルに差し替えられる形にしておきます。また、簡易版だからこそ見える「検索が外れる場面」も正直にお見せします。

前回の記事:【第2回】PostgreSQLとpgvectorをDocker Composeで動かし、文書の断片を保存する

社内文書のベクトル検索とは:キーワード検索との違い

ベクトル検索(＝文章を数値の並びに変え、数値の近さで似た文章を探す方法)は、キーワード検索とは探し方が違います。

比べる点	キーワード検索	ベクトル検索
探し方	質問の単語が文書に含まれるか	質問と文書の「数値の並び」がどれだけ近いか
得意なこと	型番・固有名詞など、決まった言葉の一致	言い換え・表現の揺れ(例:「休める?」と「取得できます」)
苦手なこと	言い換えや、話し言葉の質問	型番のような厳密な一致。近さの理由が説明しにくい
品質を左右するもの	辞書・検索の設定	埋め込みモデルの出来

この数値の並びを作る部品を「埋め込み(embedding)」と呼びます。pgvectorは、その数値の並びを保存し、距離の近い順に取り出す機能をPostgreSQLに追加します。

📰 出典：pgvector 公式リポジトリ(README)

READMEには、距離を測る演算子として <->(ユークリッド距離)・<#>(内積)・<=>(コサイン距離)があること、近い順の検索を速くする索引としてHNSWとIVFFlatがあることが説明されています(執筆時点:2026年9月)。この連載では、文章検索でよく使われるコサイン距離(向きの違いを見る距離。0に近いほど同じ向き)と、HNSW索引を使います。

この回で作るもの

部品	ファイル	役割
埋め込みの型	app/search/embedder.py	文章を数値の並びにする部品の「形」と、外部不要の簡易版
列と索引の追加	db/migrations/002_embedding.sql	chunks表にembedding列とHNSW索引を足す
マイグレーション実行	app/db/migrate.py	表の変更を番号順に流す小さな仕組み
保存と検索	app/db/store.py	断片と一緒に数値の並びを保存し、近い順に検索する
検索API・確認用コマンド	app/main.py、app/search/__main__.py	`/api/search` と `python -m app.search`

実装:埋め込みの部品を差し替え可能な形で作る

埋め込みのインターフェースと簡易版(app/search/embedder.py)

class Embedder(Protocol):
    """文章を「意味を表す数値の並び（ベクトル）」に変える部品。本物のモデルもこの形で差し替える。"""

    dimensions: int

    def embed(self, text: str) -> list[float]: ...


class HashingEmbedder:
    dimensions = DIMENSIONS  # 256

    def embed(self, text: str) -> list[float]:
        vector = [0.0] * self.dimensions
        normalized = "".join(text.split())
        for i in range(len(normalized) - 1):
            digest = hashlib.blake2b(normalized[i : i + 2].encode("utf-8"), digest_size=4).digest()
            vector[int.from_bytes(digest, "big") % self.dimensions] += 1.0
        norm = math.sqrt(sum(v * v for v in vector))
        return [v / norm for v in vector] if norm else vector

Embedder は「文章を渡すと数値のリストが返る」という約束だけを決めた型です。第0回で決めた「外部サービスに縛られない」方針を、コードの形にしたものです。

HashingEmbedder は、連載の動作確認用の簡易版です。文章を2文字ずつに区切り(「半日休暇」なら「半日」「日休」「休暇」)、その出現回数を256個の欄に振り分けて数えます。同じ2文字の並びを多く含む文章ほど、数値の並びが近くなります。

ここで大事な注意があります。この簡易版は「意味」を理解していません。「同じ文字の並びを含むか」だけで近さを決めています。本番で使う埋め込みモデルは、「休める?」と「取得できます」のような言い換えも近いと判断できるよう学習されたもので、仕組みが根本的に違います。簡易版を作るのは、外部サービスのアカウントやAPIキーがなくても、検索の流れ全体を動かして確かめられるようにするためです。

表に埋め込みの列と索引を足す(db/migrations/002_embedding.sql)

ALTER TABLE chunks ADD COLUMN IF NOT EXISTS embedding vector(256);

CREATE INDEX IF NOT EXISTS chunks_embedding_idx
    ON chunks USING hnsw (embedding vector_cosine_ops);

vector(256) は「256個の数値を入れる列」という意味です。この256は、埋め込みモデルが出す数値の個数(次元数)と必ず一致させます。本物のモデルに替えるときは次元数が変わる(数百〜数千)ので、列を作り直して全断片を入れ直します。

HNSWは、大量のデータから近いものを素早く探すための索引です。数百件程度なら索引がなくても十分速いですが、文書が増えたときに備えて最初から入れておきます。

第2回でお伝えしたとおり、db/init のSQLは初回起動のときしか流れません。そこで、表の変更は db/migrations に番号付きで置き、次の小さな部品で流す形にしました。

# app/db/migrate.py
def apply_migrations(conn: psycopg.Connection) -> list[str]:
    """db/migrations の SQL を番号順に流す。どれも何度流しても同じ結果になる書き方にしてある。"""
    applied = []
    for path in sorted(MIGRATIONS_DIR.glob("*.sql")):
        conn.execute(path.read_text(encoding="utf-8"))
        applied.append(path.name)
    conn.commit()
    return applied

SQLに IF NOT EXISTS を付けてあるので、何度流しても結果は同じです。なお、この仕組みは「流し直せる」だけで、どこまで適用済みかの記録は持ちません。実運用では、変更履歴を管理するツール(Alembicなど)の導入を検討します。

断片と一緒に埋め込みを保存し、近い順に検索する(app/db/store.py)

def to_vector_literal(vector: list[float]) -> str:
    """Pythonの数値リストを、pgvector が読める文字列 '[0.1,0.2,...]' にする。"""
    return "[" + ",".join(f"{v:.6f}" for v in vector) + "]"

def save_chunks(conn, chunks, embedder: Embedder | None = None) -> int:
    ...
    cur.executemany(
        "INSERT INTO chunks (source, heading, chunk_index, body, embedding)"
        " VALUES (%s, %s, %s, %s, %s::vector)",
        [(c.source, c.heading, c.index, c.text,
          to_vector_literal(embedder.embed(c.text)) if embedder else None) for c in chunks],
    )

def search_chunks(conn, embedder: Embedder, question: str, limit: int = 3) -> list[SearchHit]:
    """質問を数値の並びにして、近い断片を近い順に limit 件返す。"""
    query = to_vector_literal(embedder.embed(question))
    rows = conn.execute(
        "SELECT source, heading, body, embedding <=> %s::vector AS distance"
        " FROM chunks WHERE embedding IS NOT NULL"
        " ORDER BY embedding <=> %s::vector LIMIT %s",
        (query, query, limit),
    ).fetchall()
    return [SearchHit(*row) for row in rows]

検索の中身は、SQLの1文です。embedding <=> 質問のベクトル でコサイン距離を測り、ORDER BY で近い順に並べ、LIMIT で上位だけを取ります。質問も、断片を保存したときと同じ埋め込みの部品で数値にするのが決まりです。保存と検索で別のモデルを使うと、数値の意味が合わず、まともに検索できません。

SQLの値は %s で渡しており、質問文をSQLに直接つなげていません。ユーザーの入力をそのままSQLに組み込む作りは、SQLインジェクション(悪意ある入力でDBを不正に操作される攻撃)の原因になるため避けます。

検索API(app/main.py)

@app.post("/api/search")
def search(body: AskRequest, settings: Annotated[Settings, Depends(get_settings)]) -> SearchResponse:
    # 第3回：質問に近い断片を探して返す。回答文の生成は第4回。
    with connect(settings.database_url) as conn:
        return SearchResponse(hits=search_chunks(conn, HashingEmbedder(), body.question))

回答を作る /api/ask はまだ仮のままです。まず「正しい断片を見つける」ところを単独で確かめ、第4回でAIによる回答生成につなげます。

動作確認の方法

cd blogs/it_hacchu/series/rag-naibu/code
docker compose up -d
pip install -e ".[dev]"
python -m app.db                        # マイグレーション → 読み込み → 分割 → 埋め込み → 保存
python -m app.search "半日休暇は何時までですか"

実行結果(筆者の確認)

python -m app.search は、上位3件の「距離・文書名・見出し」を表示します。距離は小さいほど質問に近い値です。

Q: 半日休暇は何時までですか
0.714  vacation.md > 休暇規程（サンプル） > 年次有給休暇 > 半日休暇
0.808  vacation.md > 休暇規程（サンプル） > 年次有給休暇 > 申請方法
0.910  vacation.md > 休暇規程（サンプル）

Q: 領収書の画像はどうすればいい
0.781  expense.txt >
0.793  vacation.md > 休暇規程（サンプル）
0.890  vacation.md > 休暇規程（サンプル） > 年次有給休暇 > 申請方法

Q: 結婚したら休める？
0.765  vacation.md > 休暇規程（サンプル） > 特別休暇 > 慶弔休暇
0.871  expense.txt >
0.874  vacation.md > 休暇規程（サンプル） > 年次有給休暇 > 半日休暇

3つの質問とも、1位に期待した断片が来ました。「結婚したら休める?」は、断片に「結婚」の文字が含まれていたため1位になっています。

簡易版の限界:関係のない質問にも「一番近い断片」が返る

Q: 社長の好きな食べ物は？
0.880  vacation.md > 休暇規程（サンプル） > 特別休暇 > 慶弔休暇
0.885  expense.txt >
0.887  vacation.md > 年次有給休暇 > 半日休暇

文書に書いていない質問でも、ベクトル検索は必ず上位の断片を返します。「近いものを並べる」仕組みなので、どれも遠い場合でも1位は決まってしまうからです。ただし距離を見ると、関係のある質問(0.71〜0.78)より遠い(0.88前後)ことが分かります。第5回では、この距離を使って「根拠になる断片が無いので、答えられません」と返す判定を入れます。

また、簡易版では「休暇」と「休み」のような言い換えは近いと判定できません。言い換えに強い検索にするには、本物の埋め込みモデルへの差し替えが必要です。

筆者が確認できた範囲・できていない範囲

項目	確認状況
ruff、pytest(DBを使わないテスト12件)	確認済み
DBを使うテスト5件を含む全17件、上記の検索結果	Ubuntu標準のPostgreSQL 16とpgvector 0.6.0を直接起動して確認済み(HNSW索引の作成も確認)
docker-compose.yml と pgvector/pgvector:pg17 での起動	筆者の環境ではDockerが使えず未確認(第2回から変わらず)
本物の埋め込みモデル(外部API等)での検索	未確認。この回では簡易版のみを使用

自動テスト

埋め込み部品のテスト(DB不要)では、「関係する文の方が、関係のない文より近い」「空の文字列でも落ちない」ことを固定しました。

def test_similar_text_is_closer_than_unrelated():
    e = HashingEmbedder()
    question = e.embed("半日休暇は何時までですか")
    related = e.embed("午前休・午後休として半日単位で取得できます。半日休暇は…")
    unrelated = e.embed("領収書の画像は必ず添付してください。")
    assert cosine(question, related) > cosine(question, unrelated)

DBを使うテスト(tests/test_store.py)では、近い順に返ることと、埋め込みが未設定の断片は検索に出ないこと、/api/search が動くことを確かめています。

つまずきやすい点・セキュリティ上の注意

次元数の不一致: 列の vector(256) と埋め込みモデルの出力の個数が違うと、保存時にエラーになります。モデルを替えるときは列と全データを作り直します。
保存と検索で埋め込みを揃える: 文書を取り込んだときと違うモデル・設定で質問を数値にすると、検索結果が崩れます。モデル名と版は設定として記録しておきます。
距離の演算子と索引の組み合わせ: 索引は vector_cosine_ops(コサイン距離用)で作っているので、検索も <=> を使います。ここが食い違うと、索引が使われず遅くなります。
外部の埋め込みAPIに文書を送る場合: 社内文書の本文が外部の事業者に送信されます。利用規約(入力データの保存・学習利用の扱い)と、送ってよい文書の範囲を事前に確認してください。この連載の簡易版は外部に何も送りません。
権限はまだ入っていません: 現段階では、誰が質問しても全文書が検索対象です。部署ごとの制御は第6回で扱います。

発注者向けメモ:検索の品質を頼むときの確認点

社内ナレッジ検索では、AIの文章より前に「探す部分」の品質が、回答の正しさを左右します。探す部分が外れると、どんなに優れたAIでも、間違った資料をもとに答えてしまいます。

☐ 検索の良し悪しを測るため、自社の実際の質問(20〜50件程度)と、その正解になる文書を用意した
☐ 使う埋め込みモデルの名前・提供元・費用の考え方を聞いた
☐ 文書が外部のサービスに送られるか、送られる場合の契約上の扱いを確認した
☐ 「型番や規程番号のような厳密な一致」が必要な質問の割合を把握した(キーワード検索との併用が必要な場合があります)
☐ 検索が外れたときに、利用者が気づける表示(出典の提示など)があるか確認した

開発会社への質問例:

「検索の精度は、どうやって確かめますか。私たちの実際の質問で試す機会はありますか」
「埋め込みモデルは何を使い、後から別のモデルに替えるとき、作業と費用はどのくらいかかりますか」
「型番や社内の略語など、言葉の一致が大事な質問には、どう対応しますか」
「文書を外部のAIサービスに送る場合、送信範囲と保存の扱いを、契約書のどこで確認できますか」

工数や費用が増えやすいのは、専門用語や社内の略語が多い文書、表や図が多く文章にしにくい文書、そして日本語以外の文書も混在する場合です。検索の質は、実際の文書と質問で試さないと分かりません。見積もりの段階で「試験用の質問集を一緒に作る工程」が含まれているかを確認すると、後の手戻りを減らせます。

まとめと次回予告

第3回では、文章を数値の並びにする埋め込みの部品を差し替え可能な形で作り、pgvectorで質問に近い断片を近い順に探す検索を実装しました。簡易版でも検索の流れ全体は動きますが、言い換えに弱いこと、関係のない質問にも上位が返ることも、実行結果でお見せしました。

次回の第4回では、見つけた断片を根拠にして、生成AIに回答させます。出典つきの回答を返す仕組みを作ります(タイトル案:「見つけた文書を根拠に回答させる」)。

この連載の記事一覧

この記事は連載「FastAPIで作る社内ナレッジ検索(RAG)」の1回です。連載のほかの回は次のとおりです（連載の一覧ページ）。

システム発注の相談をする

よかったらシェアしてね！

URLをコピーしました！

URLをコピーしました！

この記事を書いた人

朝野貴朗 - TOMONO TAKAAKI

株式会社THIRD HERO代表取締役　朝野貴朗
Webシステム開発を中心に、toC向けサービスサイトの運営、ツール開発などを行ってまいりました。

問い合わせ

【FastAPIで作る社内ナレッジ検索(RAG) 第3回】質問に近い社内文書を探す(ベクトル検索)を実装する

社内文書のベクトル検索とは:キーワード検索との違い

この回で作るもの

実装:埋め込みの部品を差し替え可能な形で作る

埋め込みのインターフェースと簡易版(app/search/embedder.py)

表に埋め込みの列と索引を足す(db/migrations/002_embedding.sql)

断片と一緒に埋め込みを保存し、近い順に検索する(app/db/store.py)

検索API(app/main.py)

動作確認の方法

実行結果(筆者の確認)

簡易版の限界:関係のない質問にも「一番近い断片」が返る

筆者が確認できた範囲・できていない範囲

自動テスト

つまずきやすい点・セキュリティ上の注意

発注者向けメモ:検索の品質を頼むときの確認点

まとめと次回予告

この連載の記事一覧

システム制作・運用・保守のお問い合わせはこちら

この記事を書いた人

コメント

コメント一覧（2件）

問い合わせ

【FastAPIで作る社内ナレッジ検索(RAG) 第3回】質問に近い社内文書を探す(ベクトル検索)を実装する

社内文書のベクトル検索とは:キーワード検索との違い

この回で作るもの

実装:埋め込みの部品を差し替え可能な形で作る

埋め込みのインターフェースと簡易版(app/search/embedder.py)

表に埋め込みの列と索引を足す(db/migrations/002_embedding.sql)

断片と一緒に埋め込みを保存し、近い順に検索する(app/db/store.py)

検索API(app/main.py)

動作確認の方法

実行結果(筆者の確認)

簡易版の限界:関係のない質問にも「一番近い断片」が返る

筆者が確認できた範囲・できていない範囲

自動テスト

つまずきやすい点・セキュリティ上の注意

発注者向けメモ:検索の品質を頼むときの確認点

まとめと次回予告

この連載の記事一覧

システム制作・運用・保守のお問い合わせはこちら

この記事を書いた人

関連記事

コメント

コメント一覧 （2件）

コメント一覧（2件）