Salesforce から外部LLM へのデータパイプライン

このスキルは、チームが Salesforce データを抽出して、Salesforce Einstein プラットフォームの外部で外部パイプラインを経由させ、LLM にフィードするか、自社で所有・運用するベクトルストアに入力する必要がある場合に起動します。Bulk API v2 抽出レイヤー、データが組織の境界を離れる前の PII スクラッビング要件、外部インフラストラクチャ上に構築されたチャンキングおよびエンベッディングパイプライン、非 Data Cloud ベクトルストアのインジェストスキーマをカバーします。これは Salesforce のネイティブな Data Cloud ベクトル検索を使用するのではなく、独自の AI インフラストラクチャを構築または運用することを選択したチームのためのスキルです。

---

開始前に

このドメイン内で何かに取り組む前に、以下のコンテキストを収集してください。

• データレジデンシーと契約上の制約を確認してください。 Salesforce データが組織の境界を離れることは、顧客の Salesforce マスター サブスクリプション契約 (MSA)、該当するデータ処理附属書、および管轄権固有の規制 (GDPR、CCPA、HIPAA) の対象となります。パイプラインを設計する前に、外部 LLM プロバイダーとのデータ処理契約 (DPA) が準備されているかを確認してください。これはオプションではなく、法的前提条件です。
• PII、準識別子、または規制対象データを含むすべてのフィールドを特定してください。 スクラッビングが必要な一般的な Salesforce フィールドには、Contact と Lead の Email、Phone、MobilePhone、Name、SSN__c のようなカスタムフィールド、Health Cloud または Financial Services Cloud 組織内の PHI または MNPI にマップされるあらゆるフィールドが含まれます。PII スクラッビングは、Salesforce API レスポンスからレコードが離れる前に適用する必要があります。ベクトルストア層でのスクラッビングでは遅すぎます。
• 抽出パターンを決定します。フル ロードまたは増分デルタ。 フル ロードの実装は簡単ですが、Bulk API v2 ジョブの期間制限のため、オブジェクトあたり約 500 万レコード以上では実用的ではなくなります。増分抽出には、信頼できるハイウォーターマーク フィールド (SystemModstamp または カスタム LastSyncedAt__c フィールド) と、ハード デリートを個別に検出するための戦略が必要です。
• ボリュームと速度の範囲を確立してください。 Bulk API v2 は、接続されたアプリケーションあたり、24 時間のローリング ウィンドウあたり最大 1 億レコードをサポートします。単一の Bulk API v2 クエリ ジョブは、結果バッチあたり最大 100 MB の圧縮 CSV 出力をサポートしています。大量のレコードを持つ組織は、日付範囲またはカーディナリティの高いインデックス フィールドでジョブをパーティション分割する必要がある場合があります。

---

コア コンセプト

1. Bulk API v2 クエリ ジョブ

Bulk API v2 は、大量の Salesforce レコードを外部パイプラインのために抽出するための正しい API です。非同期です。呼び出し側は SOQL クエリを送信し、Salesforce はバックグラウンドで処理を行い、呼び出し側は完了をポーリングしてからページング分割された CSV バッチとして結果セットをダウンロードします。

主な動作:

• コンテンツ タイプはクエリ ジョブの場合のみ CSV です — JSON と XML をサポートするインジェスト ジョブとは異なり、Bulk API v2 クエリ結果は常に RFC 4180 CSV として返されます。
• 結果セットはページング分割されます。 結果エンドポイントへの各呼び出しは、最大 maxRecords 行 (デフォルト 50,000) を返します。呼び出し側は Sfdclocator カーソル ヘッダーに従って、レスポンス本文が空になるまで後続ページを取得する必要があります。
• ジョブ状態マシン: UploadComplete → InProgress → JobComplete (または Failed / Aborted)。ジョブ ステータス エンドポイントをポーリングします。経過時間に基づいて完了を想定しないでください。
• クエリ ジョブは標準的な意味での API リクエスト制限を消費しません — これらは Bulk API v2 日次バイト制限を消費し、REST および SOAP 呼び出しが消費する組織ごとの 24 時間 API 呼び出し制限ではありません。
• Bulk API v2 クエリ ジョブの SOQL は、SELECT 句でのリレーションシップ トラバーサルをサポートできません (例: SELECT Account.Name FROM Contact はサポートされていません)。事前結合フィールドは、親オブジェクトを個別にクエリしてクライアント側で結合するか、Salesforce スキーマ層で親値を非正規化するフォーミュラ フィールドを使用することで解決される必要があります。

出典: Bulk API 2.0 開発者ガイド

2. 送信前の PII スクラッビング

Salesforce レコードが API から返されると、呼び出しプロセスのメモリ内に存在します。プロセスが PII を削除する前に、レコードを外部 LLM エンドポイントに送信する場合、それ以上の実行ポイントがありません。データは組織の境界を離れています。したがって、PII スクラッビングは Bulk API 結果デコードと送信側ネットワーク呼び出し間の抽出パイプライン内の同期ステップである必要があります。

フィールド タイプ別のスクラッビング戦略:

• 直接識別子 (名前、メール、電話): LLM パイプラインで必要ない場合は、SOQL クエリから完全に除外します。重複排除のために必要な場合は、決定的疑似名に置き換えます (例: 秘密鍵で生成された元の値の HMAC-SHA256)。MD5 は使用しないでください。一般的な値のレインボー テーブルで可逆的です。
• 準識別子 (郵便番号、生年月日、職種): フィールドが意味論的に必要だが正確な値が不要な場合、k-匿名性の一般化を適用します (例: ZIP を 3 桁に切り詰める、DOB を生年に削減する)。
• 偶発的な PII を含むフリー テキスト フィールド (ケース説明、メモ、チャッター): チャンキング前に名前付きエンティティ認識 (NER) パスを適用します。Salesforce はプラットフォーム外 NER サービスを提供していません。これは、抽出パイプラインでオープン ソース ライブラリ (例: spaCy en_core_web_sm) またはプライバシー保護 NER エンドポイントを使用して実装される必要があります。

出典: Salesforce セキュリティ ガイド — データ セキュリティ

3. 外部ベクトル ストア向けチャンキングと エンベッディング

Salesforce から抽出されたテキストは、外部ベクトル ストアに保存される前にチャンキングおよびエンベッディングされる必要があります。チャンキング戦略はソース フィールド タイプによって異なります。

• 構造化テキスト (短いテキスト領域、ピックリスト文書に連結): 1 チャンク あたり 256–512 トークンで 10% の重複を伴う固定サイズ チャンキング。小さいチャンクは、構造化された事実取得の精度を向上させます。
• ロング フォーム コンテンツ (ナレッジ記事本文、ケース説明、リッチ テキスト): 固定サイズ分割よりも、意味論的な境界 (段落区切り、文末) での再帰的文字テキスト分割が推奨されます。これにより段落の一貫性が保持され、中文チャンク境界が減少します。
• HTML コンテンツ (ナレッジ記事 Body フィールド、リッチ テキスト領域): HTML はチャンキング前に削除する必要があります。Salesforce ナレッジ記事本文の生の HTML には、<p>、<ul>、<li>、  および埋め込み画像 <img> タグが含まれます。エンベッディング モデルはタグ マークアップを意味論的トークンとしてエンコードし、これは類似度スコア忠実度を低下させます。最初の変換ステップとして HTML-to-プレーン-テキスト ライブラリを使用してください。

外部ベクトル ストアに保存されるすべてのチャンクは、ソース Salesforce レコードへのトレーサビリティを実現するメタデータを含める必要があります。最小限: record_id (Salesforce 18 文字 ID)、object_api_name、field_api_name、chunk_index、および last_modified (SystemModstamp から)。このメタデータは増分再エンベッディングと監査証跡要件をサポートします。

出典: Bulk API 2.0 開発者ガイド

4. SystemModstamp による増分抽出

SystemModstamp は、すべての Salesforce オブジェクト上のシステム保守フィールドで、任意のユーザーまたはシステム プロセスによるレコード上の任意のフィールドが最後に修正された時刻を記録します。これはインデックスが付けられており、Bulk API v2 SOQL クエリで利用可能で、増分抽出のための正しいハイウォーターマーク フィールドです。

実装パターン:

1. 最初のフル抽出時に、パイプラン実行の開始タイムスタンプをハイウォーターマーク last_sync として記録します。
2. 後続の実行では、WHERE SystemModstamp >= :last_sync でクエリして、変更されたレコードのみを取得します。
3. ハイウォーターマークを永続的に保存し (例: データベース行、クラウド ストレージ内のファイル)、現在のバッチがベクトル ストアに正常に取り込まれた後でのみ更新します。その前ではなく。
4. ハード デリートは SystemModstamp フィルタリングではキャプチャされません。Salesforce Change Data Capture (CDC) API を使用して、ハード デリートを外部ベクトル ストアに伝播させる必要があるオブジェクトの ObjectDeleted イベントをサブスクライブします。

重要: LastModifiedDate はユーザー表示で、特定のインポート操作 (例: トリガー ロジックを無効にする --setbulkheader で Data Loader) によってフリーズできます。SystemModstamp はシステムで開始された変更を含む任意の修正により常に更新されるため、ETL ハイウォーターマークの信頼できる選択肢です。

出典: Change Data Capture 開発者ガイド

---

一般的なパターン

パターン 1: 夜間フル抽出ナレッジ記事パイプライン

いつ使用するか: チームが外部 RAG システム (例: LangChain + Pinecone を使用) を実行していて、Salesforce ナレッジ記事コンテンツを夜間に更新する必要があります。ボリュームは 500,000 記事未満です。レイテンシ容差は同一日の鮮度です。

動作方法:

1. スケジュール済みジョブ (例: GitHub Actions、AWS Lambda cron) は、接続されたアプリを使用した OAuth 2.0 JWT Bearer フロー経由で Salesforce に認証します。
2. Bulk API v2 クエリ ジョブを SOQL で送信します: SELECT Id, Title, Summary, Body, ArticleNumber, LastPublishedDate, SystemModstamp FROM KnowledgeArticleVersion WHERE PublishStatus = 'Online' AND Language = 'en_US'.
3. 30 秒ごとにジョブ ステータスをポーリングします。Sfdclocator カーソルを使用して結果 CSV バッチをダウンロードします。
4. 各行について: Body から HTML を削除、チャンク (512 トークン、64 トークン重複) に分割、各チャンクに Title + ": " を前置してコンテキスト一貫性を確保します。
5. 各チャンクをエンベッディング モデル (例: OpenAI Embeddings API を経由した text-embedding-3-small) で実行します。
6. Id + "_chunk_" + chunk_index をドキュメント ID として使用して Pinecone にベクトルをアップサート します。メタデータ: {record_id, article_number, title, last_published_date}。
7. 現在の抽出に存在しなくなった record_id を持つベクトルを Pinecone から削除します (前回実行 ID セットの保持または Pinecone のメタデータ フィルタリングを使用した古いレコードのパージが必要)。

なぜ代替手段ではないか: REST API を使用した個別レコード GET は、組織ごとの API 呼び出し制限 (ライセンス数によって通常 1–5 百万呼び出し/24 時間) に制限されます。300,000 記事の組織は、抽出だけで日次 API クォータのかなりの割合を消費します。Bulk API v2 はこのボリュームの正しいツールです。

パターン 2: ケース データの増分デルタ パイプライン

いつ使用するか: チームが Salesforce ケース解決メモに基づいた顧客向けチャットボットを実行しています。ケースは継続的に作成および解決されます。外部ベクトル ストアは 1 時間以内に変更を反映する必要があります。夜間のフル抽出では、鮮度要件に不十分です。

動作方法:

1. 最後の成功した抽出タイムスタンプ (last_sync) を永続的なストア (例: AWS SSM パラメータ ストア、PostgreSQL 行) に保存します。
2. 各パイプライン実行時 (15–30 分ごとにスケジュール): Bulk API v2 クエリ ジョブを SOQL で送信します: SELECT Id, CaseNumber, Subject, Description, Resolution__c, SystemModstamp FROM Case WHERE SystemModstamp >= :last_sync AND Status = 'Closed'.
3. PII スクラッビングを適用: ContactEmail を SOQL 投影から除外します。Description に存在する場合は ContactPhone をマスク (正規表現置換経由) してからチャンキングします。
4. 各ケース レコードをチャンク (連結: Subject + "\n" + Description + "\n" + Resolution__c)、エンベッディング、アップサートします。
5. 成功したアップサート後にのみ、last_sync をパイプライン実行の開始タイムスタンプに更新します。アップサートが失敗した場合、ハイウォーターマークを進めないでください。次の実行で再試行してレコード損失を回避します。
6. 別途、ケース CDC チャネル (CaseChangeEvent) をサブスクライブしてハード削除またはマージされたケースをキャプチャし、外部ストアからそのベクトルを削除します。

なぜ代替手段ではないか: CDC のみを使用したストリーミングは、唯一の抽出メカニズムとしては信頼できません。CDC イベント リプレイは 3 日に制限され、72 時間より長いパイプラインの停止によってイベント ギャップが発生します。Bulk API v2 増分抽出は信頼できるバックストップです。CDC はデリート を処理し、より低いレイテンシ補足を提供します。

パターン 3: 外部個人化モデル向けの PII セーフ Contact エンリッチメント

いつ使用するか: チームが Salesforce Contact および Account データを使用して外部個人化モデルを エンリッチしたい、ただし PII を外部サービスに送信しない場合。モデルは企業、役職、セグメント データを必要とします。個別 ID ではなく。

動作方法:

1. SOQL を構成して非 PII フィールドのみを投影: SELECT Id, Account.Industry, Account.AnnualRevenue, Account.NumberOfEmployees, Title, Department, LeadSource, SystemModstamp FROM Contact WHERE ....
2. Salesforce Id を送信前に HMAC-SHA256 疑似名に置き換え: pseudonym = hmac_sha256(secret_key, contact_id). ID 再特定が必要な場合のためにマッピング テーブルを内部に保存します。
3. 疑似名付きレコードのみを外部モデル エンドポイントに送信します。外部モデルは Salesforce レコード ID またはいかなる直接識別子も受け取ることはありません。
4. 外部モデルがセグメント スコアを返したら、疑似名から ID へのマッピング テーブルを検索して、REST API を経由してスコアを Salesforce に書き込み直します。

なぜ代替手段ではないか: 疑似名化ステップを省略して生の Salesforce ID を送信することで、再特定パスが作成されます。外部モデルの入力ログと Salesforce の両方にアクセスできる任意の当事者は、生の ID で結合して個別識別子を復元できます。疑似名化は、データを離れる境界でそのリンケージを切断します。

---

判断ガイダンス

状況	推奨されるアプローチ	理由
ボリューム 100K レコード以下、鮮度容差は日次	Bulk API v2 フル ロード夜間	実装が最も簡単。フル ロードはハイウォーターマーク ドリフト問題を回避
ボリューム 100 万レコード以上、鮮度容差は日次	Bulk API v2 増分、SystemModstamp ハイウォーターマーク	フル ロードは 100 万超で遅く費用がかかります。増分は変更レコードに限定
鮮度要件 1 時間未満	Bulk API v2 増分 + デリート向け CDC イベント サブスクリプション	Bulk API v2 短間隔は信頼できます。CDC はハード削除ギャップを埋めます
ソース フィールドは HTML (ナレッジ Body、リッチ テキスト)	チャンキング前に HTML を削除、その後ではなく	エンベッディング モデルは HTML タグを意味論的トークンとしてエンコード。トークナイザーに到達する前に削除が必須
潜在的な偶発的 PII を含むフリー テキスト フィールド (ケース メモ、チャッター)	チャンキング前の NER ベース スクラッビング	フィールド レベル除外は不十分。PII は任意のフリー テキスト フィールドに表示可能。NER はエンティティ提及をキャッチ
外部 LLM がサード パーティ SaaS (OpenAI、Anthropic)	送信前にすべての直接識別子を疑似名化	サード パーティ SaaS プロバイダーは Salesforce サブプロセッサーではありません。PII 送信には個別 DPA が必要
外部ストアのレコード削除を検出する必要がある	各オブジェクトの CDC ObjectDeleted イベントをサブスクライブ	SystemModstamp ハイウォーターマークは修正のみをキャプチャ。ハード削除は外部ストアに古いベクトルを残します
組織が GDPR 削除権義務を持つ	すべてのベクトル チャンク上に record_id メタデータを保持	完全な再インデックスなしでベクトル ストア全体の個人のすべてのチャンク削除を対象削除可能化

---

推奨ワークフロー

AI エージェント または このスキルをアクティブ化するプラクティショナーのためのステップバイステップ手順:

1. フィールドを分類して法的基礎を確認します。 対象の各オブジェクトについて、スコープ内の各フィールドをリストします。各フィールドを以下として マーク: safe (PII なし)、pseudonymize (結合キーに必要な直接識別子)、scrub (抽出から除外)、または ner-required (偶発的な PII を含む可能性のあるフリー テキスト)。外部 LLM/ベクトル ストア プロバイダーとの DPA が存在することを確認します。この分類なしで進めないでください。

2. SOQL 投影を設計します。 scrub 分類フィールドを SELECT 句から除外する SOQL クエリを記述します。ハイウォーターマーク トラッキング向けに SELECT に SystemModstamp を追加します。Developer Console で SOQL がウォーターマーク トラッキング向け LIMIT 10 を使用して正常に実行されることを確認してからバルク ジョブを送信してください。

3. 抽出ジョブを実装します。 OAuth 2.0 JWT Bearer フロー (ユーザー名/パスワードではなく) で認証します。Bulk API v2 クエリ ジョブを送信します。JobComplete までジョブ ステータスをポーリングします。Sfdclocator カーソル ヘッダーを使用してすべての結果バッチをダウンロードします。すべての結果が単一バッチに含まれていると想定しないでください。

4. プロセス内で PII スクラッビングを適用します。 レコードをディスクに書き込みまたは外部エンドポイントに送信する前に: pseudonymize とマーク されたフィ