2026.07.22 コラム
【テックコラム】Agent Search 更新時に Layout Parser の再処理対象を確認してみる
DataCurrent の金子です。今回は、Agent Search のデータストアを Cloud Storage から更新するときに、FULL インポートをどう運用するか動作を検証します。
※ この記事は、2026年7月時点の Google Cloud の仕様に基づく検証メモです。請求レポートに出た Layout Parser の料金とカウントをもとに、FULL インポート時の再処理対象を確認します。
確認したいこと
Cloud Storage のフォルダをソースにして Agent Search のデータストアを作ると、PDF などの非構造化データを取り込めます。この Cloud Storage ソースを FULL インポートで更新するとき、どの変更が Layout Parser の再処理対象になるのかを確認します。
Agent Search に統合された Document AI 機能としての Layout Parser はカウント単位で課金され、PDF では 1 ページが 1 カウントとして扱われます。そのため、FULL インポート時に変更のない PDF まで毎回再処理されるのか、URI や hash が変わった PDF だけが対象になるのかが運用上のポイントになります。
今回は Cloud Storage 側の変更パターンを分けて、実際にデータストアにインポートし、請求レポートの出方を確認することで Layout Parser の更新対象を確認してみます。
参考:
確認手順
全体の流れは次の通りです。
- 検証用 PDF を Cloud Storage に初回アップロードする
- コンソールから Agent Search のデータストアに
FULLインポートする
- 請求レポートの太平洋標準時（PST）の日付境界をまたいでから、Cloud Storage 上の PDF を更新する
- コンソールから同じ Cloud Storage フォルダをもう一度
FULLインポートする
- 翌日以降に請求レポートを SKU ごとに見て、Layout Parser のカウントを確認する
インポートでは JSONL ファイルを指定せず、Cloud Storage のフォルダをそのまま指定します。この場合、Agent Search 側の Document ID は自動で付与されます。
請求レポートは太平洋標準時基準で日付が切り替わります。初回インポートと 2 回目インポートの課金を分けて見たいので、太平洋標準時の日付境界をまたいで実行します。太平洋夏時間の期間であれば、日本時間では 16:00 以降が太平洋標準時の次の日にあたります。
ケース設計
Layout Parser の再処理対象を特定するために、Cloud Storage 側の変更パターンを分けて検証します。
✓ は 2 回目の更新で変わるもの、- は変わらないものです。
B から E は、同じ URI のまま Cloud Storage 側のどの変化が効くのかを見るケースです。F と G は URI が変わるケース、H は削除だけのケースです。内容が変わるケースは D/E/G、内容が変わらない置き換えは B/F です。
請求レポートでは、Layout Parser が処理したファイルごとの内訳ではなく、SKU ごとの合計カウントが表示されます。そのため、各ケースのページ数を 1, 2, 4, 8 … のように 2 のべき乗にしておきます。こうしておくと、2 回目の Layout Parser カウントを見たときに、どのケースが再処理されたのかを合計値から読み取りやすくなります。
執筆時点の Agent Search の料金ページ では、Layout Parser は $10.00 / 1,000 カウント です。PDF は 1 ページが 1 カウントとして扱われます。初回インポートで A から H までの 255 ページがすべて処理されると、Layout Parser の課金は $2.55 になります。
検証結果
請求レポートの Usage 列で Layout Parser のカウントを確認します。Usage は表示期間の合計なので、左は初回インポート日の 7/6 (PST)、右は更新日の 7/7 (PST) だけに絞っています。
初回の 255 カウントは、A から H までの合計ページ数と一致します。2 回目の 120 カウントは、D の 8 ページ、E の 16 ページ、F の 32 ページ、G の 64 ページを足した値と一致します。
こちらは Google Cloud Console の data store Documents 画面です。初回と更新後で ID, URI, Index Status を比較しています。
Index Status の時刻が変わっていれば、data store の index は更新されたと読めます。また、Document ID が同じであれば同じ document の更新、変わっていれば別 document として追加されたと読めます。
請求レポートと data store の Documents 画面を合わせると、ケースごとの結果は次のようになります。
✓ は 2 回目で変化あり、または再処理あり、- は変化なし、または再処理なしです。削除 は更新後の一覧から消えたものです。
Index Status は B/C/D/E/F/G で変化しました。一方で、Layout Parser の請求 count に出たのは D/E/F/G だけです。つまり、data store の index 更新と Layout Parser の再処理は同じではありません。
今回の結果では、Layout Parser の再処理は「URI が変わる」または「同じ URI でも hash が変わる」場合に発生しました。generation だけ、または metadata だけの変更では、Layout Parser の再処理対象にはなっていませんでした。
運用を考えるときのポイント
今回の検証結果により、FULL インポートを実行しても、Cloud Storage フォルダ内の PDF が毎回すべて Layout Parser で再処理されるわけではないことがわかりました。
Layout Parser の再処理対象になったのは、Cloud Storage URI が変わった PDF と、同じ URI でも hash が変わった PDF でした。これは、ユーザー側から見て「追加・差し替えされた PDF が処理対象になる」という感覚に近い結果だと言えます。
一方で、同一内容の再アップロードで generation だけが変わったケースや、metadata だけを変更したケースは Layout Parser の count には出ませんでした。削除だけのケースも、削除された PDF を Layout Parser で再処理するわけではなさそうです。
そのため、FULL インポートを定期実行する場合は、「毎回全量分」ではなく、「URI が変わる PDF」と「hash が変わる PDF」のページ数を中心にコストを見積もるのが実態に近そうです。
まとめ
Agent Search の Cloud Storage ソースを FULL インポートで更新する場合、data store の index 更新と Layout Parser の再処理は分けて考える必要があります。
そのとき、更新操作そのものではなく、Layout Parser がどの PDF を何ページ分処理した扱いになるのかがコスト上の論点になります。
今回の検証では、Cloud Storage URI が変わった PDF と、同じ URI でも hash が変わった PDF が Layout Parser の再処理対象になりました。
最後に
自社に専門人材がいない、リソースが足りない等の課題をお持ちの方に、エンジニア領域の支援サービス（Data Engineer Hub）をご提供しています。お困りごとがございましたら是非お気軽にご相談ください。
本件に関するお問い合わせは下記にて承ります。
株式会社DataCurrent
info@datacurrent.co.jp
付録: Cloud Storage 更新用の検証スクリプト
▪ 検証用スクリプトの使い方
初回アップロード:
python 20260707/vais_layout_billing/gcs_upload.py initial \ --project-id YOUR_PROJECT_ID \ --bucket YOUR_BUCKET \ --prefix vais-layout-billing-test/20260707
インポート対象:
gs://YOUR_BUCKET/vais-layout-billing-test/20260707/
2 回目の更新:
python 20260707/vais_layout_billing/gcs_upload.py second \ --project-id YOUR_PROJECT_ID \ --bucket YOUR_BUCKET \ --prefix vais-layout-billing-test/20260707
メタデータ確認用の CSV は、次の場所に出力されます。size, crc32c, md5, generation, metageneration, updated を見て、Cloud Storage 側の変更がケース設計どおりになっているかを確認します。
20260707/vais_layout_billing/logs/<bucket>/<prefix>/initial.csv 20260707/vais_layout_billing/logs/<bucket>/<prefix>/before.csv 20260707/vais_layout_billing/logs/<bucket>/<prefix>/after.csv
▪ スクリプト
import argparse
import csv
import json
import shutil
from pathlib import Path
from google.cloud import storage
from markdown_it import MarkdownIt
from weasyprint import HTML
ROOT = Path(__file__).resolve().parent
PDFS = ROOT / "pdfs"
LOGS = ROOT / "logs"
BUCKET = ""
PREFIX = ""
GCS = None
CASES_JSONL = """
{"id":"A","pages":1,"gcs":"keep","pdf":"same"}
{"id":"B","pages":2,"gcs":"replace","pdf":"same"}
{"id":"C","pages":4,"gcs":"update_metadata","pdf":"same"}
{"id":"D","pages":8,"gcs":"replace","pdf":"changed"}
{"id":"E","pages":16,"gcs":"replace","pdf":"changed_size"}
{"id":"F","pages":32,"gcs":"rename","pdf":"same"}
{"id":"G","pages":64,"gcs":"rename","pdf":"changed"}
{"id":"H","pages":128,"gcs":"delete","pdf":"same"}
"""
CASES = [json.loads(line) for line in CASES_JSONL.splitlines() if line]
def configure(args):
global BUCKET, PREFIX, GCS
BUCKET = args.bucket
PREFIX = args.prefix.strip("/")
GCS = storage.Client(project=args.project_id).bucket(BUCKET)
def initial_name(case):
return f"{case['id'].lower()}.pdf"
def second_name(case):
name = initial_name(case)
return name.replace(".pdf", "_new.pdf") if case["gcs"] == "rename" else name
def page_markdown(case, content, page, pages):
extra = "\nextra=make this PDF content longer than the initial file for size-change verification" if content == "changed_size" else ""
return f"""# VAIS Layout Billing Test
case={case}
content={content}
page={page}/{pages}
marker={case}-{content}-{page}{extra}
"""
def make_pdf(path, pages, case, content):
if path.exists():
return
path.parent.mkdir(parents=True, exist_ok=True)
md = MarkdownIt()
sections = []
for page in range(1, pages + 1):
body = md.render(page_markdown(case, content, page, pages))
sections.append(f'<section class="page">{body}</section>')
html = "<style>@page{size:Letter;margin:72px}.page{page-break-after:always}.page:last-child{page-break-after:auto}</style>"
html += "\n".join(sections)
HTML(string=html).write_pdf(path)
def pad_to(path, size):
diff = size - path.stat().st_size
if diff < 0:
raise ValueError(f"{path} is larger than target size {size}")
if diff:
with path.open("ab") as f:
f.write(b" " * diff)
def object_name(name):
return f"{PREFIX}/{name}"
def gcs_uri(name=""):
return f"gs://{BUCKET}/{PREFIX}/{name}"
def upload(path, name=None):
name = name or path.name
print(f"upload {name}")
GCS.blob(object_name(name)).upload_from_filename(path)
def delete(name):
print(f"delete {name}")
GCS.blob(object_name(name)).delete()
def patch_metadata(name):
print(f"metadata {name}")
blob = GCS.blob(object_name(name))
blob.reload()
blob.metadata = {"layout-billing-test": "metadata-only"}
blob.patch()
def snapshot(name):
fields = [
"file", "size", "crc32c", "md5", "generation", "metageneration",
"updated"
]
rows = [{
"file": Path(blob.name).name,
"size": str(blob.size),
"crc32c": blob.crc32c or "",
"md5": blob.md5_hash or "",
"generation": str(blob.generation),
"metageneration": str(blob.metageneration),
"updated": blob.updated.isoformat() if blob.updated else "",
} for blob in GCS.list_blobs(prefix=f"{PREFIX}/")
if blob.name.endswith(".pdf")]
log_dir = LOGS / BUCKET / PREFIX
log_dir.mkdir(parents=True, exist_ok=True)
with (log_dir / f"{name}.csv").open("w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fields)
writer.writeheader()
writer.writerows(sorted(rows, key=lambda row: row["file"]))
def initial():
for case in CASES:
path = PDFS / "initial" / initial_name(case)
make_pdf(path, case["pages"], case["id"], "same")
if case["id"] == "D":
changed = PDFS / "second" / initial_name(case)
make_pdf(changed, case["pages"], case["id"], "changed")
pad_to(path, changed.stat().st_size)
upload(path)
snapshot("initial")
print(f"Import this folder in Console: {gcs_uri()}")
def second():
snapshot("before")
for case in CASES:
initial = initial_name(case)
second = second_name(case)
path = PDFS / "second" / second
gcs = case["gcs"]
if gcs == "keep":
continue
if case["pdf"] == "same" and gcs in ("replace", "rename"):
path.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(PDFS / "initial" / initial, path)
elif gcs in ("replace", "rename"):
make_pdf(path, case["pages"], case["id"], case["pdf"])
if gcs == "replace":
upload(path, initial)
elif gcs == "update_metadata":
patch_metadata(initial)
elif gcs == "rename":
upload(path, second)
delete(initial)
elif gcs == "delete":
delete(initial)
else:
raise ValueError(gcs)
snapshot("after")
print(f"Compare logs under {LOGS / BUCKET / PREFIX}, then run FULL import in Console.")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=["initial", "second"])
parser.add_argument("--project-id", required=True)
parser.add_argument("--bucket", required=True)
parser.add_argument("--prefix", required=True)
args = parser.parse_args()
configure(args)
{"initial": initial, "second": second}[args.command]()
if __name__ == "__main__":
main()