INTRODUCTION
In the modern era of information overload, the ability to quickly extract meaningful insights from diverse content types has become essential. This article presents a comprehensive guide to building a sophisticated Large Language Model agent capable of summarizing documents in various formats, source code files, and even images. We will explore how to create a flexible system that supports both local and remote LLM deployments, accommodating different GPU architectures including Nvidia CUDA, AMD ROCm, Intel GPUs, and Apple Metal Performance Shaders.
The journey begins with document summarization, progresses through code analysis, and culminates in multimodal capabilities for image understanding. Throughout this exploration, we will leverage the LangChain framework, which provides robust abstractions for working with language models while maintaining clean architecture principles. Our implementation will be entirely open source, ensuring accessibility and transparency for all developers.
ARCHITECTURAL OVERVIEW AND DESIGN PRINCIPLES
Before diving into implementation details, we must establish a solid architectural foundation. The system we are building follows clean architecture principles, separating concerns into distinct layers. At the core lies the domain layer, which defines our summarization abstractions independent of any specific LLM provider or document format. The application layer orchestrates the summarization workflow, while the infrastructure layer handles concrete implementations for document parsing, LLM communication, and GPU acceleration.
The design embraces the dependency inversion principle, ensuring that high-level modules do not depend on low-level implementation details. This approach allows us to swap LLM providers, document parsers, or GPU backends without affecting the core business logic. We will also implement the strategy pattern for different summarization approaches, enabling the system to adapt its behavior based on content type and user requirements.
SETTING UP THE DEVELOPMENT ENVIRONMENT
Our implementation requires several key dependencies that form the foundation of the summarization agent. The LangChain framework serves as our primary orchestration tool, providing abstractions for LLM interactions and document processing. For local LLM inference, we will use llama-cpp-python, which offers excellent performance across different GPU architectures. The library supports CUDA for Nvidia GPUs, ROCm for AMD hardware, Metal for Apple Silicon, and even CPU-only inference for systems without dedicated accelerators.
Document parsing requires specialized libraries for each format. PyPDF2 and pdfplumber handle PDF extraction, python-docx processes Word documents, python-pptx manages PowerPoint files, and BeautifulSoup4 parses HTML content. For multimodal capabilities, we will integrate the Pillow library for image processing and transformers for vision-language models.
The following configuration establishes our project dependencies:
# requirements.txt - Project dependencies with version specifications
# Core LLM framework and utilities
langchain==0.1.0
langchain-community==0.0.13
# Local LLM inference with multi-GPU support
llama-cpp-python==0.2.27
# Remote LLM API clients
openai==1.6.1
anthropic==0.8.1
# Document parsing libraries
pypdf2==3.0.1
pdfplumber==0.10.3
python-docx==1.1.0
python-pptx==0.6.23
beautifulsoup4==4.12.2
lxml==4.9.3
markdown==3.5.1
# Image processing and multimodal support
pillow==10.1.0
transformers==4.36.2
torch==2.1.2
# Utility libraries
python-dotenv==1.0.0
pydantic==2.5.3
Installing llama-cpp-python with GPU support requires special attention to ensure proper acceleration. For Nvidia CUDA environments, we must specify the CUDA toolkit version during installation. The CMAKE_ARGS environment variable controls compilation flags that enable GPU acceleration.
# Installation commands for different GPU architectures
# For Nvidia CUDA (adjust cuda version as needed)
# CMAKE_ARGS="-DLLAMA_CUBLAS=on" pip install llama-cpp-python
# For AMD ROCm
# CMAKE_ARGS="-DLLAMA_HIPBLAS=on" pip install llama-cpp-python
# For Apple Metal (M1/M2/M3)
# CMAKE_ARGS="-DLLAMA_METAL=on" pip install llama-cpp-python
# For Intel GPUs
# CMAKE_ARGS="-DLLAMA_CLBLAST=on" pip install llama-cpp-python
DOMAIN LAYER: DEFINING CORE ABSTRACTIONS
The domain layer establishes the fundamental concepts and interfaces that define our summarization system. We begin by creating abstract base classes that represent the essential operations without coupling to specific implementations. This approach ensures that our business logic remains independent of technical details like which LLM provider we use or how documents are parsed.
The SummaryRequest class encapsulates all information needed to perform a summarization operation. It includes the content to summarize, the desired summary length, the target audience, and any specific instructions. By using a dedicated request object rather than passing multiple parameters, we create a more maintainable and extensible design.
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Dict, Any
class SummaryLength(Enum):
"""Enumeration defining standard summary length categories."""
BRIEF = "brief" # One to two sentences
SHORT = "short" # One paragraph
MEDIUM = "medium" # Two to three paragraphs
DETAILED = "detailed" # Comprehensive summary
class ContentType(Enum):
"""Enumeration of supported content types for summarization."""
DOCUMENT = "document"
CODE = "code"
IMAGE = "image"
@dataclass
class SummaryRequest:
"""
Value object representing a summarization request.
This class encapsulates all parameters needed to perform
a summarization operation, following the parameter object
pattern to reduce method signature complexity.
"""
content: str
content_type: ContentType
length: SummaryLength = SummaryLength.MEDIUM
focus_areas: Optional[str] = None
target_audience: Optional[str] = None
additional_instructions: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert request to dictionary for serialization."""
return {
'content': self.content,
'content_type': self.content_type.value,
'length': self.length.value,
'focus_areas': self.focus_areas,
'target_audience': self.target_audience,
'additional_instructions': self.additional_instructions
}
@dataclass
class SummaryResponse:
"""
Value object representing the result of a summarization operation.
Contains the generated summary along with metadata about
the operation such as token usage and processing time.
"""
summary: str
token_count: int
model_used: str
processing_time_seconds: float
confidence_score: Optional[float] = None
def __str__(self) -> str:
"""Provide human-readable string representation."""
return f"Summary ({self.token_count} tokens, {self.processing_time_seconds:.2f}s):\n{self.summary}"
The abstract LLMProvider interface defines the contract that all language model implementations must fulfill. Whether we use a local model running on GPU or a remote API service, the interface remains consistent. This abstraction enables us to switch between providers without modifying the application logic.
class LLMProvider(ABC):
"""
Abstract base class defining the interface for LLM providers.
This interface allows the system to work with different LLM
backends (local or remote) through a uniform API, following
the dependency inversion principle.
"""
@abstractmethod
def generate_summary(self, request: SummaryRequest) -> SummaryResponse:
"""
Generate a summary based on the provided request.
Args:
request: SummaryRequest containing content and parameters
Returns:
SummaryResponse with generated summary and metadata
Raises:
LLMProviderError: If generation fails
"""
pass
@abstractmethod
def is_available(self) -> bool:
"""
Check if the LLM provider is currently available.
Returns:
True if provider can accept requests, False otherwise
"""
pass
@abstractmethod
def get_model_info(self) -> Dict[str, Any]:
"""
Retrieve information about the underlying model.
Returns:
Dictionary containing model name, version, capabilities
"""
pass
The DocumentParser interface abstracts the complexity of extracting text from various document formats. Each format requires different parsing logic, but our application layer should not need to understand these details. By defining this interface, we can add support for new document types without modifying existing code.
class DocumentParser(ABC):
"""
Abstract base class for document parsing implementations.
Each concrete parser handles a specific document format,
extracting text content while preserving semantic structure.
"""
@abstractmethod
def can_parse(self, file_path: str) -> bool:
"""
Determine if this parser can handle the given file.
Args:
file_path: Path to the document file
Returns:
True if parser supports this file format
"""
pass
@abstractmethod
def parse(self, file_path: str) -> str:
"""
Extract text content from the document.
Args:
file_path: Path to the document file
Returns:
Extracted text content as a string
Raises:
ParsingError: If document cannot be parsed
"""
pass
@abstractmethod
def get_metadata(self, file_path: str) -> Dict[str, Any]:
"""
Extract metadata from the document.
Args:
file_path: Path to the document file
Returns:
Dictionary containing document metadata
"""
pass
INFRASTRUCTURE LAYER: IMPLEMENTING DOCUMENT PARSERS
With our abstractions defined, we now implement concrete document parsers for each supported format. The PDF parser demonstrates how we handle one of the most common document types. PDFs present unique challenges because they can contain complex layouts, embedded images, and multiple columns. We use pdfplumber for its robust text extraction capabilities, which preserve layout information better than simpler alternatives.
import pdfplumber
from pathlib import Path
from typing import Dict, Any
import logging
class PDFParser(DocumentParser):
"""
Concrete implementation of DocumentParser for PDF files.
Uses pdfplumber library to extract text while attempting
to preserve document structure and layout information.
"""
def __init__(self):
"""Initialize the PDF parser with logging configuration."""
self.logger = logging.getLogger(__name__)
self.supported_extensions = {'.pdf'}
def can_parse(self, file_path: str) -> bool:
"""
Check if file is a PDF by examining extension.
Args:
file_path: Path to the file to check
Returns:
True if file has .pdf extension
"""
path = Path(file_path)
return path.suffix.lower() in self.supported_extensions
def parse(self, file_path: str) -> str:
"""
Extract all text content from PDF file.
This method opens the PDF, iterates through all pages,
and concatenates the extracted text. It handles errors
gracefully and provides informative logging.
Args:
file_path: Path to the PDF file
Returns:
Concatenated text from all pages
Raises:
ParsingError: If PDF cannot be opened or read
"""
try:
self.logger.info(f"Starting PDF parsing for: {file_path}")
extracted_text = []
with pdfplumber.open(file_path) as pdf:
total_pages = len(pdf.pages)
self.logger.debug(f"PDF contains {total_pages} pages")
for page_num, page in enumerate(pdf.pages, start=1):
# Extract text from current page
page_text = page.extract_text()
if page_text:
# Add page separator for readability
extracted_text.append(f"\n--- Page {page_num} ---\n")
extracted_text.append(page_text)
self.logger.debug(f"Extracted {len(page_text)} characters from page {page_num}")
else:
self.logger.warning(f"No text found on page {page_num}")
result = ''.join(extracted_text)
self.logger.info(f"Successfully extracted {len(result)} total characters")
return result
except Exception as e:
self.logger.error(f"Failed to parse PDF {file_path}: {str(e)}")
raise ParsingError(f"PDF parsing failed: {str(e)}") from e
def get_metadata(self, file_path: str) -> Dict[str, Any]:
"""
Extract metadata from PDF file.
Retrieves document properties such as title, author,
creation date, and page count.
Args:
file_path: Path to the PDF file
Returns:
Dictionary containing PDF metadata
"""
try:
with pdfplumber.open(file_path) as pdf:
metadata = {
'page_count': len(pdf.pages),
'file_size': Path(file_path).stat().st_size,
'file_name': Path(file_path).name
}
# Extract PDF metadata if available
if pdf.metadata:
metadata.update({
'title': pdf.metadata.get('Title', 'Unknown'),
'author': pdf.metadata.get('Author', 'Unknown'),
'creator': pdf.metadata.get('Creator', 'Unknown'),
'creation_date': pdf.metadata.get('CreationDate', 'Unknown')
})
return metadata
except Exception as e:
self.logger.error(f"Failed to extract PDF metadata: {str(e)}")
return {'error': str(e)}
The Markdown parser handles a simpler format but still requires careful processing to extract clean text. Markdown files often contain formatting characters, code blocks, and links that need special handling. We use the markdown library to convert the content to plain text while preserving the semantic structure.
import markdown
from bs4 import BeautifulSoup
class MarkdownParser(DocumentParser):
"""
Parser implementation for Markdown files.
Converts Markdown to HTML, then extracts clean text
while preserving document structure and headings.
"""
def __init__(self):
"""Initialize Markdown parser with configuration."""
self.logger = logging.getLogger(__name__)
self.supported_extensions = {'.md', '.markdown'}
# Configure markdown extensions for better parsing
self.md_extensions = ['extra', 'codehilite', 'toc']
def can_parse(self, file_path: str) -> bool:
"""Check if file is a Markdown document."""
path = Path(file_path)
return path.suffix.lower() in self.supported_extensions
def parse(self, file_path: str) -> str:
"""
Extract text content from Markdown file.
This method reads the Markdown file, converts it to HTML
using the markdown library, then extracts clean text using
BeautifulSoup. This two-step process handles formatting
characters and special syntax cleanly.
Args:
file_path: Path to the Markdown file
Returns:
Extracted text content
"""
try:
self.logger.info(f"Parsing Markdown file: {file_path}")
# Read the raw Markdown content
with open(file_path, 'r', encoding='utf-8') as f:
md_content = f.read()
# Convert Markdown to HTML
html_content = markdown.markdown(
md_content,
extensions=self.md_extensions
)
# Extract clean text from HTML
soup = BeautifulSoup(html_content, 'html.parser')
text = soup.get_text(separator='\n', strip=True)
self.logger.info(f"Extracted {len(text)} characters from Markdown")
return text
except Exception as e:
self.logger.error(f"Markdown parsing failed: {str(e)}")
raise ParsingError(f"Failed to parse Markdown: {str(e)}") from e
def get_metadata(self, file_path: str) -> Dict[str, Any]:
"""Extract metadata from Markdown file."""
path = Path(file_path)
return {
'file_name': path.name,
'file_size': path.stat().st_size,
'format': 'markdown'
}
The DOCX parser handles Microsoft Word documents, which are actually ZIP archives containing XML files. The python-docx library abstracts this complexity, allowing us to iterate through paragraphs and extract text efficiently.
from docx import Document as DocxDocument
class DOCXParser(DocumentParser):
"""
Parser for Microsoft Word DOCX files.
Extracts text from Word documents while preserving
paragraph structure and handling embedded elements.
"""
def __init__(self):
"""Initialize DOCX parser."""
self.logger = logging.getLogger(__name__)
self.supported_extensions = {'.docx'}
def can_parse(self, file_path: str) -> bool:
"""Check if file is a DOCX document."""
path = Path(file_path)
return path.suffix.lower() in self.supported_extensions
def parse(self, file_path: str) -> str:
"""
Extract text from DOCX file.
Iterates through all paragraphs in the document,
extracting text while preserving paragraph breaks.
Also handles tables and other embedded content.
Args:
file_path: Path to DOCX file
Returns:
Extracted text content
"""
try:
self.logger.info(f"Parsing DOCX file: {file_path}")
doc = DocxDocument(file_path)
# Extract text from all paragraphs
paragraphs = []
for para in doc.paragraphs:
if para.text.strip(): # Skip empty paragraphs
paragraphs.append(para.text)
# Extract text from tables
for table in doc.tables:
for row in table.rows:
row_text = ' | '.join(cell.text for cell in row.cells)
if row_text.strip():
paragraphs.append(row_text)
text = '\n\n'.join(paragraphs)
self.logger.info(f"Extracted {len(text)} characters from DOCX")
return text
except Exception as e:
self.logger.error(f"DOCX parsing failed: {str(e)}")
raise ParsingError(f"Failed to parse DOCX: {str(e)}") from e
def get_metadata(self, file_path: str) -> Dict[str, Any]:
"""Extract metadata from DOCX file."""
try:
doc = DocxDocument(file_path)
core_props = doc.core_properties
return {
'file_name': Path(file_path).name,
'file_size': Path(file_path).stat().st_size,
'title': core_props.title or 'Unknown',
'author': core_props.author or 'Unknown',
'created': str(core_props.created) if core_props.created else 'Unknown',
'modified': str(core_props.modified) if core_props.modified else 'Unknown',
'paragraph_count': len(doc.paragraphs)
}
except Exception as e:
return {'error': str(e)}
IMPLEMENTING LLM PROVIDERS FOR LOCAL AND REMOTE INFERENCE
The LLM provider implementations form the core of our summarization system. We begin with a local provider that uses llama-cpp-python to run models directly on the user's hardware. This approach offers privacy, cost savings, and independence from external services. The local provider automatically detects available GPU acceleration and configures itself accordingly.
from llama_cpp import Llama
import time
from typing import Dict, Any, Optional
class LocalLLMProvider(LLMProvider):
"""
LLM provider implementation using local models via llama.cpp.
This provider supports multiple GPU backends including CUDA,
ROCm, Metal, and CPU fallback. It manages model loading,
inference, and resource cleanup.
"""
def __init__(
self,
model_path: str,
n_ctx: int = 4096,
n_gpu_layers: int = -1,
temperature: float = 0.7,
max_tokens: int = 2048
):
"""
Initialize local LLM provider with model configuration.
Args:
model_path: Path to GGUF model file
n_ctx: Context window size in tokens
n_gpu_layers: Number of layers to offload to GPU (-1 for all)
temperature: Sampling temperature for generation
max_tokens: Maximum tokens to generate
"""
self.logger = logging.getLogger(__name__)
self.model_path = model_path
self.n_ctx = n_ctx
self.n_gpu_layers = n_gpu_layers
self.temperature = temperature
self.max_tokens = max_tokens
self.model: Optional[Llama] = None
self._initialize_model()
def _initialize_model(self):
"""
Load the model into memory with GPU acceleration.
This method attempts to load the model with GPU support,
falling back to CPU if GPU is unavailable. It logs
detailed information about the loading process.
"""
try:
self.logger.info(f"Loading model from {self.model_path}")
self.logger.info(f"GPU layers: {self.n_gpu_layers}, Context: {self.n_ctx}")
start_time = time.time()
self.model = Llama(
model_path=self.model_path,
n_ctx=self.n_ctx,
n_gpu_layers=self.n_gpu_layers,
verbose=False
)
load_time = time.time() - start_time
self.logger.info(f"Model loaded successfully in {load_time:.2f} seconds")
except Exception as e:
self.logger.error(f"Failed to load model: {str(e)}")
raise LLMProviderError(f"Model initialization failed: {str(e)}") from e
def generate_summary(self, request: SummaryRequest) -> SummaryResponse:
"""
Generate summary using local LLM.
Constructs an appropriate prompt based on content type
and request parameters, then invokes the model to generate
a summary. Tracks token usage and processing time.
Args:
request: SummaryRequest with content and parameters
Returns:
SummaryResponse with generated summary
"""
if not self.is_available():
raise LLMProviderError("Model not available")
try:
start_time = time.time()
# Construct prompt based on content type and parameters
prompt = self._build_prompt(request)
self.logger.debug(f"Generating summary with {len(prompt)} character prompt")
# Generate summary using the model
response = self.model(
prompt,
max_tokens=self.max_tokens,
temperature=self.temperature,
stop=["</summary>", "\n\n\n"],
echo=False
)
# Extract generated text and metadata
summary_text = response['choices'][0]['text'].strip()
token_count = response['usage']['completion_tokens']
processing_time = time.time() - start_time
self.logger.info(f"Summary generated: {token_count} tokens in {processing_time:.2f}s")
return SummaryResponse(
summary=summary_text,
token_count=token_count,
model_used=Path(self.model_path).name,
processing_time_seconds=processing_time
)
except Exception as e:
self.logger.error(f"Summary generation failed: {str(e)}")
raise LLMProviderError(f"Failed to generate summary: {str(e)}") from e
def _build_prompt(self, request: SummaryRequest) -> str:
"""
Construct an effective prompt for summarization.
The prompt structure varies based on content type and
includes specific instructions for length, focus areas,
and target audience when provided.
Args:
request: SummaryRequest containing parameters
Returns:
Formatted prompt string
"""
# Define length guidelines
length_instructions = {
SummaryLength.BRIEF: "in 1-2 sentences",
SummaryLength.SHORT: "in a single concise paragraph",
SummaryLength.MEDIUM: "in 2-3 well-structured paragraphs",
SummaryLength.DETAILED: "in a comprehensive multi-paragraph summary"
}
length_guide = length_instructions.get(request.length, "concisely")
# Build base prompt
prompt_parts = [
f"Please provide a {request.content_type.value} summary {length_guide}.",
]
# Add optional parameters
if request.focus_areas:
prompt_parts.append(f"Focus on: {request.focus_areas}.")
if request.target_audience:
prompt_parts.append(f"Target audience: {request.target_audience}.")
if request.additional_instructions:
prompt_parts.append(request.additional_instructions)
# Add content
prompt_parts.append(f"\n\nContent to summarize:\n{request.content}\n\nSummary:")
return ' '.join(prompt_parts)
def is_available(self) -> bool:
"""Check if model is loaded and ready."""
return self.model is not None
def get_model_info(self) -> Dict[str, Any]:
"""Retrieve model information."""
return {
'model_path': self.model_path,
'context_size': self.n_ctx,
'gpu_layers': self.n_gpu_layers,
'type': 'local',
'backend': 'llama.cpp'
}
For scenarios where local resources are insufficient or specific model capabilities are required, we implement a remote LLM provider that communicates with API services. This implementation demonstrates integration with OpenAI's API, but the pattern applies to any remote service.
from openai import OpenAI
import os
class RemoteLLMProvider(LLMProvider):
"""
LLM provider implementation using remote API services.
Supports OpenAI and compatible API endpoints. Handles
authentication, rate limiting, and error recovery.
"""
def __init__(
self,
api_key: Optional[str] = None,
model_name: str = "gpt-3.5-turbo",
base_url: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2048
):
"""
Initialize remote LLM provider.
Args:
api_key: API key for authentication (reads from env if None)
model_name: Name of the model to use
base_url: Custom API endpoint (for compatible services)
temperature: Sampling temperature
max_tokens: Maximum tokens to generate
"""
self.logger = logging.getLogger(__name__)
self.model_name = model_name
self.temperature = temperature
self.max_tokens = max_tokens
# Initialize OpenAI client
self.client = OpenAI(
api_key=api_key or os.getenv('OPENAI_API_KEY'),
base_url=base_url
)
self.logger.info(f"Initialized remote provider with model: {model_name}")
def generate_summary(self, request: SummaryRequest) -> SummaryResponse:
"""
Generate summary using remote API.
Constructs messages for the chat completion API,
sends the request, and processes the response.
Args:
request: SummaryRequest with content and parameters
Returns:
SummaryResponse with generated summary
"""
try:
start_time = time.time()
# Build messages for chat completion
messages = self._build_messages(request)
self.logger.debug(f"Sending request to {self.model_name}")
# Call the API
response = self.client.chat.completions.create(
model=self.model_name,
messages=messages,
temperature=self.temperature,
max_tokens=self.max_tokens
)
# Extract response data
summary_text = response.choices[0].message.content.strip()
token_count = response.usage.completion_tokens
processing_time = time.time() - start_time
self.logger.info(f"Received summary: {token_count} tokens in {processing_time:.2f}s")
return SummaryResponse(
summary=summary_text,
token_count=token_count,
model_used=self.model_name,
processing_time_seconds=processing_time
)
except Exception as e:
self.logger.error(f"Remote API call failed: {str(e)}")
raise LLMProviderError(f"API request failed: {str(e)}") from e
def _build_messages(self, request: SummaryRequest) -> list:
"""
Construct message array for chat completion.
Creates a system message with instructions and a user
message with the content to summarize.
Args:
request: SummaryRequest with parameters
Returns:
List of message dictionaries
"""
# System message with role definition
system_content = "You are a helpful assistant specialized in creating clear, accurate summaries."
# User message with specific instructions
user_parts = [f"Please summarize the following {request.content_type.value}"]
if request.length:
length_map = {
SummaryLength.BRIEF: "very briefly (1-2 sentences)",
SummaryLength.SHORT: "in a short paragraph",
SummaryLength.MEDIUM: "in 2-3 paragraphs",
SummaryLength.DETAILED: "comprehensively"
}
user_parts.append(length_map.get(request.length, "concisely"))
if request.focus_areas:
user_parts.append(f"focusing on {request.focus_areas}")
if request.target_audience:
user_parts.append(f"for {request.target_audience}")
user_content = ' '.join(user_parts) + f":\n\n{request.content}"
return [
{"role": "system", "content": system_content},
{"role": "user", "content": user_content}
]
def is_available(self) -> bool:
"""
Check API availability.
Attempts a simple API call to verify connectivity
and authentication.
"""
try:
# Make a minimal test request
self.client.models.list()
return True
except Exception as e:
self.logger.warning(f"API availability check failed: {str(e)}")
return False
def get_model_info(self) -> Dict[str, Any]:
"""Retrieve model information."""
return {
'model_name': self.model_name,
'type': 'remote',
'provider': 'openai-compatible'
}
APPLICATION LAYER: ORCHESTRATING THE SUMMARIZATION WORKFLOW
The application layer brings together our domain abstractions and infrastructure implementations to create a cohesive summarization service. The DocumentSummarizer class serves as the primary entry point, coordinating between document parsers and LLM providers.
from typing import List
import os
class ParsingError(Exception):
"""Exception raised when document parsing fails."""
pass
class LLMProviderError(Exception):
"""Exception raised when LLM operations fail."""
pass
class DocumentSummarizer:
"""
Main application service for document summarization.
Coordinates document parsing and LLM-based summarization,
managing the complete workflow from file input to summary output.
"""
def __init__(
self,
llm_provider: LLMProvider,
parsers: Optional[List[DocumentParser]] = None
):
"""
Initialize the summarizer with dependencies.
Args:
llm_provider: LLM provider for generating summaries
parsers: List of document parsers (auto-created if None)
"""
self.logger = logging.getLogger(__name__)
self.llm_provider = llm_provider
# Initialize parsers if not provided
if parsers is None:
self.parsers = [
PDFParser(),
MarkdownParser(),
DOCXParser(),
HTMLParser(),
PPTXParser()
]
else:
self.parsers = parsers
self.logger.info(f"Initialized DocumentSummarizer with {len(self.parsers)} parsers")
def summarize_file(
self,
file_path: str,
length: SummaryLength = SummaryLength.MEDIUM,
focus_areas: Optional[str] = None,
target_audience: Optional[str] = None
) -> SummaryResponse:
"""
Summarize a document file.
This method orchestrates the complete summarization workflow:
1. Finds appropriate parser for file format
2. Extracts text content from document
3. Creates summarization request
4. Generates summary using LLM
Args:
file_path: Path to document file
length: Desired summary length
focus_areas: Optional specific topics to emphasize
target_audience: Optional audience description
Returns:
SummaryResponse with generated summary
Raises:
ParsingError: If no parser available or parsing fails
LLMProviderError: If summary generation fails
"""
self.logger.info(f"Starting summarization for: {file_path}")
# Validate file exists
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
# Find appropriate parser
parser = self._find_parser(file_path)
if parser is None:
raise ParsingError(f"No parser available for file: {file_path}")
# Extract text content
self.logger.info(f"Parsing document with {parser.__class__.__name__}")
content = parser.parse(file_path)
if not content or len(content.strip()) == 0:
raise ParsingError("No content extracted from document")
# Create summarization request
request = SummaryRequest(
content=content,
content_type=ContentType.DOCUMENT,
length=length,
focus_areas=focus_areas,
target_audience=target_audience
)
# Generate summary
self.logger.info("Generating summary with LLM")
response = self.llm_provider.generate_summary(request)
self.logger.info("Summarization completed successfully")
return response
def _find_parser(self, file_path: str) -> Optional[DocumentParser]:
"""
Find appropriate parser for given file.
Iterates through available parsers and returns the first
one that can handle the file format.
Args:
file_path: Path to file
Returns:
DocumentParser instance or None if no parser found
"""
for parser in self.parsers:
if parser.can_parse(file_path):
return parser
return None
def get_supported_formats(self) -> List[str]:
"""
Get list of supported file formats.
Returns:
List of file extensions that can be processed
"""
formats = set()
for parser in self.parsers:
if hasattr(parser, 'supported_extensions'):
formats.update(parser.supported_extensions)
return sorted(list(formats))
To complete the document parser suite, we need implementations for HTML and PowerPoint files. The HTML parser handles web content and HTML documents.
class HTMLParser(DocumentParser):
"""Parser for HTML files and web content."""
def __init__(self):
"""Initialize HTML parser."""
self.logger = logging.getLogger(__name__)
self.supported_extensions = {'.html', '.htm'}
def can_parse(self, file_path: str) -> bool:
"""Check if file is HTML."""
path = Path(file_path)
return path.suffix.lower() in self.supported_extensions
def parse(self, file_path: str) -> str:
"""
Extract text from HTML file.
Uses BeautifulSoup to parse HTML and extract text content
while removing scripts, styles, and other non-content elements.
Args:
file_path: Path to HTML file
Returns:
Extracted text content
"""
try:
self.logger.info(f"Parsing HTML file: {file_path}")
with open(file_path, 'r', encoding='utf-8') as f:
html_content = f.read()
# Parse HTML
soup = BeautifulSoup(html_content, 'lxml')
# Remove script and style elements
for element in soup(['script', 'style', 'nav', 'footer']):
element.decompose()
# Extract text
text = soup.get_text(separator='\n', strip=True)
# Clean up whitespace
lines = [line.strip() for line in text.split('\n') if line.strip()]
text = '\n'.join(lines)
self.logger.info(f"Extracted {len(text)} characters from HTML")
return text
except Exception as e:
self.logger.error(f"HTML parsing failed: {str(e)}")
raise ParsingError(f"Failed to parse HTML: {str(e)}") from e
def get_metadata(self, file_path: str) -> Dict[str, Any]:
"""Extract metadata from HTML file."""
try:
with open(file_path, 'r', encoding='utf-8') as f:
soup = BeautifulSoup(f.read(), 'lxml')
metadata = {
'file_name': Path(file_path).name,
'file_size': Path(file_path).stat().st_size
}
# Extract meta tags
if soup.title:
metadata['title'] = soup.title.string
for meta in soup.find_all('meta'):
if meta.get('name') == 'description':
metadata['description'] = meta.get('content')
elif meta.get('name') == 'author':
metadata['author'] = meta.get('content')
return metadata
except Exception as e:
return {'error': str(e)}
from pptx import Presentation
class PPTXParser(DocumentParser):
"""Parser for PowerPoint PPTX files."""
def __init__(self):
"""Initialize PPTX parser."""
self.logger = logging.getLogger(__name__)
self.supported_extensions = {'.pptx'}
def can_parse(self, file_path: str) -> bool:
"""Check if file is PPTX."""
path = Path(file_path)
return path.suffix.lower() in self.supported_extensions
def parse(self, file_path: str) -> str:
"""
Extract text from PowerPoint file.
Iterates through all slides and extracts text from shapes,
preserving slide structure and organization.
Args:
file_path: Path to PPTX file
Returns:
Extracted text content
"""
try:
self.logger.info(f"Parsing PPTX file: {file_path}")
prs = Presentation(file_path)
slides_text = []
for slide_num, slide in enumerate(prs.slides, start=1):
slide_content = [f"\n--- Slide {slide_num} ---"]
# Extract text from all shapes in slide
for shape in slide.shapes:
if hasattr(shape, "text") and shape.text.strip():
slide_content.append(shape.text)
if len(slide_content) > 1: # Has content beyond header
slides_text.append('\n'.join(slide_content))
text = '\n\n'.join(slides_text)
self.logger.info(f"Extracted text from {len(prs.slides)} slides")
return text
except Exception as e:
self.logger.error(f"PPTX parsing failed: {str(e)}")
raise ParsingError(f"Failed to parse PPTX: {str(e)}") from e
def get_metadata(self, file_path: str) -> Dict[str, Any]:
"""Extract metadata from PPTX file."""
try:
prs = Presentation(file_path)
core_props = prs.core_properties
return {
'file_name': Path(file_path).name,
'file_size': Path(file_path).stat().st_size,
'slide_count': len(prs.slides),
'title': core_props.title or 'Unknown',
'author': core_props.author or 'Unknown',
'created': str(core_props.created) if core_props.created else 'Unknown'
}
except Exception as e:
return {'error': str(e)}
EXTENDING TO CODE SUMMARIZATION
Code summarization presents unique challenges compared to natural language documents. Source code has formal syntax, semantic relationships, and domain-specific conventions that require specialized handling. Our code summarizer must understand programming language structure, identify key components like functions and classes, and explain logic in accessible terms.
We create a dedicated CodeSummarizer that extends our base architecture while adding code-specific capabilities. This class analyzes code structure, extracts documentation, and generates summaries that explain both what the code does and how it works.
import ast
from typing import Dict, List, Any
class CodeSummarizer:
"""
Specialized summarizer for source code files.
Analyzes code structure, extracts key components, and
generates summaries explaining functionality and design.
"""
def __init__(self, llm_provider: LLMProvider):
"""
Initialize code summarizer.
Args:
llm_provider: LLM provider for generating summaries
"""
self.logger = logging.getLogger(__name__)
self.llm_provider = llm_provider
# Map file extensions to language names
self.language_map = {
'.py': 'Python',
'.js': 'JavaScript',
'.ts': 'TypeScript',
'.java': 'Java',
'.cpp': 'C++',
'.c': 'C',
'.go': 'Go',
'.rs': 'Rust',
'.rb': 'Ruby',
'.php': 'PHP'
}
def summarize_code_file(
self,
file_path: str,
include_structure: bool = True,
focus_on: Optional[str] = None
) -> SummaryResponse:
"""
Generate summary of a source code file.
Analyzes the code structure, identifies key components,
and creates a comprehensive summary explaining the code's
purpose, design, and implementation.
Args:
file_path: Path to source code file
include_structure: Whether to analyze code structure
focus_on: Optional specific aspect to emphasize
Returns:
SummaryResponse with code summary
"""
self.logger.info(f"Summarizing code file: {file_path}")
# Read code content
with open(file_path, 'r', encoding='utf-8') as f:
code_content = f.read()
# Detect programming language
language = self._detect_language(file_path)
self.logger.info(f"Detected language: {language}")
# Analyze code structure if requested
structure_info = ""
if include_structure and language == 'Python':
structure_info = self._analyze_python_structure(code_content)
# Build enhanced content for summarization
enhanced_content = self._build_code_context(
code_content,
language,
structure_info,
file_path
)
# Create summarization request with code-specific instructions
instructions = self._build_code_instructions(language, focus_on)
request = SummaryRequest(
content=enhanced_content,
content_type=ContentType.CODE,
length=SummaryLength.DETAILED,
additional_instructions=instructions
)
# Generate summary
response = self.llm_provider.generate_summary(request)
self.logger.info("Code summarization completed")
return response
def _detect_language(self, file_path: str) -> str:
"""
Detect programming language from file extension.
Args:
file_path: Path to code file
Returns:
Language name or 'Unknown'
"""
ext = Path(file_path).suffix.lower()
return self.language_map.get(ext, 'Unknown')
def _analyze_python_structure(self, code: str) -> str:
"""
Analyze Python code structure using AST.
Extracts information about classes, functions, imports,
and overall code organization.
Args:
code: Python source code
Returns:
Formatted structure description
"""
try:
tree = ast.parse(code)
classes = []
functions = []
imports = []
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
# Extract class information
methods = [m.name for m in node.body if isinstance(m, ast.FunctionDef)]
classes.append({
'name': node.name,
'methods': methods,
'bases': [self._get_name(base) for base in node.bases]
})
elif isinstance(node, ast.FunctionDef):
# Only top-level functions (not methods)
if not any(isinstance(p, ast.ClassDef) for p in ast.walk(tree) if node in ast.walk(p)):
functions.append({
'name': node.name,
'args': [arg.arg for arg in node.args.args]
})
elif isinstance(node, (ast.Import, ast.ImportFrom)):
if isinstance(node, ast.Import):
imports.extend([alias.name for alias in node.names])
else:
imports.append(node.module if node.module else '')
# Format structure information
structure_parts = ["CODE STRUCTURE ANALYSIS:"]
if imports:
structure_parts.append(f"\nImports: {', '.join(set(imports))}")
if classes:
structure_parts.append(f"\nClasses ({len(classes)}):")
for cls in classes:
structure_parts.append(f" - {cls['name']}")
if cls['bases']:
structure_parts.append(f" Inherits from: {', '.join(cls['bases'])}")
if cls['methods']:
structure_parts.append(f" Methods: {', '.join(cls['methods'][:5])}")
if functions:
structure_parts.append(f"\nFunctions ({len(functions)}):")
for func in functions[:10]: # Limit to first 10
args_str = ', '.join(func['args'])
structure_parts.append(f" - {func['name']}({args_str})")
return '\n'.join(structure_parts)
except SyntaxError as e:
self.logger.warning(f"Could not parse Python code: {str(e)}")
return "CODE STRUCTURE ANALYSIS: Unable to parse code"
def _get_name(self, node) -> str:
"""Extract name from AST node."""
if isinstance(node, ast.Name):
return node.id
elif isinstance(node, ast.Attribute):
return f"{self._get_name(node.value)}.{node.attr}"
return str(node)
def _build_code_context(
self,
code: str,
language: str,
structure: str,
file_path: str
) -> str:
"""
Build enriched context for code summarization.
Combines the raw code with structural analysis and
metadata to provide comprehensive context to the LLM.
Args:
code: Source code content
language: Programming language
structure: Structural analysis results
file_path: Path to file
Returns:
Enhanced content string
"""
context_parts = [
f"File: {Path(file_path).name}",
f"Language: {language}",
f"Lines of code: {len(code.splitlines())}",
]
if structure:
context_parts.append(f"\n{structure}")
context_parts.append(f"\n\nSOURCE CODE:\n{code}")
return '\n'.join(context_parts)
def _build_code_instructions(
self,
language: str,
focus_on: Optional[str]
) -> str:
"""
Create specialized instructions for code summarization.
Args:
language: Programming language
focus_on: Optional focus area
Returns:
Instruction string for LLM
"""
instructions = [
f"Analyze this {language} code and provide a comprehensive summary.",
"Explain the main purpose, key components, and design patterns used.",
"Describe the overall architecture and how different parts interact.",
"Highlight any notable algorithms, data structures, or techniques.",
]
if focus_on:
instructions.append(f"Pay special attention to: {focus_on}")
return ' '.join(instructions)
MULTIMODAL CAPABILITIES: IMAGE SUMMARIZATION
The final extension of our summarization agent adds multimodal capabilities, enabling it to analyze and describe images. This requires integrating vision-language models that can process visual information and generate textual descriptions. We use the BLIP (Bootstrapping Language-Image Pre-training) model, which excels at image captioning and visual question answering.
from PIL import Image
from transformers import BlipProcessor, BlipForConditionalGeneration
import torch
class ImageSummarizer:
"""
Summarizer for image content using vision-language models.
Generates descriptions and summaries of images using
multimodal models that understand visual content.
"""
def __init__(
self,
llm_provider: LLMProvider,
device: Optional[str] = None
):
"""
Initialize image summarizer with vision model.
Args:
llm_provider: LLM provider for enhancing descriptions
device: Device for model inference (auto-detected if None)
"""
self.logger = logging.getLogger(__name__)
self.llm_provider = llm_provider
# Auto-detect device if not specified
if device is None:
device = self._detect_device()
self.device = device
self.logger.info(f"Initializing vision model on device: {self.device}")
# Load BLIP model for image captioning
self.processor = BlipProcessor.from_pretrained(
"Salesforce/blip-image-captioning-base"
)
self.model = BlipForConditionalGeneration.from_pretrained(
"Salesforce/blip-image-captioning-base"
).to(self.device)
self.logger.info("Vision model loaded successfully")
def _detect_device(self) -> str:
"""
Detect best available device for inference.
Checks for CUDA, MPS (Apple Silicon), and falls back to CPU.
Returns:
Device string ('cuda', 'mps', or 'cpu')
"""
if torch.cuda.is_available():
return 'cuda'
elif torch.backends.mps.is_available():
return 'mps'
else:
return 'cpu'
def summarize_image(
self,
image_path: str,
detail_level: str = 'standard',
focus_on: Optional[str] = None
) -> SummaryResponse:
"""
Generate summary of an image.
Uses vision-language model to analyze the image and
generate a descriptive summary. Can focus on specific
aspects if requested.
Args:
image_path: Path to image file
detail_level: Level of detail ('brief', 'standard', 'detailed')
focus_on: Optional specific aspect to describe
Returns:
SummaryResponse with image description
"""
self.logger.info(f"Summarizing image: {image_path}")
start_time = time.time()
# Load and preprocess image
image = Image.open(image_path).convert('RGB')
# Generate base caption using vision model
caption = self._generate_caption(image, focus_on)
# Enhance caption with LLM if detailed summary requested
if detail_level == 'detailed':
enhanced_summary = self._enhance_description(
caption,
image_path,
focus_on
)
final_summary = enhanced_summary
else:
final_summary = caption
processing_time = time.time() - start_time
return SummaryResponse(
summary=final_summary,
token_count=len(final_summary.split()),
model_used="BLIP + LLM",
processing_time_seconds=processing_time
)
def _generate_caption(
self,
image: Image.Image,
focus_on: Optional[str] = None
) -> str:
"""
Generate caption for image using vision model.
Args:
image: PIL Image object
focus_on: Optional focus area
Returns:
Generated caption
"""
try:
# Prepare inputs
if focus_on:
# Use conditional generation with focus prompt
text_prompt = f"a picture of {focus_on}"
inputs = self.processor(
image,
text=text_prompt,
return_tensors="pt"
).to(self.device)
else:
# Unconditional caption generation
inputs = self.processor(
image,
return_tensors="pt"
).to(self.device)
# Generate caption
with torch.no_grad():
output = self.model.generate(
**inputs,
max_length=50,
num_beams=5
)
# Decode and return caption
caption = self.processor.decode(
output[0],
skip_special_tokens=True
)
self.logger.debug(f"Generated caption: {caption}")
return caption
except Exception as e:
self.logger.error(f"Caption generation failed: {str(e)}")
raise LLMProviderError(f"Failed to generate image caption: {str(e)}")
def _enhance_description(
self,
base_caption: str,
image_path: str,
focus_on: Optional[str]
) -> str:
"""
Enhance basic caption with detailed description using LLM.
Takes the vision model's caption and expands it into
a more comprehensive description using the LLM.
Args:
base_caption: Initial caption from vision model
image_path: Path to image file
focus_on: Optional focus area
Returns:
Enhanced description
"""
# Build context for LLM
context_parts = [
f"Image file: {Path(image_path).name}",
f"Initial visual analysis: {base_caption}",
]
if focus_on:
context_parts.append(f"Focus area: {focus_on}")
context = '\n'.join(context_parts)
# Create request for detailed description
instructions = (
"Based on the visual analysis provided, create a detailed "
"description of the image. Expand on the key elements, "
"composition, and notable features. Be specific and descriptive."
)
request = SummaryRequest(
content=context,
content_type=ContentType.IMAGE,
length=SummaryLength.DETAILED,
additional_instructions=instructions
)
# Generate enhanced description
response = self.llm_provider.generate_summary(request)
return response.summary
UNIFIED INTERFACE: BRINGING IT ALL TOGETHER
To provide a seamless user experience, we create a unified interface that automatically routes requests to the appropriate summarizer based on content type. This facade pattern simplifies the API and hides the complexity of managing multiple specialized components.
class UnifiedSummarizationAgent:
"""
Unified interface for all summarization capabilities.
Provides a single entry point for summarizing documents,
code, and images, automatically routing to appropriate
specialized components.
"""
def __init__(
self,
llm_provider: LLMProvider,
enable_vision: bool = True
):
"""
Initialize unified summarization agent.
Args:
llm_provider: LLM provider for text generation
enable_vision: Whether to enable image summarization
"""
self.logger = logging.getLogger(__name__)
# Initialize specialized summarizers
self.document_summarizer = DocumentSummarizer(llm_provider)
self.code_summarizer = CodeSummarizer(llm_provider)
if enable_vision:
self.image_summarizer = ImageSummarizer(llm_provider)
else:
self.image_summarizer = None
# Define content type detection rules
self.code_extensions = {'.py', '.js', '.ts', '.java', '.cpp', '.c', '.go', '.rs', '.rb', '.php'}
self.image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
self.document_extensions = {'.pdf', '.md', '.docx', '.pptx', '.html', '.htm'}
self.logger.info("Unified summarization agent initialized")
def summarize(
self,
file_path: str,
length: SummaryLength = SummaryLength.MEDIUM,
focus_areas: Optional[str] = None,
**kwargs
) -> SummaryResponse:
"""
Summarize any supported file type.
Automatically detects content type and routes to
appropriate specialized summarizer.
Args:
file_path: Path to file to summarize
length: Desired summary length
focus_areas: Optional areas to emphasize
**kwargs: Additional parameters for specific summarizers
Returns:
SummaryResponse with generated summary
"""
self.logger.info(f"Processing file: {file_path}")
# Detect content type
content_type = self._detect_content_type(file_path)
self.logger.info(f"Detected content type: {content_type}")
# Route to appropriate summarizer
if content_type == 'code':
return self.code_summarizer.summarize_code_file(
file_path,
focus_on=focus_areas,
**kwargs
)
elif content_type == 'image':
if self.image_summarizer is None:
raise ValueError("Image summarization not enabled")
detail_map = {
SummaryLength.BRIEF: 'brief',
SummaryLength.SHORT: 'standard',
SummaryLength.MEDIUM: 'standard',
SummaryLength.DETAILED: 'detailed'
}
return self.image_summarizer.summarize_image(
file_path,
detail_level=detail_map.get(length, 'standard'),
focus_on=focus_areas
)
elif content_type == 'document':
return self.document_summarizer.summarize_file(
file_path,
length=length,
focus_areas=focus_areas,
**kwargs
)
else:
raise ValueError(f"Unsupported file type: {file_path}")
def _detect_content_type(self, file_path: str) -> str:
"""
Detect content type from file extension.
Args:
file_path: Path to file
Returns:
Content type string ('code', 'image', or 'document')
"""
ext = Path(file_path).suffix.lower()
if ext in self.code_extensions:
return 'code'
elif ext in self.image_extensions:
return 'image'
elif ext in self.document_extensions:
return 'document'
else:
raise ValueError(f"Unknown file extension: {ext}")
def get_capabilities(self) -> Dict[str, Any]:
"""
Get information about agent capabilities.
Returns:
Dictionary describing supported formats and features
"""
return {
'document_formats': list(self.document_extensions),
'code_formats': list(self.code_extensions),
'image_formats': list(self.image_extensions) if self.image_summarizer else [],
'vision_enabled': self.image_summarizer is not None,
'llm_info': self.document_summarizer.llm_provider.get_model_info()
}
PRACTICAL USAGE EXAMPLES
Now we demonstrate how to use the complete system in various scenarios. The first example shows basic document summarization with a local LLM.
import logging
from pathlib import Path
# Configure logging for visibility
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
def example_local_document_summarization():
"""
Demonstrate document summarization using local LLM.
This example shows how to set up a local LLM provider
and use it to summarize a PDF document.
"""
print("=" * 70)
print("EXAMPLE: Local Document Summarization")
print("=" * 70)
# Initialize local LLM provider
# Replace with actual path to your GGUF model file
model_path = "/path/to/models/mistral-7b-instruct-v0.2.Q4_K_M.gguf"
local_llm = LocalLLMProvider(
model_path=model_path,
n_ctx=4096, # Context window size
n_gpu_layers=35, # Offload layers to GPU
temperature=0.7, # Creativity level
max_tokens=1024 # Maximum summary length
)
# Create unified agent
agent = UnifiedSummarizationAgent(
llm_provider=local_llm,
enable_vision=False # Vision requires additional setup
)
# Summarize a PDF document
pdf_path = "research_paper.pdf"
try:
response = agent.summarize(
file_path=pdf_path,
length=SummaryLength.MEDIUM,
focus_areas="methodology and key findings"
)
print(f"\nSummary generated in {response.processing_time_seconds:.2f} seconds")
print(f"Tokens used: {response.token_count}")
print(f"Model: {response.model_used}")
print(f"\n{'-' * 70}")
print("SUMMARY:")
print(f"{'-' * 70}")
print(response.summary)
except Exception as e:
print(f"Error during summarization: {str(e)}")
def example_remote_code_summarization():
"""
Demonstrate code summarization using remote API.
Shows how to use a remote LLM service to analyze
and summarize source code files.
"""
print("\n" + "=" * 70)
print("EXAMPLE: Remote Code Summarization")
print("=" * 70)
# Initialize remote LLM provider
remote_llm = RemoteLLMProvider(
model_name="gpt-4",
temperature=0.3, # Lower temperature for technical content
max_tokens=2048
)
# Create agent with remote provider
agent = UnifiedSummarizationAgent(
llm_provider=remote_llm,
enable_vision=False
)
# Summarize a Python file
code_path = "complex_algorithm.py"
try:
response = agent.summarize(
file_path=code_path,
length=SummaryLength.DETAILED,
focus_areas="algorithm complexity and optimization techniques"
)
print(f"\nCode analysis completed in {response.processing_time_seconds:.2f} seconds")
print(f"\n{'-' * 70}")
print("CODE SUMMARY:")
print(f"{'-' * 70}")
print(response.summary)
except Exception as e:
print(f"Error during code summarization: {str(e)}")
def example_multimodal_summarization():
"""
Demonstrate image summarization with vision model.
Shows complete multimodal workflow for analyzing
and describing image content.
"""
print("\n" + "=" * 70)
print("EXAMPLE: Multimodal Image Summarization")
print("=" * 70)
# Use remote LLM for enhanced descriptions
llm = RemoteLLMProvider(
model_name="gpt-3.5-turbo",
temperature=0.7
)
# Create agent with vision enabled
agent = UnifiedSummarizationAgent(
llm_provider=llm,
enable_vision=True
)
# Summarize an image
image_path = "diagram.png"
try:
response = agent.summarize(
file_path=image_path,
length=SummaryLength.DETAILED,
focus_areas="technical components and relationships"
)
print(f"\nImage analysis completed in {response.processing_time_seconds:.2f} seconds")
print(f"\n{'-' * 70}")
print("IMAGE DESCRIPTION:")
print(f"{'-' * 70}")
print(response.summary)
except Exception as e:
print(f"Error during image summarization: {str(e)}")
def example_batch_processing():
"""
Demonstrate batch processing of multiple files.
Shows how to efficiently process multiple documents
of different types in a single workflow.
"""
print("\n" + "=" * 70)
print("EXAMPLE: Batch Processing Multiple Files")
print("=" * 70)
# Initialize with local LLM for cost efficiency
llm = LocalLLMProvider(
model_path="/path/to/model.gguf",
n_gpu_layers=-1 # Use all available GPU
)
agent = UnifiedSummarizationAgent(llm_provider=llm)
# List of files to process
files_to_process = [
"report.pdf",
"analysis.py",
"presentation.pptx",
"documentation.md"
]
results = []
for file_path in files_to_process:
if not Path(file_path).exists():
print(f"Skipping {file_path} - file not found")
continue
try:
print(f"\nProcessing: {file_path}")
response = agent.summarize(
file_path=file_path,
length=SummaryLength.SHORT
)
results.append({
'file': file_path,
'summary': response.summary,
'tokens': response.token_count,
'time': response.processing_time_seconds
})
print(f" Completed in {response.processing_time_seconds:.2f}s")
except Exception as e:
print(f" Error: {str(e)}")
# Print consolidated results
print(f"\n{'-' * 70}")
print("BATCH PROCESSING RESULTS")
print(f"{'-' * 70}")
total_time = sum(r['time'] for r in results)
total_tokens = sum(r['tokens'] for r in results)
print(f"\nProcessed {len(results)} files")
print(f"Total time: {total_time:.2f} seconds")
print(f"Total tokens: {total_tokens}")
print(f"Average time per file: {total_time/len(results):.2f} seconds")
CONFIGURATION AND DEPLOYMENT
For production deployment, proper configuration management is essential. We create a configuration system that handles environment-specific settings, model paths, and API credentials.
from dataclasses import dataclass
from typing import Optional
import os
from dotenv import load_dotenv
@dataclass
class LLMConfig:
"""Configuration for LLM providers."""
provider_type: str # 'local' or 'remote'
model_path: Optional[str] = None
model_name: Optional[str] = None
api_key: Optional[str] = None
base_url: Optional[str] = None
n_ctx: int = 4096
n_gpu_layers: int = -1
temperature: float = 0.7
max_tokens: int = 2048
@dataclass
class AgentConfig:
"""Configuration for summarization agent."""
enable_vision: bool = True
enable_code_analysis: bool = True
log_level: str = 'INFO'
cache_dir: Optional[str] = None
class ConfigurationManager:
"""
Manages configuration for the summarization system.
Loads settings from environment variables and config files,
providing validated configuration objects.
"""
def __init__(self, env_file: Optional[str] = None):
"""
Initialize configuration manager.
Args:
env_file: Path to .env file (optional)
"""
if env_file:
load_dotenv(env_file)
else:
load_dotenv() # Load from default .env
self.logger = logging.getLogger(__name__)
def get_llm_config(self) -> LLMConfig:
"""
Load LLM configuration from environment.
Returns:
LLMConfig object with settings
"""
provider_type = os.getenv('LLM_PROVIDER', 'local')
if provider_type == 'local':
return LLMConfig(
provider_type='local',
model_path=os.getenv('LOCAL_MODEL_PATH'),
n_ctx=int(os.getenv('CONTEXT_SIZE', '4096')),
n_gpu_layers=int(os.getenv('GPU_LAYERS', '-1')),
temperature=float(os.getenv('TEMPERATURE', '0.7')),
max_tokens=int(os.getenv('MAX_TOKENS', '2048'))
)
else:
return LLMConfig(
provider_type='remote',
model_name=os.getenv('REMOTE_MODEL_NAME', 'gpt-3.5-turbo'),
api_key=os.getenv('OPENAI_API_KEY'),
base_url=os.getenv('API_BASE_URL'),
temperature=float(os.getenv('TEMPERATURE', '0.7')),
max_tokens=int(os.getenv('MAX_TOKENS', '2048'))
)
def get_agent_config(self) -> AgentConfig:
"""
Load agent configuration from environment.
Returns:
AgentConfig object with settings
"""
return AgentConfig(
enable_vision=os.getenv('ENABLE_VISION', 'true').lower() == 'true',
enable_code_analysis=os.getenv('ENABLE_CODE', 'true').lower() == 'true',
log_level=os.getenv('LOG_LEVEL', 'INFO'),
cache_dir=os.getenv('CACHE_DIR')
)
def create_agent(self) -> UnifiedSummarizationAgent:
"""
Create configured summarization agent.
Instantiates and configures all components based on
loaded configuration settings.
Returns:
Configured UnifiedSummarizationAgent instance
"""
llm_config = self.get_llm_config()
agent_config = self.get_agent_config()
# Configure logging
logging.basicConfig(level=getattr(logging, agent_config.log_level))
# Create LLM provider
if llm_config.provider_type == 'local':
if not llm_config.model_path:
raise ValueError("LOCAL_MODEL_PATH not configured")
llm_provider = LocalLLMProvider(
model_path=llm_config.model_path,
n_ctx=llm_config.n_ctx,
n_gpu_layers=llm_config.n_gpu_layers,
temperature=llm_config.temperature,
max_tokens=llm_config.max_tokens
)
else:
llm_provider = RemoteLLMProvider(
api_key=llm_config.api_key,
model_name=llm_config.model_name,
base_url=llm_config.base_url,
temperature=llm_config.temperature,
max_tokens=llm_config.max_tokens
)
# Create and return agent
return UnifiedSummarizationAgent(
llm_provider=llm_provider,
enable_vision=agent_config.enable_vision
)
An example environment configuration file demonstrates the expected settings:
# .env.example - Environment configuration template
# LLM Provider Configuration
LLM_PROVIDER=local
# Options: 'local' or 'remote'
# Local LLM Settings (when LLM_PROVIDER=local)
LOCAL_MODEL_PATH=/path/to/models/mistral-7b-instruct.gguf
CONTEXT_SIZE=4096
GPU_LAYERS=-1
# -1 means use all available GPU layers
# 0 means CPU only
# Positive number specifies exact layer count
# Remote LLM Settings (when LLM_PROVIDER=remote)
OPENAI_API_KEY=your_api_key_here
REMOTE_MODEL_NAME=gpt-3.5-turbo
API_BASE_URL=https://api.openai.com/v1
# Leave API_BASE_URL empty for default OpenAI endpoint
# Generation Parameters
TEMPERATURE=0.7
MAX_TOKENS=2048
# Agent Features
ENABLE_VISION=true
ENABLE_CODE=true
# Logging
LOG_LEVEL=INFO
# Options: DEBUG, INFO, WARNING, ERROR
# Optional: Cache directory for models
CACHE_DIR=/path/to/cache
PERFORMANCE OPTIMIZATION AND BEST PRACTICES
Performance optimization requires attention to several key areas. For local LLM inference, GPU utilization is critical. The number of layers offloaded to GPU significantly impacts speed. We can implement dynamic layer adjustment based on available VRAM.
class OptimizedLocalLLMProvider(LocalLLMProvider):
"""
Enhanced local LLM provider with performance optimizations.
Includes automatic GPU layer optimization, batch processing
support, and memory management.
"""
def __init__(self, *args, **kwargs):
"""Initialize with automatic GPU optimization."""
# Detect optimal GPU layers before initialization
if kwargs.get('n_gpu_layers') == -1:
kwargs['n_gpu_layers'] = self._detect_optimal_layers()
super().__init__(*args, **kwargs)
def _detect_optimal_layers(self) -> int:
"""
Detect optimal number of GPU layers based on available VRAM.
This is a simplified heuristic. Production systems should
implement more sophisticated memory profiling.
Returns:
Recommended number of GPU layers
"""
try:
if torch.cuda.is_available():
# Get available GPU memory in GB
gpu_mem_gb = torch.cuda.get_device_properties(0).total_memory / (1024**3)
# Rough heuristic: 1 layer per 0.5GB VRAM
# Adjust based on model size
recommended_layers = int(gpu_mem_gb * 2)
self.logger.info(f"Detected {gpu_mem_gb:.1f}GB VRAM")
self.logger.info(f"Recommending {recommended_layers} GPU layers")
return recommended_layers
else:
self.logger.info("No CUDA GPU detected, using CPU")
return 0
except Exception as e:
self.logger.warning(f"GPU detection failed: {str(e)}")
return 0
For document processing, implementing caching prevents redundant parsing of the same files.
import hashlib
import json
from typing import Optional
class CachedDocumentSummarizer(DocumentSummarizer):
"""
Document summarizer with caching support.
Caches parsed document content and generated summaries
to improve performance for repeated operations.
"""
def __init__(self, llm_provider: LLMProvider, cache_dir: str = ".cache"):
"""
Initialize with cache directory.
Args:
llm_provider: LLM provider for generation
cache_dir: Directory for cache storage
"""
super().__init__(llm_provider)
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
self.logger.info(f"Cache directory: {self.cache_dir}")
def _get_file_hash(self, file_path: str) -> str:
"""
Calculate hash of file content for cache key.
Args:
file_path: Path to file
Returns:
SHA256 hash of file content
"""
hasher = hashlib.sha256()
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
hasher.update(chunk)
return hasher.hexdigest()
def _get_cache_path(self, file_hash: str, cache_type: str) -> Path:
"""
Get path for cached data.
Args:
file_hash: Hash of file content
cache_type: Type of cache ('parsed' or 'summary')
Returns:
Path to cache file
"""
return self.cache_dir / f"{file_hash}_{cache_type}.json"
def summarize_file(
self,
file_path: str,
length: SummaryLength = SummaryLength.MEDIUM,
focus_areas: Optional[str] = None,
target_audience: Optional[str] = None,
use_cache: bool = True
) -> SummaryResponse:
"""
Summarize file with caching support.
Checks cache before parsing and generating summary.
Args:
file_path: Path to document
length: Summary length
focus_areas: Focus areas
target_audience: Target audience
use_cache: Whether to use cache
Returns:
SummaryResponse with summary
"""
if use_cache:
# Calculate file hash
file_hash = self._get_file_hash(file_path)
# Create cache key from parameters
cache_key = f"{file_hash}_{length.value}_{focus_areas}_{target_audience}"
cache_key_hash = hashlib.md5(cache_key.encode()).hexdigest()
# Check for cached summary
summary_cache = self._get_cache_path(cache_key_hash, 'summary')
if summary_cache.exists():
self.logger.info("Loading summary from cache")
with open(summary_cache, 'r') as f:
cached_data = json.load(f)
return SummaryResponse(**cached_data)
# Generate new summary
response = super().summarize_file(
file_path,
length,
focus_areas,
target_audience
)
# Cache the result
if use_cache:
self.logger.info("Caching summary")
with open(summary_cache, 'w') as f:
json.dump({
'summary': response.summary,
'token_count': response.token_count,
'model_used': response.model_used,
'processing_time_seconds': response.processing_time_seconds
}, f)
return response
TESTING AND QUALITY ASSURANCE
Comprehensive testing ensures reliability across different scenarios. We implement unit tests for core components and integration tests for end-to-end workflows.
import unittest
from unittest.mock import Mock, patch, MagicMock
class TestDocumentParsers(unittest.TestCase):
"""Test suite for document parser implementations."""
def setUp(self):
"""Set up test fixtures."""
self.pdf_parser = PDFParser()
self.md_parser = MarkdownParser()
def test_pdf_parser_can_parse(self):
"""Test PDF parser file detection."""
self.assertTrue(self.pdf_parser.can_parse("document.pdf"))
self.assertTrue(self.pdf_parser.can_parse("file.PDF"))
self.assertFalse(self.pdf_parser.can_parse("document.docx"))
def test_markdown_parser_can_parse(self):
"""Test Markdown parser file detection."""
self.assertTrue(self.md_parser.can_parse("readme.md"))
self.assertTrue(self.md_parser.can_parse("doc.markdown"))
self.assertFalse(self.md_parser.can_parse("file.txt"))
@patch('pdfplumber.open')
def test_pdf_parser_extraction(self, mock_pdf_open):
"""Test PDF text extraction."""
# Mock PDF with two pages
mock_page1 = Mock()
mock_page1.extract_text.return_value = "Page 1 content"
mock_page2 = Mock()
mock_page2.extract_text.return_value = "Page 2 content"
mock_pdf = Mock()
mock_pdf.pages = [mock_page1, mock_page2]
mock_pdf.__enter__ = Mock(return_value=mock_pdf)
mock_pdf.__exit__ = Mock(return_value=False)
mock_pdf_open.return_value = mock_pdf
# Test extraction
result = self.pdf_parser.parse("test.pdf")
self.assertIn("Page 1 content", result)
self.assertIn("Page 2 content", result)
class TestLLMProviders(unittest.TestCase):
"""Test suite for LLM provider implementations."""
def test_summary_request_creation(self):
"""Test SummaryRequest value object."""
request = SummaryRequest(
content="Test content",
content_type=ContentType.DOCUMENT,
length=SummaryLength.BRIEF
)
self.assertEqual(request.content, "Test content")
self.assertEqual(request.content_type, ContentType.DOCUMENT)
self.assertEqual(request.length, SummaryLength.BRIEF)
@patch('openai.OpenAI')
def test_remote_provider_summary_generation(self, mock_openai):
"""Test remote LLM provider summary generation."""
# Mock API response
mock_response = Mock()
mock_response.choices = [Mock()]
mock_response.choices[0].message.content = "Generated summary"
mock_response.usage.completion_tokens = 50
mock_client = Mock()
mock_client.chat.completions.create.return_value = mock_response
mock_openai.return_value = mock_client
# Create provider and test
provider = RemoteLLMProvider(api_key="test_key")
request = SummaryRequest(
content="Test document",
content_type=ContentType.DOCUMENT
)
response = provider.generate_summary(request)
self.assertEqual(response.summary, "Generated summary")
self.assertEqual(response.token_count, 50)
class TestUnifiedAgent(unittest.TestCase):
"""Integration tests for unified summarization agent."""
def setUp(self):
"""Set up test agent with mock LLM."""
mock_llm = Mock(spec=LLMProvider)
mock_llm.generate_summary.return_value = SummaryResponse(
summary="Test summary",
token_count=100,
model_used="test-model",
processing_time_seconds=1.0
)
mock_llm.is_available.return_value = True
self.agent = UnifiedSummarizationAgent(
llm_provider=mock_llm,
enable_vision=False
)
def test_content_type_detection(self):
"""Test automatic content type detection."""
self.assertEqual(
self.agent._detect_content_type("file.pdf"),
'document'
)
self.assertEqual(
self.agent._detect_content_type("script.py"),
'code'
)
self.assertEqual(
self.agent._detect_content_type("image.jpg"),
'image'
)
def test_get_capabilities(self):
"""Test capabilities reporting."""
caps = self.agent.get_capabilities()
self.assertIn('document_formats', caps)
self.assertIn('code_formats', caps)
self.assertFalse(caps['vision_enabled'])
CONCLUSION AND FUTURE DIRECTIONS
This guide has demonstrated how to build a sophisticated LLM-based summarization agent that handles documents, source code, and images. The system architecture follows clean code principles, separating concerns into distinct layers and using dependency injection to maintain flexibility. By supporting both local and remote LLM providers, the system accommodates different deployment scenarios and resource constraints.
The local LLM implementation using llama-cpp-python provides privacy and cost benefits while supporting multiple GPU architectures including Nvidia CUDA, AMD ROCm, Intel GPUs, and Apple Metal. This multi-platform support ensures the system can run efficiently on diverse hardware configurations. The remote provider integration demonstrates how to leverage powerful cloud-based models when local resources are insufficient or specific capabilities are required.
Document parsing capabilities cover the most common formats including PDF, Word, PowerPoint, Markdown, and HTML. Each parser is implemented as a separate component following the single responsibility principle, making the system easy to extend with additional format support. The code summarization features go beyond simple text extraction, analyzing program structure and providing insights into design patterns and implementation details.
The multimodal image summarization capability represents the cutting edge of AI applications, combining vision models with language models to understand and describe visual content. This demonstrates the extensibility of the architecture and shows how new capabilities can be integrated without disrupting existing functionality.
Future enhancements could include support for video summarization using temporal analysis, integration with retrieval-augmented generation for more accurate summaries of technical documents, and fine-tuning capabilities to adapt the system to domain-specific terminology and requirements. The modular architecture makes these extensions straightforward to implement while maintaining system stability and reliability.
