Retrieve Visual Factory Schematics by Content Similarity with ColPali and LlamaIndex

"""
Production implementation for retrieving visual factory schematics based on content similarity.
Utilizes ColPali and LlamaIndex for efficient data processing.
"""

from typing import Dict, Any, List, Optional
import os
import logging
import requests
import time
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class for environment variables.
    """
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///./test.db')
    colpali_url: str = os.getenv('COLPALI_URL', 'http://localhost:8000')
    llama_index_url: str = os.getenv('LLAMA_INDEX_URL', 'http://localhost:8001')

# Create a database engine and session
engine = create_engine(Config.database_url)
session_factory = sessionmaker(bind=engine)

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

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

def fetch_data(url: str, params: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Fetch data from a given URL with retries.
    
    Args:
        url: The URL to fetch data from
        params: Parameters to include in the request
    Returns:
        Response data as a dictionary
    """
    for attempt in range(5):  # Retry up to 5 times
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            logger.warning(f'Fetch attempt {attempt + 1} failed: {e}')
            time.sleep(2 ** attempt)  # Exponential backoff
    logger.error('All fetch attempts failed')
    return None

def save_to_db(data: Dict[str, Any]) -> None:
    """Save processed data to the database.
    
    Args:
        data: Data to save
    """
    with session_factory() as session:
        session.execute(text('INSERT INTO schematics (content) VALUES (:content)'), {'content': data['content']})
        session.commit()
        logger.info('Data saved to database')

def normalize_data(schematics: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Normalize schematics data.
    
    Args:
        schematics: List of schematics to normalize
    Returns:
        Normalized list of schematics
    """
    return [{'id': s['id'], 'content': s['content'].lower()} for s in schematics]

def process_batch(data: List[Dict[str, Any]]) -> None:
    """Process a batch of data.
    
    Args:
        data: List of data to process
    """
    for item in data:
        save_to_db(item)

def format_output(data: List[Dict[str, Any]]) -> str:
    """Format output for display.
    
    Args:
        data: List of data to format
    Returns:
        Formatted string output
    """
    return '\n'.join([f'ID: {d['id']}, Content: {d['content']}' for d in data])

class VisualFactorySchematicsRetriever:
    """Main orchestrator class for retrieving schematics by content similarity.
    """
    def __init__(self) -> None:
        self.colpali_url = Config.colpali_url
        self.llama_index_url = Config.llama_index_url

    def retrieve_schematics(self, query: str) -> List[Dict[str, Any]]:
        """Main business logic to retrieve schematics.
        
        Args:
            query: Search query for schematics
        Returns:
            List of schematics matching the query
        """
        try:
            params = {'query': query}
            # Validate and sanitize input
            validate_input(params)
            params = sanitize_fields(params)
            # Fetch data from ColPali
            colpali_data = fetch_data(self.colpali_url, params)
            if colpali_data is None:
                raise RuntimeError('Failed to retrieve data from ColPali')
            # Normalize data
            normalized_data = normalize_data(colpali_data)
            # Process batch to save data to DB
            process_batch(normalized_data)
            logger.info('Schematics retrieval successful')
            return normalized_data
        except Exception as e:
            logger.error(f'Error during schematics retrieval: {e}')
            raise

if __name__ == '__main__':
    # Example usage
    retriever = VisualFactorySchematicsRetriever()
    try:
        result = retriever.retrieve_schematics('example query')
        output = format_output(result)
        print(output)
    except Exception as e:
        logger.error(f'Failed to execute retrieval: {e}')