---
source_url: "https://blog.stackademic.com/building-a-serverless-llm-pipeline-with-amazon-bedrock-and-sagemaker-fine-tuning-using-aws-cdk-f2627ad04e15"
title: Medium
mirrored_at: 2026-08-31T13:01:41.931Z
host: blog.stackademic.com
cited_in_42a: true
mirror_canonical: "https://index.42a.ai/blog.stackademic.com/building-a-serverless-llm-pipeline-with-amazon-bedrock-and-sagemaker-fine-tuning-using-aws-cdk-f2627ad04e15"
---

> **Original source:** https://blog.stackademic.com/building-a-serverless-llm-pipeline-with-amazon-bedrock-and-sagemaker-fine-tuning-using-aws-cdk-f2627ad04e15

31 min read

Feb 22, 2026

\--

Large-language models (LLMs) can support a wide range of use cases such as classification, summaries, etc. However they can require additional customization to incorporate domain-specific knowledge and up-to-date information.

In this blog we will build serverless pipelines that fine-tuning LLM Models using Amazon SageMaker, and deploying these models. Using AWS CDK as infrastructure as code, the solution separates training workflow from inference workflow, ensuring the production workloads remain stable and unaffected during model training and update. Additionally, leveraging Amazon AppConfig allows dynamic configuration updates without requiring redeployment.

The app is built using Kiro🔥

For non-member please read it from [here](https://dev.to/katevu/building-a-serverless-llm-pipeline-with-amazon-bedrock-and-sagemaker-fine-tuning-using-aws-cdk-4125)

## **Architecture Overview**

The system is composed of two main pipeline as the diagram below:

Press enter or click to view image in full size

-   Training/Fine-tuning pipeline: responsible for data preparation, model fine-tuning, evaluation, and approval.
-   Inference pipeline: responsible for serving production request using the approved model.

1.  **Training pipeline**

The training process is responsible for fine-tuning LLM models using AWS resources. While we rely on AWS resources to do the heavy job. The workflow is manually initiated.

-   Data Preparation:  
    \- Training datasets are downloaded from 1 in three sources: Hugging Face, Amazon public data, or synthesis data.  
    \- The data is formatted and splitted into 3 small sets:  
    Training dataset  
    Validation dataset  
    Test dataset  
    \- The datasets are then uploaded to S3 bucket to be ready for training process
-   Model Fine-Tuning:  
    \- Fine tuning is executed using Amazon SageMaker, and triggered by a python script.  
    \- The script supports both full-training and LoRA options with LoRa as default.  
    \- After the training job completes, evaluation metrics will be generated. If you satisfy with the result, register the model in SageMaker model registry and wait for approval.
-   Automated deployment trigger: Once the model is approved, a lambda function will be triggered automatically to:  
    \- Create a new SageMaker endpoint.  
    \- Update AWS Systems Manager Parameter Store with the new endpoint.

**2\. Inference pipeline**

This pipeline is responsible for handling realtime review summary requests from users. Incoming requests will be received via API Gateway, which invokes a lambda function to process it. The generated summaries will be stored in S3 bucket for later purposes such as auditing, monitoring, or analytical purposes.

To enable comparison between the foundation LLM model and the fine-tune model, the Lambda function first invokes a Foundation model. It then invokes the Amazon SageMaker endpoint created by the training pipeline above.

AWS AppConfig is used to manage runtime settings such as which model to invoke. This approach enables dynamic model switching without redeploying the whole application.

## Building the app

### **1\. AppConfig stack**

This will leverage Amazon AppConfig to store the config for runtime.First we define the json for each environment:

{  
  "bedrock": {  
    "modelId": "anthropic.claude-3-haiku-20240307-v1:0",  
    "maxTokens": 200,  
    "temperature": 0.5,  
    "topP": 0.9  
  },  
  "sagemaker": {  
    "enabled": true,  
    "timeout": 30000,  
    "models": {  
      "stable": {  
        "endpointName": "endpoint-kate",  
        "description": "Kate's development model",  
        "weight": 100  
      }  
    },  
    "strategy": "weighted"  
  },  
  "rag": {  
    "enabled": false,  
    "topK": 3  
  },  
  "features": {  
    "sentimentAnalysis": true,  
    "caching": false,  
    "useNewSummarizationPrompt": false,  
    "enableAdvancedRAG": false,  
    "useMultiModelEnsemble": false  
  },  
  "abTesting": {  
    "enabled": false,  
    "rules": \[\]  
  },  
  "monitoring": {  
    "logABTestAssignments": true,  
    "trackModelPerformance": true,  
    "metricsNamespace": "LLMPipeline/Kate"  
  }  
}

Then we create the stack

import \* as cdk from 'aws-cdk-lib';  
import { Construct } from 'constructs';  
import \* as appconfig from 'aws-cdk-lib/aws-appconfig';  
import \* as iam from 'aws-cdk-lib/aws-iam';  
import \* as fs from 'fs';  
import \* as path from 'path';  
import { EnvironmentConfig } from './utils';

export interface AppConfigStackProps extends cdk.StackProps {  
  config: EnvironmentConfig;  
}

export class AppConfigStack extends cdk.Stack {  
  public readonly application: appconfig.CfnApplication;  
  public readonly appConfigEnvironment: appconfig.CfnEnvironment;  
  public readonly configurationProfile: appconfig.CfnConfigurationProfile;

  constructor(scope: Construct, id: string, props: AppConfigStackProps) {  
    super(scope, id, props);

    const { config } = props;

    // Create AppConfig Application  
    this.application = new appconfig.CfnApplication(this, 'Application', {  
      name: \`llm-pipeline-${config.environmentName}\`,  
      description: 'Configuration for LLM Pipeline',  
    });

    // Create AppConfig Environment  
    this.appConfigEnvironment = new appconfig.CfnEnvironment(this, 'Environment', {  
      applicationId: this.application.ref,  
      name: config.environmentName,  
      description: \`${config.environmentName} environment\`,  
    });

    // Create Configuration Profile  
    this.configurationProfile = new appconfig.CfnConfigurationProfile(this, 'ConfigProfile', {  
      applicationId: this.application.ref,  
      name: 'runtime-config',  
      description: 'Runtime configuration for Lambda functions',  
      locationUri: 'hosted',  
      type: 'AWS.Freeform',  
    });

    // Initial configuration with A/B testing support  
    // These are RUNTIME settings that can be updated without redeployment  
    // Loaded from config/appconfig-{environment}.json  
    const configPath = path.join(\_\_dirname, \`../config/appconfig-${config.environmentName}.json\`);

        let configContent: string;  
    if (!fs.existsSync(configPath)) {  
      throw new Error(  
        \`\\n========================================\\n\` +  
        \`ERROR: AppConfig file missing for environment "${config.environmentName}"\\n\` +  
        \`========================================\\n\` +  
        \`Expected file: config/appconfig-${config.environmentName}.json\\n\` +  
        \`Full path: ${configPath}\\n\\n\` +  
        \`Please create this file with runtime configuration.\\n\` +  
        \`You can copy from an existing environment:\\n\` +  
        \`  cp config/appconfig-kate.json config/appconfig-${config.environmentName}.json\\n\` +  
        \`========================================\\n\`  
      );  
    }

        try {  
      configContent = fs.readFileSync(configPath, 'utf8');  
      // Validate it's valid JSON  
      JSON.parse(configContent);  
      console.log(\`✓ Loaded AppConfig for "${config.environmentName}" from: ${configPath}\`);  
    } catch (error) {  
      throw new Error(  
        \`\\n========================================\\n\` +  
        \`ERROR: Invalid AppConfig JSON for environment "${config.environmentName}"\\n\` +  
        \`========================================\\n\` +  
        \`File: config/appconfig-${config.environmentName}.json\\n\` +  
        \`Error: ${error instanceof Error ? error.message : String(error)}\\n\\n\` +  
        \`Please ensure the file contains valid JSON.\\n\` +  
        \`Check for:\\n\` +  
        \`  - Missing commas\\n\` +  
        \`  - Trailing commas\\n\` +  
        \`  - Unquoted keys\\n\` +  
        \`  - Invalid escape sequences\\n\` +  
        \`========================================\\n\`  
      );  
    }

    // Create deployment strategy (immediate deployment)  
    const deploymentStrategy = new appconfig.CfnDeploymentStrategy(this, 'DeploymentStrategy', {  
      name: \`immediate-${config.environmentName}\`,  
      deploymentDurationInMinutes: 0,  
      growthFactor: 100,  
      replicateTo: 'NONE',  
      finalBakeTimeInMinutes: 0,  
    });

    // Create hosted configuration version  
    const configVersion = new appconfig.CfnHostedConfigurationVersion(this, 'ConfigVersion', {  
      applicationId: this.application.ref,  
      configurationProfileId: this.configurationProfile.ref,  
      content: configContent,  
      contentType: 'application/json',  
      description: 'Initial configuration',  
    });

    // Automatically deploy the configuration  
    new appconfig.CfnDeployment(this, 'Deployment', {  
      applicationId: this.application.ref,  
      environmentId: this.appConfigEnvironment.ref,  
      deploymentStrategyId: deploymentStrategy.ref,  
      configurationProfileId: this.configurationProfile.ref,  
      configurationVersion: configVersion.ref,  
      description: 'Automatic deployment from CDK',  
    });

    // Outputs  
    new cdk.CfnOutput(this, 'ApplicationId', {  
      value: this.application.ref,  
      description: 'AppConfig Application ID',  
      exportName: \`${config.environmentName}-appconfig-app-id\`,  
    });

    new cdk.CfnOutput(this, 'EnvironmentId', {  
      value: this.appConfigEnvironment.ref,  
      description: 'AppConfig Environment ID',  
      exportName: \`${config.environmentName}-appconfig-env-id\`,  
    });

    new cdk.CfnOutput(this, 'ConfigurationProfileId', {  
      value: this.configurationProfile.ref,  
      description: 'AppConfig Configuration Profile ID',  
      exportName: \`${config.environmentName}-appconfig-profile-id\`,  
    });  
  }

  /\*\*  
   \* Grant Lambda function permission to read AppConfig  
   \*/  
  public grantRead(grantee: iam.IGrantable): void {  
    grantee.grantPrincipal.addToPrincipalPolicy(  
      new iam.PolicyStatement({  
        effect: iam.Effect.ALLOW,  
        actions: \[  
          'appconfig:GetConfiguration',  
          'appconfig:GetLatestConfiguration',  
          'appconfig:StartConfigurationSession',  
        \],  
        resources: \['\*'\],  
      })  
    );  
  }  
}

### 2\. Fine-Tuning Model Pipeline

**2.1 Create the pipeline**

This pipeline will create these AWS resources below:

-   S3 Buckets: training data S3 bucket and model artifact S3 bucket
-   SageMaker IAM role used for training job
-   SSM parameter store to store the endpoint version
-   EventBridge rule to trigger the process of when the model is approved
-   Lambda function to deploy the approved models
-   Cloudwatch logs

import \* as cdk from 'aws-cdk-lib';  
import { Construct } from 'constructs';  
import \* as s3 from 'aws-cdk-lib/aws-s3';  
import \* as lambda from 'aws-cdk-lib/aws-lambda';  
import { PythonFunction } from '@aws-cdk/aws-lambda-python-alpha';  
import \* as iam from 'aws-cdk-lib/aws-iam';  
import \* as events from 'aws-cdk-lib/aws-events';  
import \* as targets from 'aws-cdk-lib/aws-events-targets';  
import \* as ssm from 'aws-cdk-lib/aws-ssm';  
import \* as logs from 'aws-cdk-lib/aws-logs';  
import { EnvironmentConfig } from './utils';

export interface TrainingPipelineStackProps extends cdk.StackProps {  
  config: EnvironmentConfig;  
}

export class TrainingPipelineStack extends cdk.Stack {  
  public readonly trainingBucket: s3.Bucket;  
  public readonly modelBucket: s3.Bucket;  
  public readonly endpointParameter: ssm.StringParameter;

  constructor(scope: Construct, id: string, props: TrainingPipelineStackProps) {  
    super(scope, id, props);

    const { config } = props;

    // ========================================  
    // S3 Buckets for Training  
    // ========================================

    // NOTE: Using DESTROY for cost-saving during development  
    // For production, change to RETAIN to preserve training data and models  
    this.trainingBucket = new s3.Bucket(this, 'TrainingDataBucket', {  
      bucketName: \`training-data-${config.environmentName}-${cdk.Aws.ACCOUNT\_ID}\`,  
      removalPolicy: cdk.RemovalPolicy.DESTROY,  
      autoDeleteObjects: true,  
      versioned: true,  
      encryption: s3.BucketEncryption.S3\_MANAGED,  
      lifecycleRules: \[  
        {  
          id: 'DeleteOldVersions',  
          noncurrentVersionExpiration: cdk.Duration.days(90),  
        },  
      \],  
    });

    // NOTE: Using DESTROY for cost-saving during development  
    // For production, change to RETAIN to preserve model artifacts  
    this.modelBucket = new s3.Bucket(this, 'ModelArtifactsBucket', {  
      bucketName: \`model-artifacts-${config.environmentName}-${cdk.Aws.ACCOUNT\_ID}\`,  
      removalPolicy: cdk.RemovalPolicy.DESTROY,  
      autoDeleteObjects: true,  
      versioned: true,  
      encryption: s3.BucketEncryption.S3\_MANAGED,  
    });

    // ========================================  
    // Parameter Store for Active Endpoint  
    // ========================================

    this.endpointParameter = new ssm.StringParameter(this, 'ActiveEndpointParameter', {  
      parameterName: \`/summarizer/${config.environmentName}/active-endpoint\`,  
      stringValue: 'none',  
      description: 'Active SageMaker endpoint name for inference',  
      tier: ssm.ParameterTier.STANDARD,  
    });

    // ========================================  
    // IAM Role for SageMaker Training  
    // ========================================

    const sagemakerRole = new iam.Role(this, 'SageMakerTrainingRole', {  
      assumedBy: new iam.ServicePrincipal('sagemaker.amazonaws.com'),  
      managedPolicies: \[  
        iam.ManagedPolicy.fromAwsManagedPolicyName('AmazonSageMakerFullAccess'),  
      \],  
    });

    this.trainingBucket.grantReadWrite(sagemakerRole);  
    this.modelBucket.grantReadWrite(sagemakerRole);

    // ========================================  
    // Lambda: Update Endpoint on Model Approval  
    // ========================================

    const updateEndpointLogGroup = new logs.LogGroup(this, 'UpdateEndpointLogGroup', {  
      logGroupName: \`/aws/lambda/update-endpoint-${config.environmentName}\`,  
      retention: logs.RetentionDays.ONE\_WEEK,  
      removalPolicy: cdk.RemovalPolicy.DESTROY,  
    });

    const updateEndpointFn = new PythonFunction(this, 'UpdateEndpointFunction', {  
      functionName: \`update-endpoint-${config.environmentName}\`,  
      entry: 'src/lambdas/update-endpoint',  
      runtime: lambda.Runtime.PYTHON\_3\_11,  
      index: 'handler.py',  
      handler: 'handler',  
      description: \`Update SageMaker endpoint for ${config.environmentName}\`,  
      timeout: cdk.Duration.minutes(5),  
      memorySize: 256,  
      environment: {  
        PARAMETER\_NAME: this.endpointParameter.parameterName,  
        ENVIRONMENT: config.environmentName,  
        SAGEMAKER\_ROLE\_ARN: sagemakerRole.roleArn,  
      },  
      logGroup: updateEndpointLogGroup,  
    });

    // Grant permissions  
    this.endpointParameter.grantRead(updateEndpointFn);  
    this.endpointParameter.grantWrite(updateEndpointFn);

        updateEndpointFn.addToRolePolicy(new iam.PolicyStatement({  
      effect: iam.Effect.ALLOW,  
      actions: \[  
        'sagemaker:DescribeModelPackage',  
        'sagemaker:CreateModel',  
        'sagemaker:CreateEndpoint',  
        'sagemaker:CreateEndpointConfig',  
        'sagemaker:UpdateEndpoint',  
        'sagemaker:DescribeEndpoint',  
      \],  
      resources: \['\*'\],  
    }));

        // Grant permission to pass the SageMaker execution role  
    updateEndpointFn.addToRolePolicy(new iam.PolicyStatement({  
      effect: iam.Effect.ALLOW,  
      actions: \['iam:PassRole'\],  
      resources: \[sagemakerRole.roleArn\],  
    }));

    // ========================================  
    // EventBridge: Trigger on Model Approval  
    // ========================================

    const modelApprovalRule = new events.Rule(this, 'ModelApprovalRule', {  
      ruleName: \`model-approval-${config.environmentName}\`,  
      description: 'Trigger endpoint update when SageMaker model is approved',  
      eventPattern: {  
        source: \['aws.sagemaker'\],  
        detailType: \['SageMaker Model Package State Change'\],  
        detail: {  
          ModelApprovalStatus: \['Approved'\],  
        },  
      },  
    });

    modelApprovalRule.addTarget(new targets.LambdaFunction(updateEndpointFn));

    // ========================================  
    // Outputs  
    // ========================================

    new cdk.CfnOutput(this, 'TrainingBucketName', {  
      value: this.trainingBucket.bucketName,  
      description: 'S3 bucket for training data',  
      exportName: \`${config.environmentName}-training-bucket\`,  
    });

    new cdk.CfnOutput(this, 'ModelBucketName', {  
      value: this.modelBucket.bucketName,  
      description: 'S3 bucket for model artifacts',  
      exportName: \`${config.environmentName}-model-bucket\`,  
    });

    new cdk.CfnOutput(this, 'EndpointParameterName', {  
      value: this.endpointParameter.parameterName,  
      description: 'Parameter Store key for active endpoint',  
      exportName: \`${config.environmentName}-endpoint-parameter\`,  
    });

    new cdk.CfnOutput(this, 'SageMakerRoleArn', {  
      value: sagemakerRole.roleArn,  
      description: 'IAM role for SageMaker training jobs',  
      exportName: \`${config.environmentName}-sagemaker-role\`,  
    });  
  }  
}

**2.2 Create the scripts:**

-   **Prepare training datasets**

Create a python script to download the datasets. The datasets can be downloaded from huggingface, amazon reviews, or generate

#!/usr/bin/env python3  
"""  
Download and prepare training data from public datasets

This script downloads customer review data and formats it for SageMaker training.  
It supports multiple sources:  
1\. Hugging Face Datasets (recommended - easy and reliable)  
2\. Amazon Customer Reviews (real data from AWS Open Data Registry)  
3\. Synthetic data (generated for testing)

Output: training\_data/ folder with train.jsonl, validation.jsonl, test.jsonl

Usage:  
    # Download from Hugging Face (recommended)  
    python scripts/download\_training\_data.py --source huggingface --dataset amazon\_polarity --num-samples 5000

    # Generate synthetic data for testing  
    python scripts/download\_training\_data.py --source synthetic --num-samples 1000

    # Download real Amazon reviews  
    python scripts/download\_training\_data.py --source amazon --max-samples 5000  
"""

import os  
import json  
import gzip  
import argparse  
import urllib.request  
import ssl  
from pathlib import Path  
from typing import List, Dict  
import random

\# Fix SSL certificate verification issue on macOS  
ssl.\_create\_default\_https\_context = ssl.\_create\_unverified\_context

def download\_huggingface\_dataset(  
    output\_dir: Path, dataset\_name: str = "amazon\_polarity", max\_samples: int = 5000  
):  
    """  
    Download dataset from Hugging Face  
    Source: https://huggingface.co/datasets

    Popular datasets:  
    - amazon\_polarity: Amazon reviews (positive/negative) - NO SUMMARIES  
    - yelp\_review\_full: Yelp reviews with 1-5 star ratings - NO SUMMARIES  
    - imdb: Movie reviews - NO SUMMARIES  
    - rotten\_tomatoes: Movie reviews - NO SUMMARIES  
    - app\_reviews: Mobile app reviews - NO SUMMARIES  
    - cnn\_dailymail: News articles WITH SUMMARIES (recommended for summarization)  
    - xsum: News WITH SUMMARIES (extreme summarization)  
    - samsum: Dialogues WITH SUMMARIES  
    """  
    print(f"\\n📦 Downloading from Hugging Face: {dataset\_name}")  
    print(f"This may take a few minutes...")

    try:  
        from datasets import load\_dataset  
    except ImportError:  
        print("\\n❌ Error: 'datasets' library not installed")  
        print("Install it with: pip install datasets")  
        return \[\]

    try:  
        # Load dataset with config if needed  
        print(f"Loading dataset '{dataset\_name}'...")

                # Datasets that need config versions  
        if dataset\_name == 'cnn\_dailymail':  
            dataset = load\_dataset(dataset\_name, '3.0.0')  
        elif dataset\_name == 'xsum':  
            dataset = load\_dataset(dataset\_name)  
        elif dataset\_name == 'samsum':  
            dataset = load\_dataset(dataset\_name)  
        else:  
            # Regular datasets (reviews)  
            dataset = load\_dataset(dataset\_name)

        # Get train split  
        train\_data = dataset\["train"\]

        # Process samples  
        reviews = \[\]  
        count = 0

        print(f"Processing samples...")  
        for item in train\_data:  
            if count >= max\_samples:  
                break

            # Handle summarization datasets differently  
            if dataset\_name == 'cnn\_dailymail':  
                text = item.get('article', '')  
                summary = item.get('highlights', '')  
                sentiment = 'neutral'  
            elif dataset\_name == 'xsum':  
                text = item.get('document', '')  
                summary = item.get('summary', '')  
                sentiment = 'neutral'  
            elif dataset\_name == 'samsum':  
                text = item.get('dialogue', '')  
                summary = item.get('summary', '')  
                sentiment = 'neutral'  
            else:  
                # Review datasets - extract text and label  
                text = None  
                label = None

                # Try common field names  
                if "content" in item:  
                    text = item\["content"\]  
                elif "text" in item:  
                    text = item\["text"\]  
                elif "review" in item:  
                    text = item\["review"\]

                if "label" in item:  
                    label = item\["label"\]  
                elif "sentiment" in item:  
                    label = item\["sentiment"\]  
                elif "stars" in item:  
                    label = item\["stars"\]

                if not text:  
                    continue

                # Skip very short reviews  
                if len(text) < 50:  
                    continue

                # Determine sentiment from label  
                sentiment = "neutral"  
                if isinstance(label, int):  
                    if label >= 4 or label == 1:  # 5-star or positive binary  
                        sentiment = "positive"  
                    elif label <= 2 or label == 0:  # 1-2 star or negative binary  
                        sentiment = "negative"  
                    else:  
                        sentiment = "neutral"  
                elif isinstance(label, str):  
                    sentiment = label.lower()

                # Create summary (first 150 chars or extract key points)  
                # NOTE: This is NOT a real summary, just for demo purposes  
                summary = create\_summary\_from\_text(text)

            # Skip if no text or summary  
            if not text or not summary or len(text) < 50:  
                continue

            reviews.append(  
                {  
                    "text": text,  
                    "summary": summary,  
                    "sentiment": sentiment,  
                    "source": dataset\_name,  
                }  
            )

            count += 1  
            if count % 500 == 0:  
                print(f"Processed {count} samples...")

        print(f"✅ Processed {len(reviews)} samples from Hugging Face dataset")  
        return reviews

    except Exception as e:  
        print(f"\\n⚠️  Error loading dataset: {str(e)}")  
        print(f"\\nAvailable datasets:")  
        print("  Summarization (recommended):")  
        print("    - cnn\_dailymail (news articles with summaries)")  
        print("    - xsum (news with one-sentence summaries)")  
        print("    - samsum (dialogues with summaries)")  
        print("  Reviews (no real summaries):")  
        print("    - amazon\_polarity")  
        print("    - yelp\_review\_full")  
        print("    - imdb")  
        print("    - rotten\_tomatoes")  
        print("    - app\_reviews")  
        print(  
            "\\nTry: python scripts/download\_training\_data.py --source huggingface --dataset cnn\_dailymail"  
        )  
        return \[\]

def create\_summary\_from\_text(text: str, max\_length: int = 150) -> str:  
    """  
    Create a simple summary from review text  
    Takes first sentence or first N characters  
    """  
    # Try to get first sentence  
    sentences = text.split(".")  
    if sentences and len(sentences\[0\]) > 20:  
        summary = sentences\[0\].strip() + "."  
        if len(summary) <= max\_length:  
            return summary

    # Otherwise, take first N characters  
    if len(text) <= max\_length:  
        return text

    return text\[:max\_length\].rsplit(" ", 1)\[0\] + "..."

def download\_file(url: str, output\_path: str):  
    """Download file from URL with progress"""  
    print(f"Downloading from {url}...")

    def progress\_hook(count, block\_size, total\_size):  
        percent = int(count \* block\_size \* 100 / total\_size)  
        print(f"\\rProgress: {percent}%", end="", flush=True)

    urllib.request.urlretrieve(url, output\_path, progress\_hook)  
    print("\\nDownload complete!")

def download\_amazon\_reviews(  
    output\_dir: Path, category: str = "Electronics", max\_samples: int = 10000  
):  
    """  
    Download Amazon Customer Reviews dataset  
    Source: https://registry.opendata.aws/amazon-reviews/  
    """  
    print(f"\\n📦 Downloading Amazon Reviews - {category} category")  
    print(f"This may take a few minutes...")

    # Amazon Reviews Open Data URLs  
    base\_url = "https://s3.amazonaws.com/amazon-reviews-pds/tsv"  
    filename = f"amazon\_reviews\_us\_{category}\_v1\_00.tsv.gz"  
    url = f"{base\_url}/{filename}"

    # Download  
    temp\_file = output\_dir / filename

    try:  
        download\_file(url, str(temp\_file))  
    except Exception as e:  
        print(f"\\n⚠️  Download failed: {str(e)}")  
        print(f"\\nTrying alternative method using AWS CLI...")

        # Try using AWS CLI as fallback  
        import subprocess

        try:  
            result = subprocess.run(  
                \[  
                    "aws",  
                    "s3",  
                    "cp",  
                    f"s3://amazon-reviews-pds/tsv/{filename}",  
                    str(temp\_file),  
                \],  
                capture\_output=True,  
                text=True,  
            )  
            if result.returncode != 0:  
                print(f"AWS CLI also failed: {result.stderr}")  
                print(f"\\n💡 Tip: You can manually download from:")  
                print(f"   {url}")  
                print(f"   Save to: {temp\_file}")  
                return \[\]  
        except FileNotFoundError:  
            print(f"AWS CLI not found. Please install it or download manually from:")  
            print(f"   {url}")  
            return \[\]

    # Parse and convert to JSONL  
    print(f"\\nProcessing reviews...")  
    reviews = \[\]

    with gzip.open(temp\_file, "rt", encoding="utf-8") as f:  
        # Skip header  
        header = f.readline().strip().split("\\t")

        # Find column indices  
        try:  
            review\_idx = header.index("review\_body")  
            headline\_idx = header.index("review\_headline")  
            rating\_idx = header.index("star\_rating")  
        except ValueError as e:  
            print(f"Error: Could not find required columns in dataset")  
            return \[\]

        count = 0  
        for line in f:  
            if count >= max\_samples:  
                break

            try:  
                fields = line.strip().split("\\t")  
                if len(fields) <= max(review\_idx, headline\_idx, rating\_idx):  
                    continue

                review\_text = fields\[review\_idx\]  
                headline = fields\[headline\_idx\]  
                rating = int(fields\[rating\_idx\])

                # Skip empty reviews  
                if not review\_text or len(review\_text) < 50:  
                    continue

                # Determine sentiment from rating  
                if rating >= 4:  
                    sentiment = "positive"  
                elif rating <= 2:  
                    sentiment = "negative"  
                else:  
                    sentiment = "neutral"

                # Use headline as summary (not perfect but works for training)  
                # In production, you'd want human-written summaries  
                summary = headline if headline else review\_text\[:100\]

                reviews.append(  
                    {  
                        "text": review\_text,  
                        "summary": summary,  
                        "sentiment": sentiment,  
                        "rating": rating,  
                    }  
                )

                count += 1  
                if count % 1000 == 0:  
                    print(f"Processed {count} reviews...")

            except Exception as e:  
                continue

    # Clean up temp file  
    temp\_file.unlink()

    print(f"✅ Processed {len(reviews)} reviews from Amazon dataset")  
    return reviews

def create\_synthetic\_data(num\_samples: int = 1000) -> List\[Dict\]:  
    """  
    Create synthetic training data for testing  
    Use this if you can't download real data  
    """  
    print(f"\\n🔧 Generating {num\_samples} synthetic reviews...")

    templates = {  
        "positive": \[  
            (  
                "This product is absolutely amazing! {feature1} and {feature2}. Highly recommend to anyone looking for quality.",  
                "Excellent product with great {feature1} and {feature2}. Highly recommended.",  
            ),  
            (  
                "I'm very impressed with this purchase. The {feature1} exceeded my expectations and {feature2}. Worth every penny!",  
                "Very satisfied with {feature1} and {feature2}. Great value.",  
            ),  
            (  
                "Outstanding quality! {feature1} is incredible and {feature2}. Best purchase I've made this year.",  
                "Outstanding {feature1} and {feature2}. Excellent purchase.",  
            ),  
        \],  
        "negative": \[  
            (  
                "Very disappointed with this product. {issue1} and {issue2}. Would not recommend.",  
                "Poor quality with {issue1} and {issue2}. Not recommended.",  
            ),  
            (  
                "This is a waste of money. {issue1} after just a few days and {issue2}. Terrible experience.",  
                "Product failed quickly with {issue1} and {issue2}. Waste of money.",  
            ),  
            (  
                "Do not buy this! {issue1} and {issue2}. Customer service was unhelpful too.",  
                "Major issues with {issue1} and {issue2}. Poor support.",  
            ),  
        \],  
        "neutral": \[  
            (  
                "It's okay for the price. {aspect1} but {aspect2}. Nothing special.",  
                "Average product. {aspect1} but {aspect2}.",  
            ),  
            (  
                "Does what it's supposed to do. {aspect1} though {aspect2}. Acceptable.",  
                "Functional product. {aspect1} with {aspect2}.",  
            ),  
            (  
                "Mixed feelings about this. {aspect1} but {aspect2}. Could be better.",  
                "Mixed quality. {aspect1} but {aspect2}.",  
            ),  
        \],  
    }

    features = \[  
        "The battery life is excellent",  
        "The build quality feels premium",  
        "The performance is outstanding",  
        "The design is beautiful",  
        "The screen quality is amazing",  
        "The sound quality is superb",  
        "The camera takes great photos",  
        "The speed is impressive",  
    \]

    issues = \[  
        "It stopped working",  
        "The battery drains quickly",  
        "The build quality is poor",  
        "It feels cheap and flimsy",  
        "The performance is sluggish",  
        "It overheats constantly",  
        "The screen is dim",  
        "The sound quality is terrible",  
    \]

    aspects = \[  
        "The price is reasonable",  
        "It works as advertised",  
        "The design is acceptable",  
        "The features are basic",  
        "The quality is average",  
        "The performance is adequate",  
    \]

    reviews = \[\]  
    sentiments = \["positive", "negative", "neutral"\]

    for i in range(num\_samples):  
        sentiment = random.choice(sentiments)  
        template\_text, template\_summary = random.choice(templates\[sentiment\])

        if sentiment == "positive":  
            text = template\_text.format(  
                feature1=random.choice(features), feature2=random.choice(features)  
            )  
            summary = template\_summary.format(  
                feature1=random.choice(features).lower(),  
                feature2=random.choice(features).lower(),  
            )  
        elif sentiment == "negative":  
            text = template\_text.format(  
                issue1=random.choice(issues), issue2=random.choice(issues)  
            )  
            summary = template\_summary.format(  
                issue1=random.choice(issues).lower(),  
                issue2=random.choice(issues).lower(),  
            )  
        else:  
            text = template\_text.format(  
                aspect1=random.choice(aspects), aspect2=random.choice(aspects)  
            )  
            summary = template\_summary.format(  
                aspect1=random.choice(aspects).lower(),  
                aspect2=random.choice(aspects).lower(),  
            )

        reviews.append({"text": text, "summary": summary, "sentiment": sentiment})

    print(f"✅ Generated {len(reviews)} synthetic reviews")  
    return reviews

def split\_and\_save\_data(  
    reviews: List\[Dict\], output\_dir: Path, train\_ratio=0.8, val\_ratio=0.1  
):  
    """Split data into train/val/test and save as JSONL"""

    # Shuffle  
    random.shuffle(reviews)

    # Calculate splits  
    total = len(reviews)  
    train\_size = int(total \* train\_ratio)  
    val\_size = int(total \* val\_ratio)

    train\_data = reviews\[:train\_size\]  
    val\_data = reviews\[train\_size : train\_size + val\_size\]  
    test\_data = reviews\[train\_size + val\_size :\]

    # Save files  
    output\_dir.mkdir(parents=True, exist\_ok=True)

    def save\_jsonl(data, filename):  
        filepath = output\_dir / filename  
        with open(filepath, "w") as f:  
            for item in data:  
                f.write(json.dumps(item) + "\\n")  
        print(f"  ✓ {filename}: {len(data)} samples")

    print(f"\\n💾 Saving data to {output\_dir}/")  
    save\_jsonl(train\_data, "train.jsonl")  
    save\_jsonl(val\_data, "validation.jsonl")  
    save\_jsonl(test\_data, "test.jsonl")

    print(f"\\n📊 Data split:")  
    print(f"  Training:   {len(train\_data)} samples ({train\_ratio\*100:.0f}%)")  
    print(f"  Validation: {len(val\_data)} samples ({val\_ratio\*100:.0f}%)")  
    print(  
        f"  Test:       {len(test\_data)} samples ({(1-train\_ratio-val\_ratio)\*100:.0f}%)"  
    )

def main():  
    parser = argparse.ArgumentParser(  
        description="Download and prepare training data for review summarization",  
        formatter\_class=argparse.RawDescriptionHelpFormatter,  
        epilog="""  
Examples:  
  # Download from Hugging Face (recommended)  
  python scripts/download\_training\_data.py --source huggingface --dataset amazon\_polarity --num-samples 5000

  # Download different Hugging Face dataset  
  python scripts/download\_training\_data.py --source huggingface --dataset yelp\_review\_full --num-samples 3000

  # Download real Amazon reviews (Electronics)  
  python scripts/download\_training\_data.py --source amazon --max-samples 5000

  # Generate synthetic data for testing  
  python scripts/download\_training\_data.py --source synthetic --num-samples 1000

  # Custom output directory  
  python scripts/download\_training\_data.py --source huggingface --dataset imdb --output-dir my\_data/  
        """,  
    )

    parser.add\_argument(  
        "--source",  
        type=str,  
        default="huggingface",  
        choices=\["huggingface", "amazon", "synthetic"\],  
        help="Data source (default: huggingface)",  
    )  
    parser.add\_argument(  
        "--dataset",  
        type=str,  
        default="amazon\_polarity",  
        help="Hugging Face dataset name (default: amazon\_polarity)",  
    )  
    parser.add\_argument(  
        "--output-dir",  
        type=str,  
        default="training\_data",  
        help="Output directory (default: training\_data)",  
    )  
    parser.add\_argument(  
        "--max-samples",  
        type=int,  
        default=5000,  
        help="Max samples to download from Amazon (default: 5000)",  
    )  
    parser.add\_argument(  
        "--num-samples",  
        type=int,  
        default=5000,  
        help="Number of samples to generate/download (default: 5000)",  
    )  
    parser.add\_argument(  
        "--category",  
        type=str,  
        default="Electronics",  
        help="Amazon reviews category (default: Electronics)",  
    )

    args = parser.parse\_args()

    output\_dir = Path(args.output\_dir)

    print("=" \* 60)  
    print("📚 Training Data Preparation")  
    print("=" \* 60)

    # Get data based on source  
    if args.source == "huggingface":  
        reviews = download\_huggingface\_dataset(  
            output\_dir=output\_dir,  
            dataset\_name=args.dataset,  
            max\_samples=args.num\_samples,  
        )  
        if not reviews:  
            print("\\n❌ Failed to download from Hugging Face.")  
            print("Please check your internet connection or try a different dataset.")  
            return  
    elif args.source == "amazon":  
        reviews = download\_amazon\_reviews(  
            output\_dir=output\_dir, category=args.category, max\_samples=args.max\_samples  
        )  
        if not reviews:  
            print("\\n❌ Failed to download Amazon reviews.")  
            print("Please check your internet connection or AWS CLI configuration.")  
            return  
    else:  
        reviews = create\_synthetic\_data(args.num\_samples)

    # Split and save  
    if reviews:  
        split\_and\_save\_data(reviews, output\_dir)

        print("\\n" + "=" \* 60)  
        print("✅ Data preparation complete!")  
        print("=" \* 60)  
        print(f"\\nNext steps:")  
        print(f"1. Review the data in {output\_dir}/")  
        print(f"2. Upload to S3:")  
        print(f"   python scripts/upload\_training\_data.py")  
        print(f"3. Start training:")  
        print(f"   python scripts/start\_training.py")  
    else:  
        print("\\n❌ No data was generated")

if \_\_name\_\_ == "\_\_main\_\_":  
    main()

The dataset will be downloaded from the source indicated when running the script, if not it will get from hugging face.

## Get KateVu’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Since we use instruction fine-tuning, the dataset will be format as:

                {  
                    "text": text,  
                    "summary": summary,  
                    "sentiment": sentiment,  
                    "source": dataset\_name,  
                }

before splitting into training, test and validation datasets

-   **Upload to S3 bucket**

Let’s create an script to help us upload the datasets into S3 bucket

#!/usr/bin/env python3  
"""  
Upload training data to S3 training bucket

This script uploads your prepared training data to the S3 bucket created by  
the training pipeline stack. It automatically finds the correct bucket name  
from CloudFormation outputs.

Prerequisites:  
    1. Deploy training pipeline: cdk deploy TrainingPipeline  
    2. Prepare data: python scripts/download\_training\_data.py

Usage:  
    # Upload data for kate environment  
    python scripts/upload\_training\_data.py

        # Upload for different environment  
    python scripts/upload\_training\_data.py --environment dev  
"""

import boto3  
import argparse  
from pathlib import Path  
import os

def get\_training\_bucket(environment='kate'):  
    """Get training bucket name from CloudFormation stack"""  
    cfn = boto3.client('cloudformation')  
    stack\_name = f'training-pipeline-{environment}'

        try:  
        response = cfn.describe\_stacks(StackName=stack\_name)  
        outputs = response\['Stacks'\]\[0\]\['Outputs'\]

                for output in outputs:  
            if output\['OutputKey'\] == 'TrainingBucketName':  
                return output\['OutputValue'\]

                print(f"❌ Error: Could not find TrainingBucketName in stack outputs")  
        return None

            except Exception as e:  
        print(f"❌ Error: Could not find stack '{stack\_name}'")  
        print(f"Make sure you've deployed the training pipeline first:")  
        print(f"  cdk deploy TrainingPipeline")  
        return None

def upload\_directory(local\_dir: Path, bucket\_name: str, s3\_prefix: str = ''):  
    """Upload directory contents to S3"""  
    s3 = boto3.client('s3')

        if not local\_dir.exists():  
        print(f"❌ Error: Directory not found: {local\_dir}")  
        print(f"\\nRun this first to download training data:")  
        print(f"  python scripts/download\_training\_data.py")  
        return False

        # Get list of files  
    files = list(local\_dir.glob('\*.jsonl'))

        if not files:  
        print(f"❌ Error: No .jsonl files found in {local\_dir}")  
        print(f"\\nExpected files:")  
        print(f"  - train.jsonl")  
        print(f"  - validation.jsonl")  
        print(f"  - test.jsonl")  
        return False

        print(f"\\n📤 Uploading {len(files)} files to s3://{bucket\_name}/{s3\_prefix}")  
    print("=" \* 60)

        uploaded = 0  
    for file\_path in files:  
        s3\_key = f"{s3\_prefix}{file\_path.name}" if s3\_prefix else file\_path.name

                try:  
            # Get file size  
            file\_size = file\_path.stat().st\_size  
            file\_size\_mb = file\_size / (1024 \* 1024)

                        print(f"  Uploading {file\_path.name} ({file\_size\_mb:.2f} MB)...", end='', flush=True)

                        # Upload with progress  
            s3.upload\_file(  
                str(file\_path),  
                bucket\_name,  
                s3\_key,  
                Callback=lambda bytes\_transferred: None  
            )

                        print(" ✓")  
            uploaded += 1

                    except Exception as e:  
            print(f" ✗")  
            print(f"    Error: {str(e)}")

        print("=" \* 60)  
    print(f"✅ Uploaded {uploaded}/{len(files)} files successfully")

        return uploaded == len(files)

def verify\_upload(bucket\_name: str, s3\_prefix: str = ''):  
    """Verify files were uploaded correctly"""  
    s3 = boto3.client('s3')

        print(f"\\n🔍 Verifying upload...")

        try:  
        response = s3.list\_objects\_v2(  
            Bucket=bucket\_name,  
            Prefix=s3\_prefix  
        )

                if 'Contents' not in response:  
            print("❌ No files found in bucket")  
            return False

                print(f"\\n📁 Files in s3://{bucket\_name}/{s3\_prefix}")  
        print("=" \* 60)

                total\_size = 0  
        for obj in response\['Contents'\]:  
            key = obj\['Key'\]  
            size = obj\['Size'\]  
            size\_mb = size / (1024 \* 1024)  
            total\_size += size  
            print(f"  ✓ {key} ({size\_mb:.2f} MB)")

                total\_size\_mb = total\_size / (1024 \* 1024)  
        print("=" \* 60)  
        print(f"Total: {len(response\['Contents'\])} files, {total\_size\_mb:.2f} MB")

                return True

            except Exception as e:  
        print(f"❌ Error verifying upload: {str(e)}")  
        return False

def main():  
    parser = argparse.ArgumentParser(  
        description='Upload training data to S3',  
        formatter\_class=argparse.RawDescriptionHelpFormatter,  
        epilog="""  
Examples:  
  # Upload data for kate environment  
  python scripts/upload\_training\_data.py

  # Upload for different environment  
  python scripts/upload\_training\_data.py --environment dev

  # Upload from custom directory  
  python scripts/upload\_training\_data.py --data-dir my\_data/

  # Upload to specific S3 prefix  
  python scripts/upload\_training\_data.py --s3-prefix data/v1/  
        """  
    )

        parser.add\_argument('--environment', type=str, default='kate',  
                        help='Environment name (default: kate)')  
    parser.add\_argument('--data-dir', type=str, default='training\_data',  
                        help='Local data directory (default: training\_data)')  
    parser.add\_argument('--s3-prefix', type=str, default='',  
                        help='S3 prefix/folder (default: root)')

        args = parser.parse\_args()

        print("=" \* 60)  
    print("📤 Upload Training Data to S3")  
    print("=" \* 60)

        # Get training bucket  
    print(f"\\n🔍 Looking up training bucket for environment: {args.environment}")  
    bucket\_name = get\_training\_bucket(args.environment)

        if not bucket\_name:  
        return

        print(f"✓ Found bucket: {bucket\_name}")

        # Upload files  
    local\_dir = Path(args.data\_dir)  
    success = upload\_directory(local\_dir, bucket\_name, args.s3\_prefix)

        if not success:  
        return

        # Verify upload  
    verify\_upload(bucket\_name, args.s3\_prefix)

        print("\\n" + "=" \* 60)  
    print("✅ Upload complete!")  
    print("=" \* 60)  
    print(f"\\nNext steps:")  
    print(f"1. Start training job:")  
    print(f"   python scripts/start\_training.py --environment {args.environment}")  
    print(f"\\n2. Monitor training:")  
    print(f"   - AWS Console: https://console.aws.amazon.com/sagemaker/home#/jobs")  
    print(f"   - CLI: aws sagemaker list-training-jobs --sort-by CreationTime --sort-order Descending")

if \_\_name\_\_ == '\_\_main\_\_':  
    main()

-   **Training script**

#!/usr/bin/env python3  
"""  
Start a SageMaker training job for fine-tuning review summarization model

This script starts a SageMaker training job that fine-tunes a T5 or DistilBERT  
model on your review data. It automatically configures the job using resources  
from your deployed training pipeline stack.

Prerequisites:  
    1. Deploy training pipeline: cdk deploy TrainingPipeline  
    2. Prepare data: python scripts/download\_training\_data.py  
    3. Upload data: python scripts/upload\_training\_data.py

Usage:  
    # Start training with defaults (t5-small, 3 epochs, ml.g4dn.xlarge GPU)  
    python scripts/start\_training.py

        # Custom hyperparameters  
    python scripts/start\_training.py --epochs 5 --batch-size 16 --learning-rate 3e-5

        # Use GPU for faster training  
    python scripts/start\_training.py --instance-type ml.p3.2xlarge  
"""

import boto3  
import argparse  
from datetime import datetime  
import os

\# Get region from environment or use default  
REGION = os.environ.get('AWS\_REGION') or os.environ.get('AWS\_DEFAULT\_REGION') or 'ap-southeast-2'

sagemaker\_client = boto3.client('sagemaker', region\_name=REGION)  
cfn = boto3.client('cloudformation', region\_name=REGION)  
s3 = boto3.client('s3', region\_name=REGION)  
sts = boto3.client('sts', region\_name=REGION)

def get\_stack\_outputs(stack\_name):  
    """Get outputs from CloudFormation stack"""  
    response = cfn.describe\_stacks(StackName=stack\_name)  
    outputs = {}  
    for output in response\['Stacks'\]\[0\]\['Outputs'\]:  
        outputs\[output\['OutputKey'\]\] = output\['OutputValue'\]  
    return outputs

def upload\_training\_code(model\_bucket):  
    """Upload training script to S3"""  
    import tarfile  
    import tempfile  
    import os

        # Create a temporary tar.gz file with the training code  
    with tempfile.NamedTemporaryFile(suffix='.tar.gz', delete=False) as tmp:  
        tmp\_path = tmp.name

        try:  
        with tarfile.open(tmp\_path, 'w:gz') as tar:  
            tar.add('sagemaker/train.py', arcname='train.py')  
            tar.add('sagemaker/requirements.txt', arcname='requirements.txt')

                # Upload to S3  
        timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')  
        s3\_key = f'code/sourcedir-{timestamp}.tar.gz'  
        s3.upload\_file(tmp\_path, model\_bucket, s3\_key)

                return f's3://{model\_bucket}/{s3\_key}'  
    finally:  
        if os.path.exists(tmp\_path):  
            os.remove(tmp\_path)

def get\_training\_image():  
    """Get the PyTorch training container image for the current region"""  
    region = boto3.session.Session().region\_name

        # PyTorch 2.0 training image  
    pytorch\_version = '2.0.1'  
    python\_version = 'py310'

        # ECR image URI format  
    image\_uri = f'763104351884.dkr.ecr.{region}.amazonaws.com/pytorch-training:{pytorch\_version}-gpu-{python\_version}-cu118-ubuntu20.04-sagemaker'

        return image\_uri

def start\_training\_job(  
    environment='kate',  
    model\_name='t5-small',  
    epochs=3,  
    batch\_size=8,  
    learning\_rate=2e-5,  
    instance\_type='ml.m5.xlarge',  
    use\_lora=True,  
    lora\_r=8,  
    lora\_alpha=32,  
    lora\_dropout=0.1  
):  
    """Start a SageMaker training job"""

        # Get stack outputs  
    stack\_name = f'training-pipeline-{environment}'  
    print(f"Getting outputs from stack: {stack\_name}")

        try:  
        outputs = get\_stack\_outputs(stack\_name)  
    except Exception:  
        print(f"Error: Could not find stack '{stack\_name}'")  
        print("Make sure you've deployed the training pipeline first:")  
        print("  cdk deploy TrainingPipeline")  
        return

        training\_bucket = outputs\['TrainingBucketName'\]  
    model\_bucket = outputs\['ModelBucketName'\]  
    sagemaker\_role = outputs\['SageMakerRoleArn'\]

        print(f"Training bucket: {training\_bucket}")  
    print(f"Model bucket: {model\_bucket}")  
    print(f"SageMaker role: {sagemaker\_role}")

        # Upload training code to S3  
    print("\\nUploading training code to S3...")  
    source\_code\_uri = upload\_training\_code(model\_bucket)  
    print(f"Training code uploaded to: {source\_code\_uri}")

        # Generate job name with timestamp  
    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')  
    job\_name = f'review-summarizer-{environment}-{timestamp}'

        # Training job configuration  
    training\_config = {  
        'TrainingJobName': job\_name,  
        'RoleArn': sagemaker\_role,  
        'AlgorithmSpecification': {  
            'TrainingImage': get\_training\_image(),  
            'TrainingInputMode': 'File',  
        },  
        'InputDataConfig': \[  
            {  
                'ChannelName': 'training',  
                'DataSource': {  
                    'S3DataSource': {  
                        'S3DataType': 'S3Prefix',  
                        'S3Uri': f's3://{training\_bucket}/',  
                        'S3DataDistributionType': 'FullyReplicated',  
                    }  
                },  
                'ContentType': 'application/json',  
                'CompressionType': 'None',  
            }  
        \],  
        'OutputDataConfig': {  
            'S3OutputPath': f's3://{model\_bucket}/models/',  
        },  
        'ResourceConfig': {  
            'InstanceType': instance\_type,  
            'InstanceCount': 1,  
            'VolumeSizeInGB': 30,  
        },  
        'StoppingCondition': {  
            'MaxRuntimeInSeconds': 86400,  # 24 hours  
        },  
        'HyperParameters': {  
            'sagemaker\_program': 'train.py',  
            'sagemaker\_submit\_directory': source\_code\_uri,  
            'epochs': str(epochs),  
            'batch\_size': str(batch\_size),  
            'learning\_rate': str(learning\_rate),  
            'model\_name': model\_name,  
            'use\_lora': str(use\_lora).lower(),  
            'lora\_r': str(lora\_r),  
            'lora\_alpha': str(lora\_alpha),  
            'lora\_dropout': str(lora\_dropout),  
        },  
        'Tags': \[  
            {'Key': 'Environment', 'Value': environment},  
            {'Key': 'Project', 'Value': 'review-summarizer'},  
        \],  
    }

        print(f"\\nStarting training job: {job\_name}")  
    print(f"Model: {model\_name}")  
    print(f"Instance: {instance\_type}")  
    print(f"Training method: {'LoRA (Parameter-Efficient)' if use\_lora else 'Full Fine-Tuning'}")  
    print(f"Hyperparameters:")  
    print(f"  - Epochs: {epochs}")  
    print(f"  - Batch size: {batch\_size}")  
    print(f"  - Learning rate: {learning\_rate}")  
    if use\_lora:  
        print(f"  - LoRA rank: {lora\_r}")  
        print(f"  - LoRA alpha: {lora\_alpha}")  
        print(f"  - LoRA dropout: {lora\_dropout}")

        try:  
        response = sagemaker\_client.create\_training\_job(\*\*training\_config)  
        print(f"\\n✅ Training job started successfully!")  
        print(f"Job ARN: {response\['TrainingJobArn'\]}")  
        print(f"\\nMonitor progress:")  
        region = boto3.session.Session().region\_name  
        print(f"  - AWS Console: https://{region}.console.aws.amazon.com/sagemaker/home?region={region}#/jobs/{job\_name}")  
        print(f"  - CLI: aws sagemaker describe-training-job --training-job-name {job\_name}")  
        print(f"\\nView logs:")  
        print(f"  aws logs tail /aws/sagemaker/TrainingJobs --follow --log-stream-name-prefix {job\_name}")

            except Exception as e:  
        print(f"\\n❌ Error starting training job: {str(e)}")  
        print("\\nTroubleshooting:")  
        print(f"1. Make sure training data exists in s3://{training\_bucket}/")  
        print("2. Check IAM role has necessary permissions")  
        print("3. Verify the training image is available in your region")  
        print("4. Check training script exists: sagemaker/train.py")

if \_\_name\_\_ == '\_\_main\_\_':  
    parser = argparse.ArgumentParser(  
        description='Start SageMaker training job for review summarization',  
        formatter\_class=argparse.RawDescriptionHelpFormatter,  
        epilog="""  
Examples:  
  # Start training with defaults  
  python scripts/start\_training.py

  # Custom hyperparameters  
  python scripts/start\_training.py --epochs 5 --batch-size 16

  # Use larger instance  
  python scripts/start\_training.py --instance-type ml.p3.2xlarge

  # Different environment  
  python scripts/start\_training.py --environment dev  
        """  
    )

        parser.add\_argument('--environment', type=str, default='kate',  
                        help='Environment name (default: kate)')  
    parser.add\_argument('--model-name', type=str, default='t5-small',  
                        help='Base model to fine-tune (default: t5-small)')  
    parser.add\_argument('--epochs', type=int, default=3,  
                        help='Number of training epochs (default: 3)')  
    parser.add\_argument('--batch-size', type=int, default=8,  
                        help='Training batch size (default: 8)')  
    parser.add\_argument('--learning-rate', type=float, default=2e-5,  
                        help='Learning rate (default: 2e-5)')  
    parser.add\_argument('--instance-type', type=str, default='ml.g4dn.xlarge',  
                        help='SageMaker instance type (default: ml.g4dn.xlarge)')  
    parser.add\_argument('--use-lora', action='store\_true', default=True,  
                        help='Enable LoRA fine-tuning (default: True)')  
    parser.add\_argument('--no-lora', dest='use\_lora', action='store\_false',  
                        help='Disable LoRA and use full fine-tuning')  
    parser.add\_argument('--lora-r', type=int, default=8,  
                        help='LoRA rank (default: 8)')  
    parser.add\_argument('--lora-alpha', type=int, default=32,  
                        help='LoRA alpha scaling (default: 32)')  
    parser.add\_argument('--lora-dropout', type=float, default=0.1,  
                        help='LoRA dropout (default: 0.1)')

        args = parser.parse\_args()

        start\_training\_job(  
        environment=args.environment,  
        model\_name=args.model\_name,  
        epochs=args.epochs,  
        batch\_size=args.batch\_size,  
        learning\_rate=args.learning\_rate,  
        instance\_type=args.instance\_type,  
        use\_lora=args.use\_lora,  
        lora\_r=args.lora\_r,  
        lora\_alpha=args.lora\_alpha,  
        lora\_dropout=args.lora\_dropout,  
    )

### 3\. Inference Pipeline

**3.1 Create the pipeline**

This pipeline will create these AWS resources below:

-   S3 result bucket
-   API Gateway
-   Lambda function
-   IAM Roles
-   Cloudwatch logs

/\*\*  
 \* Inference Pipeline Stack  
 \*   
 \* This stack creates the infrastructure for online review summarization.  
 \* It implements a multi-stage processing pipeline:  
 \*   
 \* 1. API Gateway - REST API endpoint for incoming requests  
 \* 2. Lambda Orchestrator - Coordinates the summarization pipeline  
 \* 3. Amazon Bedrock - Generates fast, general-purpose summaries  
 \* 4. Amazon OpenSearch - Retrieves relevant context via RAG (optional)  
 \* 5. SageMaker Endpoint - Refines summary with fine-tuned model (optional)  
 \* 6. S3 Results Bucket - Stores final summaries and metadata  
 \*   
 \* Request Flow:  
 \* POST /summarize → Lambda → Bedrock → OpenSearch → SageMaker → S3 → Response  
 \*   
 \* The Lambda function reads the active SageMaker endpoint from Parameter Store,  
 \* enabling zero-downtime model updates when new versions are deployed.  
 \*/

import \* as cdk from 'aws-cdk-lib';  
import { Construct } from 'constructs';  
import \* as s3 from 'aws-cdk-lib/aws-s3';  
import \* as lambda from 'aws-cdk-lib/aws-lambda';  
import { PythonFunction } from '@aws-cdk/aws-lambda-python-alpha';  
import \* as iam from 'aws-cdk-lib/aws-iam';  
import \* as apigateway from 'aws-cdk-lib/aws-apigateway';  
import \* as logs from 'aws-cdk-lib/aws-logs';  
import \* as ssm from 'aws-cdk-lib/aws-ssm';  
import { EnvironmentConfig } from './utils';

export interface InferencePipelineStackProps extends cdk.StackProps {  
  config: EnvironmentConfig;  
  endpointParameterName: string;  
  appConfigApplicationId?: string;  
  appConfigEnvironmentId?: string;  
  appConfigProfileId?: string;  
}

export class InferencePipelineStack extends cdk.Stack {  
  public readonly api: apigateway.RestApi;  
  public readonly resultsBucket: s3.Bucket;  
  public readonly summarizerFunction: lambda.Function;

  constructor(scope: Construct, id: string, props: InferencePipelineStackProps) {  
    super(scope, id, props);

    const { config, endpointParameterName, appConfigApplicationId, appConfigEnvironmentId, appConfigProfileId } = props;

    // ========================================  
    // S3 Bucket for Results  
    // ========================================

    // NOTE: Using DESTROY for cost-saving during development  
    // Results are temporary and can be safely deleted  
    this.resultsBucket = new s3.Bucket(this, 'ResultsBucket', {  
      bucketName: \`summarizer-results-${config.environmentName}-${cdk.Aws.ACCOUNT\_ID}\`,  
      removalPolicy: cdk.RemovalPolicy.DESTROY,  
      autoDeleteObjects: true,  
      encryption: s3.BucketEncryption.S3\_MANAGED,  
      lifecycleRules: \[  
        {  
          id: 'DeleteOldResults',  
          expiration: cdk.Duration.days(30),  
        },  
      \],  
    });

    // ========================================  
    // Lambda: Main Summarizer Function  
    // ========================================

    const summarizerLogGroup = new logs.LogGroup(this, 'SummarizerLogGroup', {  
      logGroupName: \`/aws/lambda/summarizer-${config.environmentName}\`,  
      retention: logs.RetentionDays.ONE\_WEEK,  
      removalPolicy: cdk.RemovalPolicy.DESTROY,  
    });

    const summarizerFn = new PythonFunction(this, 'SummarizerFunction', {  
      functionName: \`summarizer-${config.environmentName}\`,  
      entry: 'src/lambdas/summarizer',  
      runtime: lambda.Runtime.PYTHON\_3\_11,  
      index: 'handler.py',  
      handler: 'handler',  
      description: \`Review summarization function for ${config.environmentName}\`,  
      timeout: cdk.Duration.seconds(120),  
      memorySize: 1024,  
      environment: {  
        RESULTS\_BUCKET: this.resultsBucket.bucketName,  
        ENDPOINT\_PARAMETER: endpointParameterName,  
        ENVIRONMENT: config.environmentName,  
        OPENSEARCH\_ENDPOINT: process.env.OPENSEARCH\_ENDPOINT || 'none',  
        // AppConfig IDs (if provided)  
        ...(appConfigApplicationId && { APPCONFIG\_APPLICATION\_ID: appConfigApplicationId }),  
        ...(appConfigEnvironmentId && { APPCONFIG\_ENVIRONMENT\_ID: appConfigEnvironmentId }),  
        ...(appConfigProfileId && { APPCONFIG\_CONFIGURATION\_PROFILE\_ID: appConfigProfileId }),  
      },  
      logGroup: summarizerLogGroup,  
    });

    // Expose Lambda function for AppConfig permissions  
    this.summarizerFunction = summarizerFn;

    // Grant permissions  
    this.resultsBucket.grantWrite(summarizerFn);

    summarizerFn.addToRolePolicy(new iam.PolicyStatement({  
      effect: iam.Effect.ALLOW,  
      actions: \['bedrock:InvokeModel'\],  
      resources: \['\*'\],  
    }));

    summarizerFn.addToRolePolicy(new iam.PolicyStatement({  
      effect: iam.Effect.ALLOW,  
      actions: \['sagemaker:InvokeEndpoint'\],  
      resources: \['\*'\],  
    }));

    summarizerFn.addToRolePolicy(new iam.PolicyStatement({  
      effect: iam.Effect.ALLOW,  
      actions: \['ssm:GetParameter'\],  
      resources: \[  
        \`arn:aws:ssm:${cdk.Aws.REGION}:${cdk.Aws.ACCOUNT\_ID}:parameter${endpointParameterName}\`,  
      \],  
    }));

    // OpenSearch permissions (if using)  
    summarizerFn.addToRolePolicy(new iam.PolicyStatement({  
      effect: iam.Effect.ALLOW,  
      actions: \[  
        'aoss:APIAccessAll',  
        'es:ESHttpGet',  
        'es:ESHttpPost',  
      \],  
      resources: \['\*'\],  
    }));

    // ========================================  
    // API Gateway  
    // ========================================

    if (config.enableApiGateway) {  
      this.api = new apigateway.RestApi(this, 'SummarizerAPI', {  
        restApiName: \`review-summarizer-${config.environmentName}\`,  
        description: 'API for review summarization with RAG',  
        deployOptions: {  
          stageName: config.environmentName,  
          loggingLevel: apigateway.MethodLoggingLevel.INFO,  
          dataTraceEnabled: true,  
          metricsEnabled: true,  
        },  
        defaultCorsPreflightOptions: {  
          allowOrigins: apigateway.Cors.ALL\_ORIGINS,  
          allowMethods: apigateway.Cors.ALL\_METHODS,  
        },  
      });

      // POST /summarize endpoint  
      const summarize = this.api.root.addResource('summarize');  
      summarize.addMethod('POST', new apigateway.LambdaIntegration(summarizerFn), {  
        apiKeyRequired: false,  
        requestValidator: new apigateway.RequestValidator(this, 'RequestValidator', {  
          restApi: this.api,  
          validateRequestBody: true,  
        }),  
      });

      // GET /health endpoint  
      const health = this.api.root.addResource('health');  
      health.addMethod('GET', new apigateway.MockIntegration({  
        integrationResponses: \[{  
          statusCode: '200',  
          responseTemplates: {  
            'application/json': '{"status": "healthy"}',  
          },  
        }\],  
        requestTemplates: {  
          'application/json': '{"statusCode": 200}',  
        },  
      }), {  
        methodResponses: \[{ statusCode: '200' }\],  
      });

      new cdk.CfnOutput(this, 'ApiUrl', {  
        value: this.api.url,  
        description: 'API Gateway URL',  
        exportName: \`${config.environmentName}-api-url\`,  
      });  
    }

    // ========================================  
    // Outputs  
    // ========================================

    new cdk.CfnOutput(this, 'ResultsBucketName', {  
      value: this.resultsBucket.bucketName,  
      description: 'S3 bucket for summarization results',  
      exportName: \`${config.environmentName}-results-bucket\`,  
    });

    new cdk.CfnOutput(this, 'LambdaFunctionName', {  
      value: summarizerFn.functionName,  
      description: 'Lambda function for summarization',  
      exportName: \`${config.environmentName}-summarizer-function\`,  
    });  
  }  
}

**3.2 Create the scripts**

-   Lambda function

"""  
Main Lambda function for review summarization pipeline

This function orchestrates a multi-stage summarization process with A/B testing support:

Stage 1: Amazon Bedrock  
    - Generates fast, general-purpose summary  
    - Uses Claude or other foundation models  
    - Always runs (provides baseline summary)

Stage 2: RAG Retrieval (Optional)  
    - Queries OpenSearch vector index for relevant context  
    - Grounds summary in factual knowledge  
    - Only runs if OpenSearch is configured

Stage 3: SageMaker Refinement (Optional with A/B Testing)  
    - Selects model based on A/B testing rules  
    - Calls fine-tuned model for domain-specific refinement  
    - Extracts sentiment and confidence scores  
    - Supports gradual rollouts and canary deployments

Stage 4: Storage  
    - Saves results to S3 for audit trail  
    - Returns JSON response to API Gateway

The function uses AWS AppConfig for dynamic A/B testing configuration,  
enabling gradual model rollouts without code changes.  
"""

import json  
import os  
import boto3  
from datetime import datetime  
import traceback  
from appconfig\_helper import (  
    select\_model\_for\_request,  
    get\_bedrock\_config,  
    log\_ab\_test\_assignment  
)  
from bedrock\_client import summarize\_review

\# Initialize AWS clients  
bedrock\_runtime = boto3.client('bedrock-runtime', region\_name=os.environ.get('AWS\_REGION', 'us-east-1'))  
sagemaker\_runtime = boto3.client('sagemaker-runtime')  
s3\_client = boto3.client('s3')  
ssm\_client = boto3.client('ssm')

RESULTS\_BUCKET = os.environ\['RESULTS\_BUCKET'\]  
ENDPOINT\_PARAMETER = os.environ\['ENDPOINT\_PARAMETER'\]  
BEDROCK\_MODEL\_ID = os.environ.get('BEDROCK\_MODEL\_ID', 'anthropic.claude-v2')  
OPENSEARCH\_ENDPOINT = os.environ.get('OPENSEARCH\_ENDPOINT', 'none')

def handler(event, context):  
    """  
    Main handler for summarization requests

        Expected input:  
    {  
        "text": "Review text here...",  
        "options": {  
            "include\_sentiment": true,  
            "use\_rag": true  
        }  
    }  
    """  
    try:  
        # Parse input  
        if 'body' in event:  
            body = json.loads(event\['body'\])  
        else:  
            body = event

                text = body.get('text', '')  
        options = body.get('options', {})

                # Extract request context for A/B testing  
        request\_context = {  
            'category': body.get('category', 'general'),  
            'userTier': body.get('userTier', 'standard'),  
            'textLength': len(text),  
            'userId': body.get('userId'),  
            'requestId': context.request\_id if hasattr(context, 'request\_id') else datetime.now().isoformat(),  
        }

                if not text:  
            return {  
                'statusCode': 400,  
                'body': json.dumps({'error': 'Missing required field: text'})  
            }

                request\_id = request\_context\['requestId'\]

                # Step 1: Get initial summary from Bedrock using Converse API  
        print(f"\[{request\_id}\] Step 1: Calling Bedrock for initial summary")  
        bedrock\_config = get\_bedrock\_config()

                bedrock\_response = summarize\_review(  
            text=text,  
            model\_id=bedrock\_config.get('modelId', BEDROCK\_MODEL\_ID),  
            max\_tokens=bedrock\_config.get('maxTokens', 200),  
            temperature=bedrock\_config.get('temperature', 0.5)  
        )

                initial\_summary = bedrock\_response\['text'\]

                # Log token usage  
        usage = bedrock\_response\['usage'\]  
        print(f"Bedrock usage - Input: {usage\['inputTokens'\]}, Output: {usage\['outputTokens'\]}")

                # Step 2: RAG retrieval (optional)  
        context\_text = ""  
        if options.get('use\_rag', False) and OPENSEARCH\_ENDPOINT != 'none':  
            print(f"\[{request\_id}\] Step 2: Retrieving context from OpenSearch")  
            context\_text = retrieve\_context(text)  
        else:  
            print(f"\[{request\_id}\] Step 2: Skipping RAG (disabled or not configured)")

                # Step 3: Select model using A/B testing  
        print(f"\[{request\_id}\] Step 3: Selecting model via A/B testing")  
        endpoint\_name = select\_model\_for\_request(request\_context)

                # Log A/B test assignment  
        log\_ab\_test\_assignment(request\_id, endpoint\_name)

                # Step 4: Refine with fine-tuned model (if endpoint exists)  
        final\_summary = initial\_summary  
        sentiment = "neutral"  
        confidence = 0.0

                if endpoint\_name and endpoint\_name != 'none' and endpoint\_name != 'ensemble':  
            print(f"\[{request\_id}\] Step 4: Refining with SageMaker endpoint: {endpoint\_name}")  
            refinement = refine\_with\_sagemaker(  
                endpoint\_name=endpoint\_name,  
                summary=initial\_summary,  
                context=context\_text,  
                original\_text=text  
            )  
            final\_summary = refinement.get('summary', initial\_summary)  
            sentiment = refinement.get('sentiment', 'neutral')  
            confidence = refinement.get('confidence', 0.0)  
        elif endpoint\_name == 'ensemble':  
            print(f"\[{request\_id}\] Step 4: Using multi-model ensemble")  
            # TODO: Implement ensemble logic  
            final\_summary = initial\_summary  
            sentiment = "neutral"  
            confidence = 0.0  
        else:  
            print(f"\[{request\_id}\] Step 4: Skipping SageMaker refinement (no endpoint configured)")

                # Step 5: Store results  
        result = {  
            'request\_id': request\_id,  
            'timestamp': datetime.now().isoformat(),  
            'initial\_summary': initial\_summary,  
            'final\_summary': final\_summary,  
            'sentiment': sentiment,  
            'confidence': confidence,  
            'used\_rag': options.get('use\_rag', False) and OPENSEARCH\_ENDPOINT != 'none',  
            'model\_endpoint': endpoint\_name,  
            'request\_context': request\_context,  
        }

                # Save to S3  
        s3\_key = f"results/{datetime.now().strftime('%Y/%m/%d')}/{request\_id}.json"  
        s3\_client.put\_object(  
            Bucket=RESULTS\_BUCKET,  
            Key=s3\_key,  
            Body=json.dumps(result, indent=2),  
            ContentType='application/json'  
        )

                print(f"\[{request\_id}\] Complete. Results saved to s3://{RESULTS\_BUCKET}/{s3\_key}")

                return {  
            'statusCode': 200,  
            'headers': {  
                'Content-Type': 'application/json',  
                'Access-Control-Allow-Origin': '\*'  
            },  
            'body': json.dumps(result)  
        }

            except Exception as e:  
        print(f"Error: {str(e)}")  
        print(traceback.format\_exc())  
        return {  
            'statusCode': 500,  
            'body': json.dumps({  
                'error': str(e),  
                'traceback': traceback.format\_exc()  
            })  
        }

def retrieve\_context(query: str, top\_k: int = 3) -> str:  
    """  
    Retrieve relevant context from OpenSearch  
    TODO: Implement OpenSearch vector search  
    """  
    # Placeholder - implement OpenSearch integration  
    return ""

def refine\_with\_sagemaker(endpoint\_name: str, summary: str, context: str, original\_text: str) -> dict:  
    """  
    Refine summary and extract sentiment using fine-tuned SageMaker model  
    """  
    try:  
        # Send original text to the model for summarization  
        payload = {  
            "inputs": original\_text  
        }

                response = sagemaker\_runtime.invoke\_endpoint(  
            EndpointName=endpoint\_name,  
            ContentType='application/json',  
            Body=json.dumps(payload)  
        )

                result = json.loads(response\['Body'\].read().decode())

                # Extract the summary from the model's response  
        refined\_summary = result.get('summary', summary)

                return {  
            'summary': refined\_summary,  
            'sentiment': 'neutral',  # TODO: Add sentiment analysis  
            'confidence': 0.0  
        }

            except Exception as e:  
        print(f"SageMaker error: {str(e)}")  
        return {  
            'summary': summary,  
            'sentiment': 'neutral',  
            'confidence': 0.0  
        }

-   Script to test the endpoint

#!/bin/bash  
\# Test script for the news summarization API

set -e

\# Get API URL from CloudFormation stack  
STACK\_NAME="${1:-inference-pipeline-kate}"

echo "Getting API URL from stack: $STACK\_NAME"  
API\_URL=$(aws cloudformation describe-stacks \\  
  --stack-name "$STACK\_NAME" \\  
  --query 'Stacks\[0\].Outputs\[?OutputKey==\`ApiUrl\`\].OutputValue' \\  
  --output text)

if \[ -z "$API\_URL" \]; then  
  echo "Error: Could not find API URL in stack outputs"  
  exit 1  
fi

echo "API URL: $API\_URL"  
echo ""

\# Test 1: Health check  
echo "Test 1: Health Check"  
echo "===================="  
curl -s "${API\_URL}health" | jq .  
echo -e "\\n"

\# Test 2: Technology news article  
echo "Test 2: Technology News Article"  
echo "================================"  
curl -s -X POST "${API\_URL}summarize" \\  
  -H "Content-Type: application/json" \\  
  -d '{  
    "text": "Apple Inc. announced today the launch of its latest iPhone model, featuring significant improvements in camera technology and battery life. The new device includes a 48-megapixel main camera, up from the previous 12-megapixel sensor, and promises up to 20 hours of video playback. The company also introduced new AI-powered features for photo editing and enhanced security measures. Pre-orders begin next Friday, with the device hitting stores two weeks later. Industry analysts predict strong sales, particularly in the premium smartphone segment. The starting price is set at $999 for the base model.",  
    "options": {  
      "use\_rag": false  
    }  
  }' | jq .  
echo -e "\\n"

\# Test 3: Political news article  
echo "Test 3: Political News Article"  
echo "==============================="  
curl -s -X POST "${API\_URL}summarize" \\  
  -H "Content-Type: application/json" \\  
  -d '{  
    "text": "The Senate voted 65-35 today to pass a comprehensive infrastructure bill worth $1.2 trillion. The bipartisan legislation includes funding for roads, bridges, public transit, and broadband internet expansion. Supporters argue the bill will create millions of jobs and modernize aging infrastructure. Critics express concerns about the cost and potential impact on the federal deficit. The bill now moves to the House of Representatives for consideration. President Biden praised the Senate vote, calling it a historic investment in America future. The legislation has been in negotiation for months.",  
    "options": {  
      "use\_rag": false  
    }  
  }' | jq .  
echo -e "\\n"

\# Test 4: Business news article  
echo "Test 4: Business News Article"  
echo "=============================="  
curl -s -X POST "${API\_URL}summarize" \\  
  -H "Content-Type: application/json" \\  
  -d '{  
    "text": "Tesla reported record quarterly earnings today, beating Wall Street expectations. The electric vehicle maker delivered 250,000 vehicles in the quarter, a 40% increase from the same period last year. Revenue reached $13.8 billion, up from $10.4 billion a year ago. CEO Elon Musk attributed the strong performance to increased production capacity and growing demand for electric vehicles globally. The company also announced plans to build two new manufacturing facilities in Europe and Asia. Tesla stock rose 8% in after-hours trading following the earnings announcement.",  
    "options": {  
      "use\_rag": false  
    }  
  }' | jq .  
echo -e "\\n"

\# Test 5: Sports news article  
echo "Test 5: Sports News Article"  
echo "============================"  
curl -s -X POST "${API\_URL}summarize" \\  
  -H "Content-Type: application/json" \\  
  -d '{  
    "text": "In a thrilling championship game, the Lakers defeated the Celtics 108-105 to win their 18th NBA title. LeBron James led the team with 32 points, 11 rebounds, and 8 assists in what many are calling one of the greatest performances in Finals history. The victory came after the Lakers trailed by 15 points in the third quarter. Anthony Davis contributed 28 points and played crucial defense in the final minutes. This marks the Lakers first championship in over a decade. Head coach Frank Vogel praised the team resilience and determination throughout the playoffs.",  
    "options": {  
      "use\_rag": false  
    }  
  }' | jq .  
echo -e "\\n"

echo "All tests completed!"

## Deploy the app

1.  **Deploy the resource on AWS**

cdk deploy --all

It will deploy three stacks: AppConfigStack, TrainingPipelineStack and InferencePipelineStack

**2\. Download the training data**

python3 scripts/download\_training\_data.py \\  
  --source huggingface \\  
  --dataset cnn\_dailymail \\  
  --num-samples 5000

**3\. Upload the datasets to S3 bucket**

python3 scripts/upload\_training\_data.py 

**4\. Start training job**

python3 scripts/start\_training.py \\          
  --model-name t5-base \\  
  --epochs 5 \\  
  --batch-size 4 \\  
  --instance-type ml.g4dn.xlarge

The script using LoRA for fine-tuning default, if you do want to full fine-tuning explicitly put it in the command

python3 scripts/start\_training.py --no-lora

**5\. Get the metrics**

\# Download metrics from S3  
\# Get job name from previous step  
JOB\_NAME="review-summarizer-kate-xxxxxx-xxxxxx"

MODEL\_BUCKET=$(aws cloudformation describe-stacks \\  
 --stack-name training-pipeline-kate \\  
 --query 'Stacks\[0\].Outputs\[?OutputKey==\`ModelBucketName\`\].OutputValue' \\  
 --output text)

aws s3 cp s3://$MODEL\_BUCKET/models/$JOB\_NAME/output/output.tar.gz .  
tar -xzf output.tar.gz

\# View metrics  
cat metrics.json

The result will look like

{  
  "validation\_rouge\_l": 0.2694268479883026,  
  "test\_rouge\_l": 0.27888068965217705,  
  "final\_train\_loss": 0.8831174189448356,  
  "use\_lora": true,  
  "trainable\_params": 884736,  
  "model\_name": "t5-base",  
  "epochs": 5,  
  "batch\_size": 4,  
  "learning\_rate": 3e-05,  
  "lora\_config": {  
    "r": 8,  
    "alpha": 32,  
    "dropout": 0.1  
  }   
}

Depending on the result, you can choose to adjust the parameter of the training script to get better results. For example, changing pre-training models or getting more training data.

**6\. Register Model**

If you are happy with the result, register the model and wait for approval

\# Create model package  
aws sagemaker create-model-package \\  
 --model-package-group-name "review-summarizer" \\  
 --model-package-description "Fine-tuned T5 for review summarization" \\  
 --inference-specification '{  
   "Containers": \[{  
     "Image": "763104351884.dkr.ecr.us-west-2.amazonaws.com/pytorch-inference:2.0.1-gpu-py310",  
     "ModelDataUrl": "s3://'"$MODEL\_BUCKET"'/models/'"$JOB\_NAME"'/output/model.tar.gz"  
   }\],  
   "SupportedContentTypes": \["application/json"\],  
   "SupportedResponseMIMETypes": \["application/json"\]  
 }' \\  
 --model-approval-status "PendingManualApproval"

**7\. Approve the model package**

\# Get model package ARN from previous step  
aws sagemaker list-model-packages --model-package-group-name "review-summarizer"

MODEL\_PACKAGE\_ARN="arn:aws:sagemaker:ap-southeast-2:123456789012:model-package/review-summarizer/1"

\# Approve model  
aws sagemaker update-model-package \\  
 --model-package-arn $MODEL\_PACKAGE\_ARN \\  
 --model-approval-status "Approved"

This will automatically trigger a lambda function to create SageMaker endpoint and update Parameter Store with the new endpoint

**8\. Test the api**

./scripts/test\_api.sh

Result

Test 1: Health Check  
\====================  
{  
  "status": "healthy"  
}

Test 2: Technology News Article  
\================================  
{  
  "request\_id": "2026-02-15T10:33:26.300629",  
  "timestamp": "2026-02-15T10:33:30.740083",  
  "initial\_summary": "Here is a concise summary of the customer review:\\n\\nThe new iPhone model features significant upgrades, including a 48-megapixel main camera and up to 20 hours of video playback. It also includes new AI-powered photo editing features and enhanced security measures. Pre-orders begin next Friday, with the device launching two weeks later. Industry analysts predict strong sales, particularly in the premium smartphone segment, with a starting price of $999 for the base model.",  
  "final\_summary": "the new iPhone features a 48-megapixel main camera and 20 hours of video playback. the company also introduced new AI-powered features for photo editing. Industry analysts predict strong sales, particularly in the premium smartphone segment.",  
  "sentiment": "neutral",  
  "confidence": 0.0,  
  "used\_rag": false,  
  "model\_endpoint": "endpoint-kate",  
  "request\_context": {  
    "category": "general",  
    "userTier": "standard",  
    "textLength": 602,  
    "userId": null,  
    "requestId": "2026-02-15T10:33:26.300629"  
  }  
}

Test 3: Political News Article  
\===============================  
{  
  "request\_id": "2026-02-15T10:33:30.933191",  
  "timestamp": "2026-02-15T10:33:34.585612",  
  "initial\_summary": "Here is a concise, objective summary of the customer review:\\n\\nThe Senate passed a $1.2 trillion bipartisan infrastructure bill that includes funding for roads, bridges, public transit, and broadband. Supporters say it will create jobs and modernize infrastructure, while critics are concerned about the cost and impact on the federal deficit. The bill now goes to the House for consideration, and President Biden praised the Senate's historic vote.",  
  "final\_summary": "the bill includes funding for roads, bridges, public transit, and broadband internet expansion. President Biden calls the vote a historic investment in America future.",  
  "sentiment": "neutral",  
  "confidence": 0.0,  
  "used\_rag": false,  
  "model\_endpoint": "endpoint-kate",  
  "request\_context": {  
    "category": "general",  
    "userTier": "standard",  
    "textLength": 598,  
    "userId": null,  
    "requestId": "2026-02-15T10:33:30.933191"  
  }  
}

Test 4: Business News Article  
\==============================  
{  
  "request\_id": "2026-02-15T10:33:34.705612",  
  "timestamp": "2026-02-15T10:33:38.517618",  
  "initial\_summary": "Here is a concise, objective summary of the customer review:\\n\\nTesla reported record quarterly earnings, beating Wall Street expectations. The company delivered 250,000 vehicles, a 40% increase from the previous year, and revenue reached $13.8 billion. CEO Elon Musk attributed the strong performance to increased production capacity and growing global demand for electric vehicles. Tesla also announced plans to build two new manufacturing facilities in Europe and Asia, and the stock price rose 8% after the earnings announcement.",  
  "final\_summary": "Tesla delivered 250,000 vehicles in the quarter, a 40% increase from the same period last year. Revenue reached $13.8 billion, up from $10.4 billion a year ago.",  
  "sentiment": "neutral",  
  "confidence": 0.0,  
  "used\_rag": false,  
  "model\_endpoint": "endpoint-kate",  
  "request\_context": {  
    "category": "general",  
    "userTier": "standard",  
    "textLength": 570,  
    "userId": null,  
    "requestId": "2026-02-15T10:33:34.705612"  
  }  
}

Test 5: Sports News Article  
\============================  
{  
  "request\_id": "2026-02-15T10:33:38.619204",  
  "timestamp": "2026-02-15T10:33:42.190132",  
  "initial\_summary": "In a closely contested NBA Finals, the Los Angeles Lakers defeated the Boston Celtics 108-105 to win their 18th championship. LeBron James delivered an outstanding performance with 32 points, 11 rebounds, and 8 assists, while Anthony Davis added 28 points and played strong defense in the closing minutes. The Lakers overcame a 15-point deficit in the third quarter to secure the victory, showcasing their resilience and determination throughout the playoffs, as praised by head coach Frank Vogel.",  
  "final\_summary": "LeBron James led the team with 32 points, 11 rebounds, and 8 assists. This is the Lakers first championship in over a decade.",  
  "sentiment": "neutral",  
  "confidence": 0.0,  
  "used\_rag": false,  
  "model\_endpoint": "endpoint-kate",  
  "request\_context": {  
    "category": "general",  
    "userTier": "standard",  
    "textLength": 563,  
    "userId": null,  
    "requestId": "2026-02-15T10:33:38.619204"  
  }  
}

All tests completed!

Now we have a complete fine-tuning pipeline with automatic model deployment. The application automatically uses the latest approved Amazon SageMaker endpoint for inference.

In addition, we can integrate Retrieval-Augmented Generation (RAG) into the pipeline. This involves setting up Amazon OpenSearch as a vector database, embedding relevant documents, and updating the Lambda function to retrieve contextual information before generating summaries (refer to [https://medium.com/stackademic/build-a-knowledge-based-q-a-bot-using-bedrock-s3-dynamodb-opensearch-via-aws-cdk-23f805975311](https://medium.com/stackademic/build-a-knowledge-based-q-a-bot-using-bedrock-s3-dynamodb-opensearch-via-aws-cdk-23f805975311)).

Currently, the system immediately switches to the new model once approved. However, we can implement A/B testing to gradually roll out the model, reducing potential risks and ensuring smoother transitions.

Link to the [repo](https://github.com/KateVu/aws-cdk-bedrock-sagemaker-llm)