PDFから構造化データ抽出の最適解はGemini 2.0 Flashだった！
PDFから構造化データ抽出の課題は、文字の間に謎の空白があったり、2カラムの段組のテキストを抽出すると行の順序が乱れたりしますよね～マジで！
他には見出しを抽出するには？とかフッターとヘッダーを除外するには？とか、そもそもPDFのデータ構造を理解していないと、どうしてもデータの品質が低くなってしまいますよね。
そんな課題を解決してくれたのが、まさかのGemini 2.0 Flash！
だった、という記事です。
1. RAG構築あるある！PDF解析の壁にぶち当たる…
私も最初は「PDFからテキストを抽出するのは簡単やろ」と思っていました。なぜならDifyとかのGUIツールを使ったら簡単にできることを知っていたからです。
しかし実際に始めてみると、予想もしなかった問題が次々と発生。まず試したのはPythonライブラリのmarkitdownとunstructured。どちらも同じような課題に直面しました。
文字間に謎のスペースが挿入される
2カラム構成のPDFで行の順序が完全に乱れる
見出しの抽出に一貫性がない
処理は早いものの、精度が低い。今回は企業のESG報告書などのRAG構築のハッカソンでしたので、画像やグラフ、表が多くそれらを踏まえて、テキストの抽出精度を高める必要がありました。
後で知ったのですが、PDFには見た目が同じようなグラフでも、データが違うものがあるのです。
例えば、同じようなグラフでも画像データだったり、表データだったりします。
2. PDFからの構造化データ抽出にはまる…既存ツールもイマイチ
次に試したのがAdobe PDF Servicesです。APIの呼び出し用SDKが用意されていて、月500ページまで無料で使えるのは魅力的でした。画像とテキストの両方を抽出でき、さらにグラフの情報をCSVに変換できる機能も！
これまでは、テキストデータのみの抽出でしたがグラフのデータをCSVに変換できたのは感動しました。
しかし、ここでも問題が：
抽出精度が期待ほど高くない
テキストが画像として誤って抽出されることも
よく見るとグラフの情報をCSVに誤った値を変換している
Azure Notebooksのドキュメントインテリジェンスも試してみましたが、途中で断念、やっぱCursorみたいにAIが横についてくれないと効率悪い。もうちょっとレベル上がってからにします。ということでパス。
3. 考えられる手を打ったのでLLMのVision機能にかける
まず、始めに最も性能が高いであろうClaude 3.5 Sonnetを試すことに、API代が高いのが気になりますがやらないと始まらないので、試してみることにしました。
まずは、PDFを1ページずつ分割して、1ページずつをClaudeにわたす作戦です。
PDFを1ページずつ分割する参考コードはこちら。
import os
import PyPDF2
def split_pdf(input_pdf):
"""
PDFファイルを1ページずつ分割して保存する
Args:
input_pdf: 入力PDFファイルのパス
"""
# 出力ファイル名のベース部分を取得（拡張子除去）
base_name = input_pdf.rsplit('.', 1)[0]
# PDFファイルを読み込む
pdf_reader = PyPDF2.PdfReader(input_pdf)
# ページごとに分割して保存
for i, page in enumerate(pdf_reader.pages, start=1):
# 新しいPDFライターを作成
pdf_writer = PyPDF2.PdfWriter()
pdf_writer.add_page(page)
# 分割したPDFを保存
output_pdf = f"{base_name}-{i}.pdf"
with open(output_pdf, "wb") as out_file:
pdf_writer.write(out_file)
print(f"ページ {i} を {output_pdf} として保存しました。")
if __name__ == "__main__":
# 分割したいPDFファイルのパス
input_pdf = "ファイルのパスを指定してください"
# PDFの分割を実行
split_pdf(input_pdf)
print("PDFファイルの分割が完了しました。")次は、1ページずつをClaudeにわたすサンプルコードです。
"""
PDFProcessor: ClaudeのVision APIを使用してPDFを処理し、構造化データを生成する汎用クラス
"""
import os
import base64
import json
import logging
from pathlib import Path
import anthropic
from typing import Dict, List, Tuple
class PDFProcessor:
def __init__(self, api_key: str = None):
"""
PDFProcessorの初期化
Args:
api_key: Anthropic APIキー（未指定の場合は環境変数から取得）
"""
# ロガーの設定
self.logger = logging.getLogger(__name__)
# APIキーの設定
self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY")
if not self.api_key:
raise ValueError("ANTHROPIC_API_KEY is required")
self.client = anthropic.Client(api_key=self.api_key)
self.model = "claude-3-5-sonnet-20241022"
# 出力ディレクトリの設定
self.output_dir = Path("output")
self.output_dir.mkdir(parents=True, exist_ok=True)
def _encode_pdf(self, pdf_path: str) -> str:
"""PDFファイルをbase64エンコード"""
with open(pdf_path, "rb") as f:
return base64.b64encode(f.read()).decode()
def process_pdf(self, pdf_path: str, system_prompt: str = None) -> Tuple[str, Dict]:
"""
PDFを処理して構造化データを生成
Args:
pdf_path: 処理するPDFファイルのパス
system_prompt: システムプロンプト（指定がない場合はデフォルトプロンプトを使用）
Returns:
Tuple[str, Dict]: Markdown形式のテキストとJSONメタデータ
"""
try:
# デフォルトのプロンプト設定
if not system_prompt:
system_prompt = """
以下のPDFを解析し、構造化データを生成してください。
出力形式:
1. 構造化データ（JSON）:
- metadata: ドキュメントの基本情報
- content: 抽出されたコンテンツ
- extraction_info: 抽出の品質情報
2. マークダウン形式の補足情報
注意事項:
- 重要な情報を漏れなく抽出
- 構造を維持しながら整理
- 数値データは正確に抽出
"""
# PDFをbase64エンコード
pdf_data = self._encode_pdf(pdf_path)
# APIリクエストの作成と実行
response = self.client.messages.create(
model=self.model,
max_tokens=4096,
temperature=0,
system=system_prompt,
messages=[
{
"role": "user",
"content": [
{
"type": "document",
"source": {
"type": "base64",
"media_type": "application/pdf",
"data": pdf_data
}
}
]
}
]
)
# レスポンスの解析
markdown_content, metadata = self._parse_response(response.content[0].text)
metadata["file_name"] = Path(pdf_path).name
return markdown_content, metadata
except Exception as e:
self.logger.error(f"Error processing PDF {pdf_path}: {e}")
return "", self._create_error_metadata(str(e), pdf_path)
def _parse_response(self, response: str) -> Tuple[str, Dict]:
"""
APIレスポンスを解析し、マークダウンコンテンツとメタデータを抽出
"""
try:
# structured_dataタグ内のJSONを抽出
json_start = response.find("<structured_data>")
json_end = response.find("</structured_data>")
if json_start == -1 or json_end == -1:
return self._extract_json_fallback(response)
json_str = response[json_start + len("<structured_data>"):json_end].strip()
metadata = json.loads(json_str)
# markdown_supplementタグ内のマークダウンを抽出
md_start = response.find("<markdown_supplement>")
md_end = response.find("</markdown_supplement>")
if md_start != -1 and md_end != -1:
markdown_content = response[md_start + len("<markdown_supplement>"):md_end].strip()
else:
markdown_content = response.replace(f"<structured_data>{json_str}</structured_data>", "").strip()
return markdown_content, metadata
except Exception as e:
self.logger.error(f"Error parsing response: {e}")
return self._extract_json_fallback(response)
def _extract_json_fallback(self, response: str) -> Tuple[str, Dict]:
"""
代替的なJSON抽出方法
"""
try:
# JSONブロックを探す
if "```json" in response:
json_part = response.split("```json")[1].split("```")[0].strip()
metadata = json.loads(json_part)
markdown_content = response.replace(f"```json{json_part}```", "").strip()
return markdown_content, metadata
except:
pass
# 最小限のメタデータを返す
return response.strip(), self._create_minimal_metadata()
def _create_minimal_metadata(self) -> Dict:
"""最小限の有効なメタデータを生成"""
return {
"metadata": {
"document_type": "unknown",
"language": "ja"
},
"content": {
"paragraphs": []
},
"extraction_info": {
"confidence_score": 0.0,
"extraction_notes": ["Fallback to minimal metadata"]
}
}
def _create_error_metadata(self, error_message: str, pdf_path: str) -> Dict:
"""エラー時のメタデータを生成"""
return {
"error": {
"type": "processing_error",
"message": error_message,
"file_name": Path(pdf_path).name
}
}
def save_results(self, pdf_path: str, markdown_content: str, metadata: Dict):
"""
処理結果を保存
Args:
pdf_path: 処理したPDFファイルのパス
markdown_content: マークダウン形式のテキスト
metadata: 構造化メタデータ
"""
pdf_name = Path(pdf_path).stem
# JSONファイルとして保存
json_path = self.output_dir / f"{pdf_name}.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
# マークダウンファイルとして保存
md_path = self.output_dir / f"{pdf_name}.md"
with open(md_path, "w", encoding="utf-8") as f:
f.write(markdown_content)
self.logger.info(f"Saved results to {json_path} and {md_path}")
def main():
"""使用例"""
# APIキーの設定
api_key = "your-api-key" # または環境変数 ANTHROPIC_API_KEY を設定
# PDFProcessorのインスタンス化
processor = PDFProcessor(api_key)
# 処理するPDFファイル
pdf_path = "example.pdf"
# カスタムプロンプト（オプション）
custom_prompt = """
このPDFから以下の情報を抽出してください：
1. 文書のタイトルと概要
2. 主要な見出しと内容
3. 重要な数値データ
4. 図表の説明
結果は構造化データ（JSON）とマークダウン形式で出力してください。
"""
try:
# PDF処理の実行
markdown_content, metadata = processor.process_pdf(pdf_path, custom_prompt)
# 結果の保存
processor.save_results(pdf_path, markdown_content, metadata)
print("処理が完了しました")
except Exception as e:
print(f"エラーが発生しました: {e}")
if __name__ == "__main__":
main()実際に使ったコードをAIにサンプル化してもらったので、実際に動くかは保障できませんが、あくまでも参考程度にしてください。
実際にはpromt.mdを用意して、プロンプトキャッシュを作成しました。
ただ、論文でプロンプトキャッシュの脆弱性を指摘されているので、注意が必要です。
やはり、LLMのほうが精度が高いです。PDF特有の問題である、画像データや表のデータの扱いのバラツキを一旦すべて画像データに変換してから解析するので、ライブラリでは解決できなかった課題をクリアしました！
しかし、Claude 3.5 Sonnetのさぼり癖？なのか、情報が多いと70~80%くらいの抽出量でした。
改善策として同じページを2回程渡すと、一回で取得できなかった情報を統合してくれるので、精度が高くなります。
しかし、APIのコストを考えると現実的じゃないので、次はGemini 2.0 Flashを試すことにします。
成果
文字間の不要なスペース問題が解消
2カラム構成でも正確な順序で抽出
グラフや表の数値データもほぼ正確に抽出
4. Claude 3.5 Sonnet vs Gemini 2.0 Flashを比較
Claude 3.5 Sonnetのビジョン機能の精度は約70~80%程度で、決して悪くはありませんが、API利用のコストが気になるレベル。
一方、Gemini 2.0 Flashは：
より高い精度
低コスト
高速な処理
という3つの要素をバランスよく備えていました。
具体的には同じプロンプトでも、Claude 3.5 Sonnetが出力しなかった情報をGemini 2.0 Flashが出力してくれました！
しかも低コストです。
入力: 100 万トークンあたり $0.10（テキスト / 画像 / 動画）
出力: 100 万トークンあたり $0.40（テキスト）
成果
より高い精度
低コスト
高速な処理
5. RAG構築のPDF解析はGemini 2.0 Flashで決まり！(2025年2月時点)
数々のツールを試した結果、現時点でのPDF構造化データ抽出の最適解は間違いなくGemini 2.0 Flash。特に：
高精度なテキスト抽出
グラフや表の数値データ化
コストパフォーマンスの高さ
これらの要素が、RAG構築における重要な課題を解決してくれます。
まだ試していない方は、ぜひGemini 2.0 Flashを検討してみてください。PDFからの構造化データ変換に役立てたら幸いです。
6. サンプルコード
"""
PDF処理クラス: Google Gemini 2.0 Flashを使用してPDFを処理し、構造化データを生成
"""
import os
import base64
import json
import logging
from pathlib import Path
import glob
from google import generativeai as genai
from google.genai import types
from typing import Dict, List, Tuple, Optional
import datetime
import re
class PDFProcessor:
def __init__(self, api_key: str = None, output_dir: str = "processed_data"):
"""
初期化
Args:
api_key: Google AI APIキー（未指定の場合は環境変数から取得）
output_dir: 出力ディレクトリ（デフォルトは"processed_data"）
"""
# ロガーの設定
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.INFO) # INFOレベルに設定
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
self.logger.addHandler(handler)
# APIキーの設定
self.api_key = api_key or os.getenv("GOOGLE_API_KEY")
if not self.api_key:
raise ValueError("GOOGLE_API_KEY is required")
self.logger.info("APIキーの設定が完了しました")
# Gemini APIの初期化
try:
genai.configure(api_key=self.api_key)
self.model = genai.GenerativeModel('gemini-2.0-flash')
self.logger.info("Gemini APIの初期化が完了しました")
except Exception as e:
self.logger.error(f"Gemini APIの初期化に失敗しました: {e}")
raise
# 出力ディレクトリの設定
try:
self.base_dir = Path.cwd()
self.output_dir = self.base_dir / output_dir
self.output_dir.mkdir(parents=True, exist_ok=True)
self.logger.info(f"出力ディレクトリを作成しました: {self.output_dir}")
except Exception as e:
self.logger.error(f"出力ディレクトリの作成に失敗しました: {e}")
raise
def process_directory(self, pdf_dir: str):
"""
ディレクトリ内のPDFファイルを処理する
Args:
pdf_dir: PDFファイルが格納されたディレクトリ
"""
pdf_files = glob.glob(os.path.join(pdf_dir, "*.pdf"))
if not pdf_files:
self.logger.warning(f"ディレクトリ {pdf_dir} にPDFファイルが見つかりませんでした。")
return
for pdf_path in pdf_files:
try:
structured_data = self.process_pdf(pdf_path)
if structured_data:
self.save_processed_data(pdf_path, structured_data)
self.logger.info(f"処理完了: {pdf_path}")
else:
self.logger.warning(f"処理失敗: {pdf_path}")
except Exception as e:
self.logger.error(f"PDF処理中にエラーが発生しました: {pdf_path} - {e}")
self.logger.info("PDF処理が完了しました。")
def _encode_pdf(self, pdf_path: str) -> str:
"""PDFファイルをbase64エンコード"""
with open(pdf_path, "rb") as f:
return base64.b64encode(f.read()).decode()
def process_pdf(self, pdf_path: str, custom_prompt: str = None) -> Dict:
"""
PDFを処理して構造化データを生成
Args:
pdf_path: 処理するPDFファイルのパス
custom_prompt: カスタムプロンプト（指定がない場合は標準プロンプトを使用）
Returns:
Dict: 構造化されたJSONデータ
"""
try:
# ファイル名からページ番号を抽出
pdf_name = Path(pdf_path).stem
page_number = self._extract_page_number(pdf_name)
self.logger.info(f"Processing PDF {pdf_name} (Page {page_number})")
# プロンプトの取得と調整
if custom_prompt:
system_prompt = custom_prompt
else:
system_prompt = self._get_default_prompt()
# プロンプトにページ番号情報を追加
system_prompt = self._add_page_info_to_prompt(system_prompt, page_number)
# PDFファイルを読み込み
try:
with open(pdf_path, 'rb') as pdf_file:
pdf_bytes = pdf_file.read()
except Exception as e:
self.logger.error(f"PDFファイルの読み込みに失敗しました: {e}")
return self._create_error_metadata(str(e), pdf_path)
# Gemini 2.0 Flash APIリクエストの作成と実行
try:
response = self.model.generate_content(
[
{"text": system_prompt},
{
"inline_data": {
"mime_type": "application/pdf",
"data": base64.b64encode(pdf_bytes).decode()
}
}
],
safety_settings=[
{
"category": "HARM_CATEGORY_HARASSMENT",
"threshold": "BLOCK_ONLY_HIGH"
},
{
"category": "HARM_CATEGORY_HATE_SPEECH",
"threshold": "BLOCK_ONLY_HIGH"
},
{
"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
"threshold": "BLOCK_ONLY_HIGH"
},
{
"category": "HARM_CATEGORY_DANGEROUS_CONTENT",
"threshold": "BLOCK_ONLY_HIGH"
}
],
generation_config={
"candidate_count": 1,
"max_output_tokens": 8192,
"temperature": 0,
"top_p": 0.8,
"top_k": 40,
"stop_sequences": ["</structured_data>"]
}
)
except Exception as e:
self.logger.error(f"Gemini APIの呼び出しに失敗しました: {e}")
return self._create_error_metadata(str(e), pdf_path)
# レスポンスの解析
response_text = response.text
structured_data = self._parse_response(response_text)
# ページ番号を設定
if "metadata" in structured_data:
structured_data["metadata"]["page_number"] = page_number
else:
structured_data["metadata"] = {"page_number": page_number}
structured_data["file_name"] = Path(pdf_path).name
return structured_data
except Exception as e:
self.logger.error(f"Error processing PDF {pdf_path}: {e}")
return self._create_error_metadata(str(e), pdf_path)
def _get_default_prompt(self) -> str:
"""
デフォルトのプロンプトを取得
"""
default_prompt = """
あなたは、与えられたPDFドキュメントから構造化データを抽出するAIアシスタントです。
以下のガイドラインに従って、PDFの内容を解析し、構造化されたJSON形式で情報を提供してください。
## 構造化データの形式
抽出された情報は、以下のJSON形式で提供してください。
```json
{
"metadata": {
"document_type": "レポート, 論文, 記事, その他",
"page_number": 1,
"section": "セクション名",
"subsection": "サブセクション名",
"topics": ["キーワード1", "キーワード2", "キーワード3"],
"contains_financial_data": true/false,
"contains_graphs": true/false,
"language": "ja",
"document_date": "YYYY-MM-DD",
"context": {
"previous_section": "前のセクション名",
"next_section": "次のセクション名",
"related_topics": ["関連キーワード1", "関連キーワード2"]
}
},
"content": {
"headings": [
{"level": 1, "text": "見出し1"},
{"level": 2, "text": "見出し2"}
],
"paragraphs": [
{"text": "段落1のテキスト", "type": "body"},
{"text": "段落2のテキスト", "type": "body"}
],
"numerical_data": [
{"label": "指標名1", "value": 123.45, "unit": "単位"},
{"label": "指標名2", "value": 678.90, "unit": "単位"}
],
"tables": [
{"title": "表のタイトル", "headers": ["列1", "列2"], "rows": [["データ1", "データ2"], ["データ3", "データ4"]]}
],
"references": [
{"type": "URL", "source": "http://example.com"},
{"type": "論文", "source": "著者名, 論文タイトル, ジャーナル名, 年"}
]
},
"extraction_info": {
"confidence_score": 0.95,
"missing_elements": ["要素1", "要素2"],
"extraction_notes": ["特記事項1", "特記事項2"]
}
}
```
## 重要な注意事項
1. PDFドキュメント全体を網羅的に解析し、可能な限り多くの情報を抽出してください。
2. 構造化データの形式に厳密に従い、すべてのフィールドを適切に埋めてください。
3. 不明な点や判断が難しい場合は、最も可能性の高い情報を推定して記述してください。
4. JSON形式で記述された構造化データのみを返してください。
5. JSONデータの開始タグと終了タグは不要です。
6. PDFに画像が含まれている場合、その内容をテキストで記述してください（例：グラフの種類、データの傾向など）。
7. 複数の表がある場合は、それぞれを個別のJSONオブジェクトとして記述してください。
8. 数値データは、単位とともに出力してください。
9. 参照情報は、URLまたは参考文献の形式で記述してください。
10. 抽出できなかった要素は、"missing_elements" フィールドに記述してください。
11. 抽出に関する特記事項は、"extraction_notes" フィールドに記述してください。
12. 言語は "ja" (日本語) を使用してください。
13. 日付は "YYYY-MM-DD" 形式を使用してください。
14. 常に最新の情報を優先して抽出してください。
15. PDFの内容に応じて、上記以外のフィールドを追加しても構いません。
16. PDFの内容を要約しないでください。
17. PDFに記載されている情報をそのまま抽出してください。
18. PDFに記載されていない情報を推測しないでください。
19. PDFに記載されている情報が不明確な場合は、不明確であることを明記してください。
20. PDFに記載されている情報が矛盾する場合は、矛盾していることを明記してください。
## 出力例
以下は、出力例です。
```json
{
"metadata": {
"document_type": "レポート",
"page_number": 1,
"section": "概要",
"subsection": "事業概要",
"topics": ["事業戦略", "市場分析", "競合分析"],
"contains_financial_data": true,
"contains_graphs": true,
"language": "ja",
"document_date": "2023-01-01",
"context": {
"previous_section": "",
"next_section": "財務分析",
"related_topics": ["事業計画", "リスク管理"]
}
},
"content": {
"headings": [
{"level": 1, "text": "事業概要"},
{"level": 2, "text": "市場動向"}
],
"paragraphs": [
{"text": "当社の事業は、〇〇市場において、〇〇を提供しています。", "type": "body"},
{"text": "市場は、年々成長しており、今後も成長が見込まれます。", "type": "body"}
],
"numerical_data": [
{"label": "売上高", "value": 1000000000, "unit": "円"},
{"label": "営業利益", "value": 100000000, "unit": "円"}
],
"tables": [
{"title": "市場規模", "headers": ["年", "市場規模"], "rows": [["2020", "1000000000"], ["2021", "1100000000"]]}
],
"references": [
{"type": "URL", "source": "http://example.com/market_data"}
]
},
"extraction_info": {
"confidence_score": 0.95,
"missing_elements": [],
"extraction_notes": ["特になし"]
}
}
```
## 構造化データ
<structured_data>
"""
return default_prompt
def _add_page_info_to_prompt(self, prompt: str, page_number: int) -> str:
"""
プロンプトにページ番号情報を追加
Args:
prompt: 元のプロンプト
page_number: ページ番号
Returns:
str: 調整されたプロンプト
"""
page_info = f"""
## 重要な注意事項（ページ固有）
1. このPDFは{page_number}ページ目です。このページの情報のみを抽出してください。
2. 他のページの情報は含めないでください。
3. 重複する情報は含めないでください。
4. 各セクション（headings, paragraphs, numerical_data, tables）内で重複するエントリを作成しないでください。
"""
# プロンプトの先頭に追加（既存の「重要な注意事項」の前）
if "## 重要な注意事項" in prompt:
parts = prompt.split("## 重要な注意事項")
return parts[0] + page_info + "## 重要な注意事項" + parts[1]
else:
return prompt + "\n" + page_info
def _extract_page_number(self, file_name: str) -> int:
"""
ファイル名からページ番号を抽出
Args:
file_name: PDFファイルの名前（拡張子なし）
Returns:
int: 抽出されたページ番号
"""
try:
# 1-2.pdf のような形式からページ番号を抽出
parts = file_name.split('-')
if len(parts) >= 2:
return int(parts[1])
return 0
except Exception as e:
self.logger.error(f"Error extracting page number from {file_name}: {e}")
return 0
def _parse_response(self, response: str) -> Dict:
"""
APIレスポンスを解析し、構造化データを抽出
Args:
response: API応答の文字列
Returns:
Dict: 構造化データ
"""
try:
# デバッグ用にレスポンスの内容をログ
self.logger.debug(f"Raw response: {response}")
# structured_dataタグ内のJSONを抽出
start_tag = "<structured_data>"
end_tag = "</structured_data>"
start_idx = response.find(start_tag)
end_idx = response.find(end_tag)
if start_idx == -1 or end_idx == -1:
self.logger.warning("Standard tags not found, trying alternative extraction methods")
return self._extract_json_fallback(response)
json_str = response[start_idx + len(start_tag):end_idx].strip()
self.logger.debug(f"Extracted JSON string: {json_str}")
try:
structured_data = json.loads(json_str)
except json.JSONDecodeError as e:
self.logger.error(f"JSON decode error: {e}")
# JSON文字列のクリーニングと再試行
cleaned_json = self._clean_json_string(json_str)
structured_data = json.loads(cleaned_json)
# メタデータの検証と補完
structured_data = self._validate_and_complete_metadata(structured_data)
return structured_data
except Exception as e:
self.logger.error(f"Error in primary parsing method: {e}")
try:
return self._extract_json_fallback(response)
except Exception as fallback_error:
self.logger.error(f"Fallback parsing also failed: {fallback_error}")
return self._create_minimal_metadata()
def _clean_json_string(self, json_str: str) -> str:
"""
JSON文字列をクリーニング
Args:
json_str: クリーニングするJSON文字列
Returns:
str: クリーニングされたJSON文字列
"""
# コメントの削除
json_str = re.sub(r'//.*?\n', '\n', json_str)
json_str = re.sub(r'/\*.*?\*/', '', json_str, flags=re.DOTALL)
# 一般的な問題の修正
replacements = {
"'": '"', # シングルクォートをダブルクォートに
"True": "true", # Pythonの真偽値をJSONの形式に
"False": "false",
"None": "null",
"\n": " ", # 改行を空白に
"\t": " " # タブを空白に
}
for old, new in replacements.items():
json_str = json_str.replace(old, new)
# 余分な空白の削除
json_str = re.sub(r'\s+', ' ', json_str).strip()
# 末尾のカンマの削除
json_str = re.sub(r',\s*([\]}])', r'\1', json_str)
return json_str
def _validate_and_complete_metadata(self, metadata: Dict) -> Dict:
"""
メタデータの検証と必須フィールドの補完
Args:
metadata: 検証・補完するメタデータ
Returns:
Dict: 検証・補完されたメタデータ
"""
# 基本構造の確認と補完
if "metadata" not in metadata:
metadata["metadata"] = {}
if "content" not in metadata:
metadata["content"] = {}
if "extraction_info" not in metadata:
metadata["extraction_info"] = {}
# 重複データの除去（ハッシュベースの重複チェック）
if "content" in metadata:
for key in ["headings", "paragraphs", "numerical_data", "tables"]:
if key in metadata["content"]:
# 重複を除去（より堅牢なハッシュベースの方法）
unique_items = self._remove_duplicates(metadata["content"][key])
metadata["content"][key] = unique_items
# メタデータの必須フィールドの補完
default_metadata = {
"document_type": "unknown",
"page_number": 0, # これは_process_pdfで上書きされます
"section": "unknown",
"subsection": "",
"topics": [],
"contains_financial_data": False,
"contains_graphs": False,
"language": "ja",
"document_date": "",
"context": {
"previous_section": "",
"next_section": "",
"related_topics": []
}
}
metadata["metadata"] = {**default_metadata, **metadata["metadata"]}
# コンテンツの必須フィールドの補完
if "paragraphs" not in metadata["content"]:
metadata["content"]["paragraphs"] = []
# 抽出情報の必須フィールドの補完
default_extraction_info = {
"confidence_score": 0.0,
"missing_elements": [],
"extraction_notes": []
}
metadata["extraction_info"] = {**default_extraction_info, **metadata["extraction_info"]}
return metadata
def _remove_duplicates(self, items: List[Dict]) -> List[Dict]:
"""
リスト内の重複する辞書を除去（ハッシュベースの方法）
Args:
items: 重複を含む可能性のある辞書のリスト
Returns:
List[Dict]: 重複が除去されたリスト
"""
if not items:
return []
seen_hashes = set()
unique_items = []
for item in items:
# 辞書をソートしてからハッシュ化（順序に依存しない一貫したハッシュを生成）
item_hash = self._hash_dict(item)
if item_hash not in seen_hashes:
seen_hashes.add(item_hash)
unique_items.append(item)
return unique_items
def _hash_dict(self, d: Dict) -> str:
"""
辞書のハッシュ値を生成（順序に依存しない）
Args:
d: ハッシュ化する辞書
Returns:
str: 辞書のハッシュ値
"""
if isinstance(d, dict):
# 辞書の各キーをソートし、値を再帰的にハッシュ化
return hash(tuple(sorted((k, self._hash_dict(v)) for k, v in d.items())))
elif isinstance(d, list):
# リストの各要素を再帰的にハッシュ化
return hash(tuple(self._hash_dict(v) for v in d))
else:
# その他の型は直接ハッシュ化
return hash(str(d))
def _extract_json_fallback(self, response: str) -> Dict:
"""
代替的なJSON抽出方法
Args:
response: 解析する応答文字列
Returns:
Dict: 構造化データ
"""
try:
# パターン1: ```json ブロック内のJSONを探す
if "```json" in response:
parts = response.split("```json")
if len(parts) >= 2:
json_part = parts[1].split("```")[0].strip()
try:
cleaned_json = self._clean_json_string(json_part)
structured_data = json.loads(cleaned_json)
return self._validate_and_complete_metadata(structured_data)
except Exception as e:
self.logger.warning(f"JSON解析に失敗しました（パターン1）: {e}")
# パターン2: 単純な{}ブロックを探す
try:
# 最も外側の{}を探す
start_idx = response.find('{')
if start_idx != -1:
count = 1
end_idx = -1
for i in range(start_idx + 1, len(response)):
if response[i] == '{':
count += 1
elif response[i] == '}':
count -= 1
if count == 0:
end_idx = i + 1
break
if end_idx != -1:
json_str = response[start_idx:end_idx]
cleaned_json = self._clean_json_string(json_str)
structured_data = json.loads(cleaned_json)
# JSONが見つかった場合、それ以外の部分をマークダウンとして扱う
return self._validate_and_complete_metadata(structured_data)
except Exception as e:
self.logger.warning(f"JSON解析に失敗しました（パターン2）: {e}")
# パターン3: 構造化されたテキストとしての解析を試みる
try:
# レスポンスから構造化データを抽出
structured_data = {
"content": {
"paragraphs": []
}
}
# テキストを行に分割
lines = response.split('\n')
current_paragraph = ""
for line in lines:
line = line.strip()
if line:
if len(line) > 50: # 段落として扱う
if current_paragraph:
structured_data["content"]["paragraphs"].append({
"text": current_paragraph,
"type": "body"
})
current_paragraph = line
else:
if current_paragraph:
current_paragraph += " " + line
else:
current_paragraph = line
if current_paragraph: # 最後の段落を追加
structured_data["content"]["paragraphs"].append({
"text": current_paragraph,
"type": "body"
})
return structured_data
except Exception as e:
self.logger.warning(f"構造化テキスト解析に失敗しました: {e}")
# すべての方法が失敗した場合は最小限のメタデータを生成
return self._create_minimal_metadata()
except Exception as e:
self.logger.error(f"JSON抽出処理に失敗しました: {e}")
return self._create_minimal_metadata()
def _create_minimal_metadata(self) -> Dict:
"""
最小限の有効なメタデータを生成
Returns:
Dict: 最小限のメタデータ
"""
return {
"metadata": {
"document_type": "unknown",
"page_number": 0,
"section": "unknown",
"subsection": "",
"topics": [],
"contains_financial_data": False,
"contains_graphs": False,
"language": "ja",
"document_date": "",
"context": {
"previous_section": "",
"next_section": "",
"related_topics": []
}
},
"content": {
"paragraphs": []
},
"extraction_info": {
"confidence_score": 0.0,
"missing_elements": ["all"],
"extraction_notes": ["Fallback to minimal metadata due to parsing failure"]
}
}
def save_processed_data(self, pdf_path: str, structured_data: Dict):
"""
処理済みデータの保存
Args:
pdf_path: 処理したPDFファイルのパス
structured_data: 構造化データ
"""
try:
pdf_name = Path(pdf_path).stem
# メタデータの正規化
normalized_data = self._normalize_metadata(structured_data)
# 統合データの作成
integrated_data = {
"metadata": normalized_data["metadata"],
"content": normalized_data["content"],
"extraction_info": normalized_data["extraction_info"],
"processing_info": {
"source_file": pdf_name,
"processed_at": datetime.datetime.now().isoformat(),
"processor_version": "1.0.0"
}
}
# JSONファイルとして保存
json_path = self.output_dir / f"{pdf_name}.json"
self.logger.info(f"JSONファイルを保存します: {json_path}")
with open(json_path, "w", encoding="utf-8") as f:
json.dump(integrated_data, f, ensure_ascii=False, indent=2)
self.logger.info(f"JSONファイルを保存しました: {json_path}")
except Exception as e:
self.logger.error(f"データの保存に失敗しました - {pdf_path}: {e}")
raise
def _normalize_metadata(self, metadata: Dict) -> Dict:
"""
メタデータを正規化し、重複を除去
Args:
metadata: 正規化するメタデータ
Returns:
Dict: 正規化されたメタデータ
"""
normalized = {
"metadata": {},
"content": {
"structured": {}
},
"extraction_info": {}
}
# メタデータセクションの正規化
if "metadata" in metadata:
meta = metadata["metadata"]
if isinstance(meta, dict) and "metadata" in meta:
meta = meta["metadata"]
normalized["metadata"] = meta
# コンテンツの正規化
if "content" in metadata:
content = metadata["content"]
if isinstance(content, dict):
if "structured" in content:
content = content["structured"]
# 各セクションの重複除去
for key in ["headings", "paragraphs", "numerical_data", "tables", "references"]:
if key in content:
items = content[key]
# 重複除去（ハッシュベースの方法を使用）
unique_items = self._remove_duplicates_with_hierarchy(items)
normalized["content"]["structured"][key] = unique_items
# 抽出情報の正規化
if "extraction_info" in metadata:
normalized["extraction_info"] = metadata["extraction_info"]
elif "content" in metadata and "unstructured" in metadata["content"]:
normalized["extraction_info"] = metadata["content"]["unstructured"].get("extraction_info", {})
# デフォルト値の設定
if not normalized["metadata"]:
normalized["metadata"] = {
"document_type": "unknown",
"page_number": 0,
"section": "unknown",
"subsection": "",
"topics": [],
"contains_financial_data": False,
"contains_graphs": False,
"language": "ja",
"document_date": "",
"context": {
"previous_section": "",
"next_section": "",
"related_topics": []
}
}
if not normalized["extraction_info"]:
normalized["extraction_info"] = {
"confidence_score": 0.0,
"missing_elements": [],
"extraction_notes": []
}
return normalized
def _remove_duplicates_with_hierarchy(self, items: List[Dict]) -> List[Dict]:
"""
階層構造を考慮して重複を除去
Args:
items: 重複を含む可能性のある辞書のリスト
Returns:
List[Dict]: 重複が除去されたリスト
"""
if not items:
return []
# 階層構造を考慮したキー生成
def create_hierarchy_key(item):
if "level" in item and "text" in item: # headings
return f"{item['level']}:{item['text']}"
elif "title" in item: # tables
return f"table:{item['title']}:{json.dumps(item.get('headers', []))}"
elif "text" in item: # paragraphs
return f"text:{item['text']}"
elif "label" in item: # numerical_data
return f"data:{item['label']}:{item.get('value', '')}"
elif "type" in item and "source" in item: # references
return f"ref:{item['type']}:{item['source']}"
else:
return json.dumps(item, sort_keys=True)
seen = {}
unique_items = []
for item in items:
key = create_hierarchy_key(item)
if key not in seen:
seen[key] = item
unique_items.append(item)
elif "level" in item: # 見出しの場合、より高いレベルを優先
if item["level"] < seen[key]["level"]:
seen[key] = item
unique_items[unique_items.index(seen[key])] = item
return unique_items
def _create_error_metadata(self, error_message: str, pdf_path: str) -> Dict:
"""
エラー時のメタデータを生成
Args:
error_message: エラーメッセージ
pdf_path: PDFファイルのパス
Returns:
Dict: エラーメタデータ
"""
return {
"error": {
"type": "processing_error",
"message": error_message,
"file_name": Path(pdf_path).name
},
"extraction_info": {
"confidence_score": 0.0,
"quality_score": 0.0,
"missing_elements": ["all"],
"extraction_notes": [f"Processing failed: {error_message}"]
}
}
def main():
"""メイン処理"""
try:
# 設定
api_key = os.getenv("GOOGLE_API_KEY") # 環境変数からAPIキーを取得
pdf_directory = "pdf_pages" # PDFファイルが格納されたディレクトリ
output_directory = "extracted_data" # 構造化データを出力するディレクトリ
# PDFProcessorの初期化
processor = PDFProcessor(api_key=api_key, output_dir=output_directory)
# PDFディレクトリの処理
processor.process_directory(pdf_directory)
print("PDF処理が完了しました。")
except ValueError as ve:
print(f"設定エラー: {ve}")
except Exception as e:
print(f"エラーが発生しました: {e}")
if __name__ == "__main__":
main()コードの説明:
このコードは、Google Gemini 2.0 Flashを使用して、複数ページのPDFドキュメントを1ページずつ処理し、構造化されたデータを抽出するためのPythonスクリプトです。 特に、1ページずつ分割されたPDFファイルが格納されたディレクトリを処理するのに適しています。
主な機能:
PDF処理: Google Gemini 2.0 Flash APIを使用してPDFファイルを解析し、テキスト、見出し、表、数値データなどの構造化された情報を抽出します。
構造化データ: 抽出されたデータはJSON形式で保存され、他のアプリケーションやシステムで簡単に利用できます。
ディレクトリ処理: 指定されたディレクトリ内のすべてのPDFファイルを自動的に処理します。
エラー処理: エラーが発生した場合、エラーメッセージをログに記録し、処理を続行します。
カスタマイズ: APIキー、入力ディレクトリ、出力ディレクトリなどの設定を簡単に変更できます。
使用方法:
APIキーの取得: Google Cloud PlatformでGemini APIを有効にし、APIキーを取得します。
環境変数の設定: 環境変数 GOOGLE_API_KEY に取得したAPIキーを設定します。
PDFファイルの準備: 1ページずつ分割されたPDFファイルを、指定されたディレクトリ（デフォルトは pdf_pages ）に格納します。
コードの実行: Pythonスクリプトを実行します。
python your_script_name.py
出力の確認: 抽出された構造化データは、指定された出力ディレクトリ（デフォルトは extracted_data ）にJSONファイルとして保存されます。
コードの構成:
PDFProcessor クラス: PDF処理の主要なロジックをカプセル化します。
__init__: クラスの初期化、APIキーの設定、出力ディレクトリの作成などを行います。
process_directory: 指定されたディレクトリ内のすべてのPDFファイルを処理します。
process_pdf: 個々のPDFファイルを処理し、構造化データを抽出します。
_encode_pdf: PDFファイルをBase64エンコードします。
_get_default_prompt: Gemini APIに送信するデフォルトのプロンプトを取得します。
_add_page_info_to_prompt: プロンプトにページ番号情報を追加します。
_extract_page_number: ファイル名からページ番号を抽出します。
_parse_response: Gemini APIからのレスポンスを解析し、構造化データを抽出します。
_clean_json_string: JSON文字列をクリーンアップします。
_validate_and_complete_metadata: メタデータを検証し、必要なフィールドを補完します。
_remove_duplicates: リスト内の重複する辞書を除去します。
_hash_dict: 辞書のハッシュ値を生成します。
_extract_json_fallback: 代替的なJSON抽出方法を試みます。
_create_minimal_metadata: 最小限のメタデータを生成します。
save_processed_data: 処理済みデータをJSONファイルとして保存します。
_normalize_metadata: メタデータを正規化します。
_remove_duplicates_with_hierarchy: 階層構造を考慮して重複を除去します。
_create_error_metadata: エラー時のメタデータを生成します。
main 関数: スクリプトのエントリーポイントです。
APIキー、入力ディレクトリ、出力ディレクトリなどの設定を行います。
PDFProcessor のインスタンスを作成します。
process_directory メソッドを呼び出して、PDFファイルを処理します。
カスタマイズ:
APIキー: 環境変数 GOOGLE_API_KEY を設定するか、PDFProcessor のコンストラクタに直接APIキーを渡します。
入力ディレクトリ: main 関数内の pdf_directory 変数を変更します。
出力ディレクトリ: main 関数内の output_directory 変数を変更するか、PDFProcessor のコンストラクタに output_dir パラメータを渡します。
プロンプト: _get_default_prompt メソッドを編集して、Gemini APIに送信するプロンプトをカスタマイズします。
注意点:
このスクリプトを使用するには、Google Gemini APIへのアクセス権が必要です。
PDFファイルの構造や内容によっては、抽出されるデータの品質が異なる場合があります。
大規模なPDFドキュメントを処理する場合、APIの使用量に注意してください。
このコードは、PDFドキュメントから構造化されたデータを抽出するための出発点として使用できます。必要に応じて、コードを拡張または変更して、特定のニーズに合わせてカスタマイズしてください。