INTRODUCTION TO THE CHALLENGE
Imagine you are a researcher who needs to understand hundreds of scientific papers, or a business analyst reviewing lengthy reports, or a student preparing for exams with mountains of reading material. Reading everything thoroughly would take weeks or months. This is where an intelligent text summarization system becomes invaluable. Such a system can digest large documents and produce concise, accurate summaries that capture the essential information.
In this comprehensive tutorial, we will build a sophisticated text summarization system powered by Large Language Models. We will start with the simplest approach and progressively enhance it to handle real-world challenges. By the end, you will have created a production-ready system that can process documents in multiple formats, understand complex relationships between concepts, and leverage multiple AI agents working in parallel to summarize even the largest documents efficiently.
UNDERSTANDING LARGE LANGUAGE MODELS FOR SUMMARIZATION
Before we dive into building our system, let us understand what Large Language Models are and why they excel at summarization tasks. A Large Language Model is an artificial intelligence system trained on vast amounts of text data. Through this training, the model learns patterns in language, including grammar, facts, reasoning abilities, and even some level of common sense.
When we ask an LLM to summarize text, it does not simply extract sentences randomly. Instead, it comprehends the content, identifies key themes and arguments, and generates a coherent summary in natural language. This is fundamentally different from older statistical methods that merely ranked sentences by importance.
The models we will use can run either locally on your computer or remotely via API services. Local models give you privacy and control, while remote models often provide more powerful capabilities. Our system will support both approaches.
SETTING UP THE FOUNDATION
Let us begin by setting up our development environment. We will use Python as our programming language because it has excellent libraries for working with LLMs and document processing.
First, we need to install the necessary dependencies. Create a new directory for your project and set up a virtual environment:
python -m venv summarization_env
source summarization_env/bin/activate # On Windows: summarization_env\Scripts\activate
Now install the required packages:
pip install torch transformers sentence-transformers
pip install langchain langchain-community langchain-openai
pip install pypdf python-docx beautifulsoup4 markdown
pip install faiss-cpu chromadb
pip install networkx matplotlib
pip install openai anthropic
These packages provide everything we need: PyTorch for running models, Transformers for accessing pre-trained LLMs, LangChain for building LLM applications, document parsers for different file formats, vector databases for retrieval, and graph libraries for advanced analysis.
SUPPORTING MULTIPLE GPU ARCHITECTURES
One critical aspect of our system is supporting various hardware configurations. Different users have different GPUs, and we want our system to work efficiently on all of them. Let us create a device detection module that automatically configures the system for the available hardware.
import torch
import platform
class DeviceManager:
"""
Manages device selection and configuration for different GPU architectures.
Supports NVIDIA CUDA, AMD ROCm, Apple MPS, and CPU fallback.
"""
def __init__(self):
self.device = self._detect_device()
self.device_name = self._get_device_name()
def _detect_device(self):
"""
Automatically detects the best available compute device.
Priority: CUDA > ROCm > MPS > CPU
"""
# Check for NVIDIA CUDA
if torch.cuda.is_available():
return torch.device("cuda")
# Check for Apple Metal Performance Shaders
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return torch.device("mps")
# Check for AMD ROCm (ROCm uses CUDA API compatibility)
if torch.version.hip is not None:
return torch.device("cuda") # ROCm uses cuda device string
# Fallback to CPU
return torch.device("cpu")
def _get_device_name(self):
"""Returns a human-readable name for the current device."""
if self.device.type == "cuda":
if torch.version.hip is not None:
return f"AMD ROCm GPU: {torch.cuda.get_device_name(0)}"
else:
return f"NVIDIA CUDA GPU: {torch.cuda.get_device_name(0)}"
elif self.device.type == "mps":
return "Apple Metal Performance Shaders"
else:
return "CPU"
def get_device(self):
"""Returns the PyTorch device object."""
return self.device
def print_device_info(self):
"""Prints detailed information about the current device."""
print(f"Using device: {self.device_name}")
if self.device.type == "cuda":
print(f"Memory allocated: {torch.cuda.memory_allocated(0) / 1024**3:.2f} GB")
print(f"Memory reserved: {torch.cuda.memory_reserved(0) / 1024**3:.2f} GB")
This DeviceManager class encapsulates all the complexity of hardware detection. When you create an instance, it automatically determines whether you have an NVIDIA GPU with CUDA support, an AMD GPU with ROCm, an Apple Silicon chip with Metal Performance Shaders, or if it should fall back to CPU processing. This abstraction means the rest of our code does not need to worry about hardware specifics.
PROCESSING DIFFERENT DOCUMENT FORMATS
Documents come in many formats. Some are PDFs, others are Word documents, HTML pages, or Markdown files. Our system needs to extract text from all these formats reliably. Let us build a document loader that handles this complexity.
from pypdf import PdfReader
from docx import Document
from bs4 import BeautifulSoup
import markdown
import os
class DocumentLoader:
"""
Loads and extracts text from various document formats.
Supports PDF, DOCX, HTML, MD, and plain text files.
"""
def __init__(self):
self.supported_formats = ['.pdf', '.docx', '.html', '.htm', '.md', '.txt']
def load_document(self, file_path):
"""
Loads a document and returns its text content.
Args:
file_path: Path to the document file
Returns:
Extracted text as a string
Raises:
ValueError: If file format is not supported
FileNotFoundError: If file does not exist
"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
file_extension = os.path.splitext(file_path)[1].lower()
if file_extension not in self.supported_formats:
raise ValueError(f"Unsupported format: {file_extension}")
if file_extension == '.pdf':
return self._load_pdf(file_path)
elif file_extension == '.docx':
return self._load_docx(file_path)
elif file_extension in ['.html', '.htm']:
return self._load_html(file_path)
elif file_extension == '.md':
return self._load_markdown(file_path)
else: # .txt
return self._load_text(file_path)
def _load_pdf(self, file_path):
"""Extracts text from PDF files."""
text_parts = []
reader = PdfReader(file_path)
for page_num, page in enumerate(reader.pages):
page_text = page.extract_text()
if page_text.strip():
text_parts.append(f"--- Page {page_num + 1} ---\n{page_text}")
return "\n\n".join(text_parts)
def _load_docx(self, file_path):
"""Extracts text from Word documents."""
doc = Document(file_path)
text_parts = []
for paragraph in doc.paragraphs:
if paragraph.text.strip():
text_parts.append(paragraph.text)
return "\n\n".join(text_parts)
def _load_html(self, file_path):
"""Extracts text from HTML files."""
with open(file_path, 'r', encoding='utf-8') as file:
html_content = file.read()
soup = BeautifulSoup(html_content, 'html.parser')
# Remove script and style elements
for script in soup(["script", "style"]):
script.decompose()
text = soup.get_text()
# Clean up whitespace
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = '\n'.join(chunk for chunk in chunks if chunk)
return text
def _load_markdown(self, file_path):
"""Extracts text from Markdown files."""
with open(file_path, 'r', encoding='utf-8') as file:
md_content = file.read()
# Convert markdown to HTML first
html = markdown.markdown(md_content)
# Then extract text from HTML
soup = BeautifulSoup(html, 'html.parser')
return soup.get_text()
def _load_text(self, file_path):
"""Loads plain text files."""
with open(file_path, 'r', encoding='utf-8') as file:
return file.read()
This DocumentLoader class provides a unified interface for loading documents regardless of their format. When you call load_document with a file path, it automatically detects the format and uses the appropriate extraction method. For PDFs, it extracts text page by page. For Word documents, it processes paragraphs. For HTML and Markdown, it removes formatting and extracts clean text. This abstraction is crucial because the rest of our system can work with plain text without worrying about the original format.
APPROACH ONE: DIRECT SUMMARIZATION WITH LLMS
Now that we can load documents and detect hardware, let us build our first summarization approach. This is the simplest method: we load the entire document and ask the LLM to summarize it in one go.
from transformers import AutoTokenizer, AutoModelForCausalLM
import openai
class DirectSummarizer:
"""
Performs direct summarization by sending the entire document to an LLM.
Supports both local models (via Transformers) and remote APIs (OpenAI, Anthropic).
"""
def __init__(self, model_type="local", model_name=None, api_key=None, device_manager=None):
"""
Initializes the summarizer.
Args:
model_type: "local" or "remote"
model_name: Name of the model to use
api_key: API key for remote services
device_manager: DeviceManager instance for local models
"""
self.model_type = model_type
self.device_manager = device_manager
if model_type == "local":
if model_name is None:
model_name = "facebook/opt-1.3b" # Default small model
print(f"Loading local model: {model_name}")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16 if device_manager.device.type != "cpu" else torch.float32,
low_cpu_mem_usage=True
)
self.model.to(device_manager.get_device())
self.model.eval()
elif model_type == "remote":
if model_name is None:
model_name = "gpt-3.5-turbo"
self.model_name = model_name
self.api_key = api_key
openai.api_key = api_key
def summarize(self, text, max_summary_length=500):
"""
Generates a summary of the input text.
Args:
text: The text to summarize
max_summary_length: Maximum length of the summary in words
Returns:
Summary as a string
"""
if self.model_type == "local":
return self._summarize_local(text, max_summary_length)
else:
return self._summarize_remote(text, max_summary_length)
def _summarize_local(self, text, max_summary_length):
"""Generates summary using a local model."""
prompt = self._create_prompt(text, max_summary_length)
inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=2048)
inputs = {k: v.to(self.device_manager.get_device()) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_summary_length * 2, # Rough token estimate
temperature=0.7,
do_sample=True,
top_p=0.9
)
summary = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
# Extract only the generated summary (remove the prompt)
summary = summary[len(prompt):].strip()
return summary
def _summarize_remote(self, text, max_summary_length):
"""Generates summary using a remote API."""
prompt = self._create_prompt(text, max_summary_length)
response = openai.ChatCompletion.create(
model=self.model_name,
messages=[
{"role": "system", "content": "You are a helpful assistant that creates concise, accurate summaries."},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=max_summary_length * 2
)
return response.choices[0].message.content.strip()
def _create_prompt(self, text, max_summary_length):
"""Creates an effective summarization prompt."""
return f"""Please provide a comprehensive summary of the following text.
The summary should capture the main ideas, key points, and important details. Keep the summary to approximately {max_summary_length} words.
Text to summarize: {text}
Summary:"""
This DirectSummarizer class demonstrates the basic approach to summarization. We create a carefully crafted prompt that instructs the LLM to summarize the text, then we send this prompt to the model. For local models, we use the Transformers library to load the model onto our detected device and generate text. For remote models, we use the OpenAI API. The key advantage of this approach is its simplicity. The main disadvantage, which we will address next, is that it only works for documents that fit within the model's context window.
UNDERSTANDING CONTEXT WINDOW LIMITATIONS
Every LLM has a maximum amount of text it can process at once, called the context window. Think of it as the model's working memory. Older models might only handle two thousand tokens (roughly fifteen hundred words), while newer models can handle much more, but there is always a limit.
When your document exceeds this limit, you cannot simply feed it to the model in one go. You will get an error, or worse, the model will silently truncate your document and summarize only the beginning. This is where we need a more sophisticated approach.
APPROACH TWO: RETRIEVAL-AUGMENTED GENERATION FOR LARGE DOCUMENTS
Retrieval-Augmented Generation, commonly called RAG, solves the context window problem elegantly. Instead of sending the entire document to the LLM, we break it into smaller chunks, convert these chunks into numerical representations called embeddings, and store them in a vector database. When we want to create a summary, we retrieve the most relevant chunks and summarize those.
Let us understand embeddings first. An embedding is a list of numbers that represents the semantic meaning of text. Similar texts have similar embeddings. For example, the embeddings for "The cat sat on the mat" and "A feline rested on the rug" would be very close to each other in the numerical space, even though the words are different.
from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.config import Settings
import uuid
class RAGSummarizer:
"""
Performs summarization using Retrieval-Augmented Generation.
Breaks documents into chunks, stores them in a vector database,
and retrieves relevant chunks for summarization.
"""
def __init__(self, embedding_model_name="all-MiniLM-L6-v2",
llm_summarizer=None, chunk_size=1000, chunk_overlap=200):
"""
Initializes the RAG summarizer.
Args:
embedding_model_name: Name of the sentence transformer model
llm_summarizer: DirectSummarizer instance for generating summaries
chunk_size: Size of text chunks in characters
chunk_overlap: Overlap between chunks in characters
"""
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.llm_summarizer = llm_summarizer
print(f"Loading embedding model: {embedding_model_name}")
self.embedding_model = SentenceTransformer(embedding_model_name)
# Initialize ChromaDB for vector storage
self.chroma_client = chromadb.Client(Settings(
anonymized_telemetry=False,
allow_reset=True
))
# Create a collection for this session
self.collection_name = f"documents_{uuid.uuid4().hex[:8]}"
self.collection = self.chroma_client.create_collection(
name=self.collection_name,
metadata={"description": "Document chunks for summarization"}
)
def chunk_text(self, text):
"""
Splits text into overlapping chunks.
Args:
text: The text to chunk
Returns:
List of text chunks
"""
chunks = []
start = 0
text_length = len(text)
while start < text_length:
end = start + self.chunk_size
# Try to break at sentence boundaries
if end < text_length:
# Look for sentence ending punctuation
last_period = text.rfind('.', start, end)
last_question = text.rfind('?', start, end)
last_exclamation = text.rfind('!', start, end)
sentence_end = max(last_period, last_question, last_exclamation)
if sentence_end > start:
end = sentence_end + 1
chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)
start = end - self.chunk_overlap
return chunks
def index_document(self, text):
"""
Chunks the document and stores chunks in the vector database.
Args:
text: The document text to index
Returns:
Number of chunks created
"""
chunks = self.chunk_text(text)
print(f"Created {len(chunks)} chunks from document")
# Generate embeddings for all chunks
embeddings = self.embedding_model.encode(chunks, show_progress_bar=True)
# Store in ChromaDB
ids = [f"chunk_{i}" for i in range(len(chunks))]
self.collection.add(
embeddings=embeddings.tolist(),
documents=chunks,
ids=ids
)
return len(chunks)
def retrieve_relevant_chunks(self, query, top_k=5):
"""
Retrieves the most relevant chunks for a query.
Args:
query: The query text
top_k: Number of chunks to retrieve
Returns:
List of relevant text chunks
"""
query_embedding = self.embedding_model.encode([query])[0]
results = self.collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=top_k
)
return results['documents'][0]
def summarize(self, text, max_summary_length=500):
"""
Summarizes a document using RAG approach.
Args:
text: The document text
max_summary_length: Maximum summary length in words
Returns:
Summary as a string
"""
# Index the document
num_chunks = self.index_document(text)
# Retrieve relevant chunks for summarization
# We use a generic query that asks for main points
query = "What are the main topics, key points, and important information in this document?"
# Retrieve more chunks for longer documents
top_k = min(10, num_chunks)
relevant_chunks = self.retrieve_relevant_chunks(query, top_k=top_k)
# Combine retrieved chunks
combined_text = "\n\n".join(relevant_chunks)
# Generate summary using the LLM
summary = self.llm_summarizer.summarize(combined_text, max_summary_length)
return summary
def cleanup(self):
"""Removes the temporary collection from ChromaDB."""
self.chroma_client.delete_collection(self.collection_name)
This RAGSummarizer introduces several important concepts. First, the chunk_text method splits our document into manageable pieces, trying to break at sentence boundaries to preserve meaning. Second, we use a SentenceTransformer model to convert each chunk into an embedding vector. Third, we store these embeddings in ChromaDB, a vector database that allows us to quickly find similar chunks. When summarizing, we retrieve the most relevant chunks and send only those to the LLM.
The beauty of this approach is that it scales to documents of any size. A thousand-page book is no problem because we only send the most relevant chunks to the LLM at any given time. However, this approach has a limitation: it treats each chunk independently and might miss important connections between different parts of the document.
APPROACH THREE: GRAPH-BASED RETRIEVAL-AUGMENTED GENERATION
To understand connections between concepts in a document, we need to go beyond simple chunk retrieval. This is where GraphRAG comes in. GraphRAG builds a knowledge graph from the document, where nodes represent entities like people, places, concepts, and events, and edges represent relationships between them.
For example, in a document about climate change, we might have nodes for "carbon dioxide," "global temperature," and "fossil fuels," with edges showing how they relate to each other. This graph structure helps us understand not just individual facts, but how everything connects together.
import networkx as nx
import re
from collections import defaultdict
class GraphRAGSummarizer:
"""
Performs summarization using Graph-based Retrieval-Augmented Generation.
Extracts entities and relationships to build a knowledge graph,
then uses graph analysis to identify key concepts for summarization.
"""
def __init__(self, llm_summarizer, embedding_model_name="all-MiniLM-L6-v2"):
"""
Initializes the GraphRAG summarizer.
Args:
llm_summarizer: DirectSummarizer instance
embedding_model_name: Name of the embedding model
"""
self.llm_summarizer = llm_summarizer
self.embedding_model = SentenceTransformer(embedding_model_name)
self.graph = nx.Graph()
self.entity_chunks = defaultdict(list) # Maps entities to chunks they appear in
self.chunks = []
def extract_entities(self, text):
"""
Extracts potential entities from text using simple heuristics.
In production, you would use a proper NER model.
Args:
text: Text to extract entities from
Returns:
List of entities
"""
# Simple entity extraction: capitalized words and phrases
# This is a simplified approach; production systems use NER models
entities = []
# Find capitalized words (potential proper nouns)
words = text.split()
current_entity = []
for word in words:
# Remove punctuation for checking
clean_word = re.sub(r'[^\w\s]', '', word)
if clean_word and clean_word[0].isupper() and len(clean_word) > 1:
current_entity.append(word)
else:
if current_entity:
entity = ' '.join(current_entity)
if len(entity) > 2: # Filter very short entities
entities.append(entity)
current_entity = []
if current_entity:
entity = ' '.join(current_entity)
if len(entity) > 2:
entities.append(entity)
return list(set(entities)) # Remove duplicates
def build_graph(self, text, chunk_size=1000):
"""
Builds a knowledge graph from the document.
Args:
text: The document text
chunk_size: Size of chunks for processing
Returns:
NetworkX graph object
"""
# Split into chunks
self.chunks = self._chunk_text(text, chunk_size)
print(f"Building knowledge graph from {len(self.chunks)} chunks...")
# Process each chunk
for chunk_idx, chunk in enumerate(self.chunks):
entities = self.extract_entities(chunk)
# Add entities as nodes
for entity in entities:
if not self.graph.has_node(entity):
self.graph.add_node(entity, mentions=0)
# Increment mention count
self.graph.nodes[entity]['mentions'] += 1
# Track which chunks contain this entity
self.entity_chunks[entity].append(chunk_idx)
# Create edges between entities that co-occur in the same chunk
for i, entity1 in enumerate(entities):
for entity2 in entities[i+1:]:
if self.graph.has_edge(entity1, entity2):
self.graph[entity1][entity2]['weight'] += 1
else:
self.graph.add_edge(entity1, entity2, weight=1)
print(f"Graph built with {self.graph.number_of_nodes()} entities and {self.graph.number_of_edges()} relationships")
return self.graph
def _chunk_text(self, text, chunk_size):
"""Simple text chunking."""
chunks = []
words = text.split()
for i in range(0, len(words), chunk_size):
chunk = ' '.join(words[i:i+chunk_size])
chunks.append(chunk)
return chunks
def identify_key_entities(self, top_k=10):
"""
Identifies the most important entities using graph centrality measures.
Args:
top_k: Number of key entities to identify
Returns:
List of key entities
"""
if self.graph.number_of_nodes() == 0:
return []
# Calculate PageRank to identify important entities
pagerank_scores = nx.pagerank(self.graph, weight='weight')
# Also consider degree centrality
degree_centrality = nx.degree_centrality(self.graph)
# Combine scores (weighted average)
combined_scores = {}
for node in self.graph.nodes():
combined_scores[node] = (
0.6 * pagerank_scores[node] +
0.4 * degree_centrality[node]
)
# Sort by combined score
sorted_entities = sorted(
combined_scores.items(),
key=lambda x: x[1],
reverse=True
)
return [entity for entity, score in sorted_entities[:top_k]]
def get_entity_context(self, entity, context_window=2):
"""
Retrieves chunks containing an entity and its neighbors.
Args:
entity: The entity to get context for
context_window: Number of neighboring chunks to include
Returns:
Combined text context
"""
if entity not in self.entity_chunks:
return ""
chunk_indices = set()
# Get chunks containing this entity
for idx in self.entity_chunks[entity]:
# Add the chunk and its neighbors
for offset in range(-context_window, context_window + 1):
neighbor_idx = idx + offset
if 0 <= neighbor_idx < len(self.chunks):
chunk_indices.add(neighbor_idx)
# Sort and combine chunks
sorted_indices = sorted(chunk_indices)
context_chunks = [self.chunks[i] for i in sorted_indices]
return "\n\n".join(context_chunks)
def summarize(self, text, max_summary_length=500):
"""
Summarizes a document using GraphRAG approach.
Args:
text: The document text
max_summary_length: Maximum summary length in words
Returns:
Summary as a string
"""
# Build the knowledge graph
self.build_graph(text)
# Identify key entities
key_entities = self.identify_key_entities(top_k=8)
print(f"Key entities identified: {', '.join(key_entities[:5])}...")
# Gather context for key entities
context_parts = []
for entity in key_entities:
entity_context = self.get_entity_context(entity, context_window=1)
if entity_context:
context_parts.append(entity_context)
# Combine and deduplicate
combined_context = "\n\n".join(context_parts)
# Create an enhanced prompt that mentions key entities
enhanced_prompt = f"""Please provide a comprehensive summary focusing on these key topics: {', '.join(key_entities)}.
The summary should explain how these topics relate to each other and their significance in the document.
Text to summarize: {combined_context}
Summary:"""
# Generate summary
summary = self.llm_summarizer._summarize_remote(enhanced_prompt, max_summary_length) if self.llm_summarizer.model_type == "remote" else self.llm_summarizer._summarize_local(enhanced_prompt, max_summary_length)
return summary
The GraphRAGSummarizer builds a knowledge graph where entities are nodes and co-occurrence relationships are edges. We use graph algorithms like PageRank to identify the most important entities. These are not just the most frequently mentioned entities, but the ones that are most central to the document's meaning. We then retrieve chunks containing these key entities and their context, giving the LLM a focused view of the most important information and how it connects.
This approach is particularly powerful for complex documents where understanding relationships is crucial, such as research papers, legal documents, or historical texts.
APPROACH FOUR: MULTI-AGENT SUMMARIZATION
Our most sophisticated approach uses multiple AI agents working in parallel. Imagine a team of readers, each assigned different sections of a long document. They read their sections simultaneously, create summaries, and then a coordinator combines these summaries into a final coherent summary. This is exactly what our multi-agent system does.
The architecture consists of worker agents that process document chunks in parallel, and an orchestrator agent that distributes work, synchronizes the workers, and combines their results.
import threading
from queue import Queue
from typing import List, Dict
import time
class WorkerAgent:
"""
A worker agent that processes and summarizes document chunks.
Each worker operates independently and can run in parallel.
"""
def __init__(self, agent_id, llm_summarizer):
"""
Initializes a worker agent.
Args:
agent_id: Unique identifier for this agent
llm_summarizer: DirectSummarizer instance
"""
self.agent_id = agent_id
self.llm_summarizer = llm_summarizer
self.processed_chunks = []
def process_chunk(self, chunk_id, chunk_text, summary_length=200):
"""
Processes a single chunk and generates a summary.
Args:
chunk_id: Identifier for the chunk
chunk_text: The text to summarize
summary_length: Target summary length
Returns:
Dictionary with chunk_id and summary
"""
print(f"Agent {self.agent_id} processing chunk {chunk_id}")
summary = self.llm_summarizer.summarize(chunk_text, max_summary_length=summary_length)
result = {
'chunk_id': chunk_id,
'agent_id': self.agent_id,
'summary': summary,
'original_length': len(chunk_text),
'summary_length': len(summary)
}
self.processed_chunks.append(result)
return result
class OrchestratorAgent:
"""
Orchestrates multiple worker agents to summarize a document in parallel.
Distributes work, synchronizes workers, and combines results.
"""
def __init__(self, num_workers, llm_summarizer):
"""
Initializes the orchestrator.
Args:
num_workers: Number of worker agents to create
llm_summarizer: DirectSummarizer instance for workers and final summary
"""
self.num_workers = num_workers
self.llm_summarizer = llm_summarizer
self.workers = [WorkerAgent(i, llm_summarizer) for i in range(num_workers)]
self.work_queue = Queue()
self.results_queue = Queue()
self.all_results = []
def chunk_document(self, text, num_chunks=None):
"""
Divides document into chunks for parallel processing.
Args:
text: The document text
num_chunks: Number of chunks to create (defaults to num_workers * 2)
Returns:
List of text chunks
"""
if num_chunks is None:
num_chunks = self.num_workers * 2 # Give each worker multiple chunks
words = text.split()
chunk_size = len(words) // num_chunks
chunks = []
for i in range(num_chunks):
start_idx = i * chunk_size
end_idx = start_idx + chunk_size if i < num_chunks - 1 else len(words)
chunk = ' '.join(words[start_idx:end_idx])
chunks.append(chunk)
return chunks
def worker_thread(self, worker):
"""
Thread function for a worker agent.
Continuously processes chunks from the work queue.
Args:
worker: WorkerAgent instance
"""
while True:
try:
# Get work from queue (with timeout to allow thread to exit)
work_item = self.work_queue.get(timeout=1)
if work_item is None: # Poison pill to stop worker
break
chunk_id, chunk_text = work_item
# Process the chunk
result = worker.process_chunk(chunk_id, chunk_text)
# Put result in results queue
self.results_queue.put(result)
# Mark task as done
self.work_queue.task_done()
except:
# Queue is empty, continue waiting
continue
def distribute_work(self, chunks):
"""
Distributes chunks to the work queue.
Args:
chunks: List of text chunks
"""
for chunk_id, chunk in enumerate(chunks):
self.work_queue.put((chunk_id, chunk))
def collect_results(self, expected_count):
"""
Collects results from all workers.
Args:
expected_count: Number of results to collect
Returns:
List of results sorted by chunk_id
"""
results = []
for _ in range(expected_count):
result = self.results_queue.get()
results.append(result)
# Sort by chunk_id to maintain document order
results.sort(key=lambda x: x['chunk_id'])
return results
def combine_summaries(self, summaries, max_final_length=500):
"""
Combines individual chunk summaries into a final coherent summary.
Args:
summaries: List of summary dictionaries
max_final_length: Maximum length of final summary
Returns:
Final combined summary
"""
# Extract just the summary texts
summary_texts = [s['summary'] for s in summaries]
# Combine all summaries
combined = "\n\n".join(summary_texts)
# Create a meta-summary prompt
meta_prompt = f"""The following are summaries of different sections of a document.
Please create a single, coherent summary that integrates all the key information. The final summary should flow naturally and avoid repetition.
Section summaries: {combined}
Integrated summary:"""
# Generate final summary
if self.llm_summarizer.model_type == "remote":
final_summary = self.llm_summarizer._summarize_remote(meta_prompt, max_final_length)
else:
final_summary = self.llm_summarizer._summarize_local(meta_prompt, max_final_length)
return final_summary
def summarize(self, text, max_summary_length=500):
"""
Summarizes a document using parallel worker agents.
Args:
text: The document text
max_summary_length: Maximum length of final summary
Returns:
Final summary
"""
print(f"Starting multi-agent summarization with {self.num_workers} workers")
# Chunk the document
chunks = self.chunk_document(text)
print(f"Document divided into {len(chunks)} chunks")
# Distribute work
self.distribute_work(chunks)
# Start worker threads
threads = []
for worker in self.workers:
thread = threading.Thread(target=self.worker_thread, args=(worker,))
thread.start()
threads.append(thread)
# Wait for all work to be processed
self.work_queue.join()
# Send poison pills to stop workers
for _ in range(self.num_workers):
self.work_queue.put(None)
# Wait for all threads to finish
for thread in threads:
thread.join()
# Collect all results
results = self.collect_results(len(chunks))
print(f"All chunks processed. Combining {len(results)} summaries...")
# Combine summaries into final summary
final_summary = self.combine_summaries(results, max_summary_length)
return final_summary
The multi-agent architecture provides several advantages. First, it dramatically speeds up processing for large documents by utilizing parallel computation. Second, it creates a hierarchical summarization structure where local summaries are combined into a global summary, which often produces better results than trying to summarize everything at once. Third, it is highly scalable because you can adjust the number of workers based on your hardware capabilities.
The OrchestratorAgent manages the entire process. It chunks the document, creates a work queue, spawns worker threads, collects results, and finally combines the individual summaries into a coherent whole. Each WorkerAgent operates independently, processing chunks from the queue without needing to coordinate with other workers.
PUTTING IT ALL TOGETHER: A COMPLETE SYSTEM
Now let us create a unified interface that brings all these approaches together. This SummarizationSystem class allows users to choose which approach to use based on their needs.
class SummarizationSystem:
"""
Unified interface for document summarization.
Supports multiple approaches: direct, RAG, GraphRAG, and multi-agent.
"""
def __init__(self, approach="direct", model_type="local", model_name=None,
api_key=None, num_agents=4):
"""
Initializes the summarization system.
Args:
approach: "direct", "rag", "graphrag", or "multiagent"
model_type: "local" or "remote"
model_name: Name of the model to use
api_key: API key for remote models
num_agents: Number of agents for multi-agent approach
"""
self.approach = approach
# Initialize device manager for local models
self.device_manager = DeviceManager()
self.device_manager.print_device_info()
# Initialize document loader
self.document_loader = DocumentLoader()
# Initialize base LLM summarizer
self.llm_summarizer = DirectSummarizer(
model_type=model_type,
model_name=model_name,
api_key=api_key,
device_manager=self.device_manager
)
# Initialize approach-specific components
if approach == "rag":
self.summarizer = RAGSummarizer(llm_summarizer=self.llm_summarizer)
elif approach == "graphrag":
self.summarizer = GraphRAGSummarizer(llm_summarizer=self.llm_summarizer)
elif approach == "multiagent":
self.summarizer = OrchestratorAgent(
num_workers=num_agents,
llm_summarizer=self.llm_summarizer
)
else: # direct
self.summarizer = self.llm_summarizer
def summarize_file(self, file_path, max_summary_length=500):
"""
Summarizes a document file.
Args:
file_path: Path to the document
max_summary_length: Maximum summary length in words
Returns:
Summary string
"""
print(f"Loading document: {file_path}")
text = self.document_loader.load_document(file_path)
print(f"Document loaded. Length: {len(text)} characters")
print(f"Using {self.approach} approach for summarization")
summary = self.summarizer.summarize(text, max_summary_length)
return summary
def summarize_text(self, text, max_summary_length=500):
"""
Summarizes raw text.
Args:
text: The text to summarize
max_summary_length: Maximum summary length in words
Returns:
Summary string
"""
print(f"Text length: {len(text)} characters")
print(f"Using {self.approach} approach for summarization")
summary = self.summarizer.summarize(text, max_summary_length)
return summary
This unified interface makes it easy to switch between different approaches. You simply specify which approach you want when creating the system, and it handles all the complexity behind the scenes.
COMPLETE PRODUCTION-READY IMPLEMENTATION
Here is the complete, production-ready implementation that integrates all components into a fully functional system. This code is ready to use without any placeholders or mocks.
#!/usr/bin/env python3
"""
Production-Ready LLM-Powered Text Summarization System
This system provides multiple approaches for document summarization:
- Direct summarization for small documents
- RAG for large documents exceeding context windows
- GraphRAG for understanding entity relationships
- Multi-agent parallel processing for maximum efficiency
Supports multiple document formats: PDF, DOCX, HTML, Markdown, TXT
Supports multiple GPU architectures: NVIDIA CUDA, AMD ROCm, Apple MPS, CPU
Supports both local and remote LLM models
"""
import torch
import platform
import os
import sys
from typing import List, Dict, Optional, Tuple
import threading
from queue import Queue
import uuid
import re
from collections import defaultdict
import warnings
# Document processing imports
try:
from pypdf import PdfReader
except ImportError:
print("Warning: pypdf not installed. PDF support disabled.")
PdfReader = None
try:
from docx import Document as DocxDocument
except ImportError:
print("Warning: python-docx not installed. DOCX support disabled.")
DocxDocument = None
try:
from bs4 import BeautifulSoup
except ImportError:
print("Warning: beautifulsoup4 not installed. HTML support disabled.")
BeautifulSoup = None
try:
import markdown
except ImportError:
print("Warning: markdown not installed. Markdown support disabled.")
markdown = None
# LLM and embedding imports
try:
from transformers import AutoTokenizer, AutoModelForCausalLM
except ImportError:
print("Error: transformers not installed. Please install: pip install transformers")
sys.exit(1)
try:
from sentence_transformers import SentenceTransformer
except ImportError:
print("Error: sentence-transformers not installed. Please install: pip install sentence-transformers")
sys.exit(1)
try:
import chromadb
from chromadb.config import Settings
except ImportError:
print("Warning: chromadb not installed. RAG features disabled.")
chromadb = None
try:
import networkx as nx
except ImportError:
print("Warning: networkx not installed. GraphRAG features disabled.")
nx = None
# Suppress warnings for cleaner output
warnings.filterwarnings('ignore')
class DeviceManager:
"""
Manages device selection and configuration for different GPU architectures.
Automatically detects and configures the best available compute device.
"""
def __init__(self):
"""Initializes device manager and detects available hardware."""
self.device = self._detect_device()
self.device_name = self._get_device_name()
def _detect_device(self) -> torch.device:
"""
Detects the best available compute device.
Priority: CUDA (NVIDIA/AMD) > MPS (Apple) > CPU
"""
if torch.cuda.is_available():
return torch.device("cuda")
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return torch.device("mps")
if torch.version.hip is not None:
return torch.device("cuda")
return torch.device("cpu")
def _get_device_name(self) -> str:
"""Returns human-readable device name."""
if self.device.type == "cuda":
if torch.version.hip is not None:
return f"AMD ROCm GPU: {torch.cuda.get_device_name(0)}"
else:
return f"NVIDIA CUDA GPU: {torch.cuda.get_device_name(0)}"
elif self.device.type == "mps":
return "Apple Metal Performance Shaders (MPS)"
else:
return f"CPU: {platform.processor()}"
def get_device(self) -> torch.device:
"""Returns the PyTorch device object."""
return self.device
def print_device_info(self):
"""Prints detailed device information."""
print(f"\n{'='*60}")
print(f"Device Information")
print(f"{'='*60}")
print(f"Using device: {self.device_name}")
if self.device.type == "cuda":
print(f"CUDA version: {torch.version.cuda}")
print(f"Available memory: {torch.cuda.get_device_properties(0).total_memory / 1024**3:.2f} GB")
print(f"{'='*60}\n")
class DocumentLoader:
"""
Loads and extracts text from various document formats.
Provides unified interface for PDF, DOCX, HTML, Markdown, and plain text.
"""
def __init__(self):
"""Initializes document loader with supported formats."""
self.supported_formats = []
if PdfReader is not None:
self.supported_formats.extend(['.pdf'])
if DocxDocument is not None:
self.supported_formats.extend(['.docx'])
if BeautifulSoup is not None:
self.supported_formats.extend(['.html', '.htm'])
if markdown is not None:
self.supported_formats.extend(['.md'])
self.supported_formats.extend(['.txt'])
def load_document(self, file_path: str) -> str:
"""
Loads a document and extracts its text content.
Args:
file_path: Path to the document file
Returns:
Extracted text as string
Raises:
FileNotFoundError: If file does not exist
ValueError: If file format is not supported
"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
file_extension = os.path.splitext(file_path)[1].lower()
if file_extension not in self.supported_formats:
raise ValueError(
f"Unsupported format: {file_extension}. "
f"Supported formats: {', '.join(self.supported_formats)}"
)
if file_extension == '.pdf':
return self._load_pdf(file_path)
elif file_extension == '.docx':
return self._load_docx(file_path)
elif file_extension in ['.html', '.htm']:
return self._load_html(file_path)
elif file_extension == '.md':
return self._load_markdown(file_path)
else:
return self._load_text(file_path)
def _load_pdf(self, file_path: str) -> str:
"""Extracts text from PDF files."""
if PdfReader is None:
raise RuntimeError("PDF support not available. Install pypdf.")
text_parts = []
reader = PdfReader(file_path)
for page_num, page in enumerate(reader.pages):
page_text = page.extract_text()
if page_text.strip():
text_parts.append(page_text)
return "\n\n".join(text_parts)
def _load_docx(self, file_path: str) -> str:
"""Extracts text from Word documents."""
if DocxDocument is None:
raise RuntimeError("DOCX support not available. Install python-docx.")
doc = DocxDocument(file_path)
text_parts = []
for paragraph in doc.paragraphs:
if paragraph.text.strip():
text_parts.append(paragraph.text)
return "\n\n".join(text_parts)
def _load_html(self, file_path: str) -> str:
"""Extracts text from HTML files."""
if BeautifulSoup is None:
raise RuntimeError("HTML support not available. Install beautifulsoup4.")
with open(file_path, 'r', encoding='utf-8', errors='ignore') as file:
html_content = file.read()
soup = BeautifulSoup(html_content, 'html.parser')
for script in soup(["script", "style"]):
script.decompose()
text = soup.get_text()
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
text = '\n'.join(chunk for chunk in chunks if chunk)
return text
def _load_markdown(self, file_path: str) -> str:
"""Extracts text from Markdown files."""
if markdown is None:
raise RuntimeError("Markdown support not available. Install markdown.")
with open(file_path, 'r', encoding='utf-8', errors='ignore') as file:
md_content = file.read()
html = markdown.markdown(md_content)
soup = BeautifulSoup(html, 'html.parser')
return soup.get_text()
def _load_text(self, file_path: str) -> str:
"""Loads plain text files."""
with open(file_path, 'r', encoding='utf-8', errors='ignore') as file:
return file.read()
class DirectSummarizer:
"""
Performs direct summarization by sending entire document to LLM.
Supports both local models (via Transformers) and remote APIs.
"""
def __init__(self, model_type: str = "local", model_name: Optional[str] = None,
api_key: Optional[str] = None, device_manager: Optional[DeviceManager] = None):
"""
Initializes the summarizer.
Args:
model_type: "local" or "remote"
model_name: Name of the model to use
api_key: API key for remote services
device_manager: DeviceManager instance for local models
"""
self.model_type = model_type
self.device_manager = device_manager or DeviceManager()
if model_type == "local":
self._init_local_model(model_name)
elif model_type == "remote":
self._init_remote_model(model_name, api_key)
else:
raise ValueError(f"Invalid model_type: {model_type}")
def _init_local_model(self, model_name: Optional[str]):
"""Initializes a local transformer model."""
if model_name is None:
model_name = "facebook/opt-350m"
print(f"Loading local model: {model_name}")
print("This may take a few minutes on first run...")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
dtype = torch.float16 if self.device_manager.device.type != "cpu" else torch.float32
self.model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=dtype,
low_cpu_mem_usage=True
)
self.model.to(self.device_manager.get_device())
self.model.eval()
print(f"Model loaded successfully on {self.device_manager.device_name}")
def _init_remote_model(self, model_name: Optional[str], api_key: Optional[str]):
"""Initializes remote API configuration."""
self.model_name = model_name or "gpt-3.5-turbo"
self.api_key = api_key
if api_key:
try:
import openai
openai.api_key = api_key
self.openai = openai
except ImportError:
print("Warning: openai package not installed. Remote API unavailable.")
self.openai = None
else:
print("Warning: No API key provided for remote model.")
self.openai = None
def summarize(self, text: str, max_summary_length: int = 500) -> str:
"""
Generates a summary of the input text.
Args:
text: The text to summarize
max_summary_length: Maximum length of summary in words
Returns:
Summary as string
"""
if self.model_type == "local":
return self._summarize_local(text, max_summary_length)
else:
return self._summarize_remote(text, max_summary_length)
def _summarize_local(self, text: str, max_summary_length: int) -> str:
"""Generates summary using local model."""
prompt = self._create_prompt(text, max_summary_length)
inputs = self.tokenizer(
prompt,
return_tensors="pt",
truncation=True,
max_length=1024
)
inputs = {k: v.to(self.device_manager.get_device()) for k, v in inputs.items()}
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=min(max_summary_length * 2, 512),
temperature=0.7,
do_sample=True,
top_p=0.9,
pad_token_id=self.tokenizer.eos_token_id
)
summary = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
summary = summary[len(prompt):].strip()
return summary if summary else "Summary generation failed. Text may be too long."
def _summarize_remote(self, text: str, max_summary_length: int) -> str:
"""Generates summary using remote API."""
if self.openai is None:
return "Remote API not available. Please install openai and provide API key."
prompt = self._create_prompt(text, max_summary_length)
try:
response = self.openai.ChatCompletion.create(
model=self.model_name,
messages=[
{"role": "system", "content": "You are an expert at creating concise, accurate summaries."},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=max_summary_length * 2
)
return response.choices[0].message.content.strip()
except Exception as e:
return f"API error: {str(e)}"
def _create_prompt(self, text: str, max_summary_length: int) -> str:
"""Creates an effective summarization prompt."""
return f"""Please provide a comprehensive summary of the following text.
The summary should:
- Capture the main ideas and key points
- Be approximately {max_summary_length} words
- Be clear and well-organized
- Preserve important details
Text to summarize:
{text}
Summary:"""
class RAGSummarizer:
"""
Performs summarization using Retrieval-Augmented Generation.
Handles documents exceeding context window limits.
"""
def __init__(self, llm_summarizer: DirectSummarizer,
embedding_model_name: str = "all-MiniLM-L6-v2",
chunk_size: int = 1000, chunk_overlap: int = 200):
"""
Initializes RAG summarizer.
Args:
llm_summarizer: DirectSummarizer instance
embedding_model_name: Name of sentence transformer model
chunk_size: Size of text chunks in characters
chunk_overlap: Overlap between chunks in characters
"""
self.llm_summarizer = llm_summarizer
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
print(f"Loading embedding model: {embedding_model_name}")
self.embedding_model = SentenceTransformer(embedding_model_name)
if chromadb is not None:
self.chroma_client = chromadb.Client(Settings(
anonymized_telemetry=False,
allow_reset=True
))
self.collection_name = f"docs_{uuid.uuid4().hex[:8]}"
self.collection = self.chroma_client.create_collection(
name=self.collection_name
)
else:
self.chroma_client = None
print("Warning: ChromaDB not available. Using simple chunking.")
def chunk_text(self, text: str) -> List[str]:
"""
Splits text into overlapping chunks at sentence boundaries.
Args:
text: Text to chunk
Returns:
List of text chunks
"""
chunks = []
start = 0
text_length = len(text)
while start < text_length:
end = start + self.chunk_size
if end < text_length:
last_period = text.rfind('.', start, end)
last_question = text.rfind('?', start, end)
last_exclamation = text.rfind('!', start, end)
sentence_end = max(last_period, last_question, last_exclamation)
if sentence_end > start:
end = sentence_end + 1
chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)
start = end - self.chunk_overlap
return chunks
def index_document(self, text: str) -> int:
"""
Chunks document and stores in vector database.
Args:
text: Document text to index
Returns:
Number of chunks created
"""
chunks = self.chunk_text(text)
print(f"Created {len(chunks)} chunks")
if self.chroma_client is not None:
embeddings = self.embedding_model.encode(chunks, show_progress_bar=False)
ids = [f"chunk_{i}" for i in range(len(chunks))]
self.collection.add(
embeddings=embeddings.tolist(),
documents=chunks,
ids=ids
)
self.chunks = chunks
return len(chunks)
def retrieve_relevant_chunks(self, query: str, top_k: int = 5) -> List[str]:
"""
Retrieves most relevant chunks for a query.
Args:
query: Query text
top_k: Number of chunks to retrieve
Returns:
List of relevant chunks
"""
if self.chroma_client is not None:
query_embedding = self.embedding_model.encode([query])[0]
results = self.collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=min(top_k, len(self.chunks))
)
return results['documents'][0]
else:
return self.chunks[:top_k]
def summarize(self, text: str, max_summary_length: int = 500) -> str:
"""
Summarizes document using RAG approach.
Args:
text: Document text
max_summary_length: Maximum summary length
Returns:
Summary string
"""
num_chunks = self.index_document(text)
query = "What are the main topics, key points, and important information?"
top_k = min(10, num_chunks)
relevant_chunks = self.retrieve_relevant_chunks(query, top_k=top_k)
combined_text = "\n\n".join(relevant_chunks)
summary = self.llm_summarizer.summarize(combined_text, max_summary_length)
if self.chroma_client is not None:
self.chroma_client.delete_collection(self.collection_name)
return summary
class GraphRAGSummarizer:
"""
Performs summarization using Graph-based RAG.
Builds knowledge graph to understand entity relationships.
"""
def __init__(self, llm_summarizer: DirectSummarizer,
embedding_model_name: str = "all-MiniLM-L6-v2"):
"""
Initializes GraphRAG summarizer.
Args:
llm_summarizer: DirectSummarizer instance
embedding_model_name: Name of embedding model
"""
self.llm_summarizer = llm_summarizer
self.embedding_model = SentenceTransformer(embedding_model_name)
if nx is None:
raise RuntimeError("NetworkX required for GraphRAG. Install: pip install networkx")
self.graph = nx.Graph()
self.entity_chunks = defaultdict(list)
self.chunks = []
def extract_entities(self, text: str) -> List[str]:
"""
Extracts entities from text using capitalization heuristics.
Args:
text: Text to extract entities from
Returns:
List of entities
"""
entities = []
words = text.split()
current_entity = []
for word in words:
clean_word = re.sub(r'[^\w\s]', '', word)
if clean_word and len(clean_word) > 1 and clean_word[0].isupper():
current_entity.append(word)
else:
if current_entity:
entity = ' '.join(current_entity)
if len(entity) > 2:
entities.append(entity)
current_entity = []
if current_entity:
entity = ' '.join(current_entity)
if len(entity) > 2:
entities.append(entity)
return list(set(entities))
def build_graph(self, text: str, chunk_size: int = 500) -> nx.Graph:
"""
Builds knowledge graph from document.
Args:
text: Document text
chunk_size: Size of chunks in words
Returns:
NetworkX graph
"""
words = text.split()
self.chunks = []
for i in range(0, len(words), chunk_size):
chunk = ' '.join(words[i:i+chunk_size])
self.chunks.append(chunk)
print(f"Building knowledge graph from {len(self.chunks)} chunks...")
for chunk_idx, chunk in enumerate(self.chunks):
entities = self.extract_entities(chunk)
for entity in entities:
if not self.graph.has_node(entity):
self.graph.add_node(entity, mentions=0)
self.graph.nodes[entity]['mentions'] += 1
self.entity_chunks[entity].append(chunk_idx)
for i, entity1 in enumerate(entities):
for entity2 in entities[i+1:]:
if self.graph.has_edge(entity1, entity2):
self.graph[entity1][entity2]['weight'] += 1
else:
self.graph.add_edge(entity1, entity2, weight=1)
print(f"Graph: {self.graph.number_of_nodes()} entities, {self.graph.number_of_edges()} relationships")
return self.graph
def identify_key_entities(self, top_k: int = 10) -> List[str]:
"""
Identifies most important entities using graph centrality.
Args:
top_k: Number of key entities
Returns:
List of key entities
"""
if self.graph.number_of_nodes() == 0:
return []
pagerank_scores = nx.pagerank(self.graph, weight='weight')
degree_centrality = nx.degree_centrality(self.graph)
combined_scores = {}
for node in self.graph.nodes():
combined_scores[node] = (
0.6 * pagerank_scores[node] +
0.4 * degree_centrality[node]
)
sorted_entities = sorted(
combined_scores.items(),
key=lambda x: x[1],
reverse=True
)
return [entity for entity, score in sorted_entities[:top_k]]
def get_entity_context(self, entity: str, context_window: int = 1) -> str:
"""
Retrieves context for an entity.
Args:
entity: Entity to get context for
context_window: Number of neighboring chunks
Returns:
Combined context text
"""
if entity not in self.entity_chunks:
return ""
chunk_indices = set()
for idx in self.entity_chunks[entity]:
for offset in range(-context_window, context_window + 1):
neighbor_idx = idx + offset
if 0 <= neighbor_idx < len(self.chunks):
chunk_indices.add(neighbor_idx)
sorted_indices = sorted(chunk_indices)
context_chunks = [self.chunks[i] for i in sorted_indices]
return "\n\n".join(context_chunks)
def summarize(self, text: str, max_summary_length: int = 500) -> str:
"""
Summarizes document using GraphRAG.
Args:
text: Document text
max_summary_length: Maximum summary length
Returns:
Summary string
"""
self.build_graph(text)
key_entities = self.identify_key_entities(top_k=8)
print(f"Key entities: {', '.join(key_entities[:5])}...")
context_parts = []
for entity in key_entities:
entity_context = self.get_entity_context(entity, context_window=1)
if entity_context:
context_parts.append(entity_context)
combined_context = "\n\n".join(context_parts)
enhanced_prompt = f"""Please provide a comprehensive summary focusing on these key topics: {', '.join(key_entities)}.
The summary should explain how these topics relate to each other and their significance.
Text to summarize:
{combined_context}
Summary:"""
if self.llm_summarizer.model_type == "remote":
summary = self.llm_summarizer._summarize_remote(enhanced_prompt, max_summary_length)
else:
summary = self.llm_summarizer._summarize_local(enhanced_prompt, max_summary_length)
return summary
class WorkerAgent:
"""
Worker agent that processes document chunks independently.
Designed for parallel execution.
"""
def __init__(self, agent_id: int, llm_summarizer: DirectSummarizer):
"""
Initializes worker agent.
Args:
agent_id: Unique identifier
llm_summarizer: DirectSummarizer instance
"""
self.agent_id = agent_id
self.llm_summarizer = llm_summarizer
self.processed_chunks = []
def process_chunk(self, chunk_id: int, chunk_text: str,
summary_length: int = 200) -> Dict:
"""
Processes and summarizes a chunk.
Args:
chunk_id: Chunk identifier
chunk_text: Text to summarize
summary_length: Target summary length
Returns:
Dictionary with results
"""
print(f"Agent {self.agent_id} processing chunk {chunk_id}")
summary = self.llm_summarizer.summarize(chunk_text, max_summary_length=summary_length)
result = {
'chunk_id': chunk_id,
'agent_id': self.agent_id,
'summary': summary,
'original_length': len(chunk_text),
'summary_length': len(summary)
}
self.processed_chunks.append(result)
return result
class OrchestratorAgent:
"""
Orchestrates multiple worker agents for parallel summarization.
Manages work distribution, synchronization, and result combination.
"""
def __init__(self, num_workers: int, llm_summarizer: DirectSummarizer):
"""
Initializes orchestrator.
Args:
num_workers: Number of worker agents
llm_summarizer: DirectSummarizer instance
"""
self.num_workers = num_workers
self.llm_summarizer = llm_summarizer
self.workers = [WorkerAgent(i, llm_summarizer) for i in range(num_workers)]
self.work_queue = Queue()
self.results_queue = Queue()
def chunk_document(self, text: str, num_chunks: Optional[int] = None) -> List[str]:
"""
Divides document into chunks for parallel processing.
Args:
text: Document text
num_chunks: Number of chunks (defaults to num_workers * 2)
Returns:
List of chunks
"""
if num_chunks is None:
num_chunks = self.num_workers * 2
words = text.split()
chunk_size = max(len(words) // num_chunks, 100)
chunks = []
for i in range(num_chunks):
start_idx = i * chunk_size
end_idx = start_idx + chunk_size if i < num_chunks - 1 else len(words)
chunk = ' '.join(words[start_idx:end_idx])
if chunk.strip():
chunks.append(chunk)
return chunks
def worker_thread(self, worker: WorkerAgent):
"""
Thread function for worker agent.
Args:
worker: WorkerAgent instance
"""
while True:
try:
work_item = self.work_queue.get(timeout=1)
if work_item is None:
break
chunk_id, chunk_text = work_item
result = worker.process_chunk(chunk_id, chunk_text)
self.results_queue.put(result)
self.work_queue.task_done()
except:
continue
def distribute_work(self, chunks: List[str]):
"""
Distributes chunks to work queue.
Args:
chunks: List of text chunks
"""
for chunk_id, chunk in enumerate(chunks):
self.work_queue.put((chunk_id, chunk))
def collect_results(self, expected_count: int) -> List[Dict]:
"""
Collects results from workers.
Args:
expected_count: Number of results to collect
Returns:
List of results sorted by chunk_id
"""
results = []
for _ in range(expected_count):
result = self.results_queue.get()
results.append(result)
results.sort(key=lambda x: x['chunk_id'])
return results
def combine_summaries(self, summaries: List[Dict],
max_final_length: int = 500) -> str:
"""
Combines chunk summaries into final summary.
Args:
summaries: List of summary dictionaries
max_final_length: Maximum final summary length
Returns:
Final combined summary
"""
summary_texts = [s['summary'] for s in summaries]
combined = "\n\n".join(summary_texts)
meta_prompt = f"""The following are summaries of different sections of a document.
Please create a single, coherent summary that integrates all key information.
The final summary should flow naturally and avoid repetition.
Section summaries:
{combined}
Integrated summary:"""
if self.llm_summarizer.model_type == "remote":
final_summary = self.llm_summarizer._summarize_remote(meta_prompt, max_final_length)
else:
final_summary = self.llm_summarizer._summarize_local(meta_prompt, max_final_length)
return final_summary
def summarize(self, text: str, max_summary_length: int = 500) -> str:
"""
Summarizes document using parallel agents.
Args:
text: Document text
max_summary_length: Maximum summary length
Returns:
Final summary
"""
print(f"\nStarting multi-agent summarization with {self.num_workers} workers")
chunks = self.chunk_document(text)
print(f"Document divided into {len(chunks)} chunks")
self.distribute_work(chunks)
threads = []
for worker in self.workers:
thread = threading.Thread(target=self.worker_thread, args=(worker,))
thread.start()
threads.append(thread)
self.work_queue.join()
for _ in range(self.num_workers):
self.work_queue.put(None)
for thread in threads:
thread.join()
results = self.collect_results(len(chunks))
print(f"All chunks processed. Combining {len(results)} summaries...")
final_summary = self.combine_summaries(results, max_summary_length)
return final_summary
class SummarizationSystem:
"""
Unified interface for document summarization.
Provides access to all summarization approaches.
"""
def __init__(self, approach: str = "direct", model_type: str = "local",
model_name: Optional[str] = None, api_key: Optional[str] = None,
num_agents: int = 4):
"""
Initializes summarization system.
Args:
approach: "direct", "rag", "graphrag", or "multiagent"
model_type: "local" or "remote"
model_name: Name of model to use
api_key: API key for remote models
num_agents: Number of agents for multi-agent approach
"""
self.approach = approach
self.device_manager = DeviceManager()
self.device_manager.print_device_info()
self.document_loader = DocumentLoader()
self.llm_summarizer = DirectSummarizer(
model_type=model_type,
model_name=model_name,
api_key=api_key,
device_manager=self.device_manager
)
if approach == "rag":
self.summarizer = RAGSummarizer(llm_summarizer=self.llm_summarizer)
elif approach == "graphrag":
self.summarizer = GraphRAGSummarizer(llm_summarizer=self.llm_summarizer)
elif approach == "multiagent":
self.summarizer = OrchestratorAgent(
num_workers=num_agents,
llm_summarizer=self.llm_summarizer
)
else:
self.summarizer = self.llm_summarizer
def summarize_file(self, file_path: str, max_summary_length: int = 500) -> str:
"""
Summarizes a document file.
Args:
file_path: Path to document
max_summary_length: Maximum summary length
Returns:
Summary string
"""
print(f"\n{'='*60}")
print(f"Loading document: {file_path}")
print(f"{'='*60}")
text = self.document_loader.load_document(file_path)
print(f"Document loaded: {len(text)} characters")
print(f"Using {self.approach} approach")
print(f"{'='*60}\n")
summary = self.summarizer.summarize(text, max_summary_length)
return summary
def summarize_text(self, text: str, max_summary_length: int = 500) -> str:
"""
Summarizes raw text.
Args:
text: Text to summarize
max_summary_length: Maximum summary length
Returns:
Summary string
"""
print(f"\n{'='*60}")
print(f"Text length: {len(text)} characters")
print(f"Using {self.approach} approach")
print(f"{'='*60}\n")
summary = self.summarizer.summarize(text, max_summary_length)
return summary
def main():
"""
Main function demonstrating system usage.
"""
print("\n" + "="*60)
print("LLM-Powered Text Summarization System")
print("="*60)
sample_text = """
Artificial Intelligence has transformed numerous industries over the past decade.
Machine learning algorithms now power everything from recommendation systems to
autonomous vehicles. Deep learning, a subset of machine learning, has been
particularly revolutionary in areas such as computer vision and natural language
processing. Companies like Google, Microsoft, and OpenAI have invested billions
in AI research and development.
The impact of AI extends beyond technology companies. Healthcare providers use
AI for diagnostic assistance and drug discovery. Financial institutions employ
AI for fraud detection and algorithmic trading. Manufacturing companies utilize
AI for quality control and predictive maintenance. Even creative industries
are exploring AI applications in music composition, art generation, and content
creation.
However, the rapid advancement of AI also raises important ethical considerations.
Questions about privacy, bias in algorithms, job displacement, and the potential
for misuse require careful attention. Researchers and policymakers are working
to develop frameworks for responsible AI development and deployment. The goal
is to harness AI's benefits while mitigating potential risks.
Looking forward, AI is expected to become even more integrated into daily life.
Advances in areas like quantum computing and neuromorphic engineering may unlock
new AI capabilities. The development of artificial general intelligence, while
still theoretical, remains a long-term goal for many researchers. As AI continues
to evolve, its impact on society will likely grow even more profound.
"""
print("\nDemonstrating Direct Summarization:")
print("-" * 60)
system = SummarizationSystem(
approach="direct",
model_type="local",
model_name="facebook/opt-350m"
)
summary = system.summarize_text(sample_text, max_summary_length=150)
print("\nSUMMARY:")
print("-" * 60)
print(summary)
print("-" * 60)
print("\n\nSystem demonstration complete!")
print("="*60)
print("\n\nUsage Examples:")
print("-" * 60)
print("1. Direct summarization (small documents):")
print(" system = SummarizationSystem(approach='direct')")
print(" summary = system.summarize_file('document.pdf')")
print()
print("2. RAG summarization (large documents):")
print(" system = SummarizationSystem(approach='rag')")
print(" summary = system.summarize_file('large_document.pdf')")
print()
print("3. GraphRAG (understanding relationships):")
print(" system = SummarizationSystem(approach='graphrag')")
print(" summary = system.summarize_file('research_paper.pdf')")
print()
print("4. Multi-agent (parallel processing):")
print(" system = SummarizationSystem(approach='multiagent', num_agents=4)")
print(" summary = system.summarize_file('book.pdf')")
print()
print("5. Remote API (using GPT):")
print(" system = SummarizationSystem(")
print(" approach='direct',")
print(" model_type='remote',")
print(" model_name='gpt-3.5-turbo',")
print(" api_key='your-api-key'")
print(" )")
print(" summary = system.summarize_text(text)")
print("-" * 60)
if __name__ == "__main__":
main()
CONCLUSION AND FUTURE DIRECTIONS
We have built a comprehensive text summarization system that evolves from simple direct summarization to sophisticated multi-agent architectures. Each approach has its strengths and appropriate use cases.
The direct summarization approach works well for documents that fit within the model's context window. It is simple, fast, and produces good results for shorter texts like articles, blog posts, or short reports.
The RAG approach extends our capabilities to handle arbitrarily large documents by breaking them into chunks and retrieving relevant portions. This is ideal for books, lengthy research papers, or large collections of documents where you need to understand the overall content without reading everything.
GraphRAG adds another dimension by understanding relationships between entities in the text. This is particularly valuable for complex documents where connections between concepts are as important as the concepts themselves. Research papers, legal documents, and historical texts benefit greatly from this approach.
The multi-agent architecture provides maximum efficiency through parallelization. When you need to process very large documents quickly, or when you have multiple documents to summarize, the multi-agent approach can dramatically reduce processing time by utilizing multiple CPU cores or GPUs simultaneously.
The system supports various hardware configurations automatically, from high-end NVIDIA GPUs to Apple Silicon chips to standard CPUs. It handles multiple document formats transparently, allowing users to work with PDFs, Word documents, HTML pages, Markdown files, and plain text without worrying about format conversion.
Future enhancements could include support for more sophisticated entity extraction using named entity recognition models, integration with additional vector databases for improved retrieval performance, support for multilingual summarization, and the ability to generate summaries at different levels of detail based on user preferences. The modular architecture makes these extensions straightforward to implement.
This system demonstrates how modern LLM technology can be harnessed to solve real-world problems. By understanding the strengths and limitations of different approaches, and by building flexible, production-ready code, we create tools that genuinely help people work more efficiently with large amounts of text.
No comments:
Post a Comment