Módulo 5: AWS Services for AI (S3, Lambda, SageMaker Basics)

5. SageMaker Basics

Descripción

En esta cápsula vas a entender qué es Amazon SageMaker, cuándo usarlo en lugar de Lambda para inferencia AI, y cómo funciona un deploy básico de endpoint. Esta es una cápsula de "basics" deliberadamente — SageMaker es un ecosistema enorme (notebooks, training, pipelines, feature store, MLOps). Aquí no profundizamos en todo eso. Lo que necesitas saber como AI Engineer que despliega sistemas es: ¿cuándo SageMaker es la respuesta correcta? Y si lo es, ¿cómo se ve un endpoint básico?

Contexto: Has construido inferencia con Lambda invocando APIs externas (OpenAI, Anthropic). Eso funciona cuando usas modelos de terceros via API. Pero hay escenarios donde Lambda no alcanza: modelos custom que entrenaste tú, modelos open-source que quieres hostear (Llama, Mistral), o inferencia que necesita GPU. SageMaker cubre esos casos. Esta cápsula te da el criterio para decidir y la mecánica básica para empezar.


Qué Es SageMaker

SageMaker en una frase

Amazon SageMaker es una plataforma de machine learning que cubre todo el ciclo: desde entrenar modelos hasta servirlos como endpoints de inferencia. Es la respuesta de AWS a "necesito hostear mi propio modelo."

El ecosistema SageMaker

Amazon SageMaker (ecosistema completo):
├── SageMaker Studio          ← Notebooks y IDE (como Jupyter en AWS)
├── SageMaker Training        ← Entrenar modelos con GPU on-demand
├── SageMaker Processing      ← Preprocesamiento de datos
├── SageMaker Endpoints       ← Servir modelos como API (ESTE ES TU FOCO)
├── SageMaker Pipelines       ← Orquestación de ML workflows
├── SageMaker Feature Store   ← Almacenamiento de features
├── SageMaker Model Registry  ← Versionado de modelos
├── SageMaker Ground Truth    ← Etiquetado de datos
└── SageMaker Canvas          ← No-code ML

Para esta guía, lo que importa es SageMaker Endpoints: la capacidad de tomar un modelo (tuyo o pre-entrenado) y exponerlo como una API HTTP. El resto del ecosistema es MLOps — relevante si te especializas en ML, pero fuera del scope de esta guía de deployment.

Modelo mental: SageMaker vs Lambda

¿Qué tipo de inferencia necesitas?

Caso 1: Invocas API de terceros (OpenAI, Anthropic, Cohere)
├── Tu código hace HTTP request a la API del proveedor
├── No hosteas ningún modelo
├── Necesitas: compute ligero + internet
└── Solución: Lambda ✅ (cápsulas 02-04 de este módulo)

Caso 2: Hosteas tu propio modelo (Llama, Mistral, modelo custom)
├── El modelo corre en TU infraestructura
├── Necesitas GPU o compute dedicado
├── El modelo está en S3 o un registry
└── Solución: SageMaker Endpoint ✅ (esta cápsula)

Caso 3: Inferencia con modelo ligero (scikit-learn, modelo pequeño)
├── Modelo cabe en memoria de Lambda (< 10GB container)
├── No necesita GPU
├── Tiempo de inferencia < 15min
└── Solución: Lambda con container ✅ (puede funcionar)

Cuándo SageMaker vs Lambda

Decision framework

CriterioLambdaSageMaker Endpoint
Tipo de modeloAPIs externas (OpenAI, Anthropic)Modelos propios o open-source
GPU necesaria❌ Lambda no tiene GPU✅ GPU instances disponibles
Tamaño del modeloN/A (el modelo está en el proveedor)100MB - 100GB+
Latencia cold start1-15s5-10 min (startup de instancia)
Latencia inferenciaDepende de la API (1-30s)Depende del modelo (50ms-5s)
Coste en idle$0 (se apaga)$$$ (instancia siempre encendida)
Coste en usoPor invocación + duraciónPor hora de instancia
ScalingAutomático (0 a 1000+)Configurable (min/max instancias)
Max execution15 minutosSin límite
Complejidad setupBaja (zip/container + SAM)Media-alta (model artifacts, container, config)

Escenarios concretos

Escenario 1: Chatbot con GPT-4o
├── Modelo: OpenAI API (no hosteas nada)
├── Tráfico: Variable (picos y valles)
├── Decisión: Lambda ✅
└── Razón: Solo necesitas compute para hacer HTTP requests

Escenario 2: Clasificador de sentimiento custom (fine-tuned BERT)
├── Modelo: 400MB, entrenado con tus datos
├── Tráfico: Constante, 24/7
├── Decisión: SageMaker ✅
└── Razón: Modelo propio que necesita hosting dedicado

Escenario 3: RAG con embeddings de OpenAI
├── Modelo: OpenAI Embeddings API
├── Tráfico: Moderate, horario laboral
├── Decisión: Lambda ✅
└── Razón: Embeddings via API, no hosteas modelo

Escenario 4: Generación de imágenes con Stable Diffusion
├── Modelo: 5GB+, necesita GPU
├── Tráfico: Bajo pero necesita respuesta rápida
├── Decisión: SageMaker ✅
└── Razón: GPU obligatoria, modelo pesado

Escenario 5: Clasificador scikit-learn ligero
├── Modelo: 50MB, Python puro
├── Tráfico: Event-driven (emails entrantes)
├── Decisión: Lambda (container) ✅
└── Razón: Modelo pequeño, no necesita GPU, event-driven

La regla de oro

Usa Lambda cuando invocas APIs de terceros o cuando tu modelo es lo suficientemente pequeño para caber en un container Lambda (< 10GB) sin necesitar GPU.

Usa SageMaker cuando hosteas modelos propios, modelos open-source grandes, o cualquier modelo que necesite GPU.


Anatomía de un SageMaker Endpoint

Los tres componentes

Un SageMaker Endpoint tiene tres piezas:

1. Model Artifacts (en S3)
   ├── Tu modelo entrenado (model.tar.gz)
   ├── Puede incluir: pesos, config, tokenizer
   └── SageMaker lo descarga al iniciar la instancia

2. Container Image (en ECR)
   ├── El runtime que carga y sirve el modelo
   ├── AWS provee containers pre-built para frameworks comunes
   │   (PyTorch, TensorFlow, Hugging Face, scikit-learn)
   └── O puedes crear un container custom

3. Endpoint Configuration
   ├── Instance type (ml.m5.large, ml.g4dn.xlarge para GPU)
   ├── Instance count (1+ para alta disponibilidad)
   └── Auto-scaling rules
                    ┌─────────────────────────────┐
                    │ S3: model-artifacts/         │
                    │ └── model.tar.gz             │
                    └──────────┬──────────────────┘
                               │ Descarga al crear endpoint
                               ↓
                    ┌─────────────────────────────┐
                    │ SageMaker Endpoint           │
                    │ ├── Container (ECR image)    │
                    │ ├── Instance (ml.m5.large)   │
                    │ ├── Model loaded in memory   │
                    │ └── HTTPS API disponible     │
                    └──────────┬──────────────────┘
                               │
                    invoke_endpoint()
                               │
                    ┌──────────┴──────────────────┐
                    │ Tu código (Lambda, app, etc) │
                    │ sagemaker_runtime.invoke_    │
                    │ endpoint(EndpointName=...)   │
                    └─────────────────────────────┘

Deploy de un Modelo Pre-Entrenado

Ejemplo: Modelo de clasificación con scikit-learn

# deploy_sagemaker_endpoint.py
import boto3
import json
import os
import tarfile
import pickle
import numpy as np
from datetime import datetime

sagemaker = boto3.client("sagemaker")
s3 = boto3.client("s3")

BUCKET = os.environ.get("SAGEMAKER_BUCKET", "sagemaker-models-dev")
ROLE_ARN = os.environ.get("SAGEMAKER_ROLE_ARN", "arn:aws:iam::123456789012:role/SageMakerExecutionRole")


def create_dummy_model():
    """Crea un modelo dummy de scikit-learn para demostración."""
    from sklearn.linear_model import LogisticRegression

    X = np.array([[1, 2], [3, 4], [5, 6], [7, 8]])
    y = np.array([0, 0, 1, 1])
    model = LogisticRegression()
    model.fit(X, y)
    return model


def package_model(model, model_name: str) -> str:
    """Empaqueta un modelo como model.tar.gz y lo sube a S3."""
    local_path = f"/tmp/{model_name}"
    os.makedirs(local_path, exist_ok=True)

    model_file = f"{local_path}/model.pkl"
    with open(model_file, "wb") as f:
        pickle.dump(model, f)

    tar_path = f"/tmp/{model_name}.tar.gz"
    with tarfile.open(tar_path, "w:gz") as tar:
        tar.add(model_file, arcname="model.pkl")

    s3_key = f"models/{model_name}/model.tar.gz"
    s3.upload_file(tar_path, BUCKET, s3_key)

    model_uri = f"s3://{BUCKET}/{s3_key}"
    print(f"Model uploaded to: {model_uri}")
    return model_uri


def create_sagemaker_model(model_name: str, model_uri: str) -> str:
    """Crea un SageMaker Model pointing al artifact en S3."""
    response = sagemaker.create_model(
        ModelName=model_name,
        PrimaryContainer={
            "Image": "683313688378.dkr.ecr.us-east-1.amazonaws.com/sagemaker-scikit-learn:1.2-1-cpu-py3",
            "ModelDataUrl": model_uri,
        },
        ExecutionRoleArn=ROLE_ARN,
    )
    print(f"SageMaker Model created: {model_name}")
    return response["ModelArn"]


def create_endpoint_config(config_name: str, model_name: str) -> str:
    """Crea la configuración del endpoint."""
    response = sagemaker.create_endpoint_config(
        EndpointConfigName=config_name,
        ProductionVariants=[
            {
                "VariantName": "default",
                "ModelName": model_name,
                "InstanceType": "ml.m5.large",
                "InitialInstanceCount": 1,
            }
        ],
    )
    print(f"Endpoint config created: {config_name}")
    return response["EndpointConfigArn"]


def create_endpoint(endpoint_name: str, config_name: str) -> str:
    """Crea el endpoint (esto toma 5-10 minutos)."""
    response = sagemaker.create_endpoint(
        EndpointName=endpoint_name,
        EndpointConfigName=config_name,
    )
    print(f"Creating endpoint: {endpoint_name}")
    print("This takes 5-10 minutes...")
    return response["EndpointArn"]


def wait_for_endpoint(endpoint_name: str):
    """Espera a que el endpoint esté InService."""
    waiter = sagemaker.get_waiter("endpoint_in_service")
    waiter.wait(
        EndpointName=endpoint_name,
        WaiterConfig={"Delay": 30, "MaxAttempts": 30},
    )
    print(f"Endpoint {endpoint_name} is InService")

Invocar el endpoint

import boto3
import json

sagemaker_runtime = boto3.client("sagemaker-runtime")


def invoke_sagemaker_endpoint(endpoint_name: str, data: list) -> dict:
    """Invoca un SageMaker endpoint para inferencia."""
    response = sagemaker_runtime.invoke_endpoint(
        EndpointName=endpoint_name,
        ContentType="application/json",
        Body=json.dumps({"instances": data}),
    )

    result = json.loads(response["Body"].read().decode("utf-8"))
    return result


predictions = invoke_sagemaker_endpoint(
    endpoint_name="sentiment-classifier-v1",
    data=[[1.5, 2.5], [8.0, 9.0]],
)
print(f"Predictions: {predictions}")

Invocar SageMaker desde Lambda

# lambda_sagemaker_handler.py
import boto3
import json
import os

sagemaker_runtime = boto3.client("sagemaker-runtime")
ENDPOINT_NAME = os.environ.get("SAGEMAKER_ENDPOINT", "sentiment-classifier-v1")


def handler(event, context):
    """Lambda que invoca un SageMaker endpoint."""
    try:
        body = json.loads(event.get("body", "{}"))
    except (json.JSONDecodeError, TypeError):
        return _response(400, {"error": "Invalid JSON"})

    data = body.get("data")
    if not data:
        return _response(400, {"error": "data field is required"})

    try:
        response = sagemaker_runtime.invoke_endpoint(
            EndpointName=ENDPOINT_NAME,
            ContentType="application/json",
            Body=json.dumps({"instances": data}),
        )
        result = json.loads(response["Body"].read().decode("utf-8"))
    except Exception as e:
        return _response(502, {"error": f"SageMaker error: {str(e)}"})

    return _response(200, {
        "predictions": result,
        "endpoint": ENDPOINT_NAME,
    })


def _response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(body),
    }

Cleanup: No Dejes Endpoints Encendidos

SageMaker endpoints cuestan dinero mientras están activos. Un ml.m5.large cuesta $0.115/hora ($84/mes). Siempre elimina endpoints cuando termines de experimentar:

def cleanup_endpoint(endpoint_name: str):
    """Elimina endpoint, config, y model de SageMaker."""
    try:
        sagemaker.delete_endpoint(EndpointName=endpoint_name)
        print(f"Endpoint deleted: {endpoint_name}")
    except sagemaker.exceptions.ClientError:
        print(f"Endpoint {endpoint_name} not found")

    config_name = endpoint_name
    try:
        sagemaker.delete_endpoint_config(EndpointConfigName=config_name)
        print(f"Config deleted: {config_name}")
    except sagemaker.exceptions.ClientError:
        print(f"Config {config_name} not found")

    model_name = endpoint_name
    try:
        sagemaker.delete_model(ModelName=model_name)
        print(f"Model deleted: {model_name}")
    except sagemaker.exceptions.ClientError:
        print(f"Model {model_name} not found")


cleanup_endpoint("sentiment-classifier-v1")

Costes de SageMaker: Por Qué Importa

Pricing básico

SageMaker Endpoint Pricing (us-east-1):
├── ml.t3.medium (CPU, básica):     $0.05/hr  → ~$36/mes
├── ml.m5.large (CPU, producción):  $0.115/hr → ~$84/mes
├── ml.m5.xlarge (CPU, pesada):     $0.23/hr  → ~$168/mes
├── ml.g4dn.xlarge (GPU, NVIDIA T4): $0.736/hr → ~$537/mes
└── ml.g5.xlarge (GPU, NVIDIA A10G): $1.408/hr → ~$1,028/mes

Comparación con Lambda:
├── Lambda (1K inv/día, 5s, 512MB):  ~$2/mes
├── SageMaker ml.m5.large 24/7:      ~$84/mes
└── SageMaker ml.g4dn.xlarge 24/7:   ~$537/mes

SageMaker no tiene free tier para endpoints. Desde el primer minuto que tu endpoint está "InService", estás pagando. Por eso el cleanup es crítico.

Cuándo el coste de SageMaker tiene sentido

Lambda + OpenAI API (10K inv/día):
├── Lambda: ~$3/mes
├── OpenAI: ~$200/mes (depende de tokens)
└── Total: ~$203/mes

SageMaker + modelo open-source (10K inv/día):
├── SageMaker ml.g4dn.xlarge: ~$537/mes
├── No pagas API de OpenAI: $0
└── Total: ~$537/mes

¿Cuándo SageMaker es más barato?
├── A muy alto volumen (>50K inv/día), el coste de API se dispara
├── Cuando el modelo open-source es "suficientemente bueno" para tu caso
├── Cuando necesitas control total del modelo (fine-tuning, datos sensibles)
└── Cuando no puedes enviar datos a APIs externas (compliance)

Serverless Inference: El Punto Medio

SageMaker también ofrece Serverless Inference — endpoints que escalan a cero:

def create_serverless_endpoint_config(config_name: str, model_name: str) -> str:
    """Crea un endpoint config serverless (escala a cero)."""
    response = sagemaker.create_endpoint_config(
        EndpointConfigName=config_name,
        ProductionVariants=[
            {
                "VariantName": "default",
                "ModelName": model_name,
                "ServerlessConfig": {
                    "MemorySizeInMB": 2048,
                    "MaxConcurrency": 5,
                },
            }
        ],
    )
    return response["EndpointConfigArn"]
SageMaker Serverless Inference:
├── Escala a cero (no pagas cuando no hay tráfico)
├── Pagas por: duración de inferencia + memory
├── Cold start: 30s-2min (significativamente más lento que Lambda)
├── Ideal para: modelos custom con tráfico esporádico
└── No soporta GPU (solo CPU)

Troubleshooting

Problema 1: "Could not find model data" al crear endpoint

El ModelDataUrl apunta a un S3 path que no existe o el role no tiene permisos.

# Verifica que model.tar.gz existe en S3
aws s3 ls s3://sagemaker-models-dev/models/sentiment-classifier-v1/

# Verifica que el IAM role tiene permisos de S3
aws iam get-role-policy --role-name SageMakerExecutionRole --policy-name S3Access

Problema 2: Endpoint tarda >15 minutos en crear

Es normal para instancias grandes o con GPU. El startup incluye: provisionar instancia, descargar container, descargar modelo, cargar en memoria.

# Monitorea el status
response = sagemaker.describe_endpoint(EndpointName="my-endpoint")
print(response["EndpointStatus"])
# Creating → InService (éxito) o Failed (error)

Problema 3: "ModelError" al invocar endpoint

El modelo no puede procesar el input que le enviaste. Verifica el formato.

# Verifica qué formato espera el container
# Para scikit-learn: {"instances": [[1, 2], [3, 4]]}
# Para PyTorch: depende del inference script
# Para Hugging Face: {"inputs": "texto a clasificar"}

Problema 4: Factura inesperada de SageMaker

Dejaste un endpoint encendido. Lista y elimina endpoints activos.

response = sagemaker.list_endpoints(StatusEquals="InService")
for ep in response["Endpoints"]:
    print(f"Active endpoint: {ep['EndpointName']} since {ep['CreationTime']}")
    # cleanup_endpoint(ep["EndpointName"])

Ejercicios Prácticos

Ejercicio 1: Decision framework aplicado

Dado estos 5 escenarios, decide si usarías Lambda o SageMaker para cada uno. Justifica tu respuesta con al menos 2 razones.

Ver solución
Escenario A: App de customer support que usa Claude para responder tickets
├── Decisión: Lambda ✅
├── Razón 1: Invocas API de Anthropic, no hosteas modelo
└── Razón 2: Tráfico variable → Lambda escala a cero

Escenario B: Modelo de detección de fraude entrenado con datos propios (XGBoost, 200MB)
├── Decisión: SageMaker ✅ (o Lambda con container si el volumen es bajo)
├── Razón 1: Modelo propio que necesitas hostear
└── Razón 2: Inferencia en tiempo real 24/7 para transacciones

Escenario C: Procesamiento batch de 10K documentos con GPT-4o-mini
├── Decisión: Lambda ✅
├── Razón 1: Invocas API de OpenAI, no hosteas modelo
└── Razón 2: Event-driven (S3 trigger), tráfico en ráfagas

Escenario D: Generación de embeddings con modelo open-source sentence-transformers
├── Decisión: SageMaker ✅
├── Razón 1: Modelo de ~500MB que necesitas GPU para velocidad
└── Razón 2: Puede ser más barato que OpenAI a alto volumen

Escenario E: Clasificador de sentimiento con modelo fine-tuned de 50MB (scikit-learn)
├── Decisión: Lambda (container) ✅
├── Razón 1: Modelo pequeño que cabe en Lambda
└── Razón 2: No necesita GPU, event-driven, escala a cero

Ejercicio 2: Calcular coste mensual SageMaker vs Lambda+API

Calcula el coste mensual de servir un modelo de clasificación para 50K invocaciones/día en dos escenarios: (A) SageMaker con ml.m5.large y (B) Lambda + OpenAI API.

Ver solución
def compare_costs():
    """Compara costes de SageMaker vs Lambda+API para 50K inv/día."""

    daily_invocations = 50_000
    monthly_invocations = daily_invocations * 30

    # Escenario A: SageMaker ml.m5.large
    sm_hourly = 0.115
    sm_monthly = sm_hourly * 24 * 30  # $82.80
    sm_total = sm_monthly

    # Escenario B: Lambda + OpenAI
    lambda_invocations_cost = max(0, monthly_invocations - 1_000_000) * 0.0000002
    lambda_duration_gb_s = monthly_invocations * 5 * (512 / 1024)
    lambda_compute_cost = max(0, lambda_duration_gb_s - 400_000) * 0.0000166667
    lambda_total = lambda_invocations_cost + lambda_compute_cost

    avg_input_tokens = 100
    avg_output_tokens = 50
    openai_input_cost = (monthly_invocations * avg_input_tokens / 1_000_000) * 0.15
    openai_output_cost = (monthly_invocations * avg_output_tokens / 1_000_000) * 0.60
    openai_total = openai_input_cost + openai_output_cost

    lambda_api_total = lambda_total + openai_total

    print("=" * 50)
    print(f"Invocaciones: {daily_invocations:,}/día ({monthly_invocations:,}/mes)")
    print()
    print(f"Escenario A: SageMaker ml.m5.large")
    print(f"  Endpoint 24/7:       ${sm_total:>8.2f}/mes")
    print(f"  OpenAI API:          ${0:>8.2f}/mes")
    print(f"  TOTAL:               ${sm_total:>8.2f}/mes")
    print()
    print(f"Escenario B: Lambda + OpenAI API")
    print(f"  Lambda compute:      ${lambda_total:>8.2f}/mes")
    print(f"  OpenAI API:          ${openai_total:>8.2f}/mes")
    print(f"  TOTAL:               ${lambda_api_total:>8.2f}/mes")
    print()
    if sm_total < lambda_api_total:
        print(f"SageMaker es ${lambda_api_total - sm_total:.2f}/mes más barato")
    else:
        print(f"Lambda+API es ${sm_total - lambda_api_total:.2f}/mes más barato")
    print("=" * 50)

compare_costs()

Ejercicio 3: Script de cleanup de endpoints

Crea un script que liste todos los endpoints SageMaker activos, muestre cuánto tiempo llevan encendidos, estime el coste acumulado, y pregunte si quieres eliminarlos.

Ver solución
import boto3
from datetime import datetime, timezone

sagemaker = boto3.client("sagemaker")

INSTANCE_PRICES = {
    "ml.t3.medium": 0.05,
    "ml.m5.large": 0.115,
    "ml.m5.xlarge": 0.23,
    "ml.g4dn.xlarge": 0.736,
    "ml.g5.xlarge": 1.408,
}


def audit_endpoints():
    """Lista endpoints activos con coste estimado."""
    response = sagemaker.list_endpoints(StatusEquals="InService")
    endpoints = response["Endpoints"]

    if not endpoints:
        print("No active endpoints found.")
        return []

    print(f"Found {len(endpoints)} active endpoint(s):\n")
    total_estimated_cost = 0

    results = []
    for ep in endpoints:
        name = ep["EndpointName"]
        created = ep["CreationTime"]
        now = datetime.now(timezone.utc)
        hours_running = (now - created).total_seconds() / 3600

        config = sagemaker.describe_endpoint_config(EndpointConfigName=name)
        variants = config.get("ProductionVariants", [])

        for variant in variants:
            instance_type = variant.get("InstanceType", "unknown")
            instance_count = variant.get("InitialInstanceCount", 1)
            hourly_rate = INSTANCE_PRICES.get(instance_type, 0)
            estimated_cost = hourly_rate * hours_running * instance_count

            total_estimated_cost += estimated_cost
            results.append({
                "name": name,
                "instance_type": instance_type,
                "count": instance_count,
                "hours": hours_running,
                "estimated_cost": estimated_cost,
            })

            print(f"  {name}")
            print(f"    Instance: {instance_type} x{instance_count}")
            print(f"    Running:  {hours_running:.1f} hours")
            print(f"    Cost:     ${estimated_cost:.2f} estimated")
            print()

    print(f"Total estimated cost: ${total_estimated_cost:.2f}")
    return results


def cleanup_all(endpoints: list, confirm: bool = False):
    """Elimina todos los endpoints listados."""
    for ep in endpoints:
        if confirm:
            sagemaker.delete_endpoint(EndpointName=ep["name"])
            print(f"Deleted: {ep['name']}")
        else:
            print(f"Would delete: {ep['name']} (${ep['estimated_cost']:.2f})")


results = audit_endpoints()
# cleanup_all(results, confirm=False)  # Cambia a True para eliminar

Ejercicio 4: Lambda que elige entre API externa y SageMaker

Implementa un handler Lambda que decida dinámicamente si invocar OpenAI API o un SageMaker endpoint basándose en un parámetro del request.

Ver solución
import boto3
import json
import os
from openai import OpenAI

openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))
sagemaker_runtime = boto3.client("sagemaker-runtime")
SAGEMAKER_ENDPOINT = os.environ.get("SAGEMAKER_ENDPOINT", "classifier-v1")


def invoke_openai(prompt: str, max_tokens: int) -> dict:
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tokens,
    )
    return {
        "result": response.choices[0].message.content,
        "backend": "openai",
        "model": response.model,
        "tokens_used": response.usage.total_tokens,
    }


def invoke_sagemaker(data: list) -> dict:
    response = sagemaker_runtime.invoke_endpoint(
        EndpointName=SAGEMAKER_ENDPOINT,
        ContentType="application/json",
        Body=json.dumps({"instances": data}),
    )
    result = json.loads(response["Body"].read().decode("utf-8"))
    return {
        "result": result,
        "backend": "sagemaker",
        "endpoint": SAGEMAKER_ENDPOINT,
    }


def handler(event, context):
    try:
        body = json.loads(event.get("body", "{}"))
    except (json.JSONDecodeError, TypeError):
        return _response(400, {"error": "Invalid JSON"})

    backend = body.get("backend", "openai")

    try:
        if backend == "openai":
            prompt = body.get("prompt", "").strip()
            if not prompt:
                return _response(400, {"error": "prompt required for openai"})
            result = invoke_openai(prompt, body.get("max_tokens", 500))

        elif backend == "sagemaker":
            data = body.get("data")
            if not data:
                return _response(400, {"error": "data required for sagemaker"})
            result = invoke_sagemaker(data)

        else:
            return _response(400, {"error": f"Unknown backend: {backend}"})

    except Exception as e:
        return _response(502, {"error": str(e), "backend": backend})

    return _response(200, result)


def _response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(body),
    }

Resumen

  • SageMaker es para hostear modelos propios o open-source. Si invocas APIs de terceros (OpenAI, Anthropic), Lambda es suficiente.
  • Un SageMaker endpoint tiene tres piezas: model artifacts en S3, container image en ECR, y endpoint configuration (instance type, count).
  • Los costes de SageMaker son por hora de instancia — no por invocación. Un ml.m5.large cuesta ~$84/mes 24/7. Elimina endpoints que no uses.
  • Serverless Inference es el punto medio: escala a cero como Lambda, pero sirve modelos custom. Cold start significativo (30s-2min).
  • La decisión Lambda vs SageMaker depende de: tipo de modelo (API vs propio), necesidad de GPU, patrón de tráfico, y presupuesto.
  • SageMaker basics es suficiente para esta guía. Si necesitas training, pipelines, MLOps → es otra guía completa.

Recursos Adicionales

  1. SageMaker Developer Guide — Documentación oficial completa
  2. SageMaker Python SDK — SDK de alto nivel para SageMaker
  3. SageMaker Pricing — Pricing por instance type
  4. SageMaker Serverless Inference — Endpoints serverless
  5. SageMaker Built-in Algorithms — Algoritmos pre-built
  6. Hugging Face on SageMaker — Deploy de modelos HuggingFace en SageMaker
  7. SageMaker vs Lambda — Blog oficial con comparativas
  8. SageMaker Cleanup — Guía de cleanup para evitar costes