Gemini 2.5 Pro でつくるAIドキュメントアシスタント — PDF・画像・テキストを一括解析してMarkdownレポートを自動生成する

pip install google-genai pathlib2

export GOOGLE_API_KEY="YOUR_GEMINI_API_KEY"

import os
import time
import concurrent.futures
from pathlib import Path
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
 
SUPPORTED_MIME = {
    ".pdf":  "application/pdf",
    ".jpg":  "image/jpeg",
    ".jpeg": "image/jpeg",
    ".png":  "image/png",
    ".webp": "image/webp",
    ".txt":  "text/plain",
    ".md":   "text/markdown",
}
 
def upload_file(file_path: Path) -> types.File | None:
    """単一ファイルをFile APIへアップロードする。
    
    戻り値: アップロード済みFileオブジェクト（失敗時はNone）
    """
    suffix = file_path.suffix.lower()
    mime = SUPPORTED_MIME.get(suffix)
    if mime is None:
        print(f"  ⚠ スキップ（未対応形式）: {file_path.name}")
        return None
 
    try:
        uploaded = client.files.upload(
            file=file_path,
            config=types.UploadFileConfig(mime_type=mime),
        )
        # アップロード完了まで待機（処理中ステータスの場合）
        while uploaded.state == "PROCESSING":
            time.sleep(2)
            uploaded = client.files.get(name=uploaded.name)
 
        if uploaded.state == "FAILED":
            print(f"  ✗ アップロード失敗: {file_path.name}")
            return None
 
        print(f"  ✓ アップロード完了: {file_path.name} ({mime})")
        return uploaded
    except Exception as e:
        print(f"  ✗ エラー: {file_path.name} — {e}")
        return None
 
def upload_files_parallel(file_paths: list[Path], max_workers: int = 4) -> list[types.File]:
    """複数ファイルを並列アップロードする。"""
    print(f"\n📤 {len(file_paths)}件のファイルをアップロード中...")
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(upload_file, p): p for p in file_paths}
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result is not None:
                results.append(result)
    print(f"✅ {len(results)}件のアップロード完了\n")
    return results

ANALYSIS_PROMPT = """
以下のドキュメント群を解析し、Markdown形式でレポートを作成してください。
 
## レポート要件
1. **エグゼクティブサマリー**（200文字程度）— 全体の要点を一言でまとめる
2. **ドキュメント別要約** — 各ファイルの主要な情報を箇条書きで整理する
3. **重要な数値・データ** — 表・グラフ・数値情報を抽出してまとめる
4. **キーインサイト** — 複数のドキュメントを横断した共通テーマや矛盾点を指摘する
5. **アクションアイテム** — 次のステップとして推奨される具体的なアクション（あれば）
 
## 注意事項
- 図表やレイアウト情報も含めて解析してください
- 日本語のドキュメントは日本語で、英語は英語で要約してください
- 事実を正確に記述し、推測の場合は明示してください
"""
 
def analyze_documents(uploaded_files: list[types.File]) -> str:
    """アップロード済みファイルをGemini 2.5 Proで一括解析する。
    
    戻り値: Markdownレポート文字列
    """
    # ファイル参照をコンテンツリストに変換
    contents: list = [ANALYSIS_PROMPT]
    for f in uploaded_files:
        contents.append(
            types.Part.from_uri(file_uri=f.uri, mime_type=f.mime_type)
        )
 
    print("🤖 Gemini 2.5 Pro で解析中...")
    response = client.models.generate_content(
        model="gemini-2.5-pro-preview-05-06",
        contents=contents,
        config=types.GenerateContentConfig(
            temperature=0.2,  # レポートなので低め（事実重視）
            max_output_tokens=8192,
            # thinking_budget=2000,  # 複雑な推論が必要なら有効化
        ),
    )
 
    if not response.text:
        raise ValueError("モデルからの応答が空です。ファイルの内容を確認してください。")
 
    return response.text
 
def save_report(markdown_text: str, output_path: Path) -> None:
    """レポートをMarkdownファイルとして保存する。"""
    output_path.write_text(markdown_text, encoding="utf-8")
    print(f"📄 レポート保存: {output_path}")

def cleanup_uploaded_files(uploaded_files: list[types.File]) -> None:
    """使用済みのアップロードファイルをFile APIから削除する。
    
    File APIのファイルは48時間後に自動削除されるが、
    機密情報を含む場合は即時削除することを推奨する。
    """
    for f in uploaded_files:
        try:
            client.files.delete(name=f.name)
        except Exception:
            pass  # 削除失敗は無視（自動削除で対応）
    print(f"🗑 {len(uploaded_files)}件のアップロードファイルを削除しました")
 
def main(input_dir: str, output_file: str = "report.md") -> None:
    """メイン処理。
    
    Args:
        input_dir:   解析するファイルが入ったディレクトリパス
        output_file: 出力するMarkdownレポートのファイル名
    """
    input_path = Path(input_dir)
    if not input_path.is_dir():
        raise FileNotFoundError(f"ディレクトリが見つかりません: {input_dir}")
 
    # 対応ファイルの収集
    file_paths = [
        p for p in input_path.iterdir()
        if p.is_file() and p.suffix.lower() in SUPPORTED_MIME
    ]
    if not file_paths:
        print("解析対象のファイルが見つかりませんでした。")
        return
 
    print(f"📁 解析対象: {len(file_paths)}件")
    for p in file_paths:
        print(f"  - {p.name}")
 
    uploaded_files = []
    try:
        # Step 1: 並列アップロード
        uploaded_files = upload_files_parallel(file_paths)
        if not uploaded_files:
            print("有効なファイルがアップロードされませんでした。")
            return
 
        # Step 2: 一括解析
        markdown_report = analyze_documents(uploaded_files)
 
        # Step 3: レポート保存
        output_path = Path(output_file)
        save_report(markdown_report, output_path)
        print("\n✅ 処理完了！")
 
    finally:
        # 機密情報保護のため確実にクリーンアップ
        if uploaded_files:
            cleanup_uploaded_files(uploaded_files)
 
if __name__ == "__main__":
    import sys
    input_dir = sys.argv[1] if len(sys.argv) > 1 else "./documents"
    output_file = sys.argv[2] if len(sys.argv) > 2 else "report.md"
    main(input_dir, output_file)

python document_assistant.py ./documents report_20260415.md

LEGAL_PROMPT = """
以下の契約書・法律文書を解析し、以下の観点でレポートを作成してください：
1. 当事者・締結日・有効期間
2. 主要な義務と権利
3. リスク条項（違約金・免責・守秘義務）
4. 不明確な表現・要確認事項
※ 法的判断ではなく情報整理として活用してください。
"""

COMPETITIVE_PROMPT = """
添付の資料（プレスリリース・製品仕様書・SNS投稿等）を解析し：
1. 各社の最新動向と発表内容
2. 機能・価格の比較マトリクス
3. 市場ポジションの変化
4. 自社への示唆とリスク
をまとめてください。
"""