一つの判断に一つの時点――AI集計のスナップショット境界

AIがCRMは10時、請求情報は10時8分、問い合わせ履歴は昨日のコピーを読むと、もっともらしい結論でも前提がそろいません。データ基盤責任者向けに、一つの判断を再現可能な時点へ固定する「スナップショット境界」を作ります。

「最新」ではなく判断時点をそろえる

同じデータベース内でも、読取方法によって見える時点は変わります。PostgreSQLのRead Committedでは、同じトランザクション内の連続したSELECTが別の状態を見ることがあります。一方、Repeatable Readでは最初の問い合わせ時点のスナップショットを後続の問い合わせも参照します。[1]

業務AIは複数のAPIや分析テーブルをまたぐため、一つのトランザクションだけでは囲えません。そこで「全ソースを完全に同時刻へ戻す」のではなく、判断の基準時刻、各ソースの版、許容する時差、失効条件を一つの記録に束ねます。版を指定できないAPIは取得応答そのものを保存し、再現性が必要な判断には使わない選択肢も残します。

コピーして使える判断スナップショット

次のYAMLは、AIへの入力本文ではなく、入力を集めた証拠です。AIはこの記録を変更できず、欠けたソースがあれば結論を出さずに停止します。

decision_snapshot:
  decision_id: renewal-priority-20261020-017
  as_of: "2026-10-20T10:00:00+09:00"
  max_skew_seconds: 300
  expires_at: "2026-10-20T10:30:00+09:00"
  sources:
    - name: crm_accounts
      captured_at: "2026-10-20T10:01:12+09:00"
      version: "pg-snapshot:000003A1-1"
      record_count: 42
    - name: billing_summary
      captured_at: "2026-10-20T10:03:40+09:00"
      snapshot_id: "891340112"
      record_count: 42
    - name: support_api
      captured_at: "2026-10-20T10:04:18+09:00"
      etag: '"case-feed-v318"'
      record_count: 117
  input_hash: "sha256:..."
  result_hash: "sha256:..."
  status: "ready"

分析テーブルにスナップショット機能がある場合は、時刻だけでなくIDを保存します。Apache IcebergのAPIは現在のスナップショット、特定IDのスナップショットを取得でき、asOfTimeやuseSnapshotで読取版を指定できます。[2] 「10時ごろのデータ」ではなく、再読込できる版を記録します。

5段階で入力を固定してからAIへ渡す

1. 判断の基準時刻を決める

業務オーナーがas_ofとmax_skew_secondsを決めます。月次締めなら時差ゼロに近づけ、朝会の優先順位付けなら数分を許すなど、誤差が判断を変えるかで設定します。

2. ソースごとに版を確保する

DBは読取専用トランザクションやエクスポート済みスナップショット、データレイクはスナップショットID、APIはETagまたは更新トークンを取得します。版を返さないAPIは応答本文、取得時刻、要求条件を保管します。

3. そろったかを機械判定する

必須ソースが全てあり、captured_atの最大差が許容値以内で、件数や主キーの突合が通ったときだけreadyにします。欠損、期限切れ、版の消失はblockedとし、AIへ部分入力を渡しません。

4. 入力と出力を結ぶ

整形後の入力へハッシュを付け、モデル、プロンプト版、結果ハッシュと一緒に保存します。結論だけを再生成するのではなく、同じ版の入力を再読込できることを受入条件にします。

5. 実行直前に版を再確認する

AIの提案がCRM更新などの書込みへ進むときは、読取時の版と現在版を比較します。HTTPのIf-Matchは、指定したETagと現在の表現が一致する場合だけ変更を実行し、並行更新による上書きを防ぐために使われます。[3] 不一致なら古い提案を適用せず、新しいスナップショットから判断を作り直します。

仮想例：更新優先顧客の抽出

【仮想例】カスタマーサクセス責任者が、更新期限30日以内の顧客から対応優先順位を作ります。10時を基準に、CRM、請求、問い合わせの三ソースを収集しました。CRMと請求は4分以内でしたが、問い合わせAPIのETagが前日版のままでした。

システムは42社の順位を推測で出さず、support_api: staleとして停止します。APIを再取得し、三ソースの顧客IDが一致してからAIへ渡します。担当者が提案先のCRM状態を更新する直前にETagが変わっていれば、その1社だけ再計算へ戻します。完了条件は、順位表の各行から入力版を追跡でき、期限内かつ現在版と一致した更新だけが成功することです。

各ソースの時点が違っていても結論が変わらず、再現や監査も不要という場合には向きません。取得と保存の費用が増えるためです。与信、支払、更新優先度、在庫配分など、入力の前後関係で判断が変わる業務に絞って使います。

明日から使える実装チェックリスト

[ ] 判断ごとにas_ofと許容時差を決めた
[ ] 必須ソースと欠損時の停止条件を列挙した
[ ] DB、データレイク、APIの版識別子を保存した
[ ] 版がないAPIは要求条件と応答本文を保管した
[ ] ソース間の時差、件数、主キーを機械照合した
[ ] 入力、モデル、プロンプト、結果をハッシュで結んだ
[ ] 判断の失効時刻を設定した
[ ] 書込み直前に現在版を再確認した

明日は既存のAI集計を一つ選び、データ取得の間に一件を更新してください。古い版と新しい版が混ざったら結論を止め、どのソースを取り直すべきか表示できれば、スナップショット境界は機能しています。

参考資料

本記事はAIを活用して作成し、公開前に自動レビューを実施しています。情報の確認日：2026-10-20。