Performance Improvements - Phase 2

for image in images:
    analysis = gemini_client.analyze_image(image)
    # Blocks for 2-5 seconds per image

tasks = [gemini_client.analyze_image_async(img) for img in images]
analyses = await asyncio.gather(*tasks)
# All images process concurrently

# Upload file 1, wait for analysis
POST /v1/analyses (file1.pdf) → 30-68s

# Upload file 2, wait for analysis
POST /v1/analyses (file2.pdf) → 30-68s

# Upload file 3, wait for analysis
POST /v1/analyses (file3.pdf) → 30-68s

# Total: 90-204 seconds

# Upload all files at once, background analysis
POST /v1/properties/{id}/upload (files=[file1, file2, file3])
  → Upload: 2-5s
  → Categorization: instant
  → Response returned: 2-5s
  → Background analysis: 15-23s per file (in parallel)

# Total user-facing time: 2-5 seconds
# Total processing time: 15-23 seconds (background)

50 documents uploaded
├─ All 50 analyzed immediately
├─ 50 docs × 10 images/doc × 1 Gemini call = 500 API calls
└─ Cost: 33% of daily free tier (1,500/day)

50 documents uploaded
├─ 50 filename checks (free, instant)
├─ ~10 AI scans for ambiguous docs = 10 API calls
├─ ~7 critical/important docs auto-analyzed × 10 images = 70 API calls
├─ ~5 optional docs (when user asks) × 10 images = 50 API calls
└─ Total: 130 API calls (10 + 70 + 50)

async def analyze_image_async(self, image_path: str) -> dict:
    """Async wrapper for synchronous Gemini Vision API"""
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor(max_workers=10) as executor:
        return await loop.run_in_executor(
            executor,
            self.analyze_image,  # Sync method
            image_path
        )

@retry_on_rate_limit(max_retries=3, base_delay=1.0)
def analyze_image(self, image_path: str) -> dict:
    """Analyze image with automatic retry on rate limit"""
    try:
        return self._client.generate_content(...)
    except Exception as e:
        if "rate limit" in str(e).lower():
            raise  # Retry decorator handles this
        logger.error(f"Failed to analyze image: {e}")
        return {"error": str(e)}

# Before
for img in images_to_analyze:
    analysis = self.gemini_client.analyze_image(img.path)
    results.append(analysis)

# After
tasks = [
    self.gemini_client.analyze_image_async(img.path)
    for img in images_to_analyze
]
analyses = await asyncio.gather(*tasks, return_exceptions=True)

# Filter out errors
results = [a for a in analyses if not isinstance(a, Exception)]

for i, analysis in enumerate(analyses):
    if isinstance(analysis, Exception):
        logger.warning(f"Image {i} failed: {analysis}")
        continue  # Continue with other images
    results.append(analysis)

async def categorize(self, filename, first_page_text=None, gemini_client=None):
    # Step 1: Try filename patterns (free, instant)
    result = self.categorize_by_filename(filename)

    # Step 2: If low confidence, use AI scan
    if result.confidence < 0.8 and first_page_text and gemini_client:
        ai_result = await self.categorize_by_ai_scan(filename, first_page_text, gemini_client)
        if ai_result.confidence > result.confidence:
            result = ai_result

    return result

# Upload phase (instant)
for file in files:
    content = await file.read()  # Non-blocking
    categorization = await categorizer.categorize(file.filename)  # Fast
    db.insert({...})  # Quick database insert
    if should_auto_analyze(categorization.category):
        background_tasks.add_task(process_document, content)  # Queue for later

# Return immediately
return {
    "documents_uploaded": len(files),
    "queued_for_analysis": queued_count,
    "documents": [...]
}

async def process_document_background(analysis_id, pdf_content, filename):
    # Extract text and images
    result = await processor.process_pdf(pdf_content, filename)

    # Generate embeddings in parallel
    embedding_tasks = [
        embedding_service.embed_text(chunk["content"])
        for chunk in result.text_chunks
    ]
    embeddings = await asyncio.gather(*embedding_tasks)

    # Store in database
    db.bulk_insert(embeddings)

    # Mark as analyzed
    db.update({"is_analyzed": true})

doc_keywords = {
    "purchase": ["purchase", "agreement", "contract"],
    "loan": ["loan", "mortgage", "financing"],
    "disclosure": ["disclosure", "seller"],
    "appraisal": ["appraisal", "valuation"],
    # ... 8 document types total
}

# Check if query mentions unanalyzed document
for doc in unanalyzed_docs:
    filename = doc["original_filename"].lower()
    for doc_type, keywords in doc_keywords.items():
        if any(kw in query.lower() for kw in keywords) and doc_type in filename:
            queue_analysis(doc["id"])  # Analyze on-demand

User: "What does the purchase agreement say about contingencies?"

Backend:
  1. Detects "purchase" and "agreement" keywords
  2. Finds Purchase_Agreement.pdf (is_analyzed: false)
  3. Queues document for background analysis
  4. Returns: "Analyzing purchase agreement... Check back in 15 seconds"

# Gemini API calls are I/O-bound (waiting for network response)
# Use ThreadPoolExecutor to parallelize
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(call_api, img) for img in images]
    results = [f.result() for f in futures]

# Run all async tasks concurrently
tasks = [analyze_image_async(img) for img in images]
results = await asyncio.gather(*tasks, return_exceptions=True)

@router.post("/upload")
async def upload(files: list[UploadFile], background_tasks: BackgroundTasks):
    # Quick response to user
    background_tasks.add_task(heavy_processing, files)
    return {"status": "queued"}

-- Auto-update property counters on document changes
CREATE TRIGGER trigger_update_property_counts
  AFTER INSERT OR UPDATE OR DELETE ON analyses
  FOR EACH ROW
  EXECUTE FUNCTION update_property_document_counts();

-- Optimize: WHERE property_id = ? AND category = ? AND is_analyzed = ?
CREATE INDEX idx_analyses_property_category
  ON analyses(property_id, category, is_analyzed);

-- Before (no index): Seq Scan (500ms for 10K rows)
-- After (composite index): Index Scan (5ms for 10K rows)
SELECT * FROM analyses
WHERE property_id = 'prop_123'
  AND category IN ('critical', 'important')
  AND is_analyzed = false;

import time
import logging

logger = logging.getLogger(__name__)

# Image analysis timing
start = time.time()
analyses = await asyncio.gather(*tasks)
elapsed = time.time() - start
logger.info(f"Analyzed {len(images)} images in {elapsed:.2f}s ({len(images)/elapsed:.1f} img/s)")

# Categorization accuracy
categorization_counts = db.query(
    "SELECT category, COUNT(*) FROM analyses GROUP BY category"
)
logger.info(f"Categorization distribution: {categorization_counts}")

# API usage tracking
gemini_calls_today = db.query(
    "SELECT COUNT(*) FROM api_usage WHERE date = today() AND service = 'gemini'"
)
logger.info(f"Gemini API calls today: {gemini_calls_today}/1500 (free tier)")

-- Average analysis time by category
SELECT
  category,
  COUNT(*) as total,
  AVG(EXTRACT(EPOCH FROM (updated_at - created_at))) as avg_seconds
FROM analyses
WHERE is_analyzed = true
GROUP BY category;

-- Documents waiting for analysis
SELECT
  category,
  COUNT(*) as waiting
FROM analyses
WHERE is_analyzed = false AND category IN ('critical', 'important')
GROUP BY category;

-- Categorization method distribution
SELECT
  categorization_metadata->>'method' as method,
  COUNT(*) as count,
  ROUND(AVG((categorization_metadata->>'confidence')::float), 2) as avg_confidence
FROM analyses
GROUP BY categorization_metadata->>'method';

# Monitor free tier usage (1,500 requests/day)
daily_usage = get_gemini_usage_today()

if daily_usage > 1200:  # 80% of daily limit
    logger.warning(f"Approaching Gemini API limit: {daily_usage}/1500")
    send_alert("Gemini API usage high", daily_usage)

if daily_usage >= 1500:
    logger.error("Gemini API daily limit reached")
    # Fall back to filename-only categorization
    disable_ai_scan_fallback()

# ThreadPoolExecutor max_workers
# Too low: Underutilized (2-3 workers)
# Too high: Overwhelms API (50+ workers)
# Sweet spot: 10-15 workers for Gemini API

with ThreadPoolExecutor(max_workers=10) as executor:
    # Optimal for Gemini rate limits

@retry_on_rate_limit(max_retries=3, base_delay=1.0)
def call_api():
    # Retry with delays: 1s, 2s, 4s
    # Prevents cascading failures

# Bad: One failure crashes entire batch
results = await asyncio.gather(*tasks)  # Raises exception

# Good: Isolated failures
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        logger.error(f"Task failed: {r}")
        continue
    process_result(r)

# Track background task backlog
pending_analyses = db.query(
    "SELECT COUNT(*) FROM analyses WHERE status = 'processing' AND is_analyzed = false"
)

if pending_analyses > 100:
    logger.warning(f"Background queue backlog: {pending_analyses} documents")
    # Consider adding more workers or rate limiting uploads

# Stream large files instead of loading into memory
async def upload_stream(file: UploadFile):
    content = await file.read(chunk_size=1024*1024)  # 1MB chunks
    # Process in chunks instead of loading entire file

# Log timing for each image
for i, img in enumerate(images):
    start = time.time()
    analysis = await analyze_image_async(img)
    elapsed = time.time() - start
    logger.info(f"Image {i}: {elapsed:.2f}s")

-- Categorization method distribution
SELECT
  categorization_metadata->>'method',
  COUNT(*)
FROM analyses
WHERE created_at > NOW() - INTERVAL '24 hours'
GROUP BY categorization_metadata->>'method';

-- Expected: 80% filename_pattern, 20% ai_scan

-- Documents stuck in processing
SELECT id, original_filename, created_at, status
FROM analyses
WHERE status = 'processing'
  AND is_analyzed = false
  AND created_at < NOW() - INTERVAL '5 minutes';

# Current: One embedding per chunk
for chunk in chunks:
    embedding = embed_text(chunk)

# Future: Batch embedding API
embeddings = embed_texts_batch(chunks)  # 5-10x faster

# Cache common queries
cache_key = hash(query + property_id)
cached_response = redis.get(cache_key)
if cached_response:
    return cached_response

response = generate_response(query)
redis.setex(cache_key, 3600, response)  # 1 hour TTL

# Analyze first 5 pages immediately, rest in background
async def progressive_analysis(pdf):
    # Quick analysis (5 pages)
    quick_result = await analyze_pages(pdf, pages=range(5))
    return quick_result

    # Full analysis in background
    background_tasks.add_task(analyze_remaining_pages, pdf, pages=range(5, total_pages))

# Cache standard forms (Loan Estimate template, etc.)
template_hash = hash(first_page_text)
if template_hash in STANDARD_TEMPLATES:
    return cached_analysis(template_hash)