Hacking the Context Window: How to Feed 10M Tokens to Gemini

import google.generativeai as genai

genai.configure(api_key="YOUR_API_KEY")
model = genai.GenerativeModel("gemini-1.5-pro-002")

def count_tokens(text: str) -> int:
    # Use the model's own tokenizer — never estimate with word count
    response = model.count_tokens(text)
    return response.total_tokens

with open("large_document.txt", "r") as f:
    content = f.read()

token_count = count_tokens(content)
print(f"Document: {token_count:,} tokens")
print(f"Window utilization: {token_count / 2_000_000:.1%}")

import re

def compress_for_context(text: str) -> str:
    # Remove blank lines — they're invisible tokens eating your budget
    text = re.sub(r'\n{3,}', '\n\n', text)
    
    # Collapse repeated whitespace
    text = re.sub(r'[ \t]{2,}', ' ', text)
    
    # Strip comment-only lines in code (optional — keep if comments carry meaning)
    text = re.sub(r'^\s*#.*\n', '', text, flags=re.MULTILINE)
    
    return text.strip()

def compress_logs(log_text: str, keep_levels: list = ["ERROR", "WARN"]) -> str:
    # For log files: keep only relevant severity levels
    lines = log_text.splitlines()
    filtered = [
        line for line in lines
        if any(level in line for level in keep_levels)
    ]
    return "\n".join(filtered)

original = open("app.log").read()
compressed = compress_logs(original, keep_levels=["ERROR", "WARN", "CRITICAL"])

before = count_tokens(original)
after = count_tokens(compressed)
print(f"Reduced from {before:,} to {after:,} tokens ({(1 - after/before):.0%} savings)")

def build_context_payload(sections: dict[str, str]) -> str:
    """
    sections: {"Section Name": "content...", ...}
    Returns a structured string optimized for large context retrieval.
    """
    
    map_header = "## CONTEXT MAP\n"
    map_header += "This context contains the following sections:\n"
    
    for i, name in enumerate(sections.keys(), 1):
        token_est = count_tokens(sections[name])
        map_header += f"{i}. {name} (~{token_est:,} tokens)\n"
    
    map_header += "\n---\n\n"
    
    # Build the full payload
    body = ""
    for name, content in sections.items():
        body += f"## SECTION: {name}\n\n"
        body += content
        body += "\n\n---\n\n"
    
    return map_header + body

# Example usage
sections = {
    "API Contracts": open("api_contracts.md").read(),
    "Database Schema": open("schema.sql").read(),
    "Application Logs (last 24h)": compress_logs(open("app.log").read()),
    "Recent Git Commits": open("git_log.txt").read(),
}

payload = build_context_payload(sections)
print(f"Total payload: {count_tokens(payload):,} tokens")

import google.generativeai as genai
from google.generativeai.types import GenerationConfig

def query_with_large_context(
    payload: str,
    question: str,
    model_name: str = "gemini-1.5-pro-002"
) -> str:
    
    model = genai.GenerativeModel(
        model_name=model_name,
        generation_config=GenerationConfig(
            temperature=0.1,      # Low temp = more deterministic on factual retrieval
            max_output_tokens=8192,
            candidate_count=1,    # Never request multiple candidates on large context
        )
    )
    
    # System instruction goes here, NOT in the context payload
    system = (
        "You are analyzing a structured context. "
        "Always cite the section name when referencing specific information. "
        "If information is not present in the context, say so explicitly."
    )
    
    prompt = f"{system}\n\n{payload}\n\n## QUESTION\n\n{question}"
    
    token_count = count_tokens(prompt)
    print(f"Sending {token_count:,} tokens ({token_count/2_000_000:.1%} of window)")
    
    # Stream the response — don't wait for full completion on large outputs
    response = model.generate_content(prompt, stream=True)
    
    result = ""
    for chunk in response:
        result += chunk.text
        print(chunk.text, end="", flush=True)
    
    return result

from dataclasses import dataclass

@dataclass
class ChunkSummary:
    chunk_id: int
    source: str
    summary: str
    key_facts: list[str]

def summarize_chunk(chunk: str, chunk_id: int, source: str) -> ChunkSummary:
    """Compress one chunk into a structured summary."""
    model = genai.GenerativeModel("gemini-1.5-pro-002")
    
    prompt = f"""Analyze this content and respond in this exact format:

SUMMARY: [2-3 sentence summary]
KEY_FACTS:
- [fact 1]
- [fact 2]
- [fact 3]
(up to 10 facts)

CONTENT:
{chunk}"""
    
    response = model.generate_content(prompt)
    text = response.text
    
    # Parse the structured response
    summary_match = re.search(r'SUMMARY:\s*(.+?)(?=KEY_FACTS:)', text, re.DOTALL)
    facts_match = re.findall(r'- (.+)', text)
    
    return ChunkSummary(
        chunk_id=chunk_id,
        source=source,
        summary=summary_match.group(1).strip() if summary_match else "",
        key_facts=facts_match
    )

def process_massive_dataset(documents: dict[str, str], question: str) -> str:
    """
    Process more data than fits in one context window.
    Map: summarize each document independently
    Reduce: combine summaries and answer the question
    """
    
    summaries = []
    
    for source, content in documents.items():
        # Chunk documents that are themselves too large
        tokens = count_tokens(content)
        
        if tokens < 800_000:  # Fits in one call at safe utilization
            summary = summarize_chunk(content, len(summaries), source)
            summaries.append(summary)
        else:
            # Split into ~500k token chunks with overlap
            words = content.split()
            chunk_size = 300_000  # Conservative estimate: ~4 chars/token
            overlap = 10_000
            
            for i in range(0, len(words), chunk_size - overlap):
                chunk = " ".join(words[i:i + chunk_size])
                summary = summarize_chunk(chunk, len(summaries), f"{source} (part {i//chunk_size + 1})")
                summaries.append(summary)
    
    # Build the reduce payload from summaries
    reduce_context = "## DOCUMENT SUMMARIES\n\n"
    for s in summaries:
        reduce_context += f"### {s.source}\n"
        reduce_context += f"{s.summary}\n\n"
        reduce_context += "Key facts:\n"
        for fact in s.key_facts:
            reduce_context += f"- {fact}\n"
        reduce_context += "\n"
    
    reduce_context += f"\n## QUESTION\n\n{question}"
    
    model = genai.GenerativeModel("gemini-1.5-pro-002")
    response = model.generate_content(reduce_context)
    return response.text

# Test with documents where you know the answer
test_docs = {
    "Q3 Report": "...revenue was $4.2M in Q3, up 18% YoY...",
    "Q4 Report": "...revenue was $5.1M in Q4, up 21% YoY...",
}

result = process_massive_dataset(
    test_docs,
    "What was total revenue across Q3 and Q4, and which quarter grew faster?"
)

print(result)