import os import re import hashlib from functools import lru_cache from typing import Iterable, List, Tuple import gradio as gr # --- Optional torch device selection for speed --- try: import torch TORCH_AVAILABLE = True if torch.backends.mps.is_available(): DEVICE = 0 TORCH_DEVICE_STR = "mps" elif torch.cuda.is_available(): DEVICE = 0 TORCH_DEVICE_STR = "cuda" else: DEVICE = -1 TORCH_DEVICE_STR = "cpu" except Exception: TORCH_AVAILABLE = False DEVICE = -1 TORCH_DEVICE_STR = "cpu" from transformers import pipeline # DeepEval (optional judge) from deepeval.metrics import SummarizationMetric, DAGMetric from deepeval.metrics.dag import DeepAcyclicGraph, BinaryJudgementNode, VerdictNode from deepeval.test_case import LLMTestCase # Lightweight NLP for entities & sentence split try: import spacy _NLP = spacy.load("en_core_web_sm") except Exception: _NLP = None # ---------- Config ---------- DEFAULT_SUMMARISER_MODEL = os.getenv("SUMMARISER_MODEL", "sshleifer/distilbart-cnn-12-6") DEFAULT_JUDGE_MODEL = os.getenv("JUDGE_MODEL", "gpt-4.1-mini") ENV_USE_DAG = os.getenv("USE_DAG", "false").lower() in {"1", "true", "yes"} # Common HF models to try; you can still type any repo id AVAILABLE_HF_MODELS = [ "sshleifer/distilbart-cnn-12-6", # fast baseline "facebook/bart-large-cnn", # stronger, slower "google/pegasus-xsum", # concise "google/pegasus-cnn_dailymail", # longer context "philschmid/bart-large-cnn-samsum" # dialogue-ish but often tidy ] # ---------- Sentence utils ---------- _SENT_SPLIT = re.compile(r"(?<=[.!?])\s+") _NUM_RE = re.compile(r"""(?: [+-]?\d{1,3}(?:,\d{3})*(?:\.\d+)? # 1,234.56 | [+-]?\d+\.\d+ # 12.34 | [+-]?\d+% # 12% | \d{4}-\d{2}-\d{2} # 2025-10-31 | \d{2}/\d{2}/\d{4} # 31/10/2025 )""", re.X) _TICKER_LIKE = re.compile(r"\b[A-Z]{1,5}\b") _ISIN_RE = re.compile(r"\b[A-Z]{2}[A-Z0-9]{9}\d\b") def split_sentences(text: str) -> List[str]: if not text: return [] sents = [s.strip() for s in _SENT_SPLIT.split(text.replace("\n", " ")) if s.strip()] if len(sents) <= 1 and len(text) > 400: sents = [text[i:i+300].strip() for i in range(0, len(text), 300)] return sents def three_sentences_or_less(s: str) -> bool: return len(split_sentences(s)) <= 3 def numbers_in(text: str) -> List[str]: return list({n.strip("%") for n in _NUM_RE.findall(text)}) def entities_in(text: str) -> List[str]: ents: List[str] = [] if _NLP: doc = _NLP(text) ents = [e.text.strip() for e in doc.ents if e.label_ in {"ORG", "PERSON", "PRODUCT"}] ents += _TICKER_LIKE.findall(text) ents += _ISIN_RE.findall(text) seen = set(); out = [] for e in ents: k = e.lower() if k not in seen: seen.add(k); out.append(e) return out # ---------- MMR extractive pre-pass ---------- def tfidf_vectorize(sentences: List[str]): from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity tfidf = TfidfVectorizer(min_df=1, max_df=0.9, ngram_range=(1,2)) X = tfidf.fit_transform(sentences) S = cosine_similarity(X) return X, S def sentence_score(sent: str) -> float: num_weight = len(numbers_in(sent)) ent_weight = len(entities_in(sent)) length_weight = min(len(sent) / 200.0, 1.0) return 0.6 * length_weight + 0.25 * num_weight + 0.15 * ent_weight def mmr_select(sentences: List[str], k: int = 8, lambda_: float = 0.72) -> List[str]: if not sentences: return [] rel = [sentence_score(s) for s in sentences] try: _, S = tfidf_vectorize(sentences) except Exception: S = [[0.0]*len(sentences) for _ in sentences] selected, candidates = [], list(range(len(sentences))) best = max(candidates, key=lambda i: rel[i]) selected.append(best); candidates.remove(best) while candidates and len(selected) < min(k, len(sentences)): def mmr_gain(i): diversity = max(S[i][j] for j in selected) if selected else 0.0 return lambda_ * rel[i] - (1 - lambda_) * diversity nxt = max(candidates, key=mmr_gain) selected.append(nxt); candidates.remove(nxt) selected.sort() return [sentences[i] for i in selected] # ---------- Summariser factory & caching ---------- @lru_cache(maxsize=16) def get_summariser(model_id: str): """ Cached transformers pipeline by model id. """ return pipeline( "summarization", model=model_id, device=DEVICE, framework="pt" if TORCH_AVAILABLE else None, ) def summarise_text_with(model_id: str, text: str, max_words: int = 120) -> str: if not text or not text.strip(): return "" max_len = max(30, int(max_words * 1.6)) min_len = max(10, int(max_len * 0.35)) summariser = get_summariser(model_id) out = summariser( text, max_length=max_len, min_length=min_len, do_sample=False, truncation=True, num_beams=4, no_repeat_ngram_size=3, length_penalty=1.0, early_stopping=True, ) return out[0]["summary_text"].strip() def hybrid_summarise(article: str, model_id: str, max_words: int = 120) -> Tuple[str, List[str], List[str]]: sents = split_sentences(article) support = mmr_select(sents, k=8, lambda_=0.72) support_text = " ".join(support) summary = summarise_text_with(model_id, support_text, max_words=max_words) fact_numbers = numbers_in(" ".join(support)) fact_entities = entities_in(" ".join(support)) return summary, fact_numbers, fact_entities # ---------- Local factuality checks ---------- def check_numeric_coverage(fact_numbers: List[str], summary: str) -> float: if not fact_numbers: return 1.0 s_nums = numbers_in(summary) matched = 0 for n in fact_numbers: n_norm = n.replace(",", "") if n_norm in [x.replace(",", "") for x in s_nums]: matched += 1 return matched / max(len(fact_numbers), 1) def check_entity_overlap(fact_entities: List[str], summary: str) -> float: if not fact_entities: return 1.0 s_ents = {e.lower() for e in entities_in(summary)} base = {e.lower() for e in fact_entities} inter = len(base & s_ents) return inter / max(len(base), 1) def local_confidence(fact_numbers: List[str], fact_entities: List[str], summary: str) -> Tuple[float, str]: num_cov = check_numeric_coverage(fact_numbers, summary) ent_cov = check_entity_overlap(fact_entities, summary) score = 0.65 * num_cov + 0.35 * ent_cov reasons = [ f"Numeric coverage: {num_cov:.2f} ({len(fact_numbers)} numbers tracked)", f"Entity retention: {ent_cov:.2f} ({len(fact_entities)} key entities)", ] if not three_sentences_or_less(summary): score = min(score, 0.7) reasons.append("Local check: summary is wordy (>3 sentences).") return float(score), " | ".join(reasons) # ---------- DeepEval optional judge ---------- def build_faithfulness_dag(judge_model: str) -> DAGMetric: faithful_check = BinaryJudgementNode( criteria=( "You are a strict summarisation judge.\n" "Given FACTS and SUMMARY in the INPUT, answer strictly YES if SUMMARY faithfully " "covers ALL FACTS without adding new information; otherwise NO.\n" "Respond with YES or NO only." ), children=[ VerdictNode(verdict=False, score=0.0), VerdictNode(verdict=True, score=1.0), ], ) dag = DeepAcyclicGraph(root_nodes=[faithful_check]) metric = DAGMetric( name="Faithfulness (single-call)", dag=dag, include_reason=True, verbose_mode=False, strict_mode=False, model=judge_model, ) return metric @lru_cache(maxsize=1024) def judge_faithfulness_cached(pair_hash: str, combined_input: str, judge_model: str) -> Tuple[float, str]: metric = build_faithfulness_dag(judge_model) case = LLMTestCase(input=combined_input, actual_output="") metric.measure(case) return float(metric.score), str(metric.reason) def evaluate_with_deepeval(original: str, summary: str, judge_model: str): concise = three_sentences_or_less(summary) facts = mmr_select(split_sentences(original), k=10, lambda_=0.7) combined_input = "FACTS:\n- " + "\n- ".join(facts) + "\n\nSUMMARY:\n" + summary try: base_score, reason = judge_faithfulness_cached( hashlib.sha256((combined_input + judge_model).encode()).hexdigest(), combined_input, judge_model, ) if base_score >= 1.0 and not concise: final = 0.7 reason = reason + " | Local check: summary is wordy (>3 sentences)." else: final = base_score return final, reason, "DAGMetric (1-call + local checks)" except Exception as e: print("DAGMetric failed, falling back to SummarizationMetric:", repr(e)) metric = SummarizationMetric(include_reason=True, threshold=0.5) case = LLMTestCase(input=original, actual_output=summary) metric.measure(case) score = float(metric.score) if score >= 1.0 and not concise: score = 0.7 return score, str(metric.reason), "SummarizationMetric (+ local concision)" # ---------- Gradio glue ---------- def run(text: str, max_words: int, model_id: str, use_dag: bool, judge_model: str): if not text or not text.strip(): return "", 0.0, "No input provided.", "—" summary, fact_numbers, fact_entities = hybrid_summarise(text, model_id, max_words=max_words) local_score, local_reason = local_confidence(fact_numbers, fact_entities, summary) metric_name = "Local guards (numbers+entities)" final_score = local_score reason = local_reason # Optional DeepEval (UI toggle dominates; env is default) if use_dag and os.getenv("OPENAI_API_KEY"): deepeval_score, deepeval_reason, metric_name = evaluate_with_deepeval(text, summary, judge_model) final_score = 0.7 * deepeval_score + 0.3 * local_score reason = f"{metric_name}: {deepeval_reason} | Local: {local_reason}" return summary, round(final_score, 3), reason, metric_name with gr.Blocks(title="Summarise + Verify (Finance-aware, Multi-model)") as demo: gr.Markdown( "# Summarise + Verify (Finance-aware)\n" "Hybrid extractive→abstractive summarisation with numeric/entity guards.\n" "Pick a summariser model below (type any Hugging Face repo id if it's not listed)." ) with gr.Row(): with gr.Column(): txt = gr.Textbox(label="Input text", lines=12, placeholder="Paste your article…") maxw = gr.Slider(40, 200, value=120, step=10, label="Target summary length (words)") model_dd = gr.Dropdown( label="Summariser model (HF)", choices=AVAILABLE_HF_MODELS, value=DEFAULT_SUMMARISER_MODEL if DEFAULT_SUMMARISER_MODEL in AVAILABLE_HF_MODELS else AVAILABLE_HF_MODELS[0], allow_custom_value=True, info="Type any HF model id, e.g., `facebook/bart-large-cnn`", ) with gr.Row(): use_dag_ck = gr.Checkbox( label="Use DeepEval DAG judge (requires OPENAI_API_KEY)", value=ENV_USE_DAG, ) judge_model_tb = gr.Textbox( label="Judge model", value=DEFAULT_JUDGE_MODEL, placeholder="e.g., gpt-4.1-mini", scale=2, ) go = gr.Button("Summarise & Verify") with gr.Column(): out_sum = gr.Textbox(label="Summary", lines=10) out_conf = gr.Number(label="Confidence (0–1)") out_reason = gr.Textbox(label="Verification details", lines=8) out_metric = gr.Textbox(label="Metric used", value="Local guards (numbers+entities)") go.click(fn=run, inputs=[txt, maxw, model_dd, use_dag_ck, judge_model_tb], outputs=[out_sum, out_conf, out_reason, out_metric]) if __name__ == "__main__": demo.launch()