📖

Vertex AI Gemini で非構造化データを扱う

に公開
Google Cloud
BigQuery
dataengineering
Vertex AI
Gemini
tech

Vertex AI Gemini で非構造化データを構造化して BigQuery に格納してみた

はじめに

てつどんです。データエンジニアやってます。

データエンジニアとして、必要な知識をインプット・アウトプットする上で、次はLLMを用いた学習をやってみました。

一般的な「チャットで質問する」使い方ではなく、データパイプラインの中に LLM を組み込むという観点でのインプットです。

一般的な使い方：人間 → LLM → 回答

データエンジニア的な使い方：データ → LLM → 構造化 → BigQuery

今回学んだ内容はこちらです。

  • Step 1：Vertex AI Gemini API の基本
  • Step 2：構造化出力（JSON 抽出）
  • Step 3：複数テキストの一括処理
  • Step 4：マルチモーダル対応（PDF からの情報抽出）
  • Step 5：BigQuery への連携

学習環境

  • GCP（個人アカウント）
  • Vertex AI Gemini 2.5 Flash
  • Python（ローカル環境）

費用感： Gemini 2.5 Flash は非常に安価で、学習目的で 100 回呼び出しても数円〜数十円程度です。個人アカウントでも気軽に試せます。

Step 1：Vertex AI Gemini API の基本

環境セットアップ

pip install google-cloud-aiplatform vertexai

# GCP 認証
gcloud auth application-default login

最初の API 呼び出し

import vertexai
from vertexai.generative_models import GenerativeModel

PROJECT_ID = "your-project-id"
LOCATION = "asia-northeast1"  # 東京リージョン

vertexai.init(project=PROJECT_ID, location=LOCATION)
model = GenerativeModel("gemini-2.5-flash")

response = model.generate_content("データエンジニアとはどんな仕事ですか？3行で教えてください。")
print(response.text)

たったこれだけで LLM が使えます。GCP ネイティブなので IAM による権限管理も BigQuery との連携もシームレスです。

Step 2：構造化出力（JSON 抽出）

テキストから決まった形式でデータを取り出すのが非構造化データ処理の核心です。

import vertexai
import json
from vertexai.generative_models import GenerativeModel, GenerationConfig

vertexai.init(project=PROJECT_ID, location=LOCATION)
model = GenerativeModel("gemini-2.5-flash")

text = """
先月の売上は500万円で、前月比110%でした。
主力商品はノートパソコンで、販売数は200台です。
"""

prompt = f"""
以下のテキストから情報を抽出して、JSON形式で返してください。
JSONのみを返し、説明文は不要です。

抽出する項目：
- sales_amount: 売上金額（数値）
- month_over_month_ratio: 前月比（小数）
- top_product: 主力商品名
- units_sold: 販売数（数値）

テキスト：
{text}
"""

response = model.generate_content(
    prompt,
    generation_config=GenerationConfig(
        response_mime_type="application/json"  # JSON形式を強制
    )
)

data = json.loads(response.text)
print(f"売上金額: {data['sales_amount']:,}円")
print(f"前月比: {data['month_over_month_ratio'] * 100:.0f}%")

ハマりポイント：コードブロック記号の問題

response_mime_type を指定しないと、Gemini が JSON の前後にコードブロック記号を付けて返すことがあります。

```json   ← これが邪魔
{...}
```       ← これも邪魔

解決策：response_mime_type="application/json" を必ず指定する

これで純粋な JSON が返ってくるので後処理が不要になります。

Step 3：複数テキストの一括処理

実務では複数件まとめて処理する場面がほとんどです。1件ずつ API を呼び出すと遅くてコストも高くなるので、まとめて1回のAPIコールで処理するのがおすすめです。

reviews = [
    "配送が遅くてがっかりしました。商品自体は良かったです。",
    "品質が素晴らしく、また購入したいと思います！",
    "説明と違う商品が届きました。返品したいです。",
]

prompt = f"""
以下の複数のレビューをそれぞれ分析して、JSON配列で返してください。

抽出する項目：
- sentiment: 感情（positive / negative / neutral）
- category: カテゴリ（配送 / 品質 / 商品説明 / その他）
- score: 満足度スコア（1〜5の整数）

レビュー一覧：
{json.dumps(reviews, ensure_ascii=False, indent=2)}
"""

response = model.generate_content(
    prompt,
    generation_config=GenerationConfig(
        response_mime_type="application/json"
    )
)

results = json.loads(response.text)
for i, (review, result) in enumerate(zip(reviews, results)):
    result["review"] = review
    print(f"レビュー{i+1}：{result}")

実行結果：

レビュー1：{'sentiment': 'negative', 'category': '配送', 'score': 2, 'review': '配送が遅くて...'}
レビュー2：{'sentiment': 'positive', 'category': '品質', 'score': 5, 'review': '品質が素晴らしく...'}
レビュー3：{'sentiment': 'negative', 'category': '商品説明', 'score': 1, 'review': '説明と違う...'}

Gemini がレビューの文脈をちゃんと読み取っているのがわかります。「商品自体は良かった」と書いてあっても配送の問題で negative・スコア 2 と判断しているのが的確です。

2つのアプローチの比較：

1件ずつ処理 まとめて処理
API呼び出し回数 件数分 1回
速度 遅い 速い
コスト 高め 安め
向いているケース 細かいエラーハンドリングが必要 件数が多い・速度重視

Step 4：マルチモーダル対応（PDF からの情報抽出）

Gemini はテキストだけでなく PDF も理解できます。Part.from_data() で PDF をバイナリとして渡すだけです。

from vertexai.generative_models import GenerativeModel, GenerationConfig, Part

with open("sample_invoice.pdf", "rb") as f:
    pdf_data = f.read()

pdf_part = Part.from_data(
    data=pdf_data,
    mime_type="application/pdf"
)

prompt = """
この請求書PDFから情報を抽出して、JSON形式で返してください。

抽出する項目：
- company: 会社名
- invoice_no: 請求書番号
- date: 請求日
- payment_due: 支払期限
- items: 明細リスト（各itemはname・quantity・unit_price・totalを含む）
- subtotal: 小計
- tax: 税額
- total: 合計金額
"""

# テキストのときと違い、リストで渡す
response = model.generate_content(
    [pdf_part, prompt],
    generation_config=GenerationConfig(
        response_mime_type="application/json"
    )
)

data = json.loads(response.text)
print(f"会社名: {data['company']}")
print(f"合計: ${data['total']}")

テキストと PDF の渡し方の違い：

# テキストのとき
response = model.generate_content(prompt)

# PDF のとき（リストで渡す）
response = model.generate_content([pdf_part, prompt])

従来の OCR との違い

従来の OCR ツールは「文字を認識するだけ」でしたが、Gemini は文脈を理解して構造化まで一気にやってくれます。

従来の OCR：文字を認識 → 後から自分でパースが必要
Gemini：内容を理解して構造化まで一気にやってくれる

請求書・契約書・レポートなど、PDF からの情報抽出が必要なケースで非常に強力です。

ハマりポイント：壊れた JSON が返ってくることがある

response_mime_type="application/json" を指定しても稀に壊れた JSON が返ってくることがあります。LLM のレスポンスは確率的なので、リトライ処理を入れておくのが実務的な書き方です。

for attempt in range(3):
    response = model.generate_content(
        [pdf_part, prompt],
        generation_config=GenerationConfig(
            response_mime_type="application/json"
        )
    )
    try:
        data = json.loads(response.text)
        print("抽出成功！")
        break
    except json.JSONDecodeError as e:
        print(f"試行{attempt + 1}回目失敗: {e}")
        if attempt == 2:
            raise Exception("JSONパースに3回失敗しました")

Step 5：BigQuery への連携

テーブル作成（bq コマンド）

# データセット作成
bq mk --dataset \
  --location=asia-northeast1 \
  your-project-id:llm_output

# テーブル作成
bq mk --table \
  your-project-id:llm_output.invoices \
  company:STRING,invoice_no:STRING,date:DATE,payment_due:DATE,subtotal:FLOAT,tax:FLOAT,total:FLOAT,inserted_at:TIMESTAMP

PDF → Gemini → BigQuery の一連のパイプライン

import vertexai
import json
from datetime import datetime
from vertexai.generative_models import GenerativeModel, GenerationConfig, Part
from google.cloud import bigquery

PROJECT_ID = "your-project-id"
LOCATION = "asia-northeast1"

vertexai.init(project=PROJECT_ID, location=LOCATION)
model = GenerativeModel("gemini-2.5-flash")
bq_client = bigquery.Client(project=PROJECT_ID)

# PDF を読み込む
with open("sample_invoice.pdf", "rb") as f:
    pdf_data = f.read()

pdf_part = Part.from_data(data=pdf_data, mime_type="application/pdf")

prompt = """
この請求書PDFから情報を抽出して、JSON形式で返してください。
抽出する項目：
- company, invoice_no, date（YYYY-MM-DD）, payment_due（YYYY-MM-DD）
- subtotal, tax, total（数値）
"""

# Gemini で抽出（リトライあり）
for attempt in range(3):
    response = model.generate_content(
        [pdf_part, prompt],
        generation_config=GenerationConfig(
            response_mime_type="application/json"
        )
    )
    try:
        data = json.loads(response.text)
        break
    except json.JSONDecodeError:
        if attempt == 2:
            raise Exception("JSONパースに3回失敗しました")

# BigQuery に書き込む
row = {
    "company":     data["company"],
    "invoice_no":  data["invoice_no"],
    "date":        data["date"],
    "payment_due": data["payment_due"],
    "subtotal":    float(data["subtotal"]),
    "tax":         float(data["tax"]),
    "total":       float(data["total"]),
    "inserted_at": datetime.utcnow().isoformat()
}

table_id = f"{PROJECT_ID}.llm_output.invoices"
errors = bq_client.insert_rows_json(table_id, [row])

if errors:
    print(f"エラー: {errors}")
else:
    print(f"BigQuery への書き込み成功！テーブル: {table_id}")

全体の流れ：

sample_invoice.pdf
        ↓ Part.from_data() で読み込む
Gemini 2.5 Flash
        ↓ JSON で構造化（リトライあり）
{company, invoice_no, date, ...}
        ↓ insert_rows_json() で書き込む
BigQuery llm_output.invoices

まとめ

今回学んだことを整理します。

Step 内容 重要なポイント
1 API 基本 vertexai.init() + generate_content() だけで使える
2 JSON 抽出 response_mime_type="application/json" は必須
3 一括処理 複数件はまとめて1回のAPIコールが速くて安い
4 PDF 対応 Part.from_data() でリスト形式で渡す
5 BQ 連携 insert_rows_json() で Python から直接書き込める

共通のハマりポイント：

  • response_mime_type を指定しないとコードブロック記号が混入する
  • LLM のレスポンスは確率的なのでリトライ処理は必須
  • PDF はリスト形式 [pdf_part, prompt] で渡す

まだまだ、必要な知識をインプットしていきたいと思います！

Discussion