Spaces:
Sleeping
Sleeping
File size: 3,785 Bytes
59ebe66 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
summarizer.py — glue logic: load → build context → call LLM → collect results
Public API
----------
- summarize_grants(current: list[dict], past_winners: list[dict] | None = None,
*, limit: int | None = None, include_context: bool = False) -> list[dict]
Returns a list of dicts with keys:
grant_id, title, summary_md, context(optional), source_path(optional)
This file intentionally stays light; exporting and CLI live elsewhere.
"""
from __future__ import annotations
from typing import Any, Dict, Iterable, List, Optional
from pathlib import Path
import logging
from .context_builder import build_context
from .llm_client import LLMClient
logger = logging.getLogger(__name__)
# ----------------------------- Core API ---------------------------------------
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]]:
"""Summarize a batch of grants using an LLM.
Parameters
----------
current : list of grant dicts (from data_loader.load_current_grants)
past_winners : optional list of past winner dicts (may be empty)
limit : if provided, process at most this many grants
include_context : whether to include the raw context text in the result
client : optional pre-initialized LLMClient
"""
client = client or LLMClient()
# For this MVP, we pass the *same* past_winners list to every grant.
# Later you can add filtering by theme if you want.
results: List[Dict[str, Any]] = []
items = current[: limit or len(current)]
for i, g in enumerate(items, 1):
grant_id = g.get("id") or g.get("title") or g.get("name") or f"grant_{i}"
title = g.get("title") or g.get("name") or g.get("competition_title") or "(untitled)"
try:
ctx = build_context(g, past_winners)
summary = client.summarize(ctx)
row = {
"grant_id": grant_id,
"title": title,
"summary_md": summary,
}
if include_context:
row["context"] = ctx
if g.get("_path"):
row["source_path"] = g["_path"]
results.append(row)
logger.info("Summarized: %s", title)
except Exception as e: # keep going even if one fails
logger.exception("Failed to summarize %s: %s", title, e)
results.append({
"grant_id": grant_id,
"title": title,
"summary_md": f"Summary failed: {e}",
})
return results
# ----------------------------- Ad-hoc test ------------------------------------
if __name__ == "__main__":
# Tiny smoke test using fake data
logging.basicConfig(level=logging.INFO)
fake_current = [
{
"id": "demo-1",
"title": "AI in Manufacturing",
"sections": {
"summary_raw": "Funding for AI-driven manufacturing improvements.",
"scope_raw": "Projects should demonstrate measurable productivity gains.",
},
"deadline": "2025-12-17",
"funding_amount": "up to £1M",
}
]
fake_history = [
{
"project_title": "Smart Factory Vision",
"lead_org": "Acme Robotics",
"award_amount": "£450,000",
"competition": "Manufacturing AI 2023",
"abstract": "Computer vision for automated QA on production lines.",
}
]
out = summarize_grants(fake_current, fake_history, limit=1, include_context=True)
from pprint import pprint
pprint(out) |