【ベクトル検索アプリ開発④】実装編-ベクトルストア作成-

目次（クリックするとジャンプします）

この記事でわかること

admin_app 側の実装フロー（Excel→チャンク→埋め込み→FAISS→メタ保存）
各ファイル（build_index.py / reader.py / vectorizer.py / config.py）の役割
設定項目の見どころと注意点
実行コマンドと典型的なエラー対処

全体像（4ステップ）

admin_app/build_index.py の main() が、以下の 4 ステップでインデックスを構築します。

Excel 読み込み＆チャンク化（reader.py）
埋め込み生成（vectorizer.py）
FAISS インデックス作成（内積：IndexFlatIP）
メタデータ保存（meta.pkl）

これらの入出力パス・列名・チャンク長などは config_admin.yaml から読み込みます（admin_app/config.py が CLI 引数の既定をこのファイルに固定）。

設定の読み込み（`admin_app/config.py`）

from __future__ import annotations
from functools import partial
from scripts.utils import load_config_from_cli as _load_config_from_cli

load_config_from_cli = partial(_load_config_from_cli, "config_admin.yaml")

ポイント

load_config_from_cli() は --config 引数が未指定なら config_admin.yaml を読むよう、既定値を束縛。
実行時に --config を変えれば別 YAML を使うことも可能。

ステップ1：Excel 読み込み＆チャンク化（`admin_app/reader.py`）

役割

Excel から行単位でデータを取得
指定列（fields.text）の長文を オーバーラップ付きでチャンク分割
チャンクごとに 元行ID・メタデータ・出自テキスト列名 を付与して records を生成
さらに ファセット候補（メタ列のユニークリスト）を集計

抜粋：チャンク分割 chunk_text()（省略なし）

def chunk_text(text: str, max_chars: int, stride: int) -> List[str]:
    # シンプルな文字数ベースのスライディングウィンドウ
    text = text or ""
    n = len(text)
    if n == 0:
        return []
    if n <= max_chars:
        return [text]
    chunks: List[str] = []
    start = 0
    while start < n:
        end = min(n, start + max_chars)
        chunks.append(text[start:end])
        if end == n:
            break
        start = start + (max_chars - stride)
        if start <= 0:  # 念のため
            start = end
    return chunks

ポイント

max_chars 文字で切り出し、stride 文字ぶんオーバーラップさせて次のチャンクへ進む。
章や段落をまたぐ文脈を適度に保持でき、検索精度の底上げに寄与。

抜粋：Excel → records 生成（中核部のみ）

def load_excel_records(cfg: Dict[str, Any]) -> Dict[str, Any]:
    # ... 省略（Excel読込や列名取得の前段処理） ...
    records: List[Dict[str, Any]] = []
    for i, row in df.iterrows():
        rid = row_ids[i]
        meta = {col: _to_str(row.get(col, ""), na_fill) for col in metadata_cols}
        # ... 省略 ...
        for text_col in text_cols:
            txt = _to_str(row.get(text_col, ""), na_fill)
            chunks = chunk_text(txt, max_chars=chunk_cfg["max_chars"], stride=chunk_cfg["stride"])
            # 空チャンクは除外
            chunks = [c for c in chunks if c.strip()]
            for c in chunks:
                records.append({
                    "row_id": rid,
                    "metadata": meta,
                    "text_col": text_col,
                    "chunk": c
                })
            # ... 省略 ...
    # ファセット候補収集
    facet_values = {}
    for col in metadata_cols:
        facet_values[col] = sorted(set([r["metadata"].get(col, "") for r in records]))

    return {
        "records": records,
        "metadata_cols": metadata_cols,
        "text_cols": text_cols,
        "facet_values": facet_values
    }

ポイント