2. Lambda Fundamentals para AI

Descripción

En esta cápsula vas a entender la anatomía de una función Lambda orientada a AI: cómo funciona el handler, cómo empaquetar dependencias Python para AI, cómo gestionar secrets (API keys de LLMs), y cuándo usar invocación síncrona vs asíncrona. No es un tutorial genérico de Lambda — cada línea de código invoca un LLM o gestiona un workflow de AI.

Contexto: La cápsula anterior estableció el landscape de serverless para AI y por qué Lambda tiene particularidades cuando trabaja con LLMs. Aquí pasas de "entender el problema" a "escribir la primera función." Al terminar, tendrás un Lambda funcional que invoca OpenAI, con dependencias empaquetadas y secrets bien gestionados.

Anatomía de un Lambda Handler

La firma del handler

Toda función Lambda en Python recibe exactamente dos argumentos:

def handler(event, context):
    # event: los datos que triggerearon la función
    # context: metadata del runtime (tiempo restante, request ID, etc.)
    return response

Estos dos argumentos te dan todo lo que necesitas para procesar una invocación:

# event — Datos de entrada
# Cuando Lambda se invoca via API Gateway:
event = {
    "httpMethod": "POST",
    "path": "/ask",
    "headers": {
        "Content-Type": "application/json",
        "x-api-key": "abc123"
    },
    "body": "{\"prompt\": \"Explica serverless en una frase\"}",
    "queryStringParameters": None,
    "pathParameters": None,
    "requestContext": {
        "requestId": "abc-123-def",
        "stage": "prod"
    }
}

# context — Metadata del runtime
# context.function_name          → "ai-endpoint-prod"
# context.memory_limit_in_mb     → 512
# context.get_remaining_time_in_millis() → 58000 (ms restantes)
# context.aws_request_id         → "abc-123-def-456"

El response format

Lambda espera un response con estructura específica cuando se invoca desde API Gateway:

# Response válido para API Gateway
response = {
    "statusCode": 200,
    "headers": {
        "Content-Type": "application/json",
        "Access-Control-Allow-Origin": "*"
    },
    "body": "{\"answer\": \"Serverless ejecuta tu código sin que gestiones servidores.\"}"
}

El body siempre es un string (JSON serializado). Si retornas un dict en body, API Gateway lanza un error. Este es uno de los errores más comunes al empezar con Lambda.

Tu Primer Lambda para AI

Handler completo que invoca OpenAI

# handler.py
import json
import os
import time
from openai import OpenAI

# Cliente inicializado FUERA del handler
# Se reutiliza entre invocaciones (warm starts)
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def handler(event, context):
    """Lambda que recibe un prompt y retorna respuesta de LLM."""
    start_time = time.time()

    try:
        body = json.loads(event.get("body", "{}"))
    except json.JSONDecodeError:
        return _error_response(400, "Invalid JSON in request body")

    prompt = body.get("prompt", "").strip()
    if not prompt:
        return _error_response(400, "Field 'prompt' is required and cannot be empty")

    max_tokens = body.get("max_tokens", 500)
    model = body.get("model", "gpt-4o-mini")

    remaining_ms = context.get_remaining_time_in_millis()
    # Reserva 5s para procesamiento post-LLM
    llm_timeout = (remaining_ms / 1000) - 5

    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=max_tokens,
            timeout=llm_timeout,
        )
    except Exception as e:
        return _error_response(502, f"LLM API error: {str(e)}")

    answer = response.choices[0].message.content
    tokens_used = response.usage.total_tokens
    duration_ms = int((time.time() - start_time) * 1000)

    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        "body": json.dumps({
            "answer": answer,
            "model": model,
            "tokens_used": tokens_used,
            "duration_ms": duration_ms,
            "request_id": context.aws_request_id,
        }),
    }


def _error_response(status_code, message):
    """Helper para responses de error consistentes."""
    return {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        "body": json.dumps({"error": message}),
    }

Por qué el cliente OpenAI está fuera del handler

# ✅ CORRECTO — Cliente fuera del handler
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def handler(event, context):
    response = client.chat.completions.create(...)
    ...

# ❌ INCORRECTO — Cliente dentro del handler
def handler(event, context):
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    response = client.chat.completions.create(...)
    ...

Lambda reutiliza el container entre invocaciones consecutivas (warm starts). El código fuera del handler se ejecuta solo una vez — en el cold start. El código dentro del handler se ejecuta en cada invocación. Inicializar el cliente OpenAI fuera significa que en warm starts no pagas el coste de crear la conexión HTTP.

Cold start (primera invocación):
├── Init: import openai, crear client  → ~500ms
└── Handler: invoke LLM, return        → ~3000ms
Total: ~3500ms

Warm start (invocaciones siguientes):
├── Init: SKIP (ya inicializado)       → 0ms
└── Handler: invoke LLM, return        → ~3000ms
Total: ~3000ms

Esta optimización parece menor pero se acumula: si tu función recibe 1000 invocaciones/hora, 990+ serán warm starts que se ahorran esos 500ms de inicialización.

Packaging de Dependencias Python

El problema

Lambda necesita que todas las dependencias Python estén incluidas en el paquete de deployment. No hace pip install al arrancar — todo debe estar pre-empaquetado. Para AI, esto es un reto porque las dependencias pueden ser pesadas:

Tamaño de dependencias comunes para AI:
├── openai (SDK):           ~5MB    → Viable en zip
├── anthropic (SDK):        ~8MB    → Viable en zip
├── httpx:                  ~3MB    → Viable en zip
├── langchain + deps:       ~80MB   → Límite de zip
├── numpy:                  ~30MB   → Requiere compilación
├── pandas:                 ~50MB   → Requiere compilación
├── torch (CPU):            ~800MB  → Imposible en zip, necesita container
└── transformers + torch:   ~2GB    → Imposible en Lambda estándar

Tres estrategias de packaging

Estrategia 1: Zip package (para dependencias ligeras)

# Para openai + httpx (< 50MB total)
mkdir package
pip install openai -t package/

cd package
zip -r ../deployment.zip .
cd ..
zip deployment.zip handler.py

# Resultado: deployment.zip (~8MB)

Límite de zip: 50MB comprimido, 250MB descomprimido. Suficiente para openai, anthropic, httpx. Insuficiente para langchain con todas sus dependencias.

Estrategia 2: Lambda Layers (dependencias compartidas)

# Layer: dependencias reutilizables entre funciones
mkdir -p python/lib/python3.11/site-packages
pip install openai -t python/lib/python3.11/site-packages/

zip -r openai-layer.zip python/

# Publicar layer
aws lambda publish-layer-version \
    --layer-name openai-sdk \
    --zip-file fileb://openai-layer.zip \
    --compatible-runtimes python3.11

# Adjuntar layer a tu función
aws lambda update-function-configuration \
    --function-name ai-endpoint \
    --layers arn:aws:lambda:us-east-1:123456789:layer:openai-sdk:1

Ventaja: el layer se comparte entre funciones y se cachea por Lambda. Límite total: 250MB (layers + código, descomprimidos).

Estrategia 3: Container image (para dependencias pesadas)

# Dockerfile
FROM public.ecr.aws/lambda/python:3.11

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY handler.py .

CMD ["handler.handler"]

# requirements.txt
openai>=1.0.0
langchain>=0.1.0
langchain-openai>=0.0.5

Ventaja: hasta 10GB de imagen. Sin límite práctico de dependencias. Trade-off: cold starts más largos (se cubre en cápsula 03).

Cuándo usar cada estrategia

Environment Variables y Secrets

Environment variables en Lambda

Lambda permite configurar environment variables que tu código lee con os.environ:

# En handler.py
import os

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
MODEL_NAME = os.environ.get("MODEL_NAME", "gpt-4o-mini")
MAX_TOKENS = int(os.environ.get("MAX_TOKENS", "500"))
ENVIRONMENT = os.environ.get("ENVIRONMENT", "production")

# Configurar via AWS CLI
aws lambda update-function-configuration \
    --function-name ai-endpoint \
    --environment "Variables={
        OPENAI_API_KEY=sk-xxx,
        MODEL_NAME=gpt-4o-mini,
        MAX_TOKENS=500,
        ENVIRONMENT=production
    }"

Secrets: no hardcodees API keys

Las environment variables de Lambda están encriptadas at rest, pero son visibles en la consola AWS para cualquiera con acceso a la función. Para secrets sensibles (API keys de LLMs), tienes opciones más seguras:

Opción 1: Environment variables (aceptable para la mayoría de casos)

# Aceptable si controlas el acceso IAM a la función
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

Opción 2: AWS Secrets Manager (producción)

import boto3
import json

def get_secret(secret_name):
    """Obtiene secret de AWS Secrets Manager."""
    sm_client = boto3.client("secretsmanager")
    response = sm_client.get_secret_value(SecretId=secret_name)
    return json.loads(response["SecretString"])

# Fuera del handler — se ejecuta solo en cold start
secrets = get_secret("ai-endpoint/openai")
client = OpenAI(api_key=secrets["api_key"])

Opción 3: AWS SSM Parameter Store (gratuito, suficiente)

import boto3

def get_parameter(name):
    """Obtiene parámetro de SSM Parameter Store."""
    ssm = boto3.client("ssm")
    response = ssm.get_parameter(Name=name, WithDecryption=True)
    return response["Parameter"]["Value"]

# Fuera del handler
api_key = get_parameter("/ai-endpoint/openai-api-key")
client = OpenAI(api_key=api_key)

Comparación de opciones para secrets

Para este módulo, usamos environment variables. En producción real, SSM Parameter Store es el sweet spot de coste y seguridad.

Invocación: Síncrona vs Asíncrona

Síncrona (RequestResponse)

El cliente espera la respuesta. Lambda ejecuta, y cuando termina, retorna el resultado directamente.

Cliente → API Gateway → Lambda → LLM → Lambda → API Gateway → Cliente
                                                                 ↑
                                              El cliente espera aquí

# Invocación síncrona via CLI
aws lambda invoke \
    --function-name ai-endpoint \
    --invocation-type RequestResponse \
    --payload '{"body": "{\"prompt\": \"¿Qué es serverless?\"}"}' \
    response.json

cat response.json
# {"statusCode": 200, "body": "{\"answer\": \"Serverless ejecuta...\"}"}

Cuándo usar para AI: Chatbots, APIs de pregunta-respuesta, cualquier caso donde el usuario espera la respuesta inmediatamente.

Asíncrona (Event)

El cliente envía la request y recibe un "OK, lo estoy procesando" inmediatamente. Lambda ejecuta en background.

Cliente → API Gateway → Lambda → (202 Accepted)
                          ↓
                    Procesa en background
                          ↓
                    Guarda resultado (S3, DB, webhook)

# Invocación asíncrona via CLI
aws lambda invoke \
    --function-name ai-endpoint \
    --invocation-type Event \
    --payload '{"body": "{\"prompt\": \"Genera un reporte largo\"}"}' \
    response.json

# Response inmediato: StatusCode 202 (sin esperar resultado)

Cuándo usar para AI:

Regla de oro: Si el usuario espera la pantalla, síncrona. Si el usuario puede continuar y recibir notificación después, asíncrona.

Async con destinos (destination)

Lambda puede enviar el resultado a otro servicio automáticamente:

# Configurar destino para invocaciones async exitosas
aws lambda put-function-event-invoke-config \
    --function-name ai-endpoint \
    --destination-config '{
        "OnSuccess": {
            "Destination": "arn:aws:sqs:us-east-1:123456:results-queue"
        },
        "OnFailure": {
            "Destination": "arn:aws:sqs:us-east-1:123456:dead-letter-queue"
        }
    }'

SAM Template: Definiendo la Infraestructura

template.yaml básico

AWS SAM (Serverless Application Model) define tu Lambda + API Gateway como código:

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: AI Endpoint — Lambda que invoca LLM

Globals:
  Function:
    Timeout: 60
    MemorySize: 512
    Runtime: python3.11

Resources:
  AIEndpointFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: handler.handler
      CodeUri: lambda-function/
      Environment:
        Variables:
          OPENAI_API_KEY: !Ref OpenAIApiKey
          MODEL_NAME: gpt-4o-mini
          MAX_TOKENS: "500"
      Events:
        AskAPI:
          Type: Api
          Properties:
            Path: /ask
            Method: post

Parameters:
  OpenAIApiKey:
    Type: String
    NoEcho: true
    Description: OpenAI API Key

Outputs:
  ApiUrl:
    Description: URL del API endpoint
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/ask"

Testing local con SAM

# Build
sam build

# Invocar localmente (sin AWS)
echo '{"body": "{\"prompt\": \"¿Qué es Lambda?\"}"}' | \
  sam local invoke AIEndpointFunction \
  --env-vars env.json

# API local (simula API Gateway)
sam local start-api
# http://localhost:3000/ask disponible

# Test
curl -X POST http://localhost:3000/ask \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explica serverless"}'

// env.json — Variables para testing local
{
  "AIEndpointFunction": {
    "OPENAI_API_KEY": "sk-tu-key-aqui",
    "MODEL_NAME": "gpt-4o-mini",
    "MAX_TOKENS": "500"
  }
}

Comparación: Lambda vs FastAPI Server para AI

Troubleshooting

Problema 1: "ModuleNotFoundError: No module named 'openai'"

Tu función no encuentra la dependencia. El paquete no está incluido en el deployment.

# Verifica que empaquetaste las dependencias
unzip -l deployment.zip | grep openai
# Debe mostrar archivos de openai/

# Si usas layers, verifica que el layer está adjunto
aws lambda get-function-configuration \
    --function-name ai-endpoint \
    --query 'Layers'

Problema 2: "Task timed out after X seconds"

Lambda terminó antes de que el LLM respondiera.

# Verifica el timeout de Lambda vs el timeout del LLM
# Lambda timeout: 60s (configurado en template.yaml)
# LLM timeout: debe ser < Lambda timeout

remaining = context.get_remaining_time_in_millis()
# Si remaining < 10000 (10s), no hagas la llamada al LLM

Problema 3: "body debe ser string, no dict"

API Gateway espera body como JSON string, no como dict Python.

# ❌ INCORRECTO
return {"statusCode": 200, "body": {"answer": "..."}}

# ✅ CORRECTO
return {"statusCode": 200, "body": json.dumps({"answer": "..."})}

Problema 4: "CORS error en el frontend"

Falta el header Access-Control-Allow-Origin en el response.

# Incluye CORS headers en TODOS los responses (éxito y error)
return {
    "statusCode": 200,
    "headers": {
        "Content-Type": "application/json",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "POST, OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type, x-api-key",
    },
    "body": json.dumps(result),
}

Problema 5: "Environment variable OPENAI_API_KEY not found"

La variable no está configurada en la función Lambda.

# Verifica variables configuradas
aws lambda get-function-configuration \
    --function-name ai-endpoint \
    --query 'Environment.Variables'

# Para testing local con SAM, crea env.json
# Para producción, usa aws lambda update-function-configuration

Ejercicios Prácticos

Ejercicio 1: Lambda con sistema de prompts

Modifica el handler para que acepte un system_prompt además del prompt del usuario. El system prompt configura el comportamiento del LLM (por ejemplo, "Responde siempre en español y en máximo 2 frases").

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

    prompt = body.get("prompt", "").strip()
    system_prompt = body.get(
        "system_prompt",
        "You are a helpful assistant. Respond concisely."
    )

    if not prompt:
        return _error_response(400, "Field 'prompt' is required")

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": prompt},
    ]

    try:
        response = client.chat.completions.create(
            model=os.environ.get("MODEL_NAME", "gpt-4o-mini"),
            messages=messages,
            max_tokens=int(os.environ.get("MAX_TOKENS", "500")),
        )
    except Exception as e:
        return _error_response(502, f"LLM error: {str(e)}")

    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        "body": json.dumps({
            "answer": response.choices[0].message.content,
            "tokens_used": response.usage.total_tokens,
        }),
    }

Ejercicio 2: Timeout-aware handler

Implementa un handler que verifica el tiempo restante de Lambda antes de llamar al LLM. Si quedan menos de 15 segundos, retorna un error en lugar de hacer la llamada (que probablemente fallaría por timeout).

MIN_REMAINING_MS = 15000

def handler(event, context):
    remaining_ms = context.get_remaining_time_in_millis()

    if remaining_ms < MIN_REMAINING_MS:
        return _error_response(
            503,
            f"Insufficient time remaining: {remaining_ms}ms. "
            f"Minimum required: {MIN_REMAINING_MS}ms"
        )

    try:
        body = json.loads(event.get("body", "{}"))
    except json.JSONDecodeError:
        return _error_response(400, "Invalid JSON")

    prompt = body.get("prompt", "").strip()
    if not prompt:
        return _error_response(400, "Field 'prompt' is required")

    llm_timeout_s = (remaining_ms - 5000) / 1000

    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=500,
            timeout=llm_timeout_s,
        )
    except Exception as e:
        return _error_response(502, f"LLM error: {str(e)}")

    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        "body": json.dumps({
            "answer": response.choices[0].message.content,
            "remaining_ms_at_start": remaining_ms,
            "llm_timeout_used": llm_timeout_s,
        }),
    }

Ejercicio 3: Multi-model handler

Crea un handler que soporte múltiples modelos (OpenAI y Anthropic) según un parámetro provider en el request. Usa el mismo endpoint para ambos.

import json
import os
from openai import OpenAI
from anthropic import Anthropic

openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
anthropic_client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])


def _invoke_openai(prompt, max_tokens):
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tokens,
    )
    return {
        "answer": response.choices[0].message.content,
        "provider": "openai",
        "model": "gpt-4o-mini",
        "tokens_used": response.usage.total_tokens,
    }


def _invoke_anthropic(prompt, max_tokens):
    response = anthropic_client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
    )
    return {
        "answer": response.content[0].text,
        "provider": "anthropic",
        "model": "claude-3-haiku-20240307",
        "tokens_used": response.usage.input_tokens + response.usage.output_tokens,
    }


PROVIDERS = {
    "openai": _invoke_openai,
    "anthropic": _invoke_anthropic,
}


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

    prompt = body.get("prompt", "").strip()
    provider = body.get("provider", "openai").lower()
    max_tokens = body.get("max_tokens", 500)

    if not prompt:
        return _error_response(400, "Field 'prompt' is required")

    if provider not in PROVIDERS:
        return _error_response(
            400,
            f"Provider '{provider}' not supported. Use: {list(PROVIDERS.keys())}"
        )

    try:
        result = PROVIDERS[provider](prompt, max_tokens)
    except Exception as e:
        return _error_response(502, f"{provider} error: {str(e)}")

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

Ejercicio 4: Empaqueta tu Lambda con Layer

Crea un Lambda Layer que contenga el SDK de OpenAI. Luego crea un zip de deployment que solo contenga tu handler.py. El objetivo es separar código de dependencias.

# Paso 1: Crear el layer
mkdir -p layer/python/lib/python3.11/site-packages
pip install openai -t layer/python/lib/python3.11/site-packages/

cd layer
zip -r ../openai-layer.zip python/
cd ..

# Verificar tamaño
ls -lh openai-layer.zip
# ~5MB esperado

# Paso 2: Publicar layer (en AWS)
aws lambda publish-layer-version \
    --layer-name openai-sdk \
    --description "OpenAI Python SDK for Lambda" \
    --zip-file fileb://openai-layer.zip \
    --compatible-runtimes python3.11 python3.12

# Guardar el ARN del output:
# arn:aws:lambda:us-east-1:123456789:layer:openai-sdk:1

# Paso 3: Empaquetar solo el handler
zip deployment.zip handler.py
# ~2KB

# Paso 4: Crear función con layer
aws lambda create-function \
    --function-name ai-endpoint \
    --runtime python3.11 \
    --handler handler.handler \
    --zip-file fileb://deployment.zip \
    --role arn:aws:iam::123456789:role/lambda-execution-role \
    --layers arn:aws:lambda:us-east-1:123456789:layer:openai-sdk:1 \
    --timeout 60 \
    --memory-size 512

# Paso 5: Verificar
aws lambda invoke \
    --function-name ai-endpoint \
    --payload '{"body": "{\"prompt\": \"test\"}"}' \
    output.json

Resumen

• El handler Lambda recibe event (datos) y context (metadata del runtime) — todo lo que necesitas para procesar la invocación.
• Inicializa clientes fuera del handler para reutilizarlos en warm starts. El código fuera del handler se ejecuta una sola vez por container.
• Packaging de dependencias tiene tres estrategias: zip (~50MB límite), layers (compartidas entre funciones), container (hasta 10GB).
• Para AI con openai/anthropic SDKs, zip o layers son suficientes. Para langchain o librerías de ML pesadas, necesitas container.
• Environment variables son la forma más simple de gestionar config. Para secrets en producción, SSM Parameter Store es el sweet spot.
• Invocación síncrona para cuando el usuario espera respuesta. Asíncrona para procesamiento en background.
• SAM CLI permite testing local sin cuenta AWS — sam local invoke y sam local start-api.

Recursos Adicionales

1. AWS Lambda Python Handler — Referencia oficial del handler
2. Lambda Layers — Documentación de Lambda Layers
3. Lambda Environment Variables — Config de environment variables
4. AWS SAM CLI — Local Testing — Testing local con SAM
5. Lambda Invocation Types — Sync vs Async invocation
6. OpenAI Python SDK — SDK oficial de OpenAI
7. AWS Secrets Manager vs SSM — Comparativa de gestión de secrets
8. Lambda Quotas — Límites de Lambda (tamaño, timeout, memory)