""" summarizer_optimized.py — High-performance grant summarization with: - PARALLELIZATION: asyncio.gather() for concurrent processing - STREAMING: Yield results as batches complete (for async contexts) - BATCH PROCESSING: 5 grants per API call via clever prompting - SMART CONTEXT: Extract only essential fields (~500 tokens per grant) - CACHING: In-memory cache with 1-hour TTL (Redis-ready pattern) - MODEL OPTIMIZATION: gpt-3.5-turbo for basic summaries (10x cheaper, 2x faster) Performance target: <30 seconds for 30 grants (vs. 7 minutes sequential) Public API ---------- - summarize_grants_optimized(current, past_winners=None, limit=None, include_context=False, client=None, cache=None, stream=False, batch_size=5) -> List[Dict] or AsyncGenerator[Dict] (if stream=True) """ from __future__ import annotations import asyncio import hashlib import json import logging import time from typing import Any, Dict, List, Optional, AsyncGenerator, Tuple from datetime import datetime, timedelta from pathlib import Path from .context_builder import build_context from .llm_client import LLMClient logger = logging.getLogger(__name__) # ================================= CACHING LAYER ================================= class SummaryCache: """Simple in-memory cache with TTL support. Redis-ready: replace dict with redis.Redis.""" def __init__(self, ttl_seconds: int = 3600): self.cache: Dict[str, Tuple[str, float]] = {} # hash -> (summary, timestamp) self.ttl_seconds = ttl_seconds def _hash_grant(self, grant: Dict[str, Any]) -> str: """Create deterministic hash of grant ID and essential fields.""" key_parts = [ grant.get("id", ""), grant.get("title", ""), grant.get("deadline", ""), ] key_str = "|".join(str(p) for p in key_parts) return hashlib.md5(key_str.encode()).hexdigest() def get(self, grant: Dict[str, Any]) -> Optional[str]: """Retrieve cached summary if exists and not expired.""" h = self._hash_grant(grant) if h in self.cache: summary, timestamp = self.cache[h] if time.time() - timestamp < self.ttl_seconds: logger.debug("Cache HIT for %s", grant.get("title", "unknown")) return summary else: del self.cache[h] # Expired return None def set(self, grant: Dict[str, Any], summary: str) -> None: """Store summary in cache.""" h = self._hash_grant(grant) self.cache[h] = (summary, time.time()) logger.debug("Cache SET for %s", grant.get("title", "unknown")) def stats(self) -> Dict[str, int]: """Return cache statistics.""" now = time.time() valid = sum(1 for _, (_, ts) in self.cache.items() if now - ts < self.ttl_seconds) return {"cached": len(self.cache), "valid": valid} # ================================= CONTEXT EXTRACTION ================================= def extract_minimal_context(grant: Dict[str, Any], past_winners: Optional[List[Dict[str, Any]]] = None) -> str: """ Extract only essential fields (~500 tokens per grant). Instead of full HTML, extract: - title, deadline, max_funding, brief description (first 200 words), eligibility """ parts = [] # Title title = grant.get("title") or grant.get("name") or grant.get("competition_title") or "(untitled)" parts.append(f"TITLE: {title}") # Deadline deadline = grant.get("deadline") if deadline: parts.append(f"DEADLINE: {deadline}") # Funding funding = grant.get("funding_amount") or grant.get("max_funding") if funding: parts.append(f"FUNDING: {funding}") # Brief summary/description (first 200 words) summary_raw = None for field in ["summary_raw", "summary", "description", "overview"]: if grant.get("sections", {}).get(field): summary_raw = grant["sections"][field] break if summary_raw: # Truncate to ~200 words words = summary_raw.split()[:200] parts.append(f"DESCRIPTION: {' '.join(words)}") # Eligibility eligibility = None for field in ["eligibility_raw", "eligibility", "who_can_apply"]: if grant.get("sections", {}).get(field): eligibility = grant["sections"][field] break if eligibility: words = eligibility.split()[:150] parts.append(f"ELIGIBILITY: {' '.join(words)}") # Past winners (if provided) if past_winners: parts.append(f"\nPAST WINNERS ({len(past_winners)} records):") for winner in past_winners[:3]: # Only first 3 to save tokens org = winner.get("lead_org", "Unknown") amount = winner.get("award_amount", "Unknown") title_w = winner.get("project_title", "Unknown") parts.append(f" • {org}: {title_w} ({amount})") return "\n".join(parts) # ================================= BATCH SUMMARIZATION ================================= def _build_batch_prompt(grants_batch: List[Dict[str, Any]], contexts: List[str]) -> str: """ Build a single prompt for summarizing multiple grants. Returns plain text summaries rather than JSON to avoid parsing issues. """ prompt_parts = [ f"Summarize these {len(grants_batch)} grants. For EACH grant, provide a DETAILED summary (150-250 words).\n", f"Use this format for each grant:\n", f"### Grant [NUMBER]: [GRANT TITLE]\n", f"[DETAILED SUMMARY]\n\n", ] for i, (grant, ctx) in enumerate(zip(grants_batch, contexts), 1): prompt_parts.append(f"\n--- GRANT {i} ---") prompt_parts.append(ctx) return "\n".join(prompt_parts) async def _summarize_batch_async( grants_batch: List[Dict[str, Any]], contexts: List[str], client: LLMClient, cache: SummaryCache, ) -> List[Dict[str, Any]]: """ Summarize a batch of grants in a single API call. Returns list of {grant_id, title, summary_md} dicts. """ results = [] # Check cache first cached_grants = [] uncached_grants = [] uncached_indices = [] for idx, (grant, ctx) in enumerate(zip(grants_batch, contexts)): cached_summary = cache.get(grant) if cached_summary: results.append({ "grant_id": grant.get("id") or grant.get("title") or f"grant_{idx}", "title": grant.get("title") or grant.get("name") or "(untitled)", "summary_md": cached_summary, }) else: uncached_grants.append(grant) uncached_indices.append(idx) if not uncached_grants: return results # Batch summarize uncached grants try: batch_prompt = _build_batch_prompt(uncached_grants, [contexts[i] for i in uncached_indices]) # Use faster model for batch summaries system_prompt = ( "You are an expert UK grant analyst. Provide DETAILED, THOROUGH summaries for each grant. " "Use markdown formatting. Be comprehensive and informative." ) # Run in thread pool to avoid blocking loop = asyncio.get_event_loop() summary_text = await loop.run_in_executor( None, lambda: client.summarize( batch_prompt, system_text=system_prompt, max_tokens=4000, # Increased to allow detailed summaries (200-250 words per grant) ) ) # Parse plain text response (no JSON) try: summary_text = summary_text.strip() # Split by "### Grant" to get individual summaries import re grant_sections = re.split(r'###\s+Grant\s+\d+:', summary_text) summaries = [] for i, section in enumerate(grant_sections[1:], 1): # Skip first empty split # Extract grant title and summary lines = section.strip().split('\n') if lines: summary = '\n'.join(lines).strip() if summary: summaries.append(summary) # If we didn't get enough summaries, fill with empty ones while len(summaries) < len(uncached_grants): summaries.append("[Summary generation failed]") except Exception as e: logger.warning("Failed to parse batch response: %s", e) # Fallback: return empty summaries summaries = ["[Summary generation failed]" for _ in uncached_grants] # Map summaries back to original grants for orig_idx, (grant, summary) in enumerate(zip(uncached_grants, summaries)): grant_id = grant.get("id") or grant.get("title") or f"grant_{orig_idx}" title = grant.get("title") or grant.get("name") or "(untitled)" result = { "grant_id": grant_id, "title": title, "summary_md": summary, } results.append(result) # Cache the summary cache.set(grant, summary) logger.info("Summarized (batch): %s", title) return results except Exception as e: logger.exception("Batch summarization failed: %s", e) # Fallback: return error messages for grant in uncached_grants: results.append({ "grant_id": grant.get("id") or grant.get("title") or "unknown", "title": grant.get("title") or grant.get("name") or "(untitled)", "summary_md": f"Summary failed: {str(e)[:100]}", }) return results # ================================= ASYNC ORCHESTRATION ================================= async def summarize_grants_async( current: List[Dict[str, Any]], past_winners: Optional[List[Dict[str, Any]]] = None, *, limit: Optional[int] = None, include_context: bool = False, client: Optional[LLMClient] = None, cache: Optional[SummaryCache] = None, batch_size: int = 5, ) -> List[Dict[str, Any]]: """ Async version: Summarize grants in parallel batches. Parameters ---------- current : list of grant dicts past_winners : optional list of past winner dicts limit : if provided, process at most this many grants include_context : whether to include raw context in result client : optional pre-initialized LLMClient cache : optional SummaryCache instance batch_size : number of grants per API call (default: 5) Returns ------- List of dicts with keys: grant_id, title, summary_md, context(optional), source_path(optional) """ client = client or LLMClient({}) cache = cache or SummaryCache(ttl_seconds=3600) items = current[: limit or len(current)] if not items: return [] # Build minimal contexts contexts = [extract_minimal_context(g, past_winners) for g in items] # Create batches batches = [ (items[i:i+batch_size], contexts[i:i+batch_size]) for i in range(0, len(items), batch_size) ] # Process batches in parallel start_time = time.time() batch_results = await asyncio.gather( *[ _summarize_batch_async(batch_items, batch_contexts, client, cache) for batch_items, batch_contexts in batches ], return_exceptions=True ) elapsed = time.time() - start_time logger.info("Summarized %d grants in %.1f seconds (%.2f sec/grant)", len(items), elapsed, elapsed / len(items) if items else 0) # Flatten results results = [] for batch_result in batch_results: if isinstance(batch_result, Exception): logger.error("Batch failed: %s", batch_result) else: results.extend(batch_result) # Add optional fields for result, grant in zip(results, items[:len(results)]): if include_context: result["context"] = extract_minimal_context(grant, past_winners) if grant.get("_path"): result["source_path"] = grant["_path"] # Log cache stats cache_stats = cache.stats() logger.info("Cache stats: %d total, %d valid entries", cache_stats["cached"], cache_stats["valid"]) return results async def summarize_grants_streaming( current: List[Dict[str, Any]], past_winners: Optional[List[Dict[str, Any]]] = None, *, limit: Optional[int] = None, include_context: bool = False, client: Optional[LLMClient] = None, cache: Optional[SummaryCache] = None, batch_size: int = 5, ) -> AsyncGenerator[Dict[str, Any], None]: """ Async generator: Yield results as each batch completes (streaming). Allows UI to display summaries in real-time. """ client = client or LLMClient({}) cache = cache or SummaryCache(ttl_seconds=3600) items = current[: limit or len(current)] if not items: return # Build minimal contexts contexts = [extract_minimal_context(g, past_winners) for g in items] # Create batches batches = [ (items[i:i+batch_size], contexts[i:i+batch_size]) for i in range(0, len(items), batch_size) ] # Process and yield as batches complete for batch_items, batch_contexts in batches: try: batch_results = await _summarize_batch_async(batch_items, batch_contexts, client, cache) for result, grant in zip(batch_results, batch_items): if include_context: result["context"] = extract_minimal_context(grant, past_winners) if grant.get("_path"): result["source_path"] = grant["_path"] yield result except Exception as e: logger.error("Batch streaming failed: %s", e) for grant in batch_items: yield { "grant_id": grant.get("id") or grant.get("title") or "unknown", "title": grant.get("title") or grant.get("name") or "(untitled)", "summary_md": f"Summary failed: {str(e)[:100]}", } # ================================= BACKWARD COMPATIBILITY ================================= def summarize_grants( current: List[Dict[str, Any]], past_winners: Optional[List[Dict[str, Any]]] = None, *, limit: Optional[int] = None, include_context: bool = False, client: Optional[LLMClient] = None, ) -> List[Dict[str, Any]]: """ DEPRECATED: Use summarize_grants_optimized() instead. Synchronous wrapper around async implementation for backward compatibility. """ cache = SummaryCache(ttl_seconds=3600) # Run async version in event loop try: loop = asyncio.get_event_loop() if loop.is_running(): # Already in async context; use synchronous fallback logger.warning("summarize_grants called from async context; performance will be degraded") return _summarize_grants_sync(current, past_winners, limit, include_context, client) except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete( summarize_grants_async(current, past_winners, limit=limit, include_context=include_context, client=client, cache=cache) ) def _summarize_grants_sync( current: List[Dict[str, Any]], past_winners: Optional[List[Dict[str, Any]]] = None, limit: Optional[int] = None, include_context: bool = False, client: Optional[LLMClient] = None, ) -> List[Dict[str, Any]]: """ Fallback synchronous implementation (less efficient). Process grants sequentially with batch API calls. """ from concurrent.futures import ThreadPoolExecutor client = client or LLMClient({}) cache = SummaryCache(ttl_seconds=3600) items = current[: limit or len(current)] contexts = [extract_minimal_context(g, past_winners) for g in items] batches = [ (items[i:i+5], contexts[i:i+5]) for i in range(0, len(items), 5) ] results = [] def process_batch(batch_items, batch_contexts): return asyncio.run( _summarize_batch_async(batch_items, batch_contexts, client, cache) ) # Process batches in thread pool with ThreadPoolExecutor(max_workers=3) as executor: batch_results = list(executor.map( lambda args: process_batch(args[0], args[1]), batches )) # Flatten for batch_result in batch_results: results.extend(batch_result) # Add optional fields for result, grant in zip(results, items[:len(results)]): if include_context: result["context"] = extract_minimal_context(grant, past_winners) if grant.get("_path"): result["source_path"] = grant["_path"] return results # ================================= CONVENIENCE ALIAS ================================= summarize_grants_optimized = summarize_grants_async # Main recommended API # ================================= SMOKE TEST ================================= if __name__ == "__main__": import asyncio logging.basicConfig(level=logging.INFO) fake_current = [ { "id": "demo-1", "title": "AI in Manufacturing", "sections": { "summary_raw": "Funding for AI-driven manufacturing improvements. " * 50, "eligibility_raw": "Open to SMEs and large enterprises.", }, "deadline": "2025-12-17", "funding_amount": "up to £1M", }, { "id": "demo-2", "title": "Green Energy Innovation", "sections": { "summary_raw": "Support for renewable energy projects. " * 50, "eligibility_raw": "Academic institutions and non-profits.", }, "deadline": "2025-11-30", "funding_amount": "£500k-£2M", }, ] # Test cache cache = SummaryCache(ttl_seconds=3600) print("Testing cache...") cache.set(fake_current[0], "Test summary") assert cache.get(fake_current[0]) == "Test summary" print("Cache works!") # Test context extraction print("\nTesting context extraction...") ctx = extract_minimal_context(fake_current[0]) print(f"Context length: {len(ctx)} chars") print(ctx[:200]) print("\nSmoke tests passed!")