Building an AI Document Assistant with Gemini 2.5 Pro — Analyze PDFs, Images & Text to Auto-Generate Markdown Reports

pip install google-genai
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:
    """Upload a single file to the File API.
    
    Returns the File object on success, None on failure.
    """
    suffix = file_path.suffix.lower()
    mime = SUPPORTED_MIME.get(suffix)
    if mime is None:
        print(f"  ⚠ Skipping unsupported format: {file_path.name}")
        return None
 
    try:
        uploaded = client.files.upload(
            file=file_path,
            config=types.UploadFileConfig(mime_type=mime),
        )
        # Wait for processing to complete
        while uploaded.state == "PROCESSING":
            time.sleep(2)
            uploaded = client.files.get(name=uploaded.name)
 
        if uploaded.state == "FAILED":
            print(f"  ✗ Upload failed: {file_path.name}")
            return None
 
        print(f"  ✓ Uploaded: {file_path.name} ({mime})")
        return uploaded
    except Exception as e:
        print(f"  ✗ Error uploading {file_path.name}: {e}")
        return None
 
def upload_files_parallel(file_paths: list[Path], max_workers: int = 4) -> list[types.File]:
    """Upload multiple files concurrently."""
    print(f"\n📤 Uploading {len(file_paths)} files...")
    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)} files uploaded successfully\n")
    return results

ANALYSIS_PROMPT = """
Analyze the following documents and produce a structured Markdown report with these sections:
 
1. **Executive Summary** (~100 words) — the single most important takeaway across all documents
2. **Per-Document Summaries** — key points from each file as bullet lists
3. **Key Numbers & Data** — tables, charts, and metrics extracted and consolidated
4. **Cross-Document Insights** — patterns, contradictions, or themes that span multiple files
5. **Recommended Actions** — concrete next steps, if applicable
 
Guidelines:
- Preserve visual context from charts and diagrams in your analysis
- Keep each language's content in its original language
- Distinguish clearly between stated facts and inferences
"""
 
def analyze_documents(uploaded_files: list[types.File]) -> str:
    """Analyze uploaded files with Gemini 2.5 Pro.
    
    Returns the generated Markdown report as a string.
    """
    # Build the content list: prompt first, then file references
    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("🤖 Analyzing with 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,   # Lower temp for factual accuracy
            max_output_tokens=8192,
            # thinking_budget=2000,  # Enable for complex inference tasks
        ),
    )
 
    if not response.text:
        raise ValueError("Empty response from model. Check your file contents and API key permissions.")
 
    return response.text
 
def save_report(markdown_text: str, output_path: Path) -> None:
    """Save the report to a Markdown file."""
    output_path.write_text(markdown_text, encoding="utf-8")
    print(f"📄 Report saved: {output_path}")

def cleanup_uploaded_files(uploaded_files: list[types.File]) -> None:
    """Delete uploaded files from the File API immediately after use.
    
    Files auto-delete after 48 hours, but for sensitive documents
    (contracts, financials, PII), immediate deletion is strongly recommended.
    """
    for f in uploaded_files:
        try:
            client.files.delete(name=f.name)
        except Exception:
            pass  # Silently ignore — auto-delete will handle it
    print(f"🗑 Deleted {len(uploaded_files)} uploaded files")
 
def main(input_dir: str, output_file: str = "report.md") -> None:
    """Main entry point.
    
    Args:
        input_dir:   Directory containing the files to analyze
        output_file: Output Markdown report filename
    """
    input_path = Path(input_dir)
    if not input_path.is_dir():
        raise FileNotFoundError(f"Directory not found: {input_dir}")
 
    # Collect supported files
    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("No supported files found in the directory.")
        return
 
    print(f"📁 Found {len(file_paths)} files to analyze:")
    for p in file_paths:
        print(f"  - {p.name}")
 
    uploaded_files = []
    try:
        # Step 1: Parallel upload
        uploaded_files = upload_files_parallel(file_paths)
        if not uploaded_files:
            print("No files were uploaded successfully.")
            return
 
        # Step 2: Analyze
        markdown_report = analyze_documents(uploaded_files)
 
        # Step 3: Save
        output_path = Path(output_file)
        save_report(markdown_report, output_path)
        print("\n✅ Done\!")
 
    finally:
        # Always clean up — even if analysis fails
        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 = """
Review the following contracts and legal documents. Extract:
1. Parties involved, effective dates, and term length
2. Key obligations for each party
3. Risk clauses (penalties, indemnification, confidentiality)
4. Ambiguous language requiring legal review
 
Note: This is an information summary, not legal advice.
"""

COMPETITIVE_PROMPT = """
Analyze the attached materials (press releases, product specs, social posts).
Produce:
1. Recent announcements and product changes per company
2. Feature and pricing comparison matrix
3. Market positioning shifts
4. Implications and risks for our business
"""