取り組みの背景
Gemini 2.5 Proのマルチモーダル機能により、単なるテキスト抽出を超えた高度なドキュメント処理が可能になりましました。PDFのレイアウト認識、表の構造化抽出、複数ドキュメント間の関連分析など、エンタープライズレベルのドキュメント処理を実装できます。
Geminiのドキュメント理解能力
従来のOCRとの違い
従来のOCR技術は文字認識に特化していますが、Geminiは画像コンテンツを理解できます：
- レイアウト理解: セクション、カラム、インデント構造の認識
- 表構造認識: 複雑な行列レイアウトを正しく解析
- グラフ解釈: チャートや図表から数値と傾向を抽出
- コンテキスト理解: テキストの意味と文脈を把握
Gemini APIは、金融ドキュメント、法務文書、医療記録など複雑な構造化ドキュメントに最適です。
File APIを使用したPDF処理
アップロードと処理ステータス確認
import google.generativeai as genai
import time
# クライアント初期化
client = genai.Client(api_key="YOUR_API_KEY")
def upload_and_process_pdf(file_path: str):
"""
PDFをアップロードして処理待機
"""
print(f"ファイルをアップロード中: {file_path}")
# ファイルアップロード
with open(file_path, "rb") as f:
pdf_file = client.files.upload(
file=f,
mime_type="application/pdf"
)
file_uri = pdf_file.uri
print(f"アップロード完了: {file_uri}")
# 処理ステータスの確認
for attempt in range(30):
file = client.files.get(pdf_file.name)
if file.state.name == "ACTIVE":
print(f"ファイル処理完了")
return file_uri
elif file.state.name == "FAILED":
raise Exception(f"ファイル処理失敗: {file.state.name}")
print(f"処理中... ({attempt + 1}/30)")
time.sleep(2)
raise TimeoutError("ファイル処理がタイムアウト")
# 使用例
pdf_uri = upload_and_process_pdf("contract.pdf")PDF全文解析とテキスト抽出
構造を保持した全文抽出
def analyze_full_pdf_content(pdf_uri: str):
"""
PDFの全文を構造を保持したまま抽出
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
このPDFドキュメントの全文を以下の構造で抽出してください:
# ドキュメント構造化解析結果
## メタデータ
- タイトル:
- 発行日:
- ページ数:
## セクション別内容
各セクションを階層構造で整理してください
### [セクション1]
内容...
### [セクション2]
内容...
## 重要な用語・固有名詞
- 用語1: 説明
- 用語2: 説明
## 数値データ
- 項目1: 値
- 項目2: 値
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": pdf_uri
}
])
return response.text
# 実装例
full_content = analyze_full_pdf_content(pdf_uri)
print(full_content)セクション識別
def identify_pdf_sections(pdf_uri: str):
"""
PDFのセクション構造を自動識別
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
このドキュメントのセクション構造をJSON形式で抽出してください:
{
"sections": [
{
"section_id": 1,
"title": "セクションタイトル",
"page_numbers": [1, 2],
"subsections": [
{
"title": "サブセクション",
"content_summary": "要約"
}
]
}
]
}
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": pdf_uri
}
])
import json
return json.loads(response.text)表の構造化抽出
テーブルをJSONに変換
def extract_tables_to_json(pdf_uri: str):
"""
PDFから全ての表を抽出してJSON形式で返す
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
このPDFに含まれるすべての表を以下のJSON形式で抽出してください:
{
"tables": [
{
"table_id": 1,
"title": "表のタイトル",
"location": "ページX",
"headers": ["列1", "列2", "列3"],
"rows": [
["データ11", "データ12", "データ13"],
["データ21", "データ22", "データ23"]
],
"notes": "注記があれば記入"
}
]
}
注意:
- 空白セルは null で表現
- 結合セルは適切に処理
- 単位情報は値に含める
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": pdf_uri
}
])
import json
return json.loads(response.text)
# 使用例
tables = extract_tables_to_json(pdf_uri)
for table in tables["tables"]:
print(f"表: {table['title']}")
print(f"位置: {table['location']}")
for row in table["rows"]:
print(row)複雑な表の処理
def extract_complex_tables_with_context(pdf_uri: str):
"""
複数列ヘッダー・結合セルのある複雑な表を処理
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
このドキュメントの全ての表を分析してください。
特に以下に注意:
1. 多段ヘッダー構造を正確にキャプチャ
2. セルの結合・分割を記録
3. 単位と補足情報を含める
4. サブトータル・合計行を識別
JSON形式で返してください:
{
"tables": [
{
"table_id": 1,
"structure": {
"header_rows": 2,
"header_columns": 1,
"data_rows": 10
},
"columns": [
{
"name": "列名",
"type": "string|number|date",
"unit": "単位"
}
],
"data": [...]
}
]
}
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": pdf_uri
}
])
import json
return json.loads(response.text)グラフ・図表の解釈
チャートから数値抽出
def analyze_charts_and_graphs(pdf_uri: str):
"""
PDFに含まれるグラフ・チャートから数値データを抽出
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
このドキュメント内のすべてのグラフ・チャート・図表を分析して、
以下のJSON形式で出力してください:
{
"charts": [
{
"chart_id": 1,
"type": "bar|line|pie|scatter",
"title": "グラフタイトル",
"x_axis": {
"label": "X軸ラベル",
"values": ["値1", "値2"]
},
"y_axis": {
"label": "Y軸ラベル",
"scale": "linear|logarithmic"
},
"series": [
{
"name": "系列名",
"data": [100, 150, 120]
}
],
"insights": "グラフから読み取れる重要な傾向・発見"
}
]
}
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": pdf_uri
}
])
import json
return json.loads(response.text)複数ドキュメントの横断比較
マルチドキュメント分析
def compare_multiple_documents(pdf_uris: list):
"""
複数のPDFドキュメントを比較分析
"""
model = genai.GenerativeModel("gemini-2.5-pro")
# プロンプト構築
prompt = """
以下の複数のドキュメントを比較分析して、共通点・相違点をまとめてください:
JSON形式で以下を出力:
{
"comparison": {
"document_count": N,
"similarities": [
{
"aspect": "比較対象",
"details": "共通点の説明"
}
],
"differences": [
{
"aspect": "比較対象",
"document_1": "値1",
"document_2": "値2",
"significance": "差異の重要性"
}
],
"summary": "総合評価"
}
}
"""
# コンテンツ組み立て
content = [prompt]
for pdf_uri in pdf_uris:
content.append({
"mime_type": "application/pdf",
"data": pdf_uri
})
response = model.generate_content(content)
import json
return json.loads(response.text)
# 使用例
comparison_result = compare_multiple_documents([pdf_uri_1, pdf_uri_2])
print(comparison_result)契約書自動審査
契約条件の抽出
def review_contract_automatically(contract_uri: str):
"""
契約書から主要条件を自動抽出・分析
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
この契約書を法的視点から分析してください。
以下をJSON形式で返してください:
{
"contract_analysis": {
"contract_type": "契約の種類",
"parties": [
{
"name": "当事者名",
"role": "役割"
}
],
"key_terms": {
"effective_date": "発効日",
"duration": "期間",
"termination_clause": "解除条件"
},
"financial_terms": {
"payment_amount": "金額",
"currency": "通貨",
"payment_schedule": "支払いスケジュール",
"penalties": "ペナルティ条項"
},
"obligations": [
{
"party": "当事者",
"obligation": "義務内容"
}
],
"risk_factors": [
{
"risk": "リスク",
"severity": "high|medium|low",
"mitigation": "軽減方法"
}
],
"compliance_requirements": [
"コンプライアンス要件"
],
"recommendations": "審査者からの推奨"
}
}
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": contract_uri
}
])
import json
return json.loads(response.text)
# 使用例
contract_review = review_contract_automatically(contract_uri)
print(f"リスク要因: {contract_review['contract_analysis']['risk_factors']}")違反条項の検出
def detect_unfavorable_clauses(contract_uri: str, company_context: str):
"""
当社に不利な条項を自動検出
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = f"""
当社の背景: {company_context}
この契約書から、当社にとって不利または懸念される条項を抽出してください:
{{
"unfavorable_clauses": [
{{
"clause_location": "第X条第Y項",
"clause_text": "条項の原文",
"concern": "懸念点",
"impact": "当社への影響",
"suggested_revision": "推奨される修正内容"
}}
],
"negotiation_priorities": [
{{
"priority": 1,
"clause": "修正対象",
"rationale": "修正理由"
}}
]
}}
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": contract_uri
}
])
import json
return json.loads(response.text)請求書処理
請求書からの構造化データ抽出
def process_invoice(invoice_uri: str):
"""
請求書から財務データを構造化形式で抽出
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
この請求書から以下の情報をJSON形式で抽出してください:
{
"invoice_metadata": {
"invoice_number": "請求書番号",
"invoice_date": "請求日",
"due_date": "期日",
"currency": "通貨"
},
"parties": {
"vendor": {
"name": "ベンダー名",
"address": "住所",
"contact": "連絡先"
},
"customer": {
"name": "顧客名",
"address": "住所"
}
},
"line_items": [
{
"description": "項目説明",
"quantity": 10,
"unit_price": 100.00,
"line_total": 1000.00
}
],
"summary": {
"subtotal": 小計,
"tax_amount": 税金,
"tax_rate": "税率",
"total": 合計,
"payment_terms": "支払い条件"
},
"payment_information": {
"bank_account": "銀行口座",
"payment_method": "支払い方法"
}
}
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": invoice_uri
}
])
import json
return json.loads(response.text)
# 使用例
invoice_data = process_invoice(invoice_uri)
print(f"合計金額: {invoice_data['summary']['total']}")決算書分析
財務指標の自動抽出
def analyze_financial_statement(statement_uri: str):
"""
決算書・財務報告書から主要指標を抽出
"""
model = genai.GenerativeModel("gemini-2.5-pro")
prompt = """
この財務報告書を分析して、以下をJSON形式で返してください:
{
"statement_info": {
"company_name": "企業名",
"fiscal_year": "会計年度",
"statement_type": "balance_sheet|income_statement|cash_flow"
},
"balance_sheet": {
"assets": {
"current_assets": 金額,
"non_current_assets": 金額,
"total_assets": 金額
},
"liabilities": {
"current_liabilities": 金額,
"non_current_liabilities": 金額,
"total_liabilities": 金額
},
"equity": {
"total_equity": 金額
}
},
"income_statement": {
"revenue": 売上,
"cost_of_revenue": 原価,
"gross_profit": 粗利益,
"operating_expenses": 営業費用,
"operating_income": 営業利益,
"net_income": 純利益
},
"key_ratios": {
"current_ratio": 流動比率,
"debt_to_equity": 負債比率,
"profit_margin": "利益率",
"return_on_assets": "資産利益率"
},
"year_over_year_analysis": {
"revenue_change": "売上変化率",
"profit_trend": "利益トレンド",
"key_observations": "重要な観察"
}
}
"""
response = model.generate_content([
prompt,
{
"mime_type": "application/pdf",
"data": statement_uri
}
])
import json
return json.loads(response.text)Context Cachingの活用
ドキュメントのキャッシング
def process_document_with_caching(pdf_uri: str, queries: list):
"""
同じドキュメントに対し複数クエリを実行する場合のキャッシング
"""
model = genai.GenerativeModel("gemini-2.5-pro")
# 初回リクエスト - ドキュメントをキャッシュ
cached_document = {
"mime_type": "application/pdf",
"data": pdf_uri
}
results = []
for i, query in enumerate(queries):
print(f"クエリ {i + 1}/{len(queries)}")
content = [
query,
cached_document
]
# キャッシング指定（2回目以降）
if i > 0:
# キャッシング対応のヘッダー設定
# （次のバージョンのAPIで対応予定）
pass
response = model.generate_content(content)
results.append({
"query": query,
"response": response.text
})
return results
# 使用例
queries = [
"このドキュメントの主要な発見は何か?",
"財務データの主な傾向は?",
"リスク要因は何か?"
]
cached_results = process_document_with_caching(pdf_uri, queries)Context cachingはAPI消費量を削減し、コストを最適化します。頻繁にアクセスするドキュメントに活用してください。
バッチ処理
大量ドキュメントの効率的処理
from concurrent.futures import ThreadPoolExecutor
import time
class BatchDocumentProcessor:
def __init__(self, max_workers=5):
self.max_workers = max_workers
self.results = []
def process_batch(self, document_uris: list, task_type: str):
"""
複数ドキュメントを並列処理
"""
def process_single_document(uri, task_type):
if task_type == "contract_review":
return review_contract_automatically(uri)
elif task_type == "invoice":
return process_invoice(uri)
elif task_type == "financial":
return analyze_financial_statement(uri)
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = [
executor.submit(process_single_document, uri, task_type)
for uri in document_uris
]
for i, future in enumerate(futures):
try:
result = future.result(timeout=300)
self.results.append({
"document_index": i,
"status": "success",
"result": result
})
print(f"処理完了: ドキュメント {i + 1}/{len(document_uris)}")
except Exception as e:
self.results.append({
"document_index": i,
"status": "failed",
"error": str(e)
})
return self.results
# 使用例
processor = BatchDocumentProcessor(max_workers=5)
results = processor.process_batch(pdf_uris, "contract_review")
success_count = sum(1 for r in results if r["status"] == "success")
print(f"成功: {success_count}/{len(pdf_uris)}")コスト管理
class DocumentProcessingCostManager:
def __init__(self):
# Gemini APIの料金（2026年3月時点）
self.input_cost_per_million = 0.075 # ドル
self.output_cost_per_million = 0.3
def estimate_cost(self, num_documents, avg_pages_per_doc, queries_per_doc=1):
"""
ドキュメント処理のコスト見積もり
"""
# ページ数をトークンに変換（1ページ≈2000トークン）
avg_tokens_per_page = 2000
total_input_tokens = (
num_documents * avg_pages_per_doc * avg_tokens_per_page *
queries_per_doc
)
# 出力トークン推定（入力の20%程度）
total_output_tokens = int(total_input_tokens * 0.2)
input_cost = (total_input_tokens / 1_000_000) * self.input_cost_per_million
output_cost = (total_output_tokens / 1_000_000) * self.output_cost_per_million
total_cost = input_cost + output_cost
return {
"num_documents": num_documents,
"estimated_input_tokens": total_input_tokens,
"estimated_output_tokens": total_output_tokens,
"input_cost": f"${input_cost:.2f}",
"output_cost": f"${output_cost:.2f}",
"total_cost": f"${total_cost:.2f}",
"cost_per_document": f"${total_cost / num_documents:.4f}"
}
# 使用例
cost_mgr = DocumentProcessingCostManager()
estimate = cost_mgr.estimate_cost(
num_documents=1000,
avg_pages_per_doc=10,
queries_per_doc=3
)
print(f"推定コスト: {estimate['total_cost']}")
print(f"ドキュメント当たり: {estimate['cost_per_document']}")ドキュメント処理パターン集
パターン1: 請求書の自動仕分け
def classify_and_process_invoices(invoice_uris: list):
"""
複数の請求書を分類してから処理
"""
model = genai.GenerativeModel("gemini-2.5-pro")
classified = {"vendors": {}, "failed": []}
for uri in invoice_uris:
# 分類
classify_prompt = "この請求書の発行者（ベンダー名）と金額を抽出"
classify_response = model.generate_content([
classify_prompt,
{"mime_type": "application/pdf", "data": uri}
])
vendor_name = extract_vendor_name(classify_response.text)
# ベンダー別に分類
if vendor_name not in classified["vendors"]:
classified["vendors"][vendor_name] = []
classified["vendors"][vendor_name].append(uri)
# ベンダー別に詳細処理
for vendor, uris in classified["vendors"].items():
print(f"\nベンダー: {vendor} ({len(uris)}件)")
for uri in uris:
invoice_data = process_invoice(uri)
print(f" 金額: {invoice_data['summary']['total']}")
return classifiedまとめ
Gemini APIのマルチモーダル機能により、エンタープライズグレードのドキュメント処理が実現できます。キーポイント：
- File API: 大型PDFの効率的なアップロードと処理
- 構造化抽出: responseSchemaでJSON形式を強制
- 複数ドキュメント分析: 横断比較や関連分析
- バッチ処理: 効率的な大量処理
- Context Caching: コスト削減と高速化
これらの技術を組み合わせることで、高精度で低コストなドキュメント処理システムを構築できます。
個人開発の現場から — 抽出は「外れ方」を先に決める
個人開発で Dolice Labs の運用を自動化している私自身、ドキュメント処理でいちばん時間を取られたのは精度そのものより「抽出に失敗したときの扱い」でした。契約書の表が崩れて読めないとき、黙って空欄を返すか人間に回すかで、後工程の信頼性がまるで変わります。
私は重要なフィールドには必ず信頼度スコアを持たせ、閾値を下回ったら自動処理を止めて確認キューへ送るようにしています。全自動を目指すより、危ういものだけ人に渡す設計のほうが、実務では安心して任せられると感じています。
PDF は特に、レイアウト崩れやスキャン品質のばらつきが結果を左右します。私は処理前にページ単位で簡単な品質チェックを挟み、明らかに読めない原稿は早い段階で弾くようにしています。入口で仕分けておくと、後段のモデル呼び出しのコストも安定します。
テンプレートが決まっている帳票なら、モデル任せにせず座標ベースの抽出と併用するのも有効です。私は定型部分はルールで、揺れる部分だけモデルで、と役割を分けるようにしています。すべてをモデルに委ねないほうが、精度もコストも安定します。