社内のSQL Serverに溜まった受発注データや顧客マスタを、生成AIで検索・要約できないかと考える現場は増えています。クラウドAPIに業務データを投げるのが社内規定で難しく、オンプレミス完結のRAG（検索拡張生成、社内データを検索してからAIに回答させる仕組み）を検討している情報システム担当やデータエンジニアに向けて、設計段階でつまずきやすい点を整理しました。
小規模なPDF要約のチュートリアルはネット上に数多くありますが、実務で扱うのは整形済みの文書ではなく、正規化された何十テーブルものリレーショナルデータです。ここで同じ発想を持ち込むと、思わぬところで事故が起きます。
何が起きるか：抽出フェーズでのメモリ枯渇とコスト膨張
最初につまずくのは、本番SQL Serverから一括でデータを抜き出す処理です。SELECT * FROM 取引明細のような単純なクエリを数百万行のテーブルに投げると、アプリケーション側のメモリを一気に使い切ります。
結果として本番DBへの接続がタイムアウトしたり、夜間バッチと競合してロックが発生したりします。小規模な検証環境では問題が出ず、本番データ量で初めて顕在化するのが厄介なところです。
もう一つの事象は、クラウドのLLM（大規模言語モデル）APIに社内データを流し続けることで発生するコストの揺れです。為替レートに連動したドル建て課金は、為替が動くたびに予算実績がぶれます。加えて、越境データ移転に関する社内規定や、業界によっては個人情報保護関連の法令チェックが後追いで入り、プロジェクトが差し戻しになるケースもあります。
なぜ起きるか：原因を段階的に分解する
原因1：ドキュメントRAGとテーブルRAGの設計思想の違い
PDFや議事録を対象にしたRAGは、文章を一定の長さで区切る「チャンキング」という処理さえ押さえれば動きます。一方、リレーショナルデータは行同士に外部キーの関連があり、単純に1行ずつ文章化すると文脈が切れます。
たとえば受注テーブルの1行を単独でベクトル化しても、顧客テーブルや商品テーブルとの結合情報がなければ「どの顧客の、どの商品の注文か」という意味が欠落します。ここを設計段階で見落とすと、検索結果は出るのに回答の精度が上がらない状態に陥ります。
原因2：ストリーミング抽出を使っていない
SQLAlchemy（Pythonの代表的なORMライブラリ）でDBに接続する際、デフォルトの挙動では結果セットを一度に全件メモリへロードします。サーバーサイドカーソルによるストリーミング取得を明示的に有効化しないと、大規模テーブルで即座にメモリを食い潰します。
原因3：チャンキングが文字数だけで区切られている
単純な文字数ベースの分割は、単語や数値の途中で文章を切ってしまうことがあります。業務データには金額や品番など、途中で切れると意味が壊れる数値列が多く含まれるため、単語境界を考慮したスライディングウィンドウ方式の分割が必要になります。
原因4：差分更新の仕組みがない
ベクトルデータベース（文章の意味的な近さで検索できるDB、ここではChromaDBを想定）への同期を、毎回全件洗い替えで行う設計も見かけます。データ量が増えるほど同期時間が伸び、夜間バッチの時間枠に収まらなくなります。
自分のプロジェクトが該当するかの確認方法
該当するかどうかは、次の観点で確認できます。
- 抽出対象のテーブルが100万行を超えるか：
SELECT COUNT(*) FROM 対象テーブルで行数を確認する - 接続コードで
fetchall()を使っていないか：Pythonであればソース中にfetchallの文字列検索をかける - SQLAlchemyの接続URLに
stream_results=Trueやサーバーサイドカーソルの指定があるか確認する - 現在のRAGパイプラインが生成AI提供元のクラウドAPIエンドポイントにデータを送信しているか、ネットワーク設定やAPIキーの発行元を確認する
- ベクトルDBへの同期処理が全件再構築か差分更新かを、同期スクリプトのロジックで確認する（更新対象の判定にハッシュ値を使っているかがポイント）
一つでも当てはまれば、本番投入前に設計を見直す価値があります。
対策の手順
手順1：抽出層をストリーミング対応に変える
SQLAlchemyの接続設定でサーバーサイドカーソルを有効にし、yield_per()などで一定件数ごとに取得する実装に変更します。これだけでメモリ使用量を一定に保てます。
from sqlalchemy import create_engine
engine = create_engine(
"mssql+pyodbc://user:pass@dsn",
connect_args={"stream_results": True}
)
with engine.connect() as conn:
result = conn.execution_options(stream_results=True).execute(query)
for row in result.yield_per(1000):
process(row)手順2：ステージング環境でスキーマを整形する
本番テーブルを直接ベクトル化せず、正規表現などで不要な制御文字やNULLを除去し、結合済みのフラットなビューを別途用意します。ここで外部キー関係を人間が読める文章に変換しておくと、後工程の検索精度が安定します。
手順3：単語境界を考慮したチャンキングに切り替える
文字数で機械的に分割するのではなく、オーバーラップを持たせたスライディングウィンドウ方式を採用します。金額や品番などの数値列は、分割境界をまたがないよう前処理で保護します。
手順4：差分同期の仕組みを入れる
各行のハッシュ値（MD5など）を保持しておき、前回同期時のハッシュと比較して変更があった行だけをベクトルDBに反映します。これにより、同期処理の実行時間をデータ増加量に対して線形ではなく「変更量」に対して線形にできます。
手順5：ローカルLLM実行層でアクセス制御をかける
部署や役職ごとに参照できるデータを制限する場合、ベクトル検索結果にメタデータでロールフィルタをかけます。あわせてOllamaなどのローカルLLM実行環境を使えば、推論処理自体も社外に出さずに完結させられます。
導入前に確認すること
社内DBをRAG化する際は、対象テーブルの行数とストリーミング抽出の有無をまず確認します。
チャンキングは文字数だけでなく単語境界を意識し、数値データが途中で分断されないか点検します。
ベクトルDBへの同期は全件洗い替えか差分更新かを確認し、データ量が増えても運用に耐えるかを見極めます。
クラウドAPIへのデータ送信有無と、社内の情報管理規程との整合性も、設計の初期段階でチェックリストに加えておくと手戻りを防げます。