Spaces:
Sleeping
Sleeping
File size: 9,722 Bytes
1ae86a7 | 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 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 | """
Text Summarization Module
Handles text summarization using Hugging Face Transformers.
"""
from transformers import pipeline, AutoTokenizer
import torch
from typing import List, Optional
import streamlit as st
import re
class TextSummarizer:
"""Class to handle text summarization using pre-trained models"""
def __init__(self, model_name: str = "facebook/bart-large-cnn"):
"""
Initialize the text summarizer
Args:
model_name: Name of the pre-trained model to use
"""
self.model_name = model_name
self.summarizer = None
self.tokenizer = None
self.max_chunk_length = 1024 # Maximum tokens per chunk
self.min_summary_length = 50
self.max_summary_length = 300
@st.cache_resource
def load_model(_self):
"""
Load the summarization model and tokenizer
"""
try:
# Check if CUDA is available
device = 0 if torch.cuda.is_available() else -1
# Show device info
if torch.cuda.is_available():
st.info(f"🚀 Using GPU acceleration: {torch.cuda.get_device_name()}")
else:
st.info("💻 Using CPU for processing (this may be slower)")
# Load the summarization pipeline
_self.summarizer = pipeline(
"summarization",
model=_self.model_name,
device=device,
torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32
)
# Load tokenizer for text chunking
_self.tokenizer = AutoTokenizer.from_pretrained(_self.model_name)
st.success(f"✅ Model loaded successfully: {_self.model_name}")
return True
except OSError as e:
if "Connection error" in str(e) or "timeout" in str(e).lower():
st.error("❌ Network error: Could not download the model. Please check your internet connection.")
else:
st.error(f"❌ Model loading error: {str(e)}")
return False
except RuntimeError as e:
if "CUDA" in str(e):
st.error("❌ GPU memory error. Trying to use CPU instead...")
try:
_self.summarizer = pipeline(
"summarization",
model=_self.model_name,
device=-1, # Force CPU
torch_dtype=torch.float32
)
_self.tokenizer = AutoTokenizer.from_pretrained(_self.model_name)
st.success("✅ Model loaded successfully on CPU")
return True
except Exception as cpu_e:
st.error(f"❌ Failed to load model on CPU: {str(cpu_e)}")
return False
else:
st.error(f"❌ Runtime error loading model: {str(e)}")
return False
except Exception as e:
st.error(f"❌ Unexpected error loading model: {str(e)}")
return False
def chunk_text(self, text: str) -> List[str]:
"""
Split long text into smaller chunks for processing
Args:
text: Input text to chunk
Returns:
List[str]: List of text chunks
"""
if not self.tokenizer:
# Fallback chunking by sentences if tokenizer not available
sentences = re.split(r'[.!?]+', text)
chunks = []
current_chunk = ""
for sentence in sentences:
if len(current_chunk) + len(sentence) < 2000: # Rough character limit
current_chunk += sentence + ". "
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sentence + ". "
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
# Use tokenizer for precise chunking
tokens = self.tokenizer.encode(text)
chunks = []
for i in range(0, len(tokens), self.max_chunk_length):
chunk_tokens = tokens[i:i + self.max_chunk_length]
chunk_text = self.tokenizer.decode(chunk_tokens, skip_special_tokens=True)
chunks.append(chunk_text)
return chunks
def summarize_chunk(self, chunk: str) -> Optional[str]:
"""
Summarize a single text chunk
Args:
chunk: Text chunk to summarize
Returns:
str: Summary of the chunk or None if summarization fails
"""
try:
# Adjust summary length based on chunk length
chunk_length = len(chunk.split())
max_length = min(self.max_summary_length, max(self.min_summary_length, chunk_length // 3))
min_length = min(self.min_summary_length, max_length // 2)
summary = self.summarizer(
chunk,
max_length=max_length,
min_length=min_length,
do_sample=False,
truncation=True
)
return summary[0]['summary_text']
except Exception as e:
st.warning(f"Error summarizing chunk: {str(e)}")
return None
def format_as_bullets(self, summary_text: str) -> str:
"""
Format summary text as bullet points
Args:
summary_text: Raw summary text
Returns:
str: Formatted bullet points
"""
# Split by sentences and create bullet points
sentences = re.split(r'[.!?]+', summary_text)
bullets = []
for sentence in sentences:
sentence = sentence.strip()
if sentence and len(sentence) > 10: # Filter out very short fragments
bullets.append(f"• {sentence}")
return '\n'.join(bullets)
def summarize_text(self, text: str) -> Optional[str]:
"""
Complete text summarization pipeline
Args:
text: Input text to summarize
Returns:
str: Formatted summary or None if summarization fails
"""
if not text or len(text.strip()) < 100:
st.error("❌ Text is too short to summarize effectively (minimum 100 characters required)")
return None
# Check text length limits
word_count = len(text.split())
if word_count > 10000:
st.warning(f"⚠️ Large text detected ({word_count:,} words). Processing may take several minutes.")
try:
# Load model if not already loaded
if not self.summarizer:
with st.spinner("🤖 Loading AI model..."):
if not self.load_model():
return None
# Chunk the text
chunks = self.chunk_text(text)
if len(chunks) == 0:
st.error("❌ Could not process the text into chunks")
return None
st.info(f"📄 Processing {len(chunks)} text chunk(s)...")
# Summarize each chunk
summaries = []
progress_bar = st.progress(0)
failed_chunks = 0
for i, chunk in enumerate(chunks):
try:
with st.spinner(f"🔄 Summarizing part {i+1} of {len(chunks)}..."):
chunk_summary = self.summarize_chunk(chunk)
if chunk_summary:
summaries.append(chunk_summary)
else:
failed_chunks += 1
except Exception as e:
st.warning(f"⚠️ Failed to summarize chunk {i+1}: {str(e)}")
failed_chunks += 1
continue
progress_bar.progress((i + 1) / len(chunks))
# Check if we have any successful summaries
if not summaries:
st.error("❌ Could not generate any summaries from the text")
return None
if failed_chunks > 0:
st.warning(f"⚠️ {failed_chunks} out of {len(chunks)} chunks failed to process")
# Combine summaries
combined_summary = ' '.join(summaries)
# If we have multiple chunks, summarize the combined summary
if len(chunks) > 1 and len(combined_summary.split()) > 200:
try:
with st.spinner("🔄 Creating final summary..."):
final_summary = self.summarize_chunk(combined_summary)
if final_summary:
combined_summary = final_summary
except Exception as e:
st.warning(f"⚠️ Could not create final summary, using combined chunks: {str(e)}")
# Format as bullet points
formatted_summary = self.format_as_bullets(combined_summary)
if not formatted_summary.strip():
st.error("❌ Generated summary is empty")
return None
return formatted_summary
except MemoryError:
st.error("❌ Out of memory. Please try with a shorter text or restart the application.")
return None
except Exception as e:
st.error(f"❌ Unexpected error during summarization: {str(e)}")
return None
|