Extract Structured Equipment Diagnostics from LLMs with DSPy and Instructor

                      
                     
"""
Production implementation for extracting structured equipment diagnostics from LLMs using DSPy and Instructor.
Provides secure, scalable operations with optimal logging and error handling.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import requests
import json
import asyncio
from sqlalchemy import create_engine, Column, Integer, String, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from tenacity import retry, stop_after_attempt, wait_exponential

# Setting up logging with appropriate levels
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class to hold environment variables and type hints
class Config:
    database_url: str = os.getenv('DATABASE_URL')
    llm_api_url: str = os.getenv('LLM_API_URL')

# SQLAlchemy base model
Base = declarative_base()

# Database model for diagnostics
class EquipmentDiagnostic(Base):
    __tablename__ = 'equipment_diagnostics'
    id = Column(Integer, primary_key=True)
    equipment_id = Column(String, index=True)
    diagnostics = Column(JSON)

# Logger setup for database interactions
engine = create_engine(Config.database_url)
session_factory = sessionmaker(bind=engine)

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

# Helper function to sanitize fields
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: v.strip() for k, v in data.items() if isinstance(v, str)}

# Helper function to call LLM API
@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=4, max=10))
async def call_llm_api(equipment_id: str) -> Dict[str, Any]:
    """Call the LLM API to fetch diagnostics.
    
    Args:
        equipment_id: ID of the equipment
    Returns:
        Diagnostics data from LLM
    Raises:
        HTTPError: If API call fails
    """
    logger.info(f'Calling LLM API for equipment_id: {equipment_id}')
    response = requests.get(f'{Config.llm_api_url}/{equipment_id}')
    response.raise_for_status()  # Raise an error for bad responses
    return response.json()

# Helper function to save diagnostics to the database
async def save_to_db(session: Session, equipment_id: str, diagnostics: Dict[str, Any]) -> None:
    """Save diagnostics to the database.
    
    Args:
        session: Active database session
        equipment_id: ID of the equipment
        diagnostics: Diagnostics data to save
    Raises:
        Exception: If database save fails
    """
    logger.info(f'Saving diagnostics for equipment_id: {equipment_id}')
    diag = EquipmentDiagnostic(equipment_id=equipment_id, diagnostics=diagnostics)
    session.add(diag)
    session.commit()  # Commit transaction

# Helper function for processing batch of equipment IDs
async def process_batch(equipment_ids: List[str]) -> None:
    """Process a batch of equipment IDs to extract diagnostics.
    
    Args:
        equipment_ids: List of equipment IDs
    """
    logger.info('Processing batch of equipment IDs')
    async with session_factory() as session:
        for eq_id in equipment_ids:
            try:
                await validate_input({'equipment_id': eq_id})  # Validate input
                sanitized_id = sanitize_fields({'equipment_id': eq_id})['equipment_id']
                diagnostics = await call_llm_api(sanitized_id)  # Call LLM API
                await save_to_db(session, sanitized_id, diagnostics)  # Save to DB
            except Exception as e:
                logger.error(f'Error processing equipment_id {eq_id}: {e}')  # Log error

# Main orchestrator class to handle the workflow
class EquipmentDiagnosticsExtractor:
    def __init__(self, equipment_ids: List[str]) -> None:
        self.equipment_ids = equipment_ids

    async def run(self) -> None:
        """Run the diagnostics extraction process.
        
        Raises:
            RuntimeError: If extraction fails
        """
        try:
            await process_batch(self.equipment_ids)  # Process all IDs
        except Exception as e:
            logger.error(f'Error in extraction process: {e}')  # Log overall error
            raise RuntimeError('Extraction process failed')  # Raise for upstream handling

if __name__ == '__main__':
    # Example usage
    equipment_ids = ['eq1', 'eq2', 'eq3']
    extractor = EquipmentDiagnosticsExtractor(equipment_ids)
    asyncio.run(extractor.run())
    # Run the diagnostics extraction process asynchronously