日: 2026年9月9日

第148回　CSV逐次読み込み→断片化→モック埋め込み→SQLite保存の最小構成

by

CSV逐次読み込み→断片化→モック埋め込み→SQLite保存の最小構成

注意（必読）

この記事内で説明する埋め込みクライアントは配線確認用のモック実装です。実API接続、認証、レート制御、堅牢な再試行・監視などの実務要件は実装していません。本文は「CSVを逐次読み込み→断片化→埋め込み（モック）→SQLiteへ保存→小規模検索へ渡す」ための最小構成を分かりやすく説明することを目的とします。示す設計説明や検証フローは実装ガイドとして有用ですが、示した実装例がそのまま本番で動くとは受け取らないでください。

目的と範囲

大きなCSVを一括でメモリに載せず逐次処理し、各行（あるいは行から生成した断片）単位で埋め込みを作ってSQLiteに保存する。保存したベクトルはベクトル配列とメタ情報（row_id, fragment_id, model）を同じ順序で読み出し、小規模検索（Faiss等）へ渡せるようにする、という最小構成を示します。主キーや保存形式、テストでの検証ポイントを明確にします。

前提・依存

  • 想定するライブラリ（説明用）: Python標準の csv/json/sqlite3、numpy（float32変換）、および任意で faiss。
  • 保存ファイルは1モジュール（例: pipeline.py）としてまとめる前提で説明します。
  • 埋め込みクライアントはモック。実APIに差し替える場合は認証・レート制御・再試行設計を追加してください。

処理の順序（概観）

  1. CSVをストリーミングで1行ずつ読み込む（メモリに全件を保持しない）。
  2. 各行を断片（fragment）に変換する。断片は少なくとも以下を含む: row_id, fragment_id, text, metadata。
  3. 断片をバッチ化（クライアントの batch_size に応じる）して埋め込みAPI（ここではモック）に投げる。
  4. 返却されたベクトルを float32 のバイナリに変換し、metadata（JSON）と共に SQLite に保存する。主キーは複合 (row_id, fragment_id, model) を採用する。
  5. 必要に応じて、SQLite からベクトルを ORDER BY row_id, fragment_id の順に読み出し、NumPy 配列（shape: N×D）と対応するレコード配列を得る。これをインデクシング（Faiss 等）や線形探索に渡す。

各段階で保存・保持する情報（明確化）

  • 断片変換段階: row_id（必須）、fragment_id（ユニーク化）、text（埋め込み対象）、metadata（title 等の補助情報）。
  • 保存段階（SQLite）: row_id, fragment_id, model を複合主キーとして、vector（float32 のバイナリ）、metadata（JSON文字列）、タイムスタンプを格納する。
  • 読み出し段階: vector は NumPy の float32 配列に復元し、records には row_id/fragment_id/metadata を同じ順序で保持する（位置対応の保証）。

主要な関数・クラス（役割と相互関係）

ここではコード本文は掲載しませんが、実装時に最低限用意するべき関数とその役割を示します。すべて1ファイルにまとめる想定です。

  • stream_csv_rows(path): CSV を逐次的に読み、1行ずつ辞書で返すイテレータ。
  • row_to_fragment(row): CSV の行を断片に変換。id 列の存在を検証し、row_id, fragment_id, text, metadata を返す。
  • chunked(iterable, size): イテラブルを指定サイズで区切るユーティリティ。
  • EmbeddingClient クラス: モック埋め込みクライアント。embed_batch(texts) と embed_batch_with_retry(texts) を提供し、バッチ単位でベクトル（例: 128次元のランダム値）を返す。
  • Create_schema(conn): SQLite にテーブルを作成する。主キーは (row_id, fragment_id, model)。
  • save_embedding(conn, row_id, fragment_id, model, vector, metadata): ベクトルを float32 のバイナリに変換して保存。重複は上書き（INSERT OR REPLACE 相当）する運用を想定。
  • load_embeddings(conn, model): 指定モデルの行を ORDER BY row_id, fragment_id で取得し、(vectors: NumPy 配列, records: メタ情報リスト) を返す。np.frombuffer の場合は安全のためコピーする。
  • embed_csv(path, conn, client, model): 上記を組み合わせる本体。CSV をストリームし、断片化→バッチ埋め込み→保存→コミットを繰り返す。
  • search_sqlite(…): 小規模検証用の検索。クエリを埋め込み、Faiss が使える環境なら IndexFlatL2 を作成して検索、未導入なら単純な線形探索で距離を計算する。

pytest による最小検証の流れ（コードは掲載しません）

実際のテストでは以下の手順でパイプラインの基本動作を検証します。テストは pipeline.py を同一ディレクトリに保存する前提です。

  1. 一時ディレクトリに小さな CSV を作る（例: id,title,body のヘッダと 1 行）。※シェルの printf 等で作成可能だが、ここではコマンドは載せません。
  2. メモリ上の SQLite を開き、Create_schema を実行する。
  3. EmbeddingClient インスタンスの embed_batch をモックして、既知のベクトル（例: [0.1]*128）を返すようにする。
  4. embed_csv を呼び出す。呼び出し後、モックの embed_batch が想定されるテキスト（例: ‘Hello\nThis is a test’ のように title と body を結合した文字列）で呼ばれたことを assert する。
  5. SQLite の embeddings テーブルを読み、保存された行の row_id、fragment_id、model が期待値と一致すること、vector BLOB のバイト長が float32 の次元数×4 バイト（例: 128*4）であることを検証する。

主なトラブルシューティング項目（具体的に確認する点）

  • CSV に id 列は存在するか。欠落すると row_to_fragment で失敗する。
  • embed_csv 実行後、バッチごとのコミットが行われているか（途中失敗時の部分保存の有無を設計に合わせる）。
  • 埋め込みクライアントが返すベクトル数がバッチ内の断片数と一致するか。ミスマッチは ValueError 相当で検出する。
  • vector を float32 で保存しているか。テストでは BLOB 長=次元×4 バイトを使って検証するのが簡便。
  • load_embeddings の戻りベクトル配列と records 配列が同じ順序か。ORDER BY を使って読み出すこと、np.frombuffer の場合は copy を取ることを確認する。
  • 検索時に次元不一致がないか。クエリ埋め込みの次元と保存済みベクトルの次元は一致させる必要がある。

Faiss は任意の次段階

小規模であれば SQLite に読み出したベクトルを NumPy で線形探索するだけで十分です。高速化したい場合は Faiss を導入して IndexFlatL2（あるいは他のインデックス）を作り、NumPy 配列を一括で追加して検索する。Faiss は環境依存の導入手順が必要なため、ここでは任意の次段階として簡潔に扱います。

実装時の注意（まとめ）

  • 本文ではプレーンな設計とテスト方針を示しましたが、コード本文は掲載していません。実装例をそのままコピペして動くと保証するものではありません。
  • 本番移行時には埋め込み API 固有のエラー処理、レート制御、認証管理、監視、テレメトリを必ず追加してください。
  • 小規模検証→段階的スケールアップの手順を作り、各段階で性能・一貫性・復旧を確認してください。

以上が「CSVを逐次読み込み→断片化→埋め込み（モック）→SQLiteへ保存→小規模検索へ渡す」ための最小構成の説明です。実装時には上に挙げた関数群を1つのモジュールにまとめ、pytest による保存結果の検証（モック差し替えと BLOB 長の確認）を行うと、配線確認と基本品質の担保に有効です。