Step-by-Step: Deploying a Multimodal AI Model with Llama 3.2 and FastAPI 0.112 on ECS 4.0

import os
import io
import json
import base64
import logging
from typing import Optional, Union

import torch
from fastapi import FastAPI, HTTPException, UploadFile, File, Form
from fastapi.responses import JSONResponse
from PIL import Image
from transformers import AutoProcessor, AutoModelForVision2Seq
import uvicorn

# Configure logging for production tracing
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Llama 3.2 Multimodal Inference API",
    description="Production-ready endpoint for Llama 3.2 Vision text + image inference",
    version="1.0.0"
)

# Model configuration - update for your deployment
MODEL_ID = "meta-llama/Llama-3.2-11B-Vision-Instruct"
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
MAX_IMAGE_SIZE = (512, 512)  # Match benchmarked input size
MAX_NEW_TOKENS = 128  # Benchmarked optimal for common use cases

# Global model and processor to avoid reloading on each request
model: Optional[AutoModelForVision2Seq] = None
processor: Optional[AutoProcessor] = None

@app.on_event("startup")
async def load_model():
    """Load model and processor on startup, with error handling for missing GPU or model access"""
    global model, processor
    try:
        logger.info(f"Loading model {MODEL_ID} to {DEVICE}")
        # Note: Ensure you have accepted the model license on HuggingFace Hub
        # and set HF_TOKEN environment variable for authenticated access
        processor = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
        model = AutoModelForVision2Seq.from_pretrained(
            MODEL_ID,
            trust_remote_code=True,
            torch_dtype=torch.bfloat16 if DEVICE == "cuda" else torch.float32,
            device_map=DEVICE
        )
        model.eval()  # Set to inference mode
        logger.info("Model loaded successfully")
    except Exception as e:
        logger.error(f"Failed to load model: {str(e)}")
        raise RuntimeError(f"Model initialization failed: {str(e)}")

def decode_image(image_data: Union[bytes, str]) -> Image.Image:
    """Decode base64 or raw bytes image to PIL Image, with size validation"""
    try:
        if isinstance(image_data, str):
            # Assume base64 encoded image
            image_bytes = base64.b64decode(image_data)
        else:
            image_bytes = image_data
        img = Image.open(io.BytesIO(image_bytes))
        # Resize to max size to prevent OOM errors
        img.thumbnail(MAX_IMAGE_SIZE, Image.Resampling.LANCZOS)
        return img.convert("RGB")
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Invalid image data: {str(e)}")

@app.post("/infer")
async def infer_multimodal(
    text_prompt: str = Form(...),
    image: UploadFile = File(...),
    max_new_tokens: Optional[int] = Form(MAX_NEW_TOKENS)
):
    """Multimodal inference endpoint for text + image inputs"""
    if not model or not processor:
        raise HTTPException(status_code=503, detail="Model not loaded")
    if max_new_tokens > 512:
        raise HTTPException(status_code=400, detail="max_new_tokens cannot exceed 512")

    try:
        # Read and decode image
        image_bytes = await image.read()
        pil_image = decode_image(image_bytes)

        # Prepare inputs
        inputs = processor(
            text=text_prompt,
            images=pil_image,
            return_tensors="pt",
            truncation=True,
            max_length=512  # Truncate input text to prevent OOM
        ).to(DEVICE)

        # Run inference with no gradient tracking
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=max_new_tokens,
                do_sample=True,
                temperature=0.7,
                top_p=0.9
            )

        # Decode output tokens
        response = processor.batch_decode(outputs, skip_special_tokens=True)[0]
        return JSONResponse(content={"response": response})
    except HTTPException:
        raise  # Re-raise FastAPI HTTP exceptions
    except Exception as e:
        logger.error(f"Inference failed: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Inference error: {str(e)}")

if __name__ == "__main__":
    # Production server config: 4 workers for T4 GPU, match ECS task CPU allocation
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8080,
        workers=4 if DEVICE == "cuda" else 1,
        log_level="info"
    )

#!/bin/bash
set -euo pipefail  # Exit on error, undefined vars, pipe failures

# Configuration - update these values for your deployment
AWS_REGION="us-east-1"
ECR_REPO_NAME="llama3.2-fastapi-inference"
ECS_CLUSTER_NAME="multimodal-inference-cluster"
ECS_SERVICE_NAME="llama3.2-service"
ECS_TASK_FAMILY="llama3.2-task"
DOCKER_IMAGE_TAG="v1.0.0"
TASK_CPU="4096"  # 4 vCPU for Llama 3.2 11B
TASK_MEMORY="32768"  # 32GB RAM for model weights + overhead
TASK_GPU_COUNT="1"  # Single T4 GPU per task

# Logging function for consistent output
log() {
    echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')] $1"
}

# Error handling function
handle_error() {
    log "ERROR: Deployment failed at line $1"
    exit 1
}
trap 'handle_error $LINENO' ERR

log "Starting ECS 4.0 deployment for Llama 3.2 + FastAPI 0.112"

# 1. Check prerequisites
log "Checking prerequisites..."
if ! command -v aws &> /dev/null; then
    log "ERROR: AWS CLI not installed. Install via: https://aws.amazon.com/cli/"
    exit 1
fi
if ! command -v docker &> /dev/null; then
    log "ERROR: Docker not installed. Install via: https://docs.docker.com/engine/install/"
    exit 1
fi
if ! aws sts get-caller-identity &> /dev/null; then
    log "ERROR: AWS credentials not configured. Run: aws configure"
    exit 1
fi

# 2. Create ECR repository if it doesn't exist
log "Creating ECR repository $ECR_REPO_NAME..."
aws ecr describe-repositories --repository-names "$ECR_REPO_NAME" --region "$AWS_REGION" &> /dev/null || \
    aws ecr create-repository --repository-name "$ECR_REPO_NAME" --region "$AWS_REGION" --image-scanning-configuration scanOnPush=true

# 3. Build and push Docker image
log "Building Docker image..."
docker build -t "$ECR_REPO_NAME:$DOCKER_IMAGE_TAG" .
docker tag "$ECR_REPO_NAME:$DOCKER_IMAGE_TAG" "$AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO_NAME:$DOCKER_IMAGE_TAG"

log "Pushing Docker image to ECR..."
aws ecr get-login-password --region "$AWS_REGION" | docker login --username AWS --password-stdin "$AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com"
docker push "$AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO_NAME:$DOCKER_IMAGE_TAG"

# 4. Create ECS cluster if it doesn't exist (ECS 4.0 uses VPC networking by default)
log "Creating ECS cluster $ECS_CLUSTER_NAME..."
aws ecs describe-clusters --clusters "$ECS_CLUSTER_NAME" --region "$AWS_REGION" &> /dev/null || \
    aws ecs create-cluster \
        --cluster-name "$ECS_CLUSTER_NAME" \
        --capacity-providers "FARGATE_GPU" \
        --default-capacity-provider-strategy "capacityProvider=FARGATE_GPU,weight=1" \
        --region "$AWS_REGION"

# 5. Register ECS task definition with GPU support (ECS 4.0 feature)
log "Registering ECS task definition..."
cat > task-def.json << EOF
{
    "family": "$ECS_TASK_FAMILY",
    "networkMode": "awsvpc",
    "requiresCompatibilities": ["FARGATE_GPU"],
    "cpu": "$TASK_CPU",
    "memory": "$TASK_MEMORY",
    "taskRoleArn": "arn:aws:iam::$AWS_ACCOUNT_ID:role/ecs-task-execution-role",
    "executionRoleArn": "arn:aws:iam::$AWS_ACCOUNT_ID:role/ecs-task-execution-role",
    "containerDefinitions": [
        {
            "name": "llama3.2-container",
            "image": "$AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/$ECR_REPO_NAME:$DOCKER_IMAGE_TAG",
            "essential": true,
            "portMappings": [
                {
                    "containerPort": 8080,
                    "protocol": "tcp"
                }
            ],
            "environment": [
                {"name": "DEVICE", "value": "cuda"},
                {"name": "MODEL_ID", "value": "meta-llama/Llama-3.2-11B-Vision-Instruct"}
            ],
            "secrets": [
                {"name": "HF_TOKEN", "valueFrom": "arn:aws:secretsmanager:$AWS_REGION:$AWS_ACCOUNT_ID:secret:hf-token"}
            ],
            "resourceRequirements": [
                {"type": "GPU", "value": "$TASK_GPU_COUNT"}
            ],
            "logConfiguration": {
                "logDriver": "awslogs",
                "options": {
                    "awslogs-group": "/ecs/llama3.2",
                    "awslogs-region": "$AWS_REGION",
                    "awslogs-stream-prefix": "ecs"
                }
            }
        }
    ]
}
EOF

aws ecs register-task-definition --cli-input-json file://task-def.json --region "$AWS_REGION"

# 6. Create or update ECS service
log "Creating ECS service $ECS_SERVICE_NAME..."
aws ecs describe-services --cluster "$ECS_CLUSTER_NAME" --services "$ECS_SERVICE_NAME" --region "$AWS_REGION" &> /dev/null && \
    UPDATE_CMD="update-service" || UPDATE_CMD="create-service"

aws ecs $UPDATE_CMD \
    --cluster "$ECS_CLUSTER_NAME" \
    --service "$ECS_SERVICE_NAME" \
    --task-definition "$ECS_TASK_FAMILY" \
    --desired-count 2 \
    --capacity-provider-strategy "capacityProvider=FARGATE_GPU,weight=1" \
    --network-configuration "awsvpcConfiguration={subnets=[subnet-12345678,subnet-87654321],securityGroups=[sg-12345678],assignPublicIp=ENABLED}" \
    --region "$AWS_REGION"

log "Deployment complete! Service is available at: http://$(aws ecs describe-services --cluster $ECS_CLUSTER_NAME --services $ECS_SERVICE_NAME --region $AWS_REGION --query 'services[0].loadBalancers[0].targetGroupArn' --output text)"

import os
import time
import base64
import json
import logging
from typing import List, Dict

import requests
import matplotlib.pyplot as plt
import pandas as pd
from PIL import Image
import io

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Configuration
API_ENDPOINT = "http://your-ecs-service-url:8080/infer"
HF_TOKEN = os.getenv("HF_TOKEN")  # For downloading test images
TEST_IMAGE_PATH = "test_image.jpg"  # 512x512 test image
TEST_PROMPTS = [
    "Describe the contents of this image in detail.",
    "What objects are present in this image?",
    "Identify the primary color scheme of this image.",
    "Is there text in this image? If yes, transcribe it."
]
NUM_RUNS = 100  # Number of benchmark runs per prompt
MAX_TOKENS = 128

def load_test_image() -> bytes:
    """Load or download a 512x512 test image for benchmarking"""
    if not os.path.exists(TEST_IMAGE_PATH):
        logger.info("Downloading test image...")
        # Download a public 512x512 image for testing
        import httpx
        response = httpx.get("https://picsum.photos/512/512", follow_redirects=True)
        response.raise_for_status()
        with open(TEST_IMAGE_PATH, "wb") as f:
            f.write(response.content)
    with open(TEST_IMAGE_PATH, "rb") as f:
        return f.read()

def encode_image(image_bytes: bytes) -> str:
    """Base64 encode image for multipart form data"""
    return base64.b64encode(image_bytes).decode("utf-8")

def run_benchmark() -> List[Dict]:
    """Run benchmark tests and collect latency metrics"""
    image_bytes = load_test_image()
    results = []

    for prompt in TEST_PROMPTS:
        logger.info(f"Running benchmark for prompt: {prompt[:50]}...")
        latencies = []
        errors = 0

        for run in range(NUM_RUNS):
            try:
                start_time = time.perf_counter()
                # Prepare multipart form data
                files = {"image": ("test_image.jpg", image_bytes, "image/jpeg")}
                data = {"text_prompt": prompt, "max_new_tokens": MAX_TOKENS}
                # Send request to API
                response = requests.post(API_ENDPOINT, files=files, data=data, timeout=10)
                response.raise_for_status()
                end_time = time.perf_counter()
                latency_ms = (end_time - start_time) * 1000
                latencies.append(latency_ms)
                if run % 10 == 0:
                    logger.info(f"Run {run+1}/{NUM_RUNS} completed: {latency_ms:.2f}ms")
            except Exception as e:
                errors += 1
                logger.error(f"Run {run+1} failed: {str(e)}")
                if errors > 10:
                    logger.error("Too many errors, aborting benchmark")
                    break

        if latencies:
            # Calculate metrics
            latencies_sorted = sorted(latencies)
            p50 = latencies_sorted[len(latencies_sorted)//2]
            p90 = latencies_sorted[int(len(latencies_sorted)*0.9)]
            p99 = latencies_sorted[int(len(latencies_sorted)*0.99)]
            avg = sum(latencies) / len(latencies)
            results.append({
                "prompt": prompt,
                "p50_latency_ms": round(p50, 2),
                "p90_latency_ms": round(p90, 2),
                "p99_latency_ms": round(p99, 2),
                "avg_latency_ms": round(avg, 2),
                "total_runs": len(latencies),
                "errors": errors
            })
    return results

def generate_report(results: List[Dict]):
    """Generate benchmark report with table and plot"""
    df = pd.DataFrame(results)
    logger.info("\n=== Benchmark Results ===")
    logger.info(df.to_markdown(index=False))

    # Save to JSON
    with open("benchmark_results.json", "w") as f:
        json.dump(results, f, indent=2)

    # Generate latency plot
    plt.figure(figsize=(10, 6))
    for metric in ["p50_latency_ms", "p90_latency_ms", "p99_latency_ms"]:
        plt.plot(df["prompt"], df[metric], label=metric.replace("_", " ").title())
    plt.xticks(rotation=45, ha="right")
    plt.ylabel("Latency (ms)")
    plt.title("Llama 3.2 + FastAPI 0.112 Inference Latency on ECS 4.0")
    plt.legend()
    plt.tight_layout()
    plt.savefig("benchmark_latency.png")
    logger.info("Report saved to benchmark_results.json and benchmark_latency.png")

if __name__ == "__main__":
    if not API_ENDPOINT.startswith("http"):
        logger.error("Set API_ENDPOINT to your ECS service URL")
        exit(1)
    results = run_benchmark()
    generate_report(results)

from pydantic import BaseModel, validator, Field
from fastapi import UploadFile, File

class MultimodalRequest(BaseModel):
    text_prompt: str = Field(..., max_length=512)
    max_new_tokens: int = Field(128, ge=1, le=512)

    @validator("text_prompt")
    async def validate_prompt(cls, v):
        if len(v.strip()) == 0:
            raise ValueError("text_prompt cannot be empty")
        return v

@app.post("/infer-optimized")
async def infer_optimized(
    request: MultimodalRequest = Depends(),
    image: UploadFile = File(...)
):
    # Inference logic here
    pass

aws cloudwatch put-metric-alarm \
    --alarm-name "llama3.2-low-gpu-util" \
    --metric-name "GPUUtilization" \
    --namespace "AWS/ECS" \
    --statistic "Average" \
    --period 300 \
    --threshold 30 \
    --comparison-operator "LessThanThreshold" \
    --dimensions "Name=ClusterName,Value=multimodal-inference-cluster" "Name=TaskDefinitionFamily,Value=llama3.2-task" \
    --evaluation-periods 2 \
    --alarm-actions "arn:aws:autoscaling:$AWS_REGION:$AWS_ACCOUNT_ID:scalingPolicy:12345678:ecs:cluster:multimodal-inference-cluster:scale-in"

model = AutoModelForVision2Seq.from_pretrained(
    MODEL_ID,
    trust_remote_code=True,
    torch_dtype=torch.bfloat16,  # 2x memory savings vs float32
    device_map="cuda"
)