🔍 Code Extractor

Handles LLM processing for multi-page documents with context awareness, automatically selecting optimal analysis strategies based on document size.

Purpose

This class orchestrates the analysis of multi-page documents by intelligently choosing between different processing strategies (single page, contextual pages, progressive summary, or chunked analysis) based on document length. It maintains context across pages, generates comprehensive summaries, and produces unified responses that synthesize information from all pages. The class is designed to handle documents ranging from single pages to large documents with hundreds of pages, optimizing token usage and processing efficiency.

Source Code

class MultiPageLLMHandler:
    """Handles LLM processing for multi-page documents with context awareness"""
    
    def __init__(self, api_key: str):
        """Initialize multi-page LLM handler"""
        self.llm_handler = LLMHandler(api_key)
        self.logger = logging.getLogger(__name__)
        
    async def analyze_multi_page_document(self, 
                                        pages: List[PageAnalysis],
                                        metadata: Dict[str, Any],
                                        conversation_context: str = "") -> MultiPageAnalysisResult:
        """
        Analyze complete multi-page document with context awareness
        
        Args:
            pages: List of page analyses
            metadata: Document metadata
            conversation_context: Previous conversation context
            
        Returns:
            MultiPageAnalysisResult with comprehensive analysis
        """
        total_pages = len(pages)
        self.logger.info(f"Starting multi-page analysis of {total_pages} pages")
        
        # Statistics tracking
        stats = {
            'total_pages': total_pages,
            'pages_processed': 0,
            'total_tokens': 0,
            'processing_time': 0,
            'analysis_methods': []
        }
        
        # Choose analysis strategy based on document size
        if total_pages == 1:
            # Single page - use standard processing
            result = await self._analyze_single_page(pages[0], metadata, conversation_context)
            page_analyses = [result]
            stats['analysis_methods'].append('single_page')
            
        elif total_pages <= 5:
            # Small document - analyze each page with context
            page_analyses = await self._analyze_pages_with_context(pages, metadata, conversation_context)
            stats['analysis_methods'].append('contextual_pages')
            
        elif total_pages <= 20:
            # Medium document - progressive analysis with summaries
            page_analyses = await self._analyze_progressive_summary(pages, metadata, conversation_context)
            stats['analysis_methods'].append('progressive_summary')
            
        else:
            # Large document - chunk-based analysis
            page_analyses = await self._analyze_chunked_document(pages, metadata, conversation_context)
            stats['analysis_methods'].append('chunked_analysis')
        
        stats['pages_processed'] = len(page_analyses)
        
        # Generate document summary
        document_summary = await self._generate_document_summary(pages, page_analyses, metadata)
        
        # Create combined response
        combined_response = self._create_combined_response(page_analyses, document_summary, metadata)
        
        return MultiPageAnalysisResult(
            page_analyses=page_analyses,
            document_summary=document_summary,
            combined_response=combined_response,
            processing_stats=stats
        )
    
    async def _analyze_single_page(self, page: PageAnalysis, metadata: Dict[str, Any], 
                                 conversation_context: str) -> str:
        """Analyze single page using standard processing"""
        # Use existing LLM handler for single page
        enhanced_metadata = {**metadata, 'conversation_context': conversation_context}
        return await self.llm_handler.analyze_and_respond(page.image_b64, enhanced_metadata)
    
    async def _analyze_pages_with_context(self, pages: List[PageAnalysis], 
                                        metadata: Dict[str, Any],
                                        conversation_context: str) -> List[str]:
        """Analyze each page with full document context"""
        page_analyses = []
        cumulative_context = conversation_context
        
        for i, page in enumerate(pages):
            self.logger.info(f"Analyzing page {i+1}/{len(pages)} with context")
            
            # Build context from previous pages
            if i > 0:
                prev_summary = f"\nPrevious pages summary:\n"
                for j in range(i):
                    prev_summary += f"Page {j+1}: {page_analyses[j][:200]}...\n"
                cumulative_context += prev_summary
            
            # Create context-aware prompt
            prompt = f"""Analyzing page {i+1} of {len(pages)} from a multi-page document.

Document Context:
- Total pages: {len(pages)}
- Current page: {i+1}
- Processing mode: Contextual analysis

{cumulative_context}

Page {i+1} Text Content:
{page.text_content[:1000]}{'...' if len(page.text_content) > 1000 else ''}

Please analyze this page considering:
1. The content on this specific page
2. How it relates to previous pages in the document
3. The overall document flow and structure
4. Key information that builds upon previous content
5. Any questions or insights for this page

Provide a comprehensive analysis that considers the document context."""

            # Analyze with enhanced metadata
            enhanced_metadata = {
                **metadata,
                'page_number': page.page_number,
                'total_pages': len(pages),
                'custom_prompt': prompt,
                'analysis_mode': 'contextual'
            }
            
            analysis = await self.llm_handler.analyze_and_respond(page.image_b64, enhanced_metadata)
            page_analyses.append(analysis)
            
            # Update page analysis result
            page.analysis_result = analysis
        
        return page_analyses
    
    async def _analyze_progressive_summary(self, pages: List[PageAnalysis],
                                         metadata: Dict[str, Any],
                                         conversation_context: str) -> List[str]:
        """Analyze with progressive summarization for medium documents"""
        page_analyses = []
        running_summary = conversation_context
        
        # Process in chunks of 3-5 pages with summaries
        chunk_size = 4
        
        for chunk_start in range(0, len(pages), chunk_size):
            chunk_end = min(chunk_start + chunk_size, len(pages))
            chunk_pages = pages[chunk_start:chunk_end]
            
            self.logger.info(f"Processing chunk {chunk_start+1}-{chunk_end} of {len(pages)}")
            
            # Analyze chunk pages
            chunk_analyses = []
            for i, page in enumerate(chunk_pages):
                global_page_num = chunk_start + i + 1
                
                prompt = f"""Analyzing page {global_page_num} of {len(pages)} (chunk page {i+1}/{len(chunk_pages)}).

Document Progress Summary:
{running_summary}

Page {global_page_num} Content:
{page.text_content[:800]}{'...' if len(page.text_content) > 800 else ''}

Analyze this page focusing on:
1. Key content and insights
2. How it builds on previous pages
3. Important details for document understanding
4. Progression of ideas or information"""

                enhanced_metadata = {
                    **metadata,
                    'page_number': global_page_num,
                    'total_pages': len(pages),
                    'custom_prompt': prompt,
                    'analysis_mode': 'progressive'
                }
                
                analysis = await self.llm_handler.analyze_and_respond(page.image_b64, enhanced_metadata)
                chunk_analyses.append(analysis)
                page_analyses.append(analysis)
                page.analysis_result = analysis
            
            # Create chunk summary for next iteration
            if chunk_end < len(pages):  # Not the last chunk
                chunk_summary = f"\nPages {chunk_start+1}-{chunk_end} Summary:\n"
                for i, analysis in enumerate(chunk_analyses):
                    chunk_summary += f"Page {chunk_start + i + 1}: {analysis[:150]}...\n"
                running_summary += chunk_summary
        
        return page_analyses
    
    async def _analyze_chunked_document(self, pages: List[PageAnalysis],
                                      metadata: Dict[str, Any],
                                      conversation_context: str) -> List[str]:
        """Analyze large documents using chunked approach"""
        page_analyses = []
        
        # For large documents, analyze representative pages and create summaries
        self.logger.info(f"Using chunked analysis for {len(pages)} pages")
        
        # Select key pages for detailed analysis
        key_pages_indices = self._select_key_pages(pages)
        
        # Analyze key pages in detail
        for page_idx in key_pages_indices:
            page = pages[page_idx]
            
            prompt = f"""Analyzing key page {page_idx + 1} of {len(pages)} from a large document.

This is a representative page selected for detailed analysis.

Page Content:
{page.text_content[:1000]}{'...' if len(page.text_content) > 1000 else ''}

Provide a comprehensive analysis focusing on:
1. Main themes and topics on this page
2. Key information and insights
3. Document structure and organization
4. Important details that represent this section"""

            enhanced_metadata = {
                **metadata,
                'page_number': page.page_number,
                'total_pages': len(pages),
                'custom_prompt': prompt,
                'analysis_mode': 'key_page'
            }
            
            analysis = await self.llm_handler.analyze_and_respond(page.image_b64, enhanced_metadata)
            page.analysis_result = analysis
        
        # Create analyses for all pages (detailed for key pages, summary for others)
        for i, page in enumerate(pages):
            if i in key_pages_indices:
                page_analyses.append(page.analysis_result)
            else:
                # Create summary analysis for non-key pages
                summary = f"Page {i+1}: Contains {len(page.text_content)} characters of content."
                if page.text_content.strip():
                    # Extract first few sentences as summary
                    sentences = page.text_content.split('.')[:3]
                    summary += f" Key content: {'. '.join(sentences)[:200]}..."
                page_analyses.append(summary)
        
        return page_analyses
    
    def _select_key_pages(self, pages: List[PageAnalysis]) -> List[int]:
        """Select key pages for detailed analysis in large documents"""
        total_pages = len(pages)
        
        # Always include first and last pages
        key_indices = [0]
        if total_pages > 1:
            key_indices.append(total_pages - 1)
        
        # Add middle pages based on content density
        content_scores = []
        for i, page in enumerate(pages):
            score = len(page.text_content.strip())
            content_scores.append((score, i))
        
        # Sort by content score and select top pages
        content_scores.sort(reverse=True)
        
        # Select additional key pages (up to 10 total for very large docs)
        max_key_pages = min(10, max(3, total_pages // 10))
        
        for score, idx in content_scores[:max_key_pages]:
            if idx not in key_indices:
                key_indices.append(idx)
        
        return sorted(key_indices)
    
    async def _generate_document_summary(self, pages: List[PageAnalysis],
                                       page_analyses: List[str],
                                       metadata: Dict[str, Any]) -> DocumentSummary:
        """Generate comprehensive document summary"""
        # Use multi-page processor for basic summary
        from multi_page_processor import MultiPagePDFProcessor
        processor = MultiPagePDFProcessor()
        
        # Update pages with analysis results
        for i, analysis in enumerate(page_analyses):
            if i < len(pages):
                pages[i].analysis_result = analysis
        
        return processor.generate_document_summary(pages, metadata)
    
    def _create_combined_response(self, page_analyses: List[str],
                                document_summary: DocumentSummary,
                                metadata: Dict[str, Any]) -> str:
        """Create combined response from all analyses"""
        total_pages = len(page_analyses)
        
        response = f"# Multi-Page Document Analysis\n\n"
        response += f"**Document:** {Path(metadata.get('source_file', 'Unknown')).name}\n"
        response += f"**Pages:** {total_pages} pages processed\n"
        response += f"**Type:** {document_summary.document_type.replace('_', ' ').title()}\n"
        response += f"**Confidence:** {document_summary.confidence_score:.0%}\n\n"
        
        # Overall summary
        response += f"## Document Summary\n\n{document_summary.overall_summary}\n\n"
        
        # Key findings
        if document_summary.key_findings:
            response += f"## Key Findings\n\n"
            for finding in document_summary.key_findings:
                response += f"• {finding}\n"
            response += "\n"
        
        # Main topics
        if document_summary.main_topics:
            response += f"## Main Topics\n\n"
            for topic in document_summary.main_topics[:10]:  # Limit to top 10
                response += f"• {topic}\n"
            response += "\n"
        
        # Page-by-page analysis (condensed for large documents)
        if total_pages <= 10:
            response += f"## Page-by-Page Analysis\n\n"
            for i, analysis in enumerate(page_analyses):
                response += f"### Page {i+1}\n\n{analysis}\n\n"
        else:
            response += f"## Key Pages Analysis\n\n"
            # Show only key analyses for large documents
            key_pages = [0, total_pages//2, total_pages-1]  # First, middle, last
            for page_idx in key_pages:
                if page_idx < len(page_analyses):
                    response += f"### Page {page_idx + 1}\n\n{page_analyses[page_idx]}\n\n"
        
        return response

Parameters

Parameter Details

api_key: API key for the LLM service (passed to the underlying LLMHandler). This is required for authentication with the LLM provider (e.g., OpenAI, Anthropic).

Return Value

The constructor returns an instance of MultiPageLLMHandler. The main method analyze_multi_page_document returns a MultiPageAnalysisResult object containing: page_analyses (list of analysis strings for each page), document_summary (DocumentSummary object with overall insights), combined_response (formatted string with complete analysis), and processing_stats (dictionary with metrics like total_pages, pages_processed, total_tokens, processing_time, and analysis_methods used).

Class Interface

Methods

Attributes

Dependencies

• asyncio
• logging
• pathlib
• typing
• dataclasses
• llm_handler
• multi_page_processor

Required Imports

import asyncio
import logging
from pathlib import Path
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from llm_handler import LLMHandler
from multi_page_processor import PageAnalysis, DocumentSummary, MultiPagePDFProcessor

Conditional/Optional Imports

These imports are only needed under specific conditions:

from multi_page_processor import MultiPagePDFProcessor

Usage Example

import asyncio
from multi_page_llm_handler import MultiPageLLMHandler
from multi_page_processor import PageAnalysis

# Initialize handler
api_key = "your-api-key-here"
handler = MultiPageLLMHandler(api_key)

# Prepare page analyses (from PDF processor)
pages = [
    PageAnalysis(page_number=1, image_b64="base64_image_data", text_content="Page 1 text..."),
    PageAnalysis(page_number=2, image_b64="base64_image_data", text_content="Page 2 text...")
]

metadata = {
    'source_file': 'document.pdf',
    'total_pages': 2,
    'file_size': 1024000
}

# Analyze document
result = await handler.analyze_multi_page_document(
    pages=pages,
    metadata=metadata,
    conversation_context="Previous conversation context if any"
)

# Access results
print(result.combined_response)
print(f"Processed {result.processing_stats['pages_processed']} pages")
print(f"Analysis method: {result.processing_stats['analysis_methods']}")
for i, analysis in enumerate(result.page_analyses):
    print(f"Page {i+1}: {analysis[:200]}...")

Best Practices

• Always use async/await when calling analyze_multi_page_document as it performs asynchronous LLM operations
• Ensure PageAnalysis objects have valid image_b64 and text_content before passing to the handler
• The class automatically selects the optimal analysis strategy: single page (1 page), contextual (2-5 pages), progressive summary (6-20 pages), or chunked (20+ pages)
• For large documents, the handler intelligently selects key pages for detailed analysis to manage token usage
• The conversation_context parameter allows maintaining context across multiple document analysis sessions
• Monitor processing_stats in the result to understand which analysis method was used and track performance
• The handler updates PageAnalysis objects in-place with analysis_result attribute
• For very large documents (100+ pages), expect longer processing times as the handler processes in chunks
• Ensure sufficient API rate limits and quotas for the LLM service when processing large documents
• The combined_response provides a formatted, human-readable summary suitable for direct presentation

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class MultiPagePDFProcessor 72.8% similar

  A class for processing multi-page PDF documents with context-aware analysis, OCR, and summarization capabilities.

  From: /tf/active/vicechatdev/e-ink-llm/multi_page_processor.py

• class DocumentProcessor_v4 64.4% similar

  Handles document processing and text extraction using llmsherpa (same approach as offline_docstore_multi_vice.py).

  From: /tf/active/vicechatdev/docchat/document_processor.py

• class MultiPageAnalysisResult 64.3% similar

  A dataclass that encapsulates the complete results of analyzing a multi-page document, including individual page analyses, document summary, combined response, and processing statistics.

  From: /tf/active/vicechatdev/e-ink-llm/multi_page_llm_handler.py

• class DocumentAnalyzer 64.2% similar

  Analyze PDF documents using OCR and LLM

  From: /tf/active/vicechatdev/mailsearch/document_analyzer.py

• class DocumentProcessor_v1 62.4% similar

  A document processing class that extracts text from PDF and Word documents using llmsherpa as the primary method with fallback support for PyPDF2, pdfplumber, and python-docx.

  From: /tf/active/vicechatdev/contract_validity_analyzer/utils/document_processor_new.py