phtgiang03's picture
Update app.py
387d494 verified
Raw History Blame Contribute Delete
19.5 kB
import gradio as gr
print(gr.__version__)
import torch
from sentence_transformers import SentenceTransformer
from transformers import pipeline
import numpy as np
import pandas as pd
from typing import List, Dict, Tuple
import re
import spaces
# Global variables to store models
MODELS = {}
@spaces.GPU
def load_models():
"""Load models on GPU with error handling"""
global MODELS
if MODELS: # Return cached models
return MODELS['similarity'], MODELS['entailment']
try:
print("Loading semantic similarity model...")
# Use a more reliable semantic model
similarity_model = SentenceTransformer('all-mpnet-base-v2')
print("Similarity model loaded successfully")
print("Loading entailment model...")
# Use a simpler, more reliable NLI setup
entailment_pipeline = pipeline(
"text-classification",
model="microsoft/DialoGPT-medium", # Fallback model
return_all_scores=False
)
# Try the better model, fallback if it fails
try:
entailment_pipeline = pipeline(
"zero-shot-classification",
model="facebook/bart-large-mnli"
)
print("BART-MNLI model loaded successfully")
except Exception as e:
print(f"BART-MNLI failed, using simpler approach: {e}")
entailment_pipeline = None
# Cache models
MODELS['similarity'] = similarity_model
MODELS['entailment'] = entailment_pipeline
return similarity_model, entailment_pipeline
except Exception as e:
print(f"Error loading models: {e}")
# Fallback to CPU models
try:
similarity_model = SentenceTransformer('all-MiniLM-L6-v2')
MODELS['similarity'] = similarity_model
MODELS['entailment'] = None
return similarity_model, None
except Exception as e2:
print(f"Fallback also failed: {e2}")
return None, None
def parse_bullet_points(text: str) -> List[str]:
"""Extract bullet points from text"""
lines = text.strip().split('\n')
bullet_points = []
for line in lines:
line = line.strip()
# Remove common bullet point markers
line = re.sub(r'^[-β€’*]\s*', '', line)
line = re.sub(r'^\d+\.\s*', '', line)
if line:
bullet_points.append(line)
return bullet_points
@spaces.GPU
def calculate_semantic_similarity(paragraph: str, bullet_point: str, model) -> float:
"""Calculate semantic similarity between paragraph and bullet point"""
if model is None:
return 0.0
try:
embeddings = model.encode([paragraph, bullet_point])
similarity = np.dot(embeddings[0], embeddings[1]) / (
np.linalg.norm(embeddings[0]) * np.linalg.norm(embeddings[1])
)
return float(similarity)
except Exception as e:
print(f"Similarity calculation error: {e}")
return 0.0
@spaces.GPU
def check_entailment_simple(paragraph: str, bullet_point: str, pipeline) -> float:
"""Simple entailment check with multiple fallback methods"""
if pipeline is None:
return calculate_word_overlap_score(paragraph, bullet_point)
try:
# Method 1: Zero-shot classification
if hasattr(pipeline, 'model') and 'bart' in str(pipeline.model):
result = pipeline(paragraph, [bullet_point])
return float(result['scores'][0])
# Method 2: Direct classification (fallback)
else:
# Simple heuristic based on word overlap
return calculate_word_overlap_score(paragraph, bullet_point)
except Exception as e:
print(f"Entailment error: {e}")
return calculate_word_overlap_score(paragraph, bullet_point)
def calculate_word_overlap_score(paragraph: str, bullet_point: str) -> float:
"""Calculate word overlap score as fallback"""
try:
# Convert to lowercase and split
para_words = set(paragraph.lower().split())
bullet_words = set(bullet_point.lower().split())
# Remove common stop words
stop_words = {
'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for',
'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be', 'been', 'being',
'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could',
'should', 'can', 'may', 'might', 'must', 'shall', 'this', 'that',
'these', 'those', 'i', 'you', 'he', 'she', 'it', 'we', 'they', 'me',
'him', 'her', 'us', 'them', 'my', 'your', 'his', 'its', 'our', 'their'
}
para_words = para_words - stop_words
bullet_words = bullet_words - stop_words
if not bullet_words:
return 0.0
# Calculate coverage of bullet point words in paragraph
overlap = len(para_words.intersection(bullet_words))
coverage = overlap / len(bullet_words)
# Also check for partial word matches (stems)
partial_matches = 0
for bullet_word in bullet_words:
for para_word in para_words:
if len(bullet_word) > 3 and len(para_word) > 3:
if bullet_word[:4] == para_word[:4] or bullet_word in para_word or para_word in bullet_word:
partial_matches += 1
break
partial_coverage = partial_matches / len(bullet_words)
# Combine exact and partial matches
final_score = (coverage * 0.7) + (partial_coverage * 0.3)
return min(final_score, 1.0)
except Exception as e:
print(f"Word overlap error: {e}")
return 0.0
def enhanced_semantic_matching(paragraph: str, bullet_point: str) -> float:
"""Enhanced semantic matching using multiple techniques"""
try:
# Method 1: Exact phrase matching
if bullet_point.lower() in paragraph.lower():
return 0.95
# Method 2: Key phrase extraction and matching
bullet_key_phrases = extract_key_phrases(bullet_point)
para_lower = paragraph.lower()
matches = 0
for phrase in bullet_key_phrases:
if phrase in para_lower:
matches += 1
if bullet_key_phrases:
phrase_match_score = matches / len(bullet_key_phrases)
else:
phrase_match_score = 0.0
# Method 3: Word overlap (calculated above)
word_score = calculate_word_overlap_score(paragraph, bullet_point)
# Combine scores
combined_score = (phrase_match_score * 0.6) + (word_score * 0.4)
return min(combined_score, 1.0)
except Exception as e:
print(f"Enhanced matching error: {e}")
return calculate_word_overlap_score(paragraph, bullet_point)
def extract_key_phrases(text: str) -> List[str]:
"""Extract key phrases from text"""
try:
# Simple key phrase extraction
words = text.lower().split()
stop_words = {
'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for',
'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be', 'been', 'being'
}
# Extract 2-3 word phrases
phrases = []
for i in range(len(words) - 1):
if words[i] not in stop_words and words[i+1] not in stop_words:
phrases.append(f"{words[i]} {words[i+1]}")
# Extract single important words (longer than 3 chars, not stop words)
important_words = [w for w in words if len(w) > 3 and w not in stop_words]
return phrases + important_words
except Exception as e:
print(f"Key phrase extraction error: {e}")
return []
@spaces.GPU
def evaluate_coverage(paragraph: str, bullet_points_text: str,
similarity_threshold: float = 0.6,
entailment_threshold: float = 0.5) -> Tuple[Dict, pd.DataFrame]:
"""Evaluate coverage with robust error handling"""
try:
# Load models with error handling
similarity_model, entailment_pipeline = load_models()
# Parse bullet points
bullet_points = parse_bullet_points(bullet_points_text)
if not bullet_points:
return {"error": "No bullet points found"}, pd.DataFrame()
results = []
coverage_summary = {"fully_covered": 0, "partially_covered": 0, "missing": 0}
for i, bullet_point in enumerate(bullet_points, 1):
try:
# Calculate semantic similarity
semantic_score = calculate_semantic_similarity(paragraph, bullet_point, similarity_model)
# Calculate entailment/overlap score
entailment_score = check_entailment_simple(paragraph, bullet_point, entailment_pipeline)
# Calculate enhanced semantic matching
enhanced_score = enhanced_semantic_matching(paragraph, bullet_point)
# Calculate composite score
if similarity_model is not None:
composite_score = (semantic_score * 0.4) + (entailment_score * 0.3) + (enhanced_score * 0.3)
else:
# Fallback when similarity model fails
composite_score = (entailment_score * 0.5) + (enhanced_score * 0.5)
# Determine coverage level
if composite_score >= 0.65:
coverage = "Fully Covered"
coverage_summary["fully_covered"] += 1
elif composite_score >= 0.35:
coverage = "Partially Covered"
coverage_summary["partially_covered"] += 1
else:
coverage = "Missing"
coverage_summary["missing"] += 1
results.append({
"Bullet Point #": i,
"Content": bullet_point[:80] + "..." if len(bullet_point) > 80 else bullet_point,
"Semantic Sim": f"{semantic_score:.3f}",
"Entailment/Overlap": f"{entailment_score:.3f}",
"Enhanced Match": f"{enhanced_score:.3f}",
"Composite Score": f"{composite_score:.3f}",
"Coverage Status": coverage
})
except Exception as e:
print(f"Error processing bullet point {i}: {e}")
results.append({
"Bullet Point #": i,
"Content": bullet_point[:80] + "..." if len(bullet_point) > 80 else bullet_point,
"Semantic Sim": "Error",
"Entailment/Overlap": "Error",
"Enhanced Match": "Error",
"Composite Score": "Error",
"Coverage Status": "Error"
})
# Create summary
total = len(bullet_points)
summary = {
"total_bullet_points": total,
"fully_covered": coverage_summary["fully_covered"],
"partially_covered": coverage_summary["partially_covered"],
"missing": coverage_summary["missing"],
"fully_covered_pct": round((coverage_summary["fully_covered"] / total) * 100, 1),
"partially_covered_pct": round((coverage_summary["partially_covered"] / total) * 100, 1),
"missing_pct": round((coverage_summary["missing"] / total) * 100, 1)
}
return summary, pd.DataFrame(results)
except Exception as e:
print(f"Main evaluation error: {e}")
return {"error": f"Evaluation failed: {str(e)}"}, pd.DataFrame()
def create_summary_text(summary: Dict) -> str:
"""Create formatted summary text"""
if "error" in summary:
return f"❌ Error: {summary['error']}"
return f"""
## πŸ“Š Coverage Summary
**Total Bullet Points:** {summary['total_bullet_points']}
βœ… **Fully Covered:** {summary['fully_covered']} ({summary['fully_covered_pct']}%)
🟑 **Partially Covered:** {summary['partially_covered']} ({summary['partially_covered_pct']}%)
❌ **Missing:** {summary['missing']} ({summary['missing_pct']}%)
### Overall Coverage Score: {summary['fully_covered_pct'] + (summary['partially_covered_pct'] * 0.5):.1f}%
"""
def process_files(paragraph_file, bullet_points_file, similarity_threshold, entailment_threshold):
"""Process uploaded files and evaluate coverage"""
try:
if paragraph_file is None:
return "Please upload a paragraph file", None, ""
with open(paragraph_file, 'r', encoding='utf-8') as f:
paragraph_content = f.read()
if bullet_points_file is None:
return "Please upload a bullet points file", None, ""
with open(bullet_points_file, 'r', encoding='utf-8') as f:
bullet_points_content = f.read()
# Evaluate coverage
summary, results_df = evaluate_coverage(
paragraph_content,
bullet_points_content,
similarity_threshold,
entailment_threshold
)
summary_text = create_summary_text(summary)
return summary_text, results_df, ""
except Exception as e:
return f"❌ Error processing files: {str(e)}", None, ""
def process_text(paragraph_text, bullet_points_text, similarity_threshold, entailment_threshold):
"""Process text inputs and evaluate coverage"""
try:
if not paragraph_text.strip():
return "Please enter a paragraph", None
if not bullet_points_text.strip():
return "Please enter bullet points", None
# Evaluate coverage
summary, results_df = evaluate_coverage(
paragraph_text,
bullet_points_text,
similarity_threshold,
entailment_threshold
)
summary_text = create_summary_text(summary)
return summary_text, results_df
except Exception as e:
return f"❌ Error: {str(e)}", None
# Create Gradio interface
with gr.Blocks(title="Robust Content Coverage Evaluator", theme=gr.themes.Soft()) as demo:
gr.Markdown("""
# πŸ“‹ Robust Content Coverage Evaluator
**Multi-Method Analysis** with error handling and fallbacks:
- **Semantic Similarity** (MPNet/MiniLM models with fallback)
- **Entailment/Word Overlap** (BART-MNLI with intelligent fallback)
- **Enhanced Matching** (phrase detection + word overlap)
- **Composite Scoring** (weighted combination of all methods)
πŸ›‘οΈ **Built with robust error handling for reliable results!**
""")
with gr.Tabs():
# File Upload Tab
with gr.TabItem("πŸ“ File Upload"):
with gr.Row():
with gr.Column():
paragraph_file = gr.File(
label="Upload Paragraph File (.txt)",
file_types=[".txt"]
)
bullet_file = gr.File(
label="Upload Bullet Points File (.txt)",
file_types=[".txt"]
)
with gr.Column():
gr.Markdown("### βš™οΈ Thresholds")
sim_threshold = gr.Slider(
minimum=0.1, maximum=1.0, value=0.6, step=0.05,
label="Semantic Similarity Threshold"
)
ent_threshold = gr.Slider(
minimum=0.1, maximum=1.0, value=0.5, step=0.05,
label="Entailment Threshold"
)
eval_btn_file = gr.Button("πŸ” Evaluate Coverage", variant="primary", size="lg")
summary_output_file = gr.Markdown()
results_output_file = gr.Dataframe(label="Detailed Results")
error_output_file = gr.Textbox(label="Status", visible=False)
# Text Input Tab
with gr.TabItem("✏️ Text Input"):
with gr.Row():
with gr.Column():
paragraph_input = gr.Textbox(
label="Paragraph (File A Content)",
placeholder="Paste your paragraph here...",
lines=8
)
with gr.Column():
bullet_input = gr.Textbox(
label="Bullet Points (File B Content)",
placeholder="β€’ First bullet point\nβ€’ Second bullet point\nβ€’ Third bullet point...",
lines=8
)
with gr.Row():
sim_threshold_text = gr.Slider(
minimum=0.1, maximum=1.0, value=0.6, step=0.05,
label="Semantic Similarity Threshold"
)
ent_threshold_text = gr.Slider(
minimum=0.1, maximum=1.0, value=0.5, step=0.05,
label="Entailment Threshold"
)
eval_btn_text = gr.Button("πŸ” Evaluate Coverage", variant="primary", size="lg")
summary_output_text = gr.Markdown()
results_output_text = gr.Dataframe(label="Detailed Results")
# Examples
with gr.Accordion("πŸ’‘ Example", open=False):
gr.Examples(
examples=[
[
"Machine learning is a subset of artificial intelligence that enables computers to learn and improve from experience without being explicitly programmed. It uses algorithms to analyze data, identify patterns, and make predictions or decisions.",
"β€’ Machine learning is part of AI\nβ€’ Computers learn from experience\nβ€’ No explicit programming needed\nβ€’ Uses algorithms for data analysis\nβ€’ Identifies patterns in data\nβ€’ Makes predictions and decisions\nβ€’ Requires large datasets for training"
]
],
inputs=[paragraph_input, bullet_input],
label="Try this example"
)
gr.Markdown("""
### πŸ”§ **Robust Features:**
- **Automatic Fallbacks**: If advanced models fail, switches to reliable word-based analysis
- **Multiple Scoring Methods**: Semantic similarity + entailment + enhanced phrase matching
- **Error Recovery**: Continues processing even if individual components fail
- **Improved Paraphrase Detection**: Better handling of rephrased content
""")
# Event handlers
eval_btn_file.click(
fn=process_files,
inputs=[paragraph_file, bullet_file, sim_threshold, ent_threshold],
outputs=[summary_output_file, results_output_file, error_output_file]
)
eval_btn_text.click(
fn=process_text,
inputs=[paragraph_input, bullet_input, sim_threshold_text, ent_threshold_text],
outputs=[summary_output_text, results_output_text]
)
if __name__ == "__main__":
demo.launch()