Build RAG Pipelines for Equipment Maintenance Manuals with LlamaIndex and LangChain

                      
                     
"""
Production implementation for building RAG pipelines for Equipment Maintenance Manuals.
This module integrates LlamaIndex and LangChain to provide efficient data processing.
"""
from typing import Dict, Any, List
import os
import logging
import requests
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from time import sleep

# Logger setup for monitoring and debugging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class for environment variables
class Config:
    database_url: str = os.getenv('DATABASE_URL')
    llama_index_url: str = os.getenv('LLAMA_INDEX_URL')

# Create a database connection pool
engine = create_engine(Config.database_url, pool_size=5, max_overflow=10)
Session = sessionmaker(bind=engine)

async 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 'manual_id' not in data:
        raise ValueError('Missing manual_id')
    if not isinstance(data['manual_id'], str):
        raise ValueError('manual_id must be a string')
    return True

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

async def fetch_data(manual_id: str) -> Dict[str, Any]:
    """Fetch equipment manual data from LlamaIndex API.
    
    Args:
        manual_id: ID of the equipment manual
    Returns:
        JSON response from LlamaIndex
    Raises:
        RuntimeError: If API call fails
    """
    try:
        response = requests.get(f'{Config.llama_index_url}/manuals/{manual_id}')
        response.raise_for_status()  # Raise an error for bad responses
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.error(f'Error fetching data: {e}')
        raise RuntimeError('Failed to fetch data from LlamaIndex')

async def transform_records(data: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Transform raw data into the desired format.
    
    Args:
        data: Raw data fetched from the API
    Returns:
        List of transformed records
    """
    # Transformations go here, e.g., adjusting field names, types, etc.
    transformed = []
    for record in data.get('records', []):
        transformed_record = {
            'id': record['id'],
            'name': record['name'],
            'description': record['description'],
        }
        transformed.append(transformed_record)
    return transformed

async def save_to_db(records: List[Dict[str, Any]]) -> None:
    """Save transformed records to the database.
    
    Args:
        records: List of records to save
    Raises:
        RuntimeError: If database operation fails
    """
    with Session() as session:
        try:
            for record in records:
                session.execute(text("INSERT INTO manuals (id, name, description) VALUES (:id, :name, :description)"), record)
            session.commit()  # Commit the transaction
        except Exception as e:
            logger.error(f'Error saving data to DB: {e}')
            session.rollback()  # Rollback if any error occurs
            raise RuntimeError('Database operation failed')

async def process_batch(manual_id: str) -> None:
    """Process a batch of equipment manual data.
    
    Args:
        manual_id: The ID of the manual to process
    """
    try:
        logger.info(f'Starting to process manual ID: {manual_id}')
        data = await fetch_data(manual_id)  # Fetch data
        sanitized_data = await sanitize_fields(data)  # Sanitize fields
        transformed_records = await transform_records(sanitized_data)  # Transform data
        await save_to_db(transformed_records)  # Save to DB
        logger.info('Processing completed successfully.')
    except Exception as e:
        logger.error(f'Error in processing batch: {e}')

async def aggregate_metrics() -> None:
    """Aggregate metrics from the database for reporting.
    
    Returns:
        None
    """
    with Session() as session:
        result = session.execute(text("SELECT COUNT(*) FROM manuals"))
        logger.info(f'Total manuals processed: {result.scalar()}')

if __name__ == '__main__':
    # Example usage
    import asyncio
    manual_id_example = '123456'
    asyncio.run(process_batch(manual_id_example))
    asyncio.run(aggregate_metrics())