概要
- 問い合わせデータをGemini APIへ投げてラベリングしてみた
- 何もわからず投げたら約1万件投げたら8時間ぐらいかかった
- 途中、Batch APIなるものの存在に気づき、投げたら180秒ぐらいでおわりました／(^o^)＼
使ったモデル
- gemini-2.5-flash
3.5はなぜかエラーで使えなかった（特に気にせず変更しました）
やったこと
必要なpackageのインストール
$ pip3 install google-genai
コードを生成する
#!/usr/bin/env python3
"""
問い合わせ CSV を Gemini でカテゴリ判定するスクリプト
処理フロー:
1. 入力 CSV を読み込む（inquiry_data_filled.csv）
2. 各問い合わせの「問い合わせ内容」（スレッド全体）を Gemini に送信
3. 下記 N カテゴリのいずれかに分類し、判定理由・確信度とともに CSV に出力
カテゴリ:
- ...省略...
事前準備:
pip install google-genai
認証:
export GEMINI_API_KEY=xxxxx
または --api-key オプションで指定
実行例:
python3 scripts/classify_inquiry_category.py \\
--input inquiry_data_filled.csv \\
--output inquiry_classified.csv
# 先頭 20 件のみで動作確認
python3 scripts/classify_inquiry_category.py --limit 20
# モデル変更・間隔調整
python3 scripts/classify_inquiry_category.py --model gemini-2.5-flash --interval 1.0
"""
import argparse
import csv
import json
import os
import subprocess
import sys
import time
from typing import Optional
# ---------------------------------------------------------------------------
# 定数
# ---------------------------------------------------------------------------
DEFAULT_INPUT = "inquiry_data_filled.csv"
DEFAULT_OUTPUT = "inquiry_classified.csv"
DEFAULT_MODEL = "gemini-3.5-flash"
DEFAULT_INTERVAL = 0.5 # リクエスト間隔（秒）
MAX_CONTENT_CHARS = 8000 # 問い合わせ内容の最大文字数（トークン節約）
MAX_RETRIES = 5 # レート制限時の最大リトライ数
CATEGORIES = [
# ...省略...
]
# Gemini に渡す JSON スキーマ
RESPONSE_SCHEMA = {
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": CATEGORIES,
"description": "判定カテゴリ",
},
"reason": {
"type": "string",
"description": "判定理由（50字以内）",
},
"confidence": {
"type": "number",
"description": "確信度（0.0〜1.0）",
},
},
"required": ["category", "reason", "confidence"],
}
# ---------------------------------------------------------------------------
# プロンプト
# ---------------------------------------------------------------------------
SYSTEM_PROMPT = """\
あなたは◯◯会社のカスタマーサポート問い合わせを分類するアシスタントです。
問い合わせスレッド（顧客メールと対応履歴）を読み、以下のカテゴリのいずれか1つに分類してください。
【大前提】
...省略...
【カテゴリ定義】
...省略...
【判定ルール】
...省略...
"""
def build_user_prompt(content: str, category: str, sub_category: str) -> str:
"""ユーザープロンプトを組み立てる。"""
hint = ""
if category or sub_category:
hint = f"\n【参考情報（既存分類）】カテゴリ: {category} / サブカテゴリ: {sub_category}\n"
truncated = content[:MAX_CONTENT_CHARS]
if len(content) > MAX_CONTENT_CHARS:
truncated += "\n... (以降省略)"
return f"{hint}\n【問い合わせスレッド】\n{truncated}"
# ---------------------------------------------------------------------------
# 処理済み問い合わせID の読み込み（再開用）
# ---------------------------------------------------------------------------
def load_done_contact_ids(output_path: str) -> set[str]:
"""既存出力 CSV から処理済み問い合わせID を収集する。"""
done: set[str] = set()
if not os.path.exists(output_path):
return done
try:
with open(output_path, encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f)
for row in reader:
tid = row.get("問い合わせID", "").strip()
if tid:
done.add(tid)
except Exception:
pass
return done
# ---------------------------------------------------------------------------
# Vertex AI 直接 HTTP 呼び出し（gcloud トークン使用）
# ---------------------------------------------------------------------------
class VertexAIClient:
"""
gcloud トークンで Vertex AI generateContent を直接呼ぶクライアント。
google-genai SDK の Vertex AI 認証問題を回避するため raw HTTP を使用。
"""
VERTEX_URL_TEMPLATE = (
"https://{location}-aiplatform.googleapis.com/v1/projects/{project}"
"/locations/{location}/publishers/google/models/{model}:generateContent"
)
GOOGLE_AI_URL_TEMPLATE = (
"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent"
"?key={api_key}"
)
def __init__(
self,
vertexai: bool,
project: Optional[str],
location: str,
api_key: str,
):
self.vertexai = vertexai
self.project = project
self.location = location
self.api_key = api_key
self._token: Optional[str] = None
def _get_token(self) -> str:
token = subprocess.check_output(
["gcloud", "auth", "print-access-token"], text=True, stderr=subprocess.DEVNULL
).strip()
self._token = token
return token
def call(self, model: str, payload: dict) -> dict:
import urllib.request, urllib.error
if self.vertexai:
url = self.VERTEX_URL_TEMPLATE.format(
location=self.location, project=self.project, model=model
)
token = self._get_token()
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
else:
url = self.GOOGLE_AI_URL_TEMPLATE.format(model=model, api_key=self.api_key)
headers = {"Content-Type": "application/json"}
body = json.dumps(payload).encode()
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=60) as resp:
return json.loads(resp.read())
def classify_with_gemini(
client: VertexAIClient,
model_name: str,
content: str,
category: str,
sub_category: str,
) -> dict:
"""
Gemini API を呼び出してカテゴリ判定を行う。
Returns: {"category": str, "reason": str, "confidence": float}
"""
import urllib.error
user_prompt = build_user_prompt(content, category, sub_category)
payload = {
"system_instruction": {"parts": [{"text": SYSTEM_PROMPT}]},
"contents": [{"role": "user", "parts": [{"text": user_prompt}]}],
"generationConfig": {
"temperature": 0.1,
"responseMimeType": "application/json",
"responseSchema": RESPONSE_SCHEMA,
},
}
for attempt in range(MAX_RETRIES):
try:
data = client.call(model_name, payload)
text = data["candidates"][0]["content"]["parts"][0]["text"]
result = json.loads(text)
if result.get("category") not in CATEGORIES:
result["category"] = "該当しない（その他）"
result.setdefault("reason", "")
result.setdefault("confidence", 0.0)
return result
except urllib.error.HTTPError as e:
err_body = e.read().decode(errors="replace")
try:
err_msg = json.loads(err_body).get("error", {}).get("message", err_body)
except Exception:
err_msg = err_body
err_str = f"HTTP {e.code}: {err_msg}"
if e.code in (429, 500, 503):
wait = 2 ** attempt * 2
print(f" [リトライ {attempt+1}/{MAX_RETRIES}] {err_str[:80]}")
print(f" {wait}秒待機...")
time.sleep(wait)
continue
raise RuntimeError(err_str) from e
except (json.JSONDecodeError, KeyError) as e:
print(f" [警告] レスポンス解析エラー: {e}")
return {
"category": "該当しない（その他）",
"reason": f"解析エラー: {str(e)[:40]}",
"confidence": 0.0,
}
return {
"category": "該当しない（その他）",
"reason": f"APIエラー（{MAX_RETRIES}回リトライ失敗）",
"confidence": 0.0,
}
# 全リトライ失敗
return {
"category": "該当しない（その他）",
"reason": f"APIエラー（{MAX_RETRIES}回リトライ失敗）",
"confidence": 0.0,
}
# ---------------------------------------------------------------------------
# メイン処理
# ---------------------------------------------------------------------------
def main(
input_path: str,
output_path: str,
api_key: str,
model_name: str,
interval: float,
limit: Optional[int],
vertexai: bool,
project: Optional[str],
location: str,
) -> None:
if not os.path.exists(input_path):
print(f"エラー: 入力ファイルが見つかりません: {input_path}", file=sys.stderr)
sys.exit(1)
# 入力 CSV 読み込み
with open(input_path, encoding="utf-8-sig", newline="") as f:
rows = list(csv.DictReader(f))
if limit:
rows = rows[:limit]
total = len(rows)
print(f"入力: {input_path} ({total} 件)")
# 再開: 処理済みをスキップ
done_ids = load_done_contact_ids(output_path)
if done_ids:
print(f"既存出力を検出: {len(done_ids)} 件をスキップして再開します → {output_path}")
# 出力 CSV を追記モードで開く
input_fieldnames = list(rows[0].keys()) if rows else []
output_fieldnames = input_fieldnames + ["判定カテゴリ", "判定理由", "確信度"]
file_exists = os.path.exists(output_path) and os.path.getsize(output_path) > 0
csv_file = open(output_path, "a", newline="", encoding="utf-8-sig")
writer = csv.DictWriter(csv_file, fieldnames=output_fieldnames)
if not file_exists:
writer.writeheader()
csv_file.flush()
print(f"出力: {output_path} に随時書き込みます")
print(f"モデル: {model_name} / リクエスト間隔: {interval}s")
print()
# Gemini クライアント初期化
client = VertexAIClient(
vertexai=vertexai,
project=project,
location=location,
api_key=api_key,
)
if vertexai:
print(f"認証: Vertex AI / gcloud トークン (project={project}, location={location})")
else:
print("認証: Google AI Studio API キー")
written = 0
skipped = 0
errors = 0
for i, row in enumerate(rows, 1):
contact_id = row.get("問い合わせID", "").strip()
content = row.get("問い合わせ内容", "") or ""
category = row.get("カテゴリ", "") or ""
sub_category = row.get("サブカテゴリ", "") or ""
# 処理済みスキップ
if contact_id in done_ids:
skipped += 1
continue
if not content.strip():
# 内容が空の場合はスキップ扱い
out_row = dict(row)
out_row["判定カテゴリ"] = "該当しない（その他）"
out_row["判定理由"] = "問い合わせ内容が空"
out_row["確信度"] = 0.0
writer.writerow(out_row)
csv_file.flush()
print(f" [{i}/{total}] {contact_id} | 問い合わせ内容なし → 該当しない（その他）")
written += 1
continue
try:
result = classify_with_gemini(client, model_name, content, category, sub_category)
except Exception as e:
print(f" [{i}/{total}] {contact_id} | エラー: {e}", file=sys.stderr)
result = {
"category": "該当しない（その他）",
"reason": f"エラー: {str(e)[:40]}",
"confidence": 0.0,
}
errors += 1
out_row = dict(row)
out_row["判定カテゴリ"] = result["category"]
out_row["判定理由"] = result["reason"]
out_row["確信度"] = result["confidence"]
writer.writerow(out_row)
csv_file.flush()
written += 1
print(
f" [{i}/{total}] {contact_id} | {result['category']} | 確信度: {result['confidence']:.2f}"
)
# リクエスト間隔（最後の1件はスリープ不要）
remaining = total - i - skipped
if remaining > 0:
time.sleep(interval)
csv_file.close()
print()
print(f"完了: 書き込み {written} 件 / スキップ {skipped} 件 / エラー {errors} 件")
# 入力 CSV の順序に合わせて出力 CSV を再ソート（JOIN のため順序保証）
print("出力CSVを入力順に並び替え中...")
order = {r.get("問い合わせID", ""): i for i, r in enumerate(rows)}
with open(output_path, encoding="utf-8-sig", newline="") as f:
out_rows = list(csv.DictReader(f))
out_rows.sort(key=lambda r: order.get(r.get("問い合わせID", ""), 999999))
with open(output_path, "w", encoding="utf-8-sig", newline="") as f:
writer2 = csv.DictWriter(f, fieldnames=output_fieldnames)
writer2.writeheader()
writer2.writerows(out_rows)
print(f"出力: {output_path}")
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="問い合わせ CSV を Gemini でカテゴリ判定する"
)
parser.add_argument(
"--input",
default=DEFAULT_INPUT,
help=f"入力 CSV ファイルパス（デフォルト: {DEFAULT_INPUT}）",
)
parser.add_argument(
"--output",
default=DEFAULT_OUTPUT,
help=f"出力 CSV ファイルパス（デフォルト: {DEFAULT_OUTPUT}）",
)
parser.add_argument(
"--api-key",
default=os.environ.get("GEMINI_API_KEY", ""),
help="Gemini API キー（省略時は環境変数 GEMINI_API_KEY を使用）",
)
parser.add_argument(
"--model",
default=DEFAULT_MODEL,
help=f"Gemini モデル名（デフォルト: {DEFAULT_MODEL}）",
)
parser.add_argument(
"--interval",
type=float,
default=DEFAULT_INTERVAL,
help=f"リクエスト間隔（秒、デフォルト: {DEFAULT_INTERVAL}）",
)
parser.add_argument(
"--limit",
type=int,
default=None,
help="処理件数の上限（動作確認用。省略時は全件処理）",
)
parser.add_argument(
"--vertexai",
action="store_true",
default=False,
help="Vertex AI (gcloud ADC) を使用する（APIキー不要）",
)
parser.add_argument(
"--project",
default=os.environ.get("GOOGLE_CLOUD_PROJECT", "default-project"),
help="Vertex AI 使用時の GCP プロジェクト ID（デフォルト: default-project）",
)
parser.add_argument(
"--location",
default="us-central1",
help="Vertex AI 使用時のリージョン（デフォルト: us-central1）",
)
args = parser.parse_args()
if not args.vertexai and not args.api_key:
print(
"エラー: GEMINI_API_KEY が設定されていません。\n"
" export GEMINI_API_KEY=xxxxx\n"
"または --api-key / --vertexai オプションで指定してください。",
file=sys.stderr,
)
sys.exit(1)
main(
input_path=args.input,
output_path=args.output,
api_key=args.api_key,
model_name=args.model,
interval=args.interval,
limit=args.limit,
vertexai=args.vertexai,
project=args.project,
location=args.location,
)
システムプロンプトを用意する
- 問い合わせを毎日捌いているプロの人達からドメインを言語化してください
- ぼくは直近携わったこともあるので、自分である程度作ることはできた
10件ずつランダムサンプリングして、評価を確認する
合計10回ぐらいはやった記憶
- 問い合わせデータから10件ランダムサンプリングして、評価する
- 評価結果のデータと問い合わせ内容を並べて確認する
- 間違ってるデータは、プロンプトを修正して正しい評価に変更する
- 変更したプロンプトを使って再評価し、1で入力して得られた結果と修正したところ以外で差がないか確認する → もし結果が変わっていたら3に戻る
6回目ぐらいから意図したとおりになってきました
実行する
$ python3 scripts/classify_inquiry_category.py \
--input inquiry_data_filled.csv \
--output inquiry_classified.csv \
--vertexai \
--model gemini-2.5-flash \
--interval 1.0
絶望
⏺ M / N件（30%） 完了です。残り約N - M件、このペースだとあと3〜4時間で完了見込みです。
なんと・・・（8時間動かしたあとの状態）
ミニバッチとか、バックグラウンドで回す系ないんかな
あったよバッチAPI！（でかした！）
あったのね。。。無知は辛い
コード生成する
「Thinking 無効化（thinkingBudget: 0）でトークン節約」としています
#!/usr/bin/env python3
"""
問い合わせ CSV を Gemini Batch API でカテゴリ判定するスクリプト
通常の classify_inquiry_category.py と同じ判定ロジックを使い、
Batch API（50% オフ、最大 24 時間以内完了）で全件を非同期処理する。
処理フロー:
1. PREPARE : 入力 CSV → JSONL（1行1リクエスト）を生成
2. UPLOAD : JSONL を File API にアップロード
3. CREATE : Batch ジョブを作成（ジョブ名をファイルに保存）
4. POLL : ジョブ完了をポーリング
5. DOWNLOAD : 結果 JSONL をダウンロード
6. CONVERT : 入力 CSV とマージし、出力 CSV を生成
コスト（gemini-2.5-flash）: $0.15/1M入力 + $1.25/1M出力（通常の50%）
事前準備:
pip install google-genai
認証:
export GEMINI_API_KEY=xxxxx
実行例:
# 全フロー実行（推奨）
python3 scripts/batch_classify_inquiry_category.py \\
--input inquiry_data_filled.csv \\
--output inquiry_classified_batch.csv
# JSONL 生成のみ（内容確認してからアップロードしたい場合）
python3 scripts/batch_classify_inquiry_category.py --prepare-only
# 既存ジョブの結果取得（ジョブ作成済みで再開する場合）
python3 scripts/batch_classify_inquiry_category.py \\
--from-job batches/xxxxx \\
--input inquiry_data_filled.csv \\
--output inquiry_classified_batch.csv
# 先頭 20 件だけで動作確認
python3 scripts/batch_classify_inquiry_category.py --limit 20
"""
import argparse
import csv
import json
import os
import sys
import time
from typing import Optional
# ---------------------------------------------------------------------------
# 定数（classify_inquiry_category.py と共通）
# ---------------------------------------------------------------------------
DEFAULT_INPUT = "inquiry_data_filled.csv"
DEFAULT_OUTPUT = "inquiry_classified_batch.csv"
DEFAULT_MODEL = "gemini-2.5-flash"
DEFAULT_JSONL = "batch_requests.jsonl"
DEFAULT_JOB_FILE = "batch_job.txt" # ジョブ名を保存するファイル（再開用）
MAX_CONTENT_CHARS = 8000
CATEGORIES = [
# ...省略...
]
# REST/JSON Schema 形式（型は大文字）
RESPONSE_SCHEMA = {
"type": "OBJECT",
"properties": {
"category": {
"type": "STRING",
"enum": CATEGORIES,
"description": "判定カテゴリ",
},
"reason": {
"type": "STRING",
"description": "判定理由（50字以内）",
},
"confidence": {
"type": "NUMBER",
"description": "確信度（0.0〜1.0）",
},
},
"required": ["category", "reason", "confidence"],
}
SYSTEM_PROMPT = """\
あなたは◯◯会社のカスタマーサポート問い合わせを分類するアシスタントです。
問い合わせスレッド（顧客メールと対応履歴）を読み、以下のカテゴリのいずれか1つに分類してください。
【大前提】
...省略...
【カテゴリ定義】
...省略...
【判定ルール】
...省略...
"""
# ---------------------------------------------------------------------------
# ユーティリティ
# ---------------------------------------------------------------------------
def build_user_prompt(content: str, category: str, sub_category: str) -> str:
hint = ""
if category or sub_category:
hint = f"\n【参考情報（既存分類）】カテゴリ: {category} / サブカテゴリ: {sub_category}\n"
truncated = content[:MAX_CONTENT_CHARS]
if len(content) > MAX_CONTENT_CHARS:
truncated += "\n... (以降省略)"
return f"{hint}\n【問い合わせスレッド】\n{truncated}"
def load_done_contact_ids(output_path: str) -> set[str]:
done: set[str] = set()
if not os.path.exists(output_path):
return done
try:
with open(output_path, encoding="utf-8-sig", newline="") as f:
for row in csv.DictReader(f):
tid = row.get("問い合わせID", "").strip()
if tid:
done.add(tid)
except Exception:
pass
return done
# ---------------------------------------------------------------------------
# フェーズ 1: JSONL 生成
# ---------------------------------------------------------------------------
def phase_prepare(
input_path: str,
jsonl_path: str,
done_ids: set[str],
limit: Optional[int],
) -> int:
"""入力 CSV から Batch API 用 JSONL を生成する。生成件数を返す。"""
with open(input_path, encoding="utf-8-sig", newline="") as f:
rows = list(csv.DictReader(f))
if limit:
rows = rows[:limit]
count = 0
with open(jsonl_path, "w", encoding="utf-8") as f:
for row in rows:
contact_id = row.get("問い合わせID", "").strip()
if not contact_id or contact_id in done_ids:
continue
content = row.get("問い合わせ内容", "") or ""
category = row.get("カテゴリ", "") or ""
sub_category = row.get("サブカテゴリ", "") or ""
user_prompt = build_user_prompt(content, category, sub_category)
# JSONL 1行 = GenerateContentRequest（REST protobuf JSON 形式・camelCase）
entry = {
"key": contact_id,
"request": {
"systemInstruction": {"parts": [{"text": SYSTEM_PROMPT}]},
"contents": [{"role": "user", "parts": [{"text": user_prompt}]}],
"generationConfig": {
"temperature": 0.1,
"responseMimeType": "application/json",
"responseSchema": RESPONSE_SCHEMA,
"thinkingConfig": {"thinkingBudget": 0},
},
},
}
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
count += 1
print(f"[1/6] JSONL 生成: {count} 件 → {jsonl_path}")
return count
# ---------------------------------------------------------------------------
# フェーズ 2: File API アップロード
# ---------------------------------------------------------------------------
def phase_upload(client, jsonl_path: str) -> str:
"""JSONL を File API にアップロードし、ファイル名（files/xxxxx）を返す。"""
from google.genai import types as genai_types
file_size_mb = os.path.getsize(jsonl_path) / 1024 / 1024
print(f"[2/6] File API アップロード中... ({file_size_mb:.1f} MB)")
uploaded = client.files.upload(
file=jsonl_path,
config=genai_types.UploadFileConfig(
display_name="batch-requests",
mime_type="application/jsonl",
),
)
print(f"[2/6] アップロード完了: {uploaded.name}")
return uploaded.name
# ---------------------------------------------------------------------------
# フェーズ 3: Batch ジョブ作成
# ---------------------------------------------------------------------------
def phase_create(client, model: str, uploaded_file_name: str, job_file: str) -> str:
"""Batch ジョブを作成し、ジョブ名を返す。ジョブ名はファイルにも保存する。"""
print(f"[3/6] Batch ジョブ作成中... (モデル: {model})")
batch_job = client.batches.create(
model=model,
src=uploaded_file_name,
config={"display_name": "inquiry-classification"},
)
job_name = batch_job.name
print(f"[3/6] ジョブ作成完了: {job_name}")
# 再開用にジョブ名を保存
with open(job_file, "w") as f:
f.write(job_name)
print(f" ジョブ名を保存: {job_file}")
return job_name
# ---------------------------------------------------------------------------
# フェーズ 4: ポーリング
# ---------------------------------------------------------------------------
COMPLETED_STATES = {"JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED", "JOB_STATE_CANCELLED", "JOB_STATE_EXPIRED"}
def phase_poll(client, job_name: str, poll_interval: int = 60) -> object:
"""ジョブ完了まで poll_interval 秒ごとにポーリングし、完了した batch_job を返す。"""
print(f"[4/6] ジョブをポーリング中: {job_name}")
print(f" ({poll_interval}秒ごとに確認。Ctrl+C で中断して後で --from-job で再開できます)")
batch_job = client.batches.get(name=job_name)
while batch_job.state.name not in COMPLETED_STATES:
print(f" 状態: {batch_job.state.name} ... {poll_interval}秒待機")
time.sleep(poll_interval)
batch_job = client.batches.get(name=job_name)
print(f"[4/6] ジョブ完了: {batch_job.state.name}")
if batch_job.state.name != "JOB_STATE_SUCCEEDED":
err = getattr(batch_job, "error", None)
print(f" エラー: {err}", file=sys.stderr)
sys.exit(1)
return batch_job
# ---------------------------------------------------------------------------
# フェーズ 5: 結果ダウンロード
# ---------------------------------------------------------------------------
def phase_download(client, batch_job) -> list[dict]:
"""
結果 JSONL をダウンロードし、{contact_id: result_dict} の辞書を返す。
result_dict = {"category": ..., "reason": ..., "confidence": ..., "error": ...}
"""
print("[5/6] 結果ダウンロード中...")
result_file_name = batch_job.dest.file_name
print(f" 結果ファイル: {result_file_name}")
raw_bytes = client.files.download(file=result_file_name)
content = raw_bytes.decode("utf-8")
results: dict[str, dict] = {}
error_count = 0
for line in content.splitlines():
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
continue
key = obj.get("key", "")
if not key:
continue
if "error" in obj:
results[key] = {
"category": "該当しない（その他）",
"reason": f"APIエラー: {str(obj['error'])[:40]}",
"confidence": 0.0,
}
error_count += 1
continue
# 成功レスポンスをパース
try:
text = obj["response"]["candidates"][0]["content"]["parts"][0]["text"]
parsed = json.loads(text)
if parsed.get("category") not in CATEGORIES:
parsed["category"] = "該当しない（その他）"
parsed.setdefault("reason", "")
parsed.setdefault("confidence", 0.0)
results[key] = parsed
except (KeyError, IndexError, json.JSONDecodeError) as e:
results[key] = {
"category": "該当しない（その他）",
"reason": f"解析エラー: {str(e)[:40]}",
"confidence": 0.0,
}
error_count += 1
print(f"[5/6] ダウンロード完了: {len(results)} 件 (エラー {error_count} 件)")
return results
# ---------------------------------------------------------------------------
# フェーズ 6: 出力 CSV 生成
# ---------------------------------------------------------------------------
def phase_convert(
input_path: str,
output_path: str,
results: dict,
done_ids: set[str],
limit: Optional[int],
) -> None:
"""入力 CSV と Batch 結果をマージし、出力 CSV を生成する。"""
print("[6/6] 出力 CSV 生成中...")
with open(input_path, encoding="utf-8-sig", newline="") as f:
rows = list(csv.DictReader(f))
if limit:
rows = rows[:limit]
input_fieldnames = list(rows[0].keys()) if rows else []
output_fieldnames = input_fieldnames + ["判定カテゴリ", "判定理由", "確信度"]
# 既存出力 CSV を読み込んでマージ（再開時）
existing_rows: dict[str, dict] = {}
if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
with open(output_path, encoding="utf-8-sig", newline="") as f:
for row in csv.DictReader(f):
tid = row.get("問い合わせID", "").strip()
if tid:
existing_rows[tid] = row
written = 0
skipped = 0
missing = 0
with open(output_path, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=output_fieldnames)
writer.writeheader()
for row in rows:
contact_id = row.get("問い合わせID", "").strip()
out_row = dict(row)
if contact_id in existing_rows:
# 既存の判定結果を使用
existing = existing_rows[contact_id]
out_row["判定カテゴリ"] = existing.get("判定カテゴリ", "該当しない（その他）")
out_row["判定理由"] = existing.get("判定理由", "")
out_row["確信度"] = existing.get("確信度", 0.0)
skipped += 1
elif contact_id in results:
r = results[contact_id]
out_row["判定カテゴリ"] = r["category"]
out_row["判定理由"] = r["reason"]
out_row["確信度"] = r["confidence"]
written += 1
else:
# Batch 結果にない問い合わせ（JSONL に含まれなかった = 空内容など）
out_row["判定カテゴリ"] = "該当しない（その他）"
out_row["判定理由"] = "バッチ結果なし"
out_row["確信度"] = 0.0
missing += 1
writer.writerow(out_row)
print(f"[6/6] 出力完了: {output_path}")
print(f" 新規: {written} 件 / スキップ（既存）: {skipped} 件 / 結果なし: {missing} 件")
# ---------------------------------------------------------------------------
# メイン
# ---------------------------------------------------------------------------
def main(
input_path: str,
output_path: str,
api_key: str,
model: str,
jsonl_path: str,
job_file: str,
limit: Optional[int],
prepare_only: bool,
from_job: Optional[str],
poll_interval: int,
) -> None:
if not os.path.exists(input_path):
print(f"エラー: 入力ファイルが見つかりません: {input_path}", file=sys.stderr)
sys.exit(1)
done_ids = load_done_contact_ids(output_path)
if done_ids:
print(f"既存出力を検出: {len(done_ids)} 件はスキップします")
# --from-job 指定時はジョブのポーリングから開始
if from_job:
from google import genai
client = genai.Client(api_key=api_key)
print(f"既存ジョブから再開: {from_job}")
batch_job = phase_poll(client, from_job, poll_interval)
results = phase_download(client, batch_job)
phase_convert(input_path, output_path, results, done_ids, limit)
return
# Phase 1: JSONL 生成
count = phase_prepare(input_path, jsonl_path, done_ids, limit)
if count == 0:
print("処理対象がありません（全件処理済み）。")
return
if prepare_only:
print(f"--prepare-only: JSONL を生成しました → {jsonl_path}")
print(f"内容確認後、以下を実行してバッチ処理を開始してください:")
print(f" python3 scripts/batch_classify_inquiry_category.py \\")
print(f" --input {input_path} --output {output_path}")
return
# Phase 2〜6: SDK を使った処理
from google import genai
client = genai.Client(api_key=api_key)
uploaded_file_name = phase_upload(client, jsonl_path)
job_name = phase_create(client, model, uploaded_file_name, job_file)
batch_job = phase_poll(client, job_name, poll_interval)
results = phase_download(client, batch_job)
phase_convert(input_path, output_path, results, done_ids, limit)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="問い合わせ CSV を Gemini Batch API でカテゴリ判定する"
)
parser.add_argument(
"--input",
default=DEFAULT_INPUT,
help=f"入力 CSV（デフォルト: {DEFAULT_INPUT}）",
)
parser.add_argument(
"--output",
default=DEFAULT_OUTPUT,
help=f"出力 CSV（デフォルト: {DEFAULT_OUTPUT}）",
)
parser.add_argument(
"--api-key",
default=os.environ.get("GEMINI_API_KEY", ""),
help="Gemini API キー（省略時は環境変数 GEMINI_API_KEY を使用）",
)
parser.add_argument(
"--model",
default=DEFAULT_MODEL,
help=f"Gemini モデル名（デフォルト: {DEFAULT_MODEL}）",
)
parser.add_argument(
"--jsonl-file",
default=DEFAULT_JSONL,
dest="jsonl_path",
help=f"JSONL ファイルパス（デフォルト: {DEFAULT_JSONL}）",
)
parser.add_argument(
"--job-file",
default=DEFAULT_JOB_FILE,
dest="job_file",
help=f"ジョブ名保存ファイル（デフォルト: {DEFAULT_JOB_FILE}）",
)
parser.add_argument(
"--limit",
type=int,
default=None,
help="処理件数の上限（動作確認用）",
)
parser.add_argument(
"--prepare-only",
action="store_true",
default=False,
help="JSONL 生成のみ行い、アップロード・ジョブ作成はしない",
)
parser.add_argument(
"--from-job",
default=None,
metavar="JOB_NAME",
help="既存ジョブ名（batches/xxxxx）からポーリング・結果取得を再開",
)
parser.add_argument(
"--poll-interval",
type=int,
default=60,
help="ポーリング間隔（秒、デフォルト: 60）",
)
args = parser.parse_args()
if not args.prepare_only and not args.api_key:
print(
"エラー: GEMINI_API_KEY が設定されていません。\n"
" export GEMINI_API_KEY=xxxxx\n"
"または --api-key オプションで指定してください。",
file=sys.stderr,
)
sys.exit(1)
main(
input_path=args.input,
output_path=args.output,
api_key=args.api_key,
model=args.model,
jsonl_path=args.jsonl_path,
job_file=args.job_file,
limit=args.limit,
prepare_only=args.prepare_only,
from_job=args.from_job,
poll_interval=args.poll_interval,
)
実行
$ $ python3 scripts/batch_classify_inquiry_category.py \
--input inquiry_data_filled.csv \
--output inquiry_classified_batch.csv
[1/6] JSONL 生成: N 件 → batch_requests.jsonl
[2/6] File API アップロード中... (63.4 MB)
[2/6] アップロード完了: files/abcdefghijklmn
[3/6] Batch ジョブ作成中... (モデル: gemini-2.5-flash)
[3/6] ジョブ作成完了: batches/abcdefghijklmn
ジョブ名を保存: batch_job.txt
[4/6] ジョブをポーリング中: batches/abcdefghijklmn
(60秒ごとに確認。Ctrl+C で中断して後で --from-job で再開できます)
状態: JOB_STATE_RUNNING ... 60秒待機
状態: JOB_STATE_RUNNING ... 60秒待機
状態: JOB_STATE_RUNNING ... 60秒待機
[4/6] ジョブ完了: JOB_STATE_SUCCEEDED
[5/6] 結果ダウンロード中...
結果ファイル: files/batch-abcdefghijklmn
[5/6] ダウンロード完了: N 件 (エラー 1 件)
[6/6] 出力 CSV 生成中...
[6/6] 出力完了: inquiry_classified_batch.csv
新規: N 件 / スキップ（既存）: 0 件 / 結果なし: 0 件
はやーい！！
気になるお値段
- Vertex AI（逐次処理の方） 1,301円
- Gemini API（バッチの方） 297円
バッチ最高でした
学び
- Gemini APIを叩くだけなら、API Keyを用意しなくてもよい
- gcloud cliで認証(gcloud auth)したアカウントがあれば、Vertex AI経由で認証して叩けた
- IAMからアカウントに「Agent Platform ユーザー」の権限を付与する必要はある
- Batch APIの場合を2通りある
- API Keyのほうが簡単そうだったから、 API Keyにした
雑感
- その前の前処理（データ整形とか正規化とか）で疲れた...
- Gemmaでもためしてみて、精度あんまりかわらないのであればGemmaでもいいなっておもった