Building Production-Ready RAG: Why Parsing Quality is Everything

# Basic file loading setup
pdf_path = "bank_q1_report.pdf"
print(f"Processing: {pdf_path}")

import pdfplumber

def extract_with_pdfplumber(pdf_path):
    with pdfplumber.open(pdf_path) as pdf:
        page = pdf.pages[0]
        
        # Extract text with layout awareness
        text = page.extract_text()
        
        # Extract tables
        tables = page.extract_tables()
        
        print(f"Text preview: {text[:500]}")
        print(f"Found {len(tables)} tables")
        
        return text, tables

# Usage
text, tables = extract_with_pdfplumber("bank_q1_report.pdf")

import camelot
import pandas as pd

def extract_tables_camelot(pdf_path):
    # Extract all tables from PDF
    tables = camelot.read_pdf(pdf_path, flavor="stream", pages="1-end")
    
    extracted_data = []
    for i, table in enumerate(tables):
        print(f"\n Table {i+1} - Accuracy: {table.accuracy:.2f}%")
        print(table.df.head())
        extracted_data.append(table.df)
    
    return extracted_data

# Usage
tables = extract_tables_camelot("bank_q1_report.pdf")

from unstructured.partition.pdf import partition_pdf

def extract_with_unstructured(pdf_path):
    # High-resolution strategy for better accuracy
    elements = partition_pdf(pdf_path, strategy="hi_res")
    
    # Organize by content type
    content_map = {
        'titles': [],
        'tables': [],
        'paragraphs': [],
        'lists': []
    }
    
    for element in elements:
        category = element.category.lower()
        content = element.text[:150] + "..." if len(element.text) > 150 else element.text
        
        print(f"{element.category}: {content}")
        
        if 'title' in category:
            content_map['titles'].append(element.text)
        elif 'table' in category:
            content_map['tables'].append(element.text)
        elif 'text' in category:
            content_map['paragraphs'].append(element.text)
    
    return content_map

# Usage
content = extract_with_unstructured("bank_q1_report.pdf")
print(f"Found: {len(content['titles'])} titles, {len(content['tables'])} tables")

from docling.document_converter import DocumentConverter

def extract_with_docling(pdf_path):
    # Initialize the converter
    converter = DocumentConverter()
    
    # Convert PDF with AI understanding
    result = converter.convert(pdf_path)
    
    # Export to structured markdown
    markdown_content = result.document.export_to_markdown()
    
    # Get structured elements
    text_elements = result.document.texts
    table_elements = result.document.tables
    
    print(f"Markdown preview:\n{markdown_content[:500]}...")
    print(f"\n Found {len(text_elements)} text elements")
    print(f"Found {len(table_elements)} tables")
    
    return {
        'markdown': markdown_content,
        'texts': text_elements,
        'tables': table_elements
    }

# Usage
docling_result = extract_with_docling("bank_q1_report.pdf")

# Visualize document structure
docling_result['document'].show_layout()
# This creates an interactive layout view

from pdf2image import convert_from_path
from pytesseract import image_to_string
import cv2
import numpy as np

def extract_with_basic_ocr(pdf_path):
    # Convert PDF to images
    images = convert_from_path(pdf_path, dpi=300)  # Higher DPI for better quality
    
    extracted_text = ""
    
    for i, img in enumerate(images):
        # Convert PIL to OpenCV format for preprocessing
        img_cv = cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)
        
        # Preprocess image for better OCR
        gray = cv2.cvtColor(img_cv, cv2.COLOR_BGR2GRAY)
        
        # Apply OCR
        page_text = image_to_string(gray, config='--psm 6')
        extracted_text += f"\n--- Page {i+1} ---\n{page_text}\n"
        
        print(f"Processed page {i+1}")
    
    return extracted_text

# Usage
ocr_text = extract_with_basic_ocr("bank_q1_report.pdf")
print(f"Extracted {len(ocr_text)} characters")

from pdf2image import convert_from_path
from surya.ocr import run_ocr
from surya.model.detection.segformer import load_model as load_det_model, load_processor as load_det_processor
from surya.model.recognition.model import load_model as load_rec_model
from surya.model.recognition.processor import load_processor as load_rec_processor

def extract_with_surya_ocr(pdf_path, max_pages=5):
    """
    State-of-the-art OCR extraction using Surya
    Offers better accuracy than traditional OCR methods
    """
    print("Loading Surya OCR models...")
    
    # Load Surya models (detection + recognition)
    det_model = load_det_model()
    det_processor = load_det_processor()
    rec_model = load_rec_model()
    rec_processor = load_rec_processor()
    
    # Convert PDF to images (limit pages for demo)
    images = convert_from_path(pdf_path)[:max_pages]
    languages = [['en']] * len(images)  # English for all pages
    
    print(f"Processing {len(images)} pages...")
    
    # Run advanced OCR
    predictions = run_ocr(
        images, 
        languages,
        det_model, 
        det_processor, 
        rec_model, 
        rec_processor
    )
    
    # Extract and organize text
    extracted_content = {
        'full_text': '',
        'page_texts': [],
        'confidence_scores': []
    }
    
    for i, page_result in enumerate(predictions):
        page_text = "\n".join([line.text for line in page_result.text_lines])
        avg_confidence = sum([line.confidence for line in page_result.text_lines]) / len(page_result.text_lines)
        
        extracted_content['page_texts'].append(page_text)
        extracted_content['confidence_scores'].append(avg_confidence)
        extracted_content['full_text'] += f"\n--- Page {i+1} ---\n{page_text}\n\n"
        
        print(f"Page {i+1}: {len(page_text)} chars, confidence: {avg_confidence:.2f}")
    
    return extracted_content

# Usage
surya_result = extract_with_surya_ocr("bank_q1_report.pdf")
print(f"\n Total extracted: {len(surya_result['full_text'])} characters")
print(f"Average confidence: {sum(surya_result['confidence_scores'])/len(surya_result['confidence_scores']):.2f}")

from docling.document_converter import DocumentConverter
import camelot
import fitz
from typing import Dict, List, Any

class HybridPDFParser:
    """Production-ready hybrid parsing pipeline"""
    
    def __init__(self):
        self.docling_converter = DocumentConverter()
        
    def parse_document(self, pdf_path: str) -> Dict[str, Any]:
        """
        Comprehensive document parsing using hybrid approach
        """
        print(f"Starting hybrid parsing for: {pdf_path}")
        
        # 1. Docling for semantic structure
        print("Phase 1: AI-native structure extraction...")
        docling_result = self.docling_converter.convert(pdf_path)
        
        text_elements = docling_result.document.texts
        docling_tables = docling_result.document.tables
        markdown_content = docling_result.document.export_to_markdown()
        
        # 2. Camelot for precise table extraction
        print("Phase 2: Precision table extraction...")
        try:
            camelot_tables = camelot.read_pdf(pdf_path, flavor="stream", pages="1-10")
            high_accuracy_tables = [t for t in camelot_tables if t.accuracy > 80]
        except Exception as e:
            print(f"Camelot extraction failed: {e}")
            high_accuracy_tables = []
        
        # 3. PyMuPDF for image extraction
        print("Phase 3: Image and metadata extraction...")
        doc = fitz.open(pdf_path)
        images = []
        metadata = doc.metadata
        
        for page_num, page in enumerate(doc):
            page_images = page.get_images()
            for img_index, img in enumerate(page_images):
                images.append({
                    'page': page_num + 1,
                    'index': img_index,
                    'xref': img[0],
                    'info': img
                })
        
        doc.close()
        
        # 4. Combine and structure results
        parsed_content = {
            'document_info': {
                'source': pdf_path,
                'total_pages': len(docling_result.document.pages) if hasattr(docling_result.document, 'pages') else 'N/A',
                'metadata': metadata
            },
            'structured_content': {
                'markdown': markdown_content,
                'text_elements': len(text_elements),
                'docling_tables': len(docling_tables),
                'precision_tables': len(high_accuracy_tables),
                'images': len(images)
            },
            'raw_data': {
                'texts': text_elements,
                'docling_tables': docling_tables,
                'camelot_tables': high_accuracy_tables,
                'images': images
            }
        }
        
        self._print_summary(parsed_content)
        return parsed_content
    
    def _print_summary(self, content: Dict[str, Any]):
        """Print parsing summary"""
        print("\n Hybrid Parsing Results:")
        print(f"  Text elements: {content['structured_content']['text_elements']}")
        print(f"  Docling tables: {content['structured_content']['docling_tables']}")
        print(f" Precision tables: {content['structured_content']['precision_tables']}")
        print(f" Images found: {content['structured_content']['images']}")
        print(f" Total pages: {content['document_info']['total_pages']}")

# Usage
hybrid_parser = HybridPDFParser()
result = hybrid_parser.parse_document("bank_q1_report.pdf")

# Access different data types
markdown_for_rag = result['structured_content']['markdown']
precision_tables = result['raw_data']['camelot_tables']
image_data = result['raw_data']['images']

from markitdown import MarkItDown
import json
from typing import Dict, Any

def extract_with_markitdown(pdf_path: str) -> Dict[str, Any]:
    """
    Microsoft's MarkItDown for intelligent document conversion
    """
    print(f"Processing {pdf_path} with MarkItDown...")
    
    # Initialize MarkItDown converter
    md_converter = MarkItDown()
    
    # Convert to structured markdown
    result = md_converter.convert(pdf_path)
    
    # Extract key information
    extracted_data = {
        'markdown_content': result.text_content,
        'content_length': len(result.text_content),
        'conversion_successful': bool(result.text_content.strip())
    }
    
    print(f"Converted to {extracted_data['content_length']} characters")
    print(f"Preview: {result.text_content[:200]}...")
    
    return extracted_data

# Usage
markdown_result = extract_with_markitdown("bank_q1_report.pdf")

import openai  # or anthropic, depending on your preference
from typing import Dict, List
import json

def llm_semantic_restructuring(raw_text: str, document_type: str = "financial_report") -> Dict[str, Any]:
    """
    Use LLM to intelligently restructure document content for RAG
    """
    
    system_prompt = f"""
    You are a document parsing expert. Transform the following {document_type} content into a structured JSON format optimized for RAG systems.
    
    Extract and organize:
    1. Key financial metrics and KPIs
    2. Executive summary points
    3. Table data with proper context
    4. Section hierarchies
    5. Important dates and figures
    
    Output clean, searchable chunks with semantic meaning.
    """
    
    user_prompt = f"""
    Parse and restructure this document content:
    
    {raw_text[:8000]}  # Limit for token constraints
    
    Return structured JSON with 'sections', 'key_metrics', 'tables', and 'summary'.
    """
    
    try:
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.1,  # Low temperature for consistent extraction
            max_tokens=2000
        )
        
        # Parse the structured response
        structured_content = json.loads(response.choices[0].message.content)
        
        print(" LLM restructuring completed")
        print(f"Extracted {len(structured_content.get('sections', []))} sections")
        
        return structured_content
        
    except Exception as e:
        print(f" LLM processing failed: {e}")
        return {"error": str(e), "raw_content": raw_text[:1000]}

# Usage with previously extracted content
raw_content = markdown_result['markdown_content']
structured_data = llm_semantic_restructuring(raw_content, "quarterly_financial_report")

# Perfect for RAG chunking
rag_chunks = structured_data.get('sections', [])
key_metrics = structured_data.get('key_metrics', {})

import os, time, gc, fitz, pdfplumber, pytesseract
import camelot
from pdf2image import convert_from_path
from unstructured.partition.pdf import partition_pdf
from docling.document_converter import DocumentConverter
from Levenshtein import ratio
import matplotlib.pyplot as plt
import numpy as np
from math import pi
import fitz

# Configuration settings for memory-safe execution in Colab
os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
PDF_PATH = "/content/sample.pdf"
TEMP_PATH = "/content/temp_first_pages.pdf"
GROUND_TRUTH_PATH = "/content/ground_truth.txt"

def measure_accuracy(extracted_text, ground_truth):
    """Calculate text similarity using Levenshtein distance ratio."""
    return ratio(extracted_text, ground_truth)

def measure_speed(func):
    """Time how long a parsing function takes to execute."""
    start = time.time()
    result = func()
    end = time.time()
    return result, end - start

def semantic_score(text):
    """
    Estimate document structure quality based on headings and paragraphs.
    Returns a score from 1 to 5.
    """
    headings = sum([1 for line in text.split("\n") if line.isupper() and len(line) < 80])
    paragraphs = len([p for p in text.split("\n\n") if len(p.strip()) > 100])
    if headings > 10: return 5
    elif headings > 5: return 4
    elif headings > 2: return 3
    else: return 2

def scanned_support_check():
    """Test if OCR can extract text from the first page of the PDF."""
    try:
        pages = convert_from_path(PDF_PATH, first_page=1, last_page=2)
        text = pytesseract.image_to_string(pages[0])
        return len(text.strip()) > 20
    except Exception:
        return False

def layout_parsing():
    """Extract text using pdfplumber and tables using Camelot."""
    text = ""
    with pdfplumber.open(PDF_PATH) as pdf:
        for i, page in enumerate(pdf.pages[:10]):
            text += page.extract_text() or ""
    try:
        tables = camelot.read_pdf(PDF_PATH, pages="1-10")
        text += "\n".join([t.df.to_string() for t in tables])
    except Exception:
        pass
    return text

def ml_parsing():
    """
    Use machine learning-based document understanding with Docling.
    Processes only the first 5 pages to avoid memory issues.
    """
    doc = fitz.open(PDF_PATH)
    new_doc = fitz.open()
    for i in range(min(5, len(doc))):
        new_doc.insert_pdf(doc, from_page=i, to_page=i)
    new_doc.save(TEMP_PATH)
    new_doc.close()

    text = ""
    try:
        converter = DocumentConverter()
        converted_doc = converter.convert(TEMP_PATH)
        text = converted_doc.document.export_to_markdown()
    except Exception as e:
        print("Docling parsing error:", e)
    return text

def ocr_parsing():
    """Convert PDF pages to images and extract text using Tesseract OCR."""
    text = ""
    pages = convert_from_path(PDF_PATH, first_page=1, last_page=5)
    for page in pages:
        text += pytesseract.image_to_string(page)
    return text

def hybrid_parsing():
    """Combine PyMuPDF for text extraction with Camelot for tables."""
    text = ""
    doc = fitz.open(PDF_PATH)
    for page in doc[:10]:
        text += page.get_text()
    try:
        tables = camelot.read_pdf(PDF_PATH, pages="1-10")
        text += "\n".join([t.df.to_string() for t in tables])
    except Exception:
        pass
    return text

def llm_augmented_parsing():
    """Use MarkItDown for LLM-enhanced document conversion."""
    try:
        from markitdown import MarkItDown
        md = MarkItDown()
        out = md.convert(PDF_PATH)
        return out.text_content
    except Exception as e:
        print("MarkItDown error:", e)
        return ""

# Define all parsing techniques to evaluate
techniques = {
    "Layout-Based": layout_parsing,
    "ML-Based": ml_parsing,
    "OCR-Based": ocr_parsing,
    "Hybrid": hybrid_parsing,
    "LLM-Augmented": llm_augmented_parsing
}

# Load ground truth if available
ground_truth = open(GROUND_TRUTH_PATH).read() if os.path.exists(GROUND_TRUTH_PATH) else ""
results = []

# Run evaluation for each parsing technique
for name, func in techniques.items():
    print(f"\nEvaluating {name} Parsing...")
    text, elapsed = measure_speed(func)
    acc = measure_accuracy(text, ground_truth) if ground_truth else np.nan
    sem = semantic_score(text)
    scan = scanned_support_check()
    results.append({
        "Technique": name,
        "Accuracy": acc,
        "Speed_per_page": elapsed / 10,
        "Semantic": sem,
        "ScannedSupport": scan
    })
    # Clean up memory after each test
    del text
    gc.collect()

# Extract metrics for visualization
techs = [r["Technique"] for r in results]
acc = [r["Accuracy"] for r in results]
spd = [r["Speed_per_page"] for r in results]
sem = [r["Semantic"] for r in results]
ocr_support = [1 if r["ScannedSupport"] else 0 for r in results]

# Create comparison charts
plt.figure(figsize=(12, 8))
plt.subplot(2,2,1)
plt.bar(techs, acc, alpha=0.7)
plt.title("Accuracy Comparison")
plt.ylabel("Levenshtein Similarity")

plt.subplot(2,2,2)
plt.bar(techs, spd, alpha=0.7, color='orange')
plt.title("Speed (Seconds per Page)")
plt.ylabel("Time (s)")

plt.subplot(2,2,3)
plt.bar(techs, sem, alpha=0.7, color='green')
plt.title("Semantic Structure Score (1–5)")

plt.subplot(2,2,4)
plt.bar(techs, ocr_support, alpha=0.7, color='purple')
plt.title("Scanned PDF Support (Yes=1, No=0)")

plt.tight_layout()
plt.show()

# Print results as a formatted table
print("\n| Technique | Accuracy | Speed (s/page) | Semantic | Scanned Support |")
print("|------------|-----------|----------------|-----------|-----------------|")
for r in results:
    acc_val = f"{r['Accuracy']:.2f}" if not np.isnan(r['Accuracy']) else "N/A"
    print(f"| {r['Technique']} | {acc_val} | {r['Speed_per_page']:.2f} | {r['Semantic']} | {r['ScannedSupport']} |")