Spaces:
Sleeping
Sleeping
| 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 ---------- | |
| 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 | |
| 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() | |