👨‍💻 Developer Guide¶

"""
Example: Building a custom cloud storage backend
"""
import json
import boto3
from typing import Dict, Any, List, Optional
from llm_evaluation_framework.persistence.base_store import BaseStore
from botocore.exceptions import ClientError

class S3PersistenceBackend(BaseStore):
    """
    AWS S3 persistence backend for scalable evaluation storage
    """

    def __init__(self, bucket_name: str, region: str = 'us-east-1', 
                 key_prefix: str = 'llm-evaluations/'):
        """
        Initialize S3 backend

        Args:
            bucket_name: S3 bucket name
            region: AWS region
            key_prefix: Prefix for all keys
        """
        self.bucket_name = bucket_name
        self.key_prefix = key_prefix

        # Initialize S3 client
        self.s3_client = boto3.client('s3', region_name=region)

        # Verify bucket access
        self._verify_bucket_access()

    def _verify_bucket_access(self):
        """Verify bucket exists and is accessible"""
        try:
            self.s3_client.head_bucket(Bucket=self.bucket_name)
        except ClientError as e:
            if e.response['Error']['Code'] == '404':
                raise ValueError(f"Bucket {self.bucket_name} does not exist")
            else:
                raise ValueError(f"Cannot access bucket {self.bucket_name}: {e}")

    def save(self, key: str, data: Dict[str, Any]) -> bool:
        """Save evaluation results to S3"""
        try:
            full_key = f"{self.key_prefix}{key}.json"

            # Add metadata for search/indexing
            metadata = self._extract_metadata(data)

            # Upload to S3
            self.s3_client.put_object(
                Bucket=self.bucket_name,
                Key=full_key,
                Body=json.dumps(data, default=str),
                ContentType='application/json',
                Metadata=metadata,
                ServerSideEncryption='AES256'  # Enable encryption
            )

            return True

        except Exception as e:
            print(f"S3 save failed: {e}")
            return False

    def load(self, key: str) -> Dict[str, Any]:
        """Load evaluation results from S3"""
        try:
            full_key = f"{self.key_prefix}{key}.json"

            response = self.s3_client.get_object(
                Bucket=self.bucket_name,
                Key=full_key
            )

            return json.loads(response['Body'].read())

        except ClientError as e:
            if e.response['Error']['Code'] == 'NoSuchKey':
                raise KeyError(f"Key not found: {key}")
            raise

    def query(self, filters: Dict[str, Any] = None) -> List[Dict[str, Any]]:
        """Query evaluations with optional filters"""
        try:
            # List objects with prefix
            response = self.s3_client.list_objects_v2(
                Bucket=self.bucket_name,
                Prefix=self.key_prefix
            )

            if 'Contents' not in response:
                return []

            results = []
            for obj in response['Contents']:
                try:
                    # Get object metadata
                    head_response = self.s3_client.head_object(
                        Bucket=self.bucket_name,
                        Key=obj['Key']
                    )

                    metadata = head_response.get('Metadata', {})

                    # Apply filters if provided
                    if filters and not self._matches_filters(metadata, filters):
                        continue

                    # Load full object if it matches filters
                    key = obj['Key'].replace(self.key_prefix, '').replace('.json', '')
                    full_data = self.load(key)
                    results.append(full_data)

                except Exception as e:
                    print(f"Error processing object {obj['Key']}: {e}")
                    continue

            return results

        except Exception as e:
            print(f"S3 query failed: {e}")
            return []

    def delete(self, key: str) -> bool:
        """Delete evaluation from S3"""
        try:
            full_key = f"{self.key_prefix}{key}.json"

            self.s3_client.delete_object(
                Bucket=self.bucket_name,
                Key=full_key
            )

            return True

        except Exception as e:
            print(f"S3 delete failed: {e}")
            return False

    def list_evaluations(self, limit: int = 100) -> List[Dict[str, str]]:
        """List available evaluations with metadata"""
        try:
            response = self.s3_client.list_objects_v2(
                Bucket=self.bucket_name,
                Prefix=self.key_prefix,
                MaxKeys=limit
            )

            if 'Contents' not in response:
                return []

            evaluations = []
            for obj in response['Contents']:
                key = obj['Key'].replace(self.key_prefix, '').replace('.json', '')
                evaluations.append({
                    'key': key,
                    'last_modified': obj['LastModified'].isoformat(),
                    'size': obj['Size']
                })

            return evaluations

        except Exception as e:
            print(f"S3 list failed: {e}")
            return []

    def _extract_metadata(self, data: Dict[str, Any]) -> Dict[str, str]:
        """Extract metadata for S3 object tagging"""
        metadata = {}

        if 'model_name' in data:
            metadata['model-name'] = data['model_name']

        if 'timestamp' in data:
            metadata['timestamp'] = data['timestamp']

        if 'aggregate_metrics' in data:
            metrics = data['aggregate_metrics']
            if 'accuracy' in metrics:
                metadata['accuracy'] = str(round(metrics['accuracy'], 3))
            if 'total_cost' in metrics:
                metadata['total-cost'] = str(round(metrics['total_cost'], 4))

        return metadata

    def _matches_filters(self, metadata: Dict[str, str], 
                        filters: Dict[str, Any]) -> bool:
        """Check if metadata matches filters"""
        for filter_key, filter_value in filters.items():
            if filter_key == 'model_name':
                if metadata.get('model-name') != filter_value:
                    return False
            elif filter_key == 'min_accuracy':
                accuracy = float(metadata.get('accuracy', 0))
                if accuracy < filter_value:
                    return False
            # Add more filter conditions as needed

        return True

# Usage example
def use_custom_persistence():
    """Example of using custom S3 persistence backend"""

    # Initialize S3 backend
    s3_backend = S3PersistenceBackend(
        bucket_name='my-llm-evaluations',
        region='us-west-2',
        key_prefix='evaluations/production/'
    )

    # Use in persistence manager
    from llm_evaluation_framework.persistence import PersistenceManager

    persistence_manager = PersistenceManager({
        's3': s3_backend,
        'local': JsonStore('./local_backup/')  # Local backup
    })

    # Use in evaluation engine
    from llm_evaluation_framework import ModelInferenceEngine, ModelRegistry

    registry = ModelRegistry()
    engine = ModelInferenceEngine(registry, persistence_manager)

    # Run evaluation - results automatically saved to S3
    results = engine.evaluate_model("gpt-3.5-turbo", test_cases)

    # Query stored evaluations
    recent_evaluations = s3_backend.query({
        'model_name': 'gpt-3.5-turbo',
        'min_accuracy': 0.8
    })

    print(f"Found {len(recent_evaluations)} high-accuracy evaluations")