AI LLM Deployment for E-commerce Demand Forecasting

import boto3
import json
import os

# --- Configuration ---
SAGEMAKER_ENDPOINT_NAME = os.environ.get('SAGEMAKER_ENDPOINT_NAME', 'your-sagemaker-endpoint-name')
AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1')

sagemaker_runtime = boto3.client('sagemaker-runtime', region_name=AWS_REGION)

def lambda_handler(event, context):
    """
    AWS Lambda handler to invoke SageMaker endpoint for demand forecasting.

    Args:
        event (dict): Lambda event payload. Expected format:
            {
                "sku": "SKU123",
                "historical_data": [ ... ], # List of past sales data points
                "forecast_periods": 30 # Number of periods to forecast
            }
        context (object): Lambda context object.

    Returns:
        dict: Lambda response containing forecast or error message.
    """
    try:
        body = json.loads(event['body'])
        sku = body.get('sku')
        historical_data = body.get('historical_data')
        forecast_periods = body.get('forecast_periods', 30)

        if not sku or not historical_data:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'Missing sku or historical_data in request body.'})
            }

        # Construct the payload for the SageMaker endpoint
        # This payload structure MUST match what your trained LLM expects
        payload = {
            "inputs": {
                "sku": sku,
                "historical_data": historical_data,
                "forecast_periods": forecast_periods
            },
            "parameters": {
                "max_new_tokens": forecast_periods, # Example parameter
                "temperature": 0.7 # Example parameter
            }
        }

        # Invoke the SageMaker endpoint
        response = sagemaker_runtime.invoke_endpoint(
            EndpointName=SAGEMAKER_ENDPOINT_NAME,
            ContentType='application/json',
            Body=json.dumps(payload)
        )

        # Parse the response from SageMaker
        result = json.loads(response['Body'].read().decode('utf-8'))

        # Assuming the LLM returns a structured forecast, e.g., a list of predicted values
        # Adjust parsing based on your specific LLM's output format
        forecast_values = result.get('generated_text', []) # Example parsing

        # Format the output to include SKU and forecast periods
        formatted_forecast = {
            'sku': sku,
            'forecast_periods': forecast_periods,
            'predicted_demand': forecast_values
        }

        return {
            'statusCode': 200,
            'body': json.dumps(formatted_forecast)
        }

    except Exception as e:
        print(f"Error invoking SageMaker endpoint: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': f'Internal server error: {str(e)}'})
        }

# Example of how to test this locally (requires AWS credentials configured)
# if __name__ == '__main__':
#     # Replace with your actual test data
#     test_event = {
#         'body': json.dumps({
#             'sku': 'TESTSKU456',
#             'historical_data': [[1678886400, 10], [1678972800, 12]], # Example: [timestamp, quantity]
#             'forecast_periods': 7
#         })
#     }
#     # Ensure SAGEMAKER_ENDPOINT_NAME and AWS_REGION are set in your environment or replace them here
#     # print(lambda_handler(test_event, None))