はじめに：現場でつまずきやすい点に寄り添って
大きなCSVやログを扱うと、いつの間にかメモリが増えてプロセスが落ちる、あるいは埋め込みAPIに一度に大量送信して失敗する――こうした現場あるあるに悩んでいませんか。第148回を読んで埋め込みの概念は理解したが、いざ現場で大規模データを安定運用する段になると「どう設計し、どう復旧するか」が課題になりがちです。本記事では、Pythonのジェネレータ／イテレータを中心に、メモリを一定に保ちながら埋め込み→ベクトルDB登録→簡易検索までつなぐための設計上の手順とチェックリストを提示します。特定の埋め込みAPIやベクトルDBに依存しない、概念と実装方針の解説です。
学習ゴール（この記事を読み終えたとき）
- ストリーミング読み出しとバッチ化でメモリを一定に保つ設計がわかる
- 埋め込みAPI呼び出しの安全なリトライやレート制御の考え方を実務で使える形で理解できる
- 障害時の途中再開（チェックポイント）や簡易メトリクスの実装ポイントが掴める
- 小さなデータでのテスト方法とCIに組み込む観点が分かる
ステップ1：概念とパターン（ジェネレータ／イテレータの実務的理解）
ジェネレータの基本と利点
Pythonのジェネレータは、yieldで値を逐次返す仕組みです。データ全体をメモリに展開せずに「必要な分だけ処理する」ことで、長時間実行のバッチ処理や大規模ログ処理で安定化します。実務では次の点を押さえます。
- 遅延評価：入力を1行ずつ（あるいは小さなチャンク単位で）処理することでメモリ使用量が一定に近づく。
- パイプライン合成：読み出し→前処理→バッチ化→API送信 の各段をジェネレータで繋ぐと、各段の責務が明確になる。
- 単体テストしやすい：入力ストリームをモック化してジェネレータ単体を検証できる。
よく使うパターン
- chunked（固定サイズバッチ）: バッチサイズNごとにまとめる
- sliding window（可変・重複あり）: 文脈ウィンドウが必要なとき
- filter/mapチェーン: 前処理→フィルタ→正規化を逐次的に行う
メモリ使用イメージ比較
| 処理方法 | 主な特徴 | メモリ使用の見積もり |
|---|---|---|
| 一括ロード（pandas.read_csv） | 実装は簡単だが全件をメモリに保持 | データサイズ＋行・列のオーバーヘッド（高） |
| ジェネレータで逐次処理 | 行単位や小チャンク単位で処理。低メモリ。 | バッファサイズ＋バッチサイズに依存（一定） |
| ハイブリッド（分割して複数ファイル） | ファイル分割で並列処理。運用と整合性管理が必要。 | プロセスごとの最大バッファを足し合わせた値 |
ステップ2：実装ガイド（現場で使える設計と小さな実装ヒント）
1) ストリーミング読み出しジェネレータ（CSV/JSONL/ログ）
ポイントは「逐次読取」と「最小限のパース」。簡単な考え方は次の通りです。
- CSV: ファイルを逐行読み、必要な列だけパースしてdictをyieldする。
- JSONL: 各行をjson.loadsしてyield。パース失敗はログに出してスキップ可能。
- 圧縮ファイル: gzip.open や streaming ライブラリで逐次解凍しながら処理。
（実装ヒント）関数名の例:
read_csv_rows(file_path) → yield {‘id’: …, ‘text’: …}
2) 固定サイズ／可変サイズバッチの作り方
固定サイズバッチ（chunked）は最も実用的です。イメージは次の通り。
| 用途 | 実装方針 | 注意点 |
|---|---|---|
| 固定サイズバッチ | ジェネレータからN個ずつ取り出してyield | 最終バッチはサイズ未満になることを許容する |
| 可変バッチ（サイズ上限のみ） | 文字数やトークン数で上限を設定して詰める | トークン推定が必要でやや複雑 |
（実装ヒント）chunkedの疑似手順: イテレータをループして一時リストにappend、サイズ到達でyieldしクリア。
3) 埋め込みAPI呼び出し：同期／非同期の使い分けと安全なリトライ
実務では次の観点で選択します。
- 同期: 単純なワンオフ処理やデバッグ時に便利。失敗時の影響範囲が分かりやすい。
- 非同期（async/await）: 高スループットが必要な場合。イベントループ設計とエラーハンドリングが必要。
リトライ戦略（実務向け）:
- タイムアウト設定を必ず設ける（APIごとに最適値）。
- 指数バックオフ＋最大試行回数。短時間の一時障害を吸収する。
- レート制御: APIのスロットリング（固定インターバル）かトークンバケット方式を採用。
- 失敗時の扱い: 一時的失敗はバッチ単位で再試行、恒久的エラーはログと失敗キューに記録。
（実装ヒント）簡易流れ: chunkedジェネレータ → API送信（タイムアウト＋リトライ）→ 成功した埋め込みは次段へyield
ステップ3：結合と耐障害設計
埋め込み失敗時の戦略
- 短時間の失敗: バッチ単位で再試行（指数バックオフ）。再試行回数はビジネス要件で決める。
- 恒久エラー（フォーマット不備等）: そのレコードをスキップして失敗ログに保管。人手確認用のCSV/JSONとして残す。
- 失敗キュー: 再処理用ファイルや小さなDBに失敗レコードを貯め、別ジョブで再投入する。
途中再開（チェックポイント）の実装案
長時間ジョブでは途中から再開できることが重要です。シンプルな手法:
- 進捗マーカー（例: 最終処理行番号、最終ID）を定期的にファイルに書く。
- 各バッチの登録完了後にチェックポイントを更新。チェックポイントは冪等に保つ（同じIDで再登録しない工夫）。
- ベクトルDB側にアップサート（upsert）やidempotentなID設計をすることで重複登録を避ける。
ロギングと簡易メトリクス
| 指標 | 何を見るか | 取得方法 |
|---|---|---|
| 処理レート | records/s や batches/min | 処理した件数を時間で割る（ジョブ内でカウント） |
| メモリ使用 | プロセスのRSSやコンテナメモリ | psutilやコンテナ監視で定期取得 |
| 失敗率 | 総バッチ数に対する失敗バッチ数 | ログから集計、アラート閾値を設定 |
ステップ4：ベクトルDB登録と簡易検索
少量インサート vs バルクインサート
- 少量インサート（逐次登録）: レイテンシは低めだが総オーバーヘッドが大きくスループットが落ちる。
- バルクインサート: スループット重視。失敗時のロールバック/再試行戦略が重要。
ID管理: アプリ側で一意のIDを発行しておくと、途中再開や重複検出が容易になります（例: ハッシュ＋元ファイルの行番号）。
簡易検索とスコア調整
- 検索テスト: 少量データでレイテンシと順位を確認（kとスコア閾値の調整）。
- スコアリングの観察点: 同一ドメインの類似度分布を確認し、閾値を決める。
テストとCIのポイント
- ジェネレータ単体テスト: ファイルストリームをモックして逐次出力を検証する。
- 埋め込みAPIはモック化して遅延・エラーをシミュレートしたテストを用意する。
- メモリ回帰テスト: 小データと大データでメモリ使用を比較し、閾値超過の検出をCIに組み込む。
- E2Eテスト: 小さなサンプル（例: 100件）で読み出し→埋め込み→登録→検索までの流れを自動化する。
運用の現実的注意点と伸ばしどころ
- Dockerでのメモリ制限: コンテナ側のメモリ上限を設定し、OOM発生時の挙動を確認する。
- 長時間ジョブのタイムアウト: ジョブ管理（AirflowやKubernetes CronJobなど）で最大実行時間を設定。
- 監視指標: 処理レート、メモリ、失敗率、最終正常実行時刻を継続監視する。
- 将来的な拡張: RAGやキャッシュ戦略との連携、並列処理によるスケールアウトの設計を検討する。
実務チェックリスト
| 項目 | 確認ポイント |
|---|---|
| 入力ストリーム | 逐次読み出しでメモリ固定化されているか。パースエラーはログ化されるか。 |
| バッチ設計 | バッチサイズとレート制御が現場負荷に合っているか。 |
| リトライ/バックオフ | タイムアウト・最大試行回数・バックオフが設定されているか。 |
| チェックポイント | 定期的に進捗を保存し、再開時に重複登録を防げるか。 |
| ベクトルDB整合性 | ID管理とupsert戦略があるか。 |
| テスト | ジェネレータ単体・APIモック・E2EのテストがCIに含まれているか。 |
| 監視 | 処理レート、メモリ、失敗率をダッシュボードで確認できるか。 |
まとめ（実務でまず手を付ける優先順）
- 1: まずはストリーミング読み出し（ジェネレータ）に置き換え、メモリ問題を排除する。
- 2: 固定サイズバッチで埋め込みを試し、同期→非同期の切替はスループット要件に応じて行う。
- 3: チェックポイントとID管理を実装して途中再開と整合性を確保する。
- 4: ロギングと簡易メトリクスを整備して運用監視を始める。
次回候補として『RAGの品質評価とログ駆動の改善ループ』を想定しています。本稿で挙げたチェックポイントやログは、RAG運用に移行する際の良い出発点になります。ここで示したパターンは過度に特殊化せず、現場の多様なデータ形態に適用できる実務的な基準を目指しました。小さなサンプルでまず動かし、観測→改善を繰り返すことをおすすめします。