Gemini 1M Token Long Context Strategies — Production Patterns for Large Document Processing

import google.generativeai as genai
from pathlib import Path
 
# Initialize Gemini API
genai.configure(api_key="YOUR_API_KEY")
 
def count_tokens_for_content(content: str, model_name: str = "gemini-2.5-pro") -> dict:
    """
    Count tokens in text content using Gemini's token counter.
    """
    model = genai.GenerativeModel(model_name)
    response = model.count_tokens(content)
    return {
        "input_tokens": response.total_tokens,
        "estimated_output_tokens": 2000,  # Conservative estimate for typical generation
    }
 
def estimate_cost(input_tokens: int, output_tokens: int, cached: bool = False) -> dict:
    """
    Estimate Gemini 2.5 Pro execution cost.
 
    Pricing (as of March 2026):
    - Input: $1.25 / 1M tokens (cache hit: $0.125 / 1M tokens)
    - Output: $5.00 / 1M tokens
    """
    if cached:
        input_cost = (input_tokens / 1_000_000) * 0.125
    else:
        input_cost = (input_tokens / 1_000_000) * 1.25
 
    output_cost = (output_tokens / 1_000_000) * 5.00
    total_cost = input_cost + output_cost
 
    return {
        "input_cost_usd": round(input_cost, 6),
        "output_cost_usd": round(output_cost, 6),
        "total_cost_usd": round(total_cost, 6),
    }
 
# Example usage
sample_content = """
Your large document or codebase content here...
This can be hundreds of thousands of tokens.
"""
 
tokens = count_tokens_for_content(sample_content)
print(f"Input tokens: {tokens['input_tokens']}")
print(f"Estimated output tokens: {tokens['estimated_output_tokens']}")
 
cost = estimate_cost(tokens['input_tokens'], tokens['estimated_output_tokens'])
print(f"Estimated cost: ${cost['total_cost_usd']}")
 
# With caching
cached_cost = estimate_cost(tokens['input_tokens'], tokens['estimated_output_tokens'], cached=True)
print(f"Cost with cache hit: ${cached_cost['total_cost_usd']}")

import os
import google.generativeai as genai
from pathlib import Path
from typing import Dict, List
 
genai.configure(api_key="YOUR_API_KEY")
 
def load_codebase(directory: str, extensions: List[str] = [".py", ".ts", ".js", ".go"]) -> Dict[str, str]:
    """
    Load all source files from a directory recursively.
 
    Args:
        directory: Root directory path
        extensions: File extensions to include
 
    Returns:
        Dictionary mapping file paths to their contents
    """
    files = {}
    base_path = Path(directory)
 
    for ext in extensions:
        for filepath in base_path.rglob(f"*{ext}"):
            # Skip hidden directories and common exclusions
            parts = filepath.parts
            if any(part.startswith(".") or part in ["node_modules", "__pycache__", "venv"]
                   for part in parts):
                continue
 
            try:
                relative_path = filepath.relative_to(base_path)
                with open(filepath, "r", encoding="utf-8") as f:
                    content = f.read()
                    files[str(relative_path)] = content
                    print(f"✓ Loaded: {relative_path} ({len(content)} chars)")
            except Exception as e:
                print(f"⚠ Failed to read {filepath}: {e}")
 
    print(f"\nTotal files loaded: {len(files)}")
    return files
 
def analyze_codebase_architecture(directory: str, focus_areas: List[str] = None) -> str:
    """
    Analyze complete codebase architecture in a single API call.
 
    Args:
        directory: Project root directory
        focus_areas: Specific areas to emphasize (e.g., ["scalability", "security"])
    """
    files = load_codebase(directory)
 
    if not files:
        return "No source files found in directory."
 
    # Format files for inclusion (limit per-file to manage token count)
    codebase_text = "\n\n".join(
        f"=== File: {path} ===\n{content[:3000]}{'...[truncated]' if len(content) > 3000 else ''}"
        for path, content in sorted(files.items())[:100]  # Limit to first 100 files
    )
 
    focus_prompt = ""
    if focus_areas:
        focus_prompt = f"\nSpecial focus areas: {', '.join(focus_areas)}"
 
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    prompt = f"""
Analyze the following codebase comprehensively. Provide insights on:
 
1. **Architecture Overview**: Main components, module organization, dependency graph
2. **Design Patterns**: Identified patterns (MVC, event-driven, microservices, etc.)
3. **Code Quality**: Strengths and areas for improvement in terms of maintainability
4. **Scalability Analysis**: Potential bottlenecks and scaling considerations
5. **Security Review**: Identified vulnerabilities, data handling concerns
6. **Testing Coverage**: Test strategy and coverage assessment
7. **Performance Optimization**: Opportunities for speed and resource efficiency
8. **Modernization**: Outdated dependencies or patterns that should be updated
 
{focus_prompt}
 
Codebase:
{codebase_text}
 
Provide a structured technical review with actionable recommendations.
"""
 
    print("\nAnalyzing codebase (this may take 1-2 minutes)...")
    response = model.generate_content(prompt, stream=True)
 
    full_response = ""
    for chunk in response:
        if chunk.text:
            print(chunk.text, end="", flush=True)
            full_response += chunk.text
 
    print("\n")
    return full_response
 
# Usage example
if __name__ == "__main__":
    analysis = analyze_codebase_architecture(
        "./my_project",
        focus_areas=["security", "scalability", "maintainability"]
    )
 
    # Optionally save to file
    with open("codebase_analysis.md", "w") as f:
        f.write(analysis)

import google.generativeai as genai
from pathlib import Path
from typing import List
 
genai.configure(api_key="YOUR_API_KEY")
 
def upload_pdf_corpus(pdf_paths: List[str]) -> List:
    """
    Upload multiple PDFs to Gemini's File API.
 
    Args:
        pdf_paths: List of absolute paths to PDF files
 
    Returns:
        List of uploaded file objects
    """
    uploaded_files = []
 
    for pdf_path in pdf_paths:
        path = Path(pdf_path)
        if not path.exists():
            print(f"✗ File not found: {pdf_path}")
            continue
 
        print(f"Uploading: {path.name}")
        try:
            file = genai.upload_file(
                path=str(pdf_path),
                mime_type="application/pdf"
            )
            uploaded_files.append(file)
            print(f"  ✓ Uploaded: {file.uri}")
        except Exception as e:
            print(f"  ✗ Upload failed: {e}")
 
    print(f"\nSuccessfully uploaded {len(uploaded_files)}/{len(pdf_paths)} files")
    return uploaded_files
 
def analyze_pdf_corpus(
    pdf_paths: List[str],
    analysis_focus: str = "key themes and actionable insights"
) -> str:
    """
    Analyze multiple PDFs for cross-document patterns and insights.
 
    Args:
        pdf_paths: List of PDF file paths
        analysis_focus: What aspects to analyze
 
    Returns:
        Analysis results as structured text
    """
    # Upload PDFs
    uploaded_files = upload_pdf_corpus(pdf_paths)
 
    if not uploaded_files:
        return "No PDFs were successfully uploaded."
 
    # Build content with file references
    content_parts = [f"Analyze these {len(uploaded_files)} documents for {analysis_focus}:\n"]
 
    for file in uploaded_files:
        content_parts.append(genai.types.FileData(
            mime_type=file.mime_type,
            file_uri=file.uri
        ))
 
    # Add analysis prompt
    analysis_prompt = """
Please analyze these documents and provide:
 
1. **Core Themes**: Common themes across documents
2. **Divergences**: Contradictions or significantly different viewpoints
3. **Information Gaps**: Areas lacking coverage or detail
4. **Cross-Document Synthesis**: Insights that only emerge from analyzing multiple documents together
5. **Recommendations**: Actionable recommendations based on the analysis
6. **Relationships**: How documents relate to and support each other
 
Format your response as structured markdown with clear sections.
"""
 
    content_parts.append(analysis_prompt)
 
    model = genai.GenerativeModel("gemini-2.5-pro")
    print("\nProcessing documents (may take 1-2 minutes)...")
    response = model.generate_content(content_parts, stream=True)
 
    full_response = ""
    for chunk in response:
        if chunk.text:
            print(chunk.text, end="", flush=True)
            full_response += chunk.text
 
    print("\n")
 
    # Cleanup
    for file in uploaded_files:
        try:
            genai.delete_file(file.name)
            print(f"Cleaned up: {file.name}")
        except Exception as e:
            print(f"Cleanup warning: {e}")
 
    return full_response
 
# Usage example
pdf_files = [
    "./documents/Q1_2024_Report.pdf",
    "./documents/Q2_2024_Report.pdf",
    "./documents/Q3_2024_Report.pdf",
    "./documents/Industry_Analysis_2024.pdf",
]
 
analysis = analyze_pdf_corpus(
    pdf_files,
    analysis_focus="quarterly trends, risk factors, and growth opportunities"
)
 
# Save results
with open("pdf_analysis_results.md", "w") as f:
    f.write(analysis)

import google.generativeai as genai
from datetime import datetime, timedelta
from typing import List
 
genai.configure(api_key="YOUR_API_KEY")
 
def create_cached_context(
    document_content: str,
    ttl_minutes: int = 30,
    system_instruction: str = "You are an expert document analyst."
) -> genai.caching.CachedContent:
    """
    Cache a document for reuse across multiple queries.
 
    Args:
        document_content: The document to cache
        ttl_minutes: Cache time-to-live in minutes (5-60)
        system_instruction: System prompt for the model
 
    Returns:
        CachedContent object for use in subsequent queries
    """
    cached_content = genai.caching.CachedContent.create(
        model="gemini-2.5-pro",
        system_instruction=system_instruction,
        contents=[{
            "role": "user",
            "parts": [{
                "text": f"Please analyze this document:\n\n{document_content}"
            }]
        }],
        ttl=timedelta(minutes=ttl_minutes)
    )
 
    return cached_content
 
def query_cached_context(
    cached_content: genai.caching.CachedContent,
    query: str,
    temperature: float = 0.3
) -> str:
    """
    Execute a query against cached context.
 
    Args:
        cached_content: CachedContent object from create_cached_context
        query: The query to execute
        temperature: Model temperature (0 = deterministic, higher = creative)
 
    Returns:
        Query response text
    """
    model = genai.GenerativeModel(
        "gemini-2.5-pro",
        generation_config=genai.types.GenerationConfig(
            temperature=temperature,
            max_output_tokens=2000
        )
    )
 
    response = model.generate_content(
        [query],
        cached_content=cached_content
    )
 
    return response.text
 
def calculate_caching_roi(
    document_tokens: int,
    num_queries: int
) -> dict:
    """
    Calculate return on investment for context caching.
 
    Pricing assumptions (March 2026):
    - Standard input: $1.25 / 1M tokens
    - Cache write: $1.25 / 1M tokens (first query)
    - Cache read: $0.125 / 1M tokens (subsequent queries)
    - Output: $5.00 / 1M tokens (always)
    """
    # Costs without caching
    cost_without_caching = (document_tokens / 1_000_000) * 1.25 * num_queries
 
    # Costs with caching
    cache_write_cost = (document_tokens / 1_000_000) * 1.25
    cache_read_cost = (document_tokens / 1_000_000) * 0.125 * (num_queries - 1)
    cost_with_caching = cache_write_cost + cache_read_cost
 
    savings = cost_without_caching - cost_with_caching
    savings_percent = (savings / cost_without_caching * 100) if cost_without_caching > 0 else 0
 
    return {
        "cost_without_caching_usd": round(cost_without_caching, 4),
        "cost_with_caching_usd": round(cost_with_caching, 4),
        "total_savings_usd": round(savings, 4),
        "savings_percent": round(savings_percent, 1),
        "breakeven_queries": 2,  # Caching pays off after just 2 queries
    }
 
# Example: Analyze a technical specification document
def multi_query_analysis_example(document: str):
    """
    Demonstrate multi-query analysis with context caching.
    """
    print("Creating cached context...")
    cache = create_cached_context(
        document,
        ttl_minutes=30,
        system_instruction="You are an expert software architect and technical analyst."
    )
 
    print(f"Cache created. Expiration: {cache.expire_time}")
 
    # Execute multiple queries
    queries = [
        "What are the main architectural components described in this document?",
        "What performance requirements or SLAs are specified?",
        "What are the potential security concerns or vulnerabilities mentioned?",
        "What integration points with external systems are described?",
        "What operational considerations or maintenance concerns are outlined?",
    ]
 
    results = []
    for i, query in enumerate(queries, 1):
        print(f"\nQuery {i}/{len(queries)}: {query[:60]}...")
        result = query_cached_context(cache, query, temperature=0.2)
        results.append({"query": query, "answer": result})
        print(f"Response length: {len(result)} characters")
 
    # Calculate savings
    tokens = genai.GenerativeModel("gemini-2.5-pro").count_tokens(document)
    roi = calculate_caching_roi(tokens.total_tokens, len(queries))
 
    print(f"\n=== Caching ROI Analysis ===")
    print(f"Cost without caching: ${roi['cost_without_caching_usd']}")
    print(f"Cost with caching: ${roi['cost_with_caching_usd']}")
    print(f"Total savings: ${roi['total_savings_usd']} ({roi['savings_percent']}%)")
 
    return results, roi

import google.generativeai as genai
from typing import List, Dict
import json
import time
 
genai.configure(api_key="YOUR_API_KEY")
 
def semantic_chunk_text(
    text: str,
    target_chunk_tokens: int = 800000,
    overlap_tokens: int = 10000,
    split_on: str = "\n\n"
) -> List[str]:
    """
    Split text into semantic chunks with overlap.
 
    Args:
        text: Input text
        target_chunk_tokens: Target size per chunk (conservative estimate)
        overlap_tokens: Overlap between chunks to preserve context
        split_on: Delimiter for semantic boundaries
 
    Returns:
        List of chunks with overlap regions
    """
    # Rough estimate: 4 characters ≈ 1 token
    chars_per_token = 4
    target_chars = target_chunk_tokens * chars_per_token
    overlap_chars = overlap_tokens * chars_per_token
 
    # Split into logical sections
    sections = text.split(split_on)
 
    chunks = []
    current_chunk = ""
 
    for section in sections:
        if len(current_chunk) + len(section) < target_chars:
            current_chunk += section + split_on
        else:
            if current_chunk:
                chunks.append(current_chunk)
            current_chunk = section + split_on
 
    if current_chunk:
        chunks.append(current_chunk)
 
    # Add overlap
    overlapped_chunks = []
    for i, chunk in enumerate(chunks):
        if i > 0:
            previous_overlap = chunks[i - 1][-overlap_chars:] if overlap_chars > 0 else ""
            chunk = previous_overlap + split_on + chunk
 
        overlapped_chunks.append(chunk)
 
    return overlapped_chunks
 
def process_chunks_sequentially(
    chunks: List[str],
    analysis_prompt_template: str
) -> List[Dict]:
    """
    Process chunks sequentially and collect results.
 
    Note: Sequential processing is safer than parallel due to API rate limits.
    """
    model = genai.GenerativeModel("gemini-2.5-pro")
    results = []
 
    for i, chunk in enumerate(chunks):
        print(f"\nProcessing chunk {i+1}/{len(chunks)}...")
        print(f"  Chunk size: {len(chunk)} chars")
 
        prompt = analysis_prompt_template.format(
            chunk_number=i + 1,
            total_chunks=len(chunks),
            content=chunk
        )
 
        response = model.generate_content(prompt)
 
        result = {
            "chunk_number": i + 1,
            "total_chunks": len(chunks),
            "analysis": response.text,
            "timestamp": datetime.now().isoformat(),
        }
 
        results.append(result)
 
        # Respect API rate limits
        if i < len(chunks) - 1:
            time.sleep(2)
 
    return results
 
def merge_chunk_analyses(
    chunk_results: List[Dict]
) -> str:
    """
    Synthesize individual chunk analyses into comprehensive findings.
    """
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    # Combine chunk analyses
    combined_analysis = "\n\n---\n\n".join(
        f"Chunk {r['chunk_number']}:\n{r['analysis']}"
        for r in chunk_results
    )
 
    synthesis_prompt = f"""
Below is analysis of a large document that was split into {len(chunk_results)} chunks.
 
Synthesize these individual chunk analyses into a comprehensive report that:
 
1. Identifies overarching themes across all chunks
2. Highlights relationships between findings in different chunks
3. Identifies any contradictions or inconsistencies
4. Provides unified recommendations
5. Creates a coherent narrative of the complete document
 
Individual chunk analyses:
{combined_analysis}
 
Provide a synthesized analysis that reads as if from a single comprehensive review.
"""
 
    response = model.generate_content(synthesis_prompt, stream=True)
 
    full_response = ""
    for chunk in response:
        if chunk.text:
            full_response += chunk.text
 
    return full_response
 
# Example usage
def process_extra_large_document(document_path: str, analysis_prompt: str):
    """
    End-to-end processing of a document larger than 1M tokens.
    """
    print("Loading document...")
    with open(document_path, "r") as f:
        document = f.read()
 
    print(f"Document size: {len(document):,} characters")
 
    # Split into chunks
    print("\nChunking document...")
    chunks = semantic_chunk_text(
        document,
        target_chunk_tokens=800000,
        overlap_tokens=15000
    )
    print(f"Created {len(chunks)} chunks")
 
    # Process chunks
    print("\nProcessing chunks...")
    chunk_results = process_chunks_sequentially(
        chunks,
        analysis_prompt_template=analysis_prompt
    )
 
    # Merge results
    print("\nSynthesizing results...")
    final_analysis = merge_chunk_analyses(chunk_results)
 
    return final_analysis

import google.generativeai as genai
import time
 
genai.configure(api_key="YOUR_API_KEY")
 
def optimized_long_context_query(
    document_content: str,
    query: str,
    max_output_tokens: int = 3000,
    temperature: float = 0.2,
    use_streaming: bool = True
) -> Dict[str, any]:
    """
    Execute optimized long-context queries with performance monitoring.
 
    Args:
        document_content: Full document content
        query: Analysis query
        max_output_tokens: Max tokens in response (conserve resources)
        temperature: Model temperature (0.2 for factual, 0.7+ for creative)
        use_streaming: Use streaming for memory efficiency
 
    Returns:
        Response with timing metrics
    """
    model = genai.GenerativeModel(
        "gemini-2.5-pro",
        generation_config=genai.types.GenerationConfig(
            max_output_tokens=max_output_tokens,
            temperature=temperature,
            top_p=0.95,
            top_k=40,
        )
    )
 
    full_prompt = f"""
{document_content}
 
---
 
Analysis Request: {query}
 
Provide concise, structured response focusing on key insights.
"""
 
    start_time = time.time()
 
    if use_streaming:
        print("Generating response (streaming)...")
        response = model.generate_content(full_prompt, stream=True)
 
        full_response = ""
        chunk_count = 0
 
        for chunk in response:
            if chunk.text:
                full_response += chunk.text
                chunk_count += 1
 
        elapsed = time.time() - start_time
 
        return {
            "response": full_response,
            "elapsed_seconds": round(elapsed, 2),
            "chunks_streamed": chunk_count,
            "tokens_per_second": len(full_response.split()) / elapsed if elapsed > 0 else 0,
        }
    else:
        response = model.generate_content(full_prompt)
        elapsed = time.time() - start_time
 
        return {
            "response": response.text,
            "elapsed_seconds": round(elapsed, 2),
        }
 
def batch_optimize_queries(
    document: str,
    queries: List[str],
    batch_size: int = 3
) -> List[Dict]:
    """
    Execute multiple optimized queries with batching.
    """
    results = []
 
    for i in range(0, len(queries), batch_size):
        batch = queries[i:i + batch_size]
        print(f"\nBatch {i // batch_size + 1}: {len(batch)} queries")
 
        for query in batch:
            result = optimized_long_context_query(
                document,
                query,
                max_output_tokens=2500,
                temperature=0.15,
                use_streaming=True
            )
 
            results.append({
                "query": query,
                "result": result,
            })
 
            print(f"✓ Query completed in {result['elapsed_seconds']}s")
 
            # Add delay between queries to respect rate limits
            time.sleep(1)
 
    return results

def comprehensive_cost_analysis(
    input_tokens: int,
    output_tokens: int,
    strategy: str = "standard",
    cache_hits: int = 0
) -> Dict[str, float]:
    """
    Calculate costs for different strategies.
 
    Strategies:
    - "standard": Single query, no caching
    - "cached_single": Cache + single follow-up query
    - "cached_multi": Cache + multiple follow-up queries
    """
    INPUT_PRICE = 1.25 / 1_000_000
    CACHE_WRITE = 1.25 / 1_000_000
    CACHE_READ = 0.125 / 1_000_000
    OUTPUT_PRICE = 5.00 / 1_000_000
 
    if strategy == "standard":
        cost = (input_tokens * INPUT_PRICE) + (output_tokens * OUTPUT_PRICE)
        return {
            "strategy": "standard",
            "total_cost_usd": round(cost, 4),
            "input_cost": round(input_tokens * INPUT_PRICE, 4),
            "output_cost": round(output_tokens * OUTPUT_PRICE, 4),
        }
 
    elif strategy == "cached_multi":
        # First query (write cache)
        write_cost = input_tokens * CACHE_WRITE
 
        # Subsequent queries (read cache)
        read_cost = input_tokens * CACHE_READ * cache_hits
 
        # Output tokens for all queries
        output_cost = output_tokens * OUTPUT_PRICE * (1 + cache_hits)
 
        total_cost = write_cost + read_cost + output_cost
 
        # Calculate break-even
        standard_cost = input_tokens * INPUT_PRICE * (1 + cache_hits) + output_cost
        savings = standard_cost - total_cost
        savings_percent = (savings / standard_cost * 100) if standard_cost > 0 else 0
 
        return {
            "strategy": "cached_multi",
            "total_cost_usd": round(total_cost, 4),
            "savings_vs_standard_usd": round(savings, 4),
            "savings_percent": round(savings_percent, 1),
            "cache_hits": cache_hits,
            "breakeven_queries": 2,
        }
 
    return {}
 
# Example calculation
print("=== Cost Analysis Scenarios ===\n")
 
# Scenario 1: Single large document analysis
single = comprehensive_cost_analysis(
    input_tokens=950000,
    output_tokens=2000,
    strategy="standard"
)
print("Scenario 1 (Standard):")
print(f"  Cost: ${single['total_cost_usd']}\n")
 
# Scenario 2: Same document, 5 follow-up queries with caching
cached = comprehensive_cost_analysis(
    input_tokens=950000,
    output_tokens=2000,
    strategy="cached_multi",
    cache_hits=5
)
print("Scenario 2 (Cached, 5 follow-up queries):")
print(f"  Cost: ${cached['total_cost_usd']}")
print(f"  Savings: ${cached['savings_vs_standard_usd']} ({cached['savings_percent']}%)\n")