Build a Technical Specification RAG Pipeline with Docling and Haystack

                      
                     
"""
Production implementation for building a Technical Specification RAG Pipeline using Docling and Haystack.
Provides secure, scalable operations for processing specifications and generating RAG metrics.
"""

from typing import Dict, Any, List
import os
import logging
import asyncio
import httpx
from sqlalchemy import create_engine, Column, Integer, String, Sequence
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from contextlib import asynccontextmanager

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Database configuration
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///rag_pipeline.db')
engine = create_engine(DATABASE_URL)
Base = declarative_base()

# SQLAlchemy ORM model for metrics
class RAGMetrics(Base):
    __tablename__ = 'rag_metrics'
    id = Column(Integer, Sequence('rag_metric_id_seq'), primary_key=True)
    specification_id = Column(String, nullable=False)
    red_count = Column(Integer, default=0)
    amber_count = Column(Integer, default=0)
    green_count = Column(Integer, default=0)

Base.metadata.create_all(engine)

# Configuration class for environment variables
class Config:
    retry_attempts: int = int(os.getenv('RETRY_ATTEMPTS', 3))
    backoff_factor: float = float(os.getenv('BACKOFF_FACTOR', 2))

# Async context manager for database session
@asynccontextmanager
async def get_db() -> AsyncGenerator[Session, None]:
    """
    Provide a database session to the caller.
    
    Yields:
        Session: SQLAlchemy session
    """  
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Helper function for input validation
async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        bool: True if valid
    Raises:
        ValueError: If validation fails
    """  
    if 'specification_id' not in data:
        raise ValueError('Missing specification_id')
    return True

# Helper function to sanitize fields
async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input data fields.
    
    Args:
        data: Input data to sanitize
    Returns:
        Dict[str, Any]: Sanitized data
    """  
    return {k: v.strip() for k, v in data.items() if isinstance(v, str)}

# Helper function to fetch data from an API
async def fetch_data(api_url: str) -> Dict[str, Any]:
    """Fetch data from the given API URL.
    
    Args:
        api_url: The URL of the API to fetch data from
    Returns:
        Dict[str, Any]: Response data
    Raises:
        httpx.HTTPStatusError: If the API request fails
    """  
    async with httpx.AsyncClient() as client:
        response = await client.get(api_url)
        response.raise_for_status()  # Raise an error for bad responses
        return response.json()

# Function to save metrics to the database
async def save_to_db(db: Session, data: RAGMetrics) -> None:
    """Save RAG metrics to the database.
    
    Args:
        db: Database session
        data: RAGMetrics instance to save
    Raises:
        Exception: If saving fails
    """  
    db.add(data)
    db.commit()
    logger.info('Metrics saved to database')

# Function to process a batch of specifications
async def process_batch(batch: List[Dict[str, Any]]) -> None:
    """Process a batch of specifications to generate RAG metrics.
    
    Args:
        batch: List of specifications to process
    """  
    async with get_db() as db:
        for item in batch:
            try:
                await validate_input(item)  # Validate input
                sanitized_data = await sanitize_fields(item)  # Sanitize data
                rag_data = RAGMetrics(
                    specification_id=sanitized_data['specification_id'],
                    red_count=sanitized_data.get('red_count', 0),
                    amber_count=sanitized_data.get('amber_count', 0),
                    green_count=sanitized_data.get('green_count', 0)
                )
                await save_to_db(db, rag_data)  # Save to DB
            except Exception as e:
                logger.error(f'Error processing item {item}: {e}')  # Log error

# Main orchestrator class for the pipeline
class RAGPipeline:
    def __init__(self, config: Config):
        self.config = config

    async def run(self, api_url: str) -> None:
        """Run the RAG Pipeline process.
        
        Args:
            api_url: URL to fetch specifications
        """  
        try:
            data = await fetch_data(api_url)  # Fetch data
            await process_batch(data)  # Process specifications
        except Exception as e:
            logger.error(f'Pipeline error: {e}')  # Log pipeline error

# Main block for execution
if __name__ == '__main__':
    api_url = os.getenv('API_URL', 'https://api.example.com/specifications')
    config = Config()  # Load configuration
    pipeline = RAGPipeline(config)  # Initialize pipeline
    asyncio.run(pipeline.run(api_url))  # Run pipeline