0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

キーワード検索で足りるか、ベクトル検索を足すか:同じ質問で検索漏れを比べる

0
Posted at

検索方式の比較を表す抽象的な幾何学アイキャッチ

2026-10-17時点の公開仕様と、本稿用にAI支援で準備・実行した架空データのローカル検証を整理します。ナレッジ検索を選ぶエンジニア・テックリード向けに、キーワード検索を改善するか、ベクトル検索を追加検証するかを判断します。実業務の成果や本番運用の報告ではありません。

結論:検索漏れの原因を分けてから方式を足す

用語・型番で探す質問はキーワード検索を基準にし、表現違いの取りこぼしが残るなら、同じ質問でベクトル検索を追加評価するのが出発点です。今回の架空fixtureでは追加の改善が得られず、ベクトル追加は保留とします。本番の採用判断は未実施です。

「文書に答えがない」「古い版を参照した」「閲覧権限がない」は、検索方式の追加だけでは解決できません。まず正解の根拠文書を確定し、検索対象に存在するのに上位へ出ないケースを切り出します。

ここでいうベクトル検索は、埋め込みモデルで文書と質問をベクトル化し、類似度で順位を付ける方式です。全文検索のBM25スコアとは尺度が異なるため、その数値を直接足して比較しません。Elasticの公式資料では、全文検索とベクトル検索の順位を統合するRRFが紹介されています。本稿ではまず2方式を単独で比較し、統合方式の採否は次の検証へ残します。

前提と選択肢

扱うのは「短いFAQから根拠の候補を探す」段階です。回答生成や自動送信は行いません。

選択肢 選ぶ条件 確認・維持するもの
FAQ目次・人による案内を継続 件数が少なく、人が根拠へ到達できる 案内時間、検索失敗の記録
既存製品の検索を使う 権限・更新反映が既存基盤で満たせる 実データでの検索品質、契約範囲、移行性
キーワード検索を改善 型番・用語が明確、同義語を管理できる トークナイズ、辞書、文書タイトル、順位
ベクトル検索を追加検証 文書と質問で表現がずれる モデル、再埋め込み、検索負荷、更新管理
ハイブリッドを検証 異なる失敗を補完できる 候補数、順位統合、回帰テスト、保守責任

これらは設計上の選択肢です。既製品の比較試用、同義語辞書、ハイブリッドの性能は今回測定していません。

同じ文書・質問で比べる

文書8件、質問13件を実行前に固定しました。架空の製品・プラン(Atlas B10/B100、Nimbus N20/N200)の制限、復元、認証などを扱う英語の診断用fixtureです。日本語の分かち書き・多言語性能は評価していません。

  • 環境:Python 3.11.8、SQLite 3.51.0、Ollama 0.20.2。
  • ベクトル:既存ローカルの nomic-embed-text:latest、768次元。モデルdigestはコードに固定し、一致しなければ停止します。latestという名前だけで同一モデルとみなしません。
  • キーワード:FTS5 unicode61、質問の英数字トークンをOR検索し、BM25で上位3件。
  • ベクトル:文書8件すべてとコサイン類似度を計算し、上位3件。近似最近傍探索は使いません。
  • 文書・質問・指定参照先ID・k=3を共通化。IDフィルタ、しきい値、LLM生成は使いません。

SQLite公式仕様に合わせ、BM25は小さい値ほど上位に並べています。ベクトル側はNomicのモデルカードの検索用接頭辞に合わせ、文書に search_document:、質問に search_query: を付けます。Ollamaの埋め込みAPIをloopbackで呼び、件数・次元・有限値を確認して正規化しています。

指標のHit@kは「事前に指定した参照先IDが上位k件に入った質問数/対象質問数」です。全質問の正しい回答率ではありません。1問1IDの診断ラベルで、別の文書にも同じ根拠がある場合を網羅していません。例えばQ9の指定先はD3ですが、D4にもB10がSAML非対応という記述があります。

正解なし3問は分母を分け、候補が返るかだけを観測します。無関係な候補を回答に採用した回数や、回答拒否の品質は測っていません。

再現コード

以下は本稿の実行に使ったコードです。Python標準ライブラリ、FTS5対応SQLite、稼働済みのローカルOllama、取得済みの同じモデルが前提です。モデル取得・インストールを自動実行せず、接続先はloopbackへ固定しています。

架空FAQ・質問・評価コードを表示
#!/usr/bin/env python3
"""Synthetic English FAQ diagnostic; stdlib + an already cached local model.

No installs, model pulls, cloud requests, ID filters, score thresholds, or LLM
answers. Fixtures and gold labels are fixed before retrieval. This is not a
Japanese-quality or production-scale benchmark. Run with Python 3.9+.
"""
import argparse
import datetime
import hashlib
import http.client
import json
import math
from pathlib import Path
import platform
import re
import sqlite3

MODEL = "nomic-embed-text:latest"
DIGEST = "0a109f422b47e3a30ba2b10eca18548e944e8a23073ee3f3e947efcf3c45e59f"
K = 3

# Fictional fixtures: do not revise after seeing retrieval scores.
DOCS = [
    {"id": "D1", "text": "Atlas B10 API rate limit is 60 requests per minute. HTTP 429 means the limit was exceeded. Wait for the Retry-After header before sending another request."},
    {"id": "D2", "text": "Atlas B100 API rate limit is 600 requests per minute. HTTP 429 means the limit was exceeded. Wait for the Retry-After header before sending another request."},
    {"id": "D3", "text": "Atlas B10 does not support SAML single sign-on. Users sign in with a password and multi-factor authentication. SAML requires the Atlas B100 plan."},
    {"id": "D4", "text": "Atlas B100 supports SAML single sign-on through a company identity provider. A workspace administrator must configure the SAML connection. Atlas B10 does not include this feature."},
    {"id": "D5", "text": "Nimbus N20 retains deleted files in the recycle bin for 7 days. Open the recycle bin and select Restore to recover a deleted file before the retention period ends."},
    {"id": "D6", "text": "Nimbus N200 retains deleted files in the recycle bin for 30 days. Open the recycle bin and select Restore to recover a deleted file before the retention period ends."},
    {"id": "D7", "text": "Atlas invoice PDF downloads are available in Finance, then Invoices. Only workspace owners can download billing invoices. The billing contact email does not grant access."},
    {"id": "D8", "text": "Atlas webhook delivery retries HTTP 503 responses with exponential backoff. The event ID stays the same across attempts; receivers must deduplicate by event ID. HTTP 400 responses are not retried."},
]
QUERIES = [
    {"id": "Q1", "category": "exact", "text": "Atlas B10 API rate limit", "relevant": ["D1"]},
    {"id": "Q2", "category": "exact", "text": "Atlas B100 API rate limit", "relevant": ["D2"]},
    {"id": "Q3", "category": "exact", "text": "Nimbus N20 deleted files retention period", "relevant": ["D5"]},
    {"id": "Q4", "category": "exact", "text": "Atlas invoice PDF downloads Finance", "relevant": ["D7"]},
    {"id": "Q5", "category": "paraphrase", "text": "Our Atlas B10 integration is sending requests too quickly. When may it try again?", "relevant": ["D1"]},
    {"id": "Q6", "category": "paraphrase", "text": "On Atlas B100, can staff use the company's identity provider to log in?", "relevant": ["D4"]},
    {"id": "Q7", "category": "paraphrase", "text": "I removed a Nimbus N200 document by mistake yesterday. How do I recover it?", "relevant": ["D6"]},
    {"id": "Q8", "category": "paraphrase", "text": "The Atlas delivery endpoint briefly says service unavailable. How should duplicate notifications be handled?", "relevant": ["D8"]},
    {"id": "Q9", "category": "negation", "text": "Does Atlas B10 support SAML single sign-on?", "relevant": ["D3"]},
    {"id": "Q10", "category": "negation", "text": "Does Atlas retry a webhook after an HTTP 400 response?", "relevant": ["D8"]},
    {"id": "Q11", "category": "no_answer", "text": "Does Atlas B100 offer SCIM directory provisioning?", "relevant": []},
    {"id": "Q12", "category": "no_answer", "text": "What is the Atlas B10 data residency region?", "relevant": []},
    {"id": "Q13", "category": "no_answer", "text": "Jupiter Europa passenger ferry timetable", "relevant": []},
]


def sha256(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False,
                                     separators=(",", ":")).encode()).hexdigest()


def local_api(path, payload=None):
    # A fixed IP + HTTPConnection bypass proxies and never follow redirects.
    connection = http.client.HTTPConnection("127.0.0.1", 11434, timeout=90)
    try:
        connection.request("GET" if payload is None else "POST", path,
                           None if payload is None else json.dumps(payload),
                           {"Content-Type": "application/json"})
        response = connection.getresponse()
        if response.status != 200:
            raise RuntimeError(f"Local API {path}: HTTP {response.status}")
        body = response.read(8_000_001)
        if len(body) > 8_000_000:
            raise ValueError("Local API response exceeds 8 MB")
        return json.loads(body)
    finally:
        connection.close()


def unit_vector(vector, expected_dimension=None):
    if not isinstance(vector, list) or not vector:
        raise ValueError("Embedding must be a nonempty list")
    if expected_dimension is not None and len(vector) != expected_dimension:
        raise ValueError("Embedding dimensions differ")
    if any(type(x) not in (int, float) or not math.isfinite(x) for x in vector):
        raise ValueError("Embedding contains a non-finite or nonnumeric value")
    norm = math.sqrt(math.fsum(x * x for x in vector))
    if not math.isfinite(norm) or norm == 0:
        raise ValueError("Invalid embedding norm")
    return [x / norm for x in vector]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, default=Path(__file__).with_name(
        "2026-10-17-search-comparison-result.json"))
    output = parser.parse_args().output
    if output.exists():
        raise FileExistsError("Result exists; choose a new --output path")

    # These hashes are established before either retrieval method runs.
    hashes = {"corpus_sha256": sha256(DOCS), "queries_sha256": sha256(QUERIES)}
    version = local_api("/api/version")["version"]
    models = local_api("/api/tags")["models"]
    matching = [m for m in models if m["name"] == MODEL]
    if len(matching) != 1 or matching[0]["digest"] != DIGEST:
        raise RuntimeError("Required cached model digest is absent; no pull attempted")
    inputs = ["search_document: " + d["text"] for d in DOCS]
    inputs += ["search_query: " + q["text"] for q in QUERIES]
    raw = local_api("/api/embed", {"model": MODEL, "input": inputs,
                                   "truncate": False, "keep_alive": "5m"})["embeddings"]
    if not isinstance(raw, list) or len(raw) != len(inputs):
        raise ValueError("Unexpected embedding count")
    first = unit_vector(raw[0])
    dimension = len(first)
    vectors = [first] + [unit_vector(v, dimension) for v in raw[1:]]

    db = sqlite3.connect(":memory:")
    db.execute("CREATE VIRTUAL TABLE faq USING fts5(doc_id UNINDEXED, body, tokenize='unicode61')")
    db.executemany("INSERT INTO faq VALUES (?, ?)", [(d["id"], d["text"]) for d in DOCS])
    rows = []
    for index, query in enumerate(QUERIES):
        # OR every English alphanumeric token, including stopwords; no ID filter.
        tokens = sorted(set(re.findall(r"[A-Za-z0-9]+", query["text"].lower())))
        match = " OR ".join('"' + token + '"' for token in tokens)
        lexical = db.execute("SELECT doc_id, bm25(faq) FROM faq WHERE faq MATCH ? "
                             "ORDER BY bm25(faq), doc_id LIMIT ?", (match, K)).fetchall() if tokens else []
        qvector = vectors[len(DOCS) + index]
        dense = [(d["id"], math.fsum(x * y for x, y in zip(qvector, vectors[i])))
                 for i, d in enumerate(DOCS)]
        if any(not math.isfinite(score) for _, score in dense):
            raise ValueError("Non-finite cosine score")
        dense.sort(key=lambda pair: (-pair[1], pair[0]))
        rows.append({**query, "fts5_match": match, "rankings": {
            "bm25": [{"doc_id": doc_id, "score": score} for doc_id, score in lexical],
            "dense": [{"doc_id": doc_id, "score": score} for doc_id, score in dense[:K]],
        }})
    db.close()

    aggregate = {}
    for category in ("exact", "paraphrase", "negation", "answerable_all"):
        subset = [r for r in rows if r["relevant"] and
                  (category == "answerable_all" or r["category"] == category)]
        aggregate[category] = {"n": len(subset)}
        for method in ("bm25", "dense"):
            aggregate[category][method] = {
                f"hit_at_{k}_count": sum(any(hit["doc_id"] in r["relevant"]
                                            for hit in r["rankings"][method][:k])
                                        for r in subset) for k in (1, K)}
    no_answer = [r for r in rows if not r["relevant"]]
    aggregate["no_answer"] = {"n": len(no_answer), "interpretation":
        "Candidate emission only; no score threshold or answer acceptance was tested."}
    for method in ("bm25", "dense"):
        aggregate["no_answer"][method] = {
            "nonempty_candidate_queries": sum(bool(r["rankings"][method]) for r in no_answer),
            "returned_candidate_count": sum(len(r["rankings"][method]) for r in no_answer)}
    result = {
        "status": "completed", "generated_at_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "environment": {"python": platform.python_version(), "sqlite": sqlite3.sqlite_version,
                        "ollama": version, "model": MODEL, "model_digest": DIGEST,
                        "embedding_dimensions": dimension},
        "settings": {"k": K, "lexical": "SQLite FTS5 unicode61 BM25; OR all English alphanumeric tokens; lower score first",
                     "dense": "nomic search_document/search_query prefixes; exhaustive cosine; higher score first",
                     "ties": "doc_id ascending", "id_filter": False, "score_threshold": None,
                     "network": "loopback Ollama only; cached model required"},
        "limitations": ["Fictional English fixtures; not a representative benchmark",
                        "No Japanese tokenization or quality claim", "No ANN, generation, or answer acceptance tested",
                        "No latency comparison; fixture labels have no independent human review"],
        **hashes, "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "documents": DOCS, "queries": rows, "aggregate": aggregate,
    }
    with output.open("x", encoding="utf-8") as handle:
        json.dump(result, handle, ensure_ascii=False, indent=2, allow_nan=False)
        handle.write("\n")
    print(json.dumps({"output": str(output), **hashes, "dimensions": dimension,
                      "aggregate": aggregate}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

実測結果

質問区分 BM25 Hit@1 ベクトル Hit@1 BM25 Hit@3 ベクトル Hit@3
用語・ID 4/4 2/4 4/4 4/4
表現違い 4/4 2/4 4/4 4/4
否定・非対応 2/2 2/2 2/2 2/2
指定参照先あり・合計 10/10 6/10 10/10 10/10

このfixtureではBM25が指定参照先10/10を1位に出し、ベクトルは6/10でした。上位3件なら両方式10/10です。今回の範囲では、ベクトル検索を追加する品質上の根拠は得られていません。 実際の対象文書での評価へ進むまでは、既存のキーワード検索を維持する判断になります。

表現違いの4問にも、製品名や文書と共通する語が残っています。「語彙が重ならない質問」への一般化はできません。少数の意図的なfixtureであり、質問分布や正解ラベルの業務担当者レビューも未実施です。

追跡する例 指定参照先 BM25上位 ベクトル上位
Q1: Atlas B10 API rate limit D1 D1,D2,D3 D1,D2,D8
Q2: Atlas B100 API rate limit D2 D2,D1,D3 D1,D2,D8
Q3: Nimbus N20 deleted files retention period D5 D5,D6 D6,D5,D8
Q7: I removed a Nimbus N200 document by mistake yesterday. How do I recover it? D6 D6,D5,D8 D5,D6,D3

Q1とQ2は型番だけを変えています。保存結果ではベクトルの上位3件とスコアが同じになり、B100の質問にもB10のD1が1位でした。内部の原因は調査していません。「このモデルなら型番差を保持する」と仮定せず、重要な識別子を別の条件として扱う必要がある、という観測です。

正解なし3問で候補が返ったのはBM25が2/3問(計6候補)、ベクトルが3/3問(計9候補)です。ベクトル側はしきい値なしで必ず3件返す実装なので、この差だけで品質の勝敗を決められません。検索結果が存在することと、回答できる根拠があることを分けます。

コードを search_comparison.py として保存し、未作成の結果ファイルを指定して実行します。接続失敗・モデル不一致・不正な埋め込みは例外で停止し、成功した場合にだけ結果JSONを新規保存します。既存結果は上書きしません。

python3 search_comparison.py --output search-result.json

文書SHA-256:91270de30b16e55850674ee8b47669aa95e65db1ecd3b93952307ebaaff4fbd3。質問・指定参照先SHA-256:b28719a034ce7c320e43eb68a73a6cf6a829e9536240d798e7566f3a1b07f104。結果JSONにも環境、設定、全問の順位とスコアを保存します。

結果から何を決め、何を保留するか

比較対象は今回の簡単なBM25設定と、特定の埋め込みモデルだけです。 キーワード検索全般と意味検索全般の優劣を測ったものではありません。BM25側のOR検索は一般語でも候補を返し、ストップワード除去や同義語展開も実装していません。改善したキーワード検索と比べなければ、方式追加の必要性は判断できません。

次の段階では、実際の対象言語・文書形式で評価セットを作り直し、開発用と未使用の確認用へ分けます。辞書・モデル・しきい値を調整した質問だけを再評価しても、新しい質問への強さは分かりません。

観測した失敗 先に試すこと 方式追加の条件
固有IDがあるのに別の文書が上位 IDを構造化フィールドとして扱い、必要なら完全一致で絞る IDの意味を理解したうえで、自然文部分の改善が必要
同義語・言い換えで見つからない 文書の見出し・用語整備、管理可能な同義語展開 未使用質問でもベクトル検索の改善が残る
答えのない質問にも候補が出る 候補と回答根拠を分け、根拠不足なら保留 正解なしの独立した評価でも誤採用を抑えられる
根拠文書が欠落・古い 文書所有者が内容・版を整備 検索対象の品質を満たしてから再比較

完全一致のIDも、本文をOR検索するだけでは強制条件になりません。反対に、ベクトルの上位1件というだけで型番や否定条件の一致は保証できません。条件が必須なら、検索の前後で明示的に検証します。

費用・品質・運用の責任を増やせるか

費用はAPI代だけで比べません。次の式は測定項目を整理するための設計案で、価格・削減効果の試算値は置いていません。

追加費用 = 初回/更新時の埋め込み + 質問の埋め込み + 索引の計算・保存 + 評価・辞書・モデル変更への対応

ローカル推論でも計算資源と保守は必要です。今回の少数文書の全件比較では、大規模索引・近似検索の品質、p95遅延、月額費用、業務工数の削減は分かりません。

担当 決める・記録すること
文書所有者 正しい根拠、適用範囲、更新日、廃止する版
開発担当 文書/質問の前処理、モデル識別子、索引版、権限フィルタ
評価担当 質問別の正解、検索漏れの分類、未使用質問での回帰
運用担当 失敗率・遅延、旧索引との照合、停止と切戻し

ログは質問の全文を無条件に残さず、許可された評価ID、検索方式・索引版、候補ID、保留理由、処理時間を中心にします。閲覧権限を満たさない候補は、LLMや利用者へ渡す前に除外する設計が必要です。本稿では権限制御の実装・検証は行っていません。

埋め込み失敗時に空のベクトルで検索を続けたり、索引の更新途中で新旧モデルを混ぜたりしないようにします。索引を版単位で用意し、確認後に切り替える方法を次の検証候補とします。切戻し先は既存検索または人の案内であり、未検証の回答を自動送信する経路ではありません。

条件付きの推奨と次の確認

キーワード検索の改善で重要な質問を満たせるなら、方式追加を見送ります。表現違いが残り、未使用の質問でも改善し、固有ID・正解なし・権限・遅延・保守の条件を満たすなら、ベクトル検索またはハイブリッドのPoCを進めます。基準を満たさなければ現状維持や人による案内を続けます。

次に行うのは、担当者が「取りこぼすと困る質問」「答えてはいけない質問」を定義し、開発担当が既存検索の改善版と比較することです。合否基準は実際の誤りの影響と運用制約から先に決めます。今回の架空データのスコアを、そのまま導入基準に流用しません。

参考資料

いずれも2026-10-17確認。仕様の記述と、本稿のローカル結果・設計案は区別しています。

この記事を書いた人✏️@YushiYamamoto
ITPRODX.com代表。AI × Web × 業務自動化の開発・検証に取り組んでいます。
技術選定・品質・運用の判断材料を、検証範囲と制約を示して発信しています。
AI導入や業務自動化の技術選定・設計整理については、対応できる範囲を確認のうえご相談を承ります。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?