Módulo 4: LocalStack — AWS Local Development

5. Pipeline S3 + Lambda Local

Siguiente →

Descripción

En esta cápsula vas a conectar S3 y Lambda en un pipeline AI completo corriendo en LocalStack. Lambda lee un documento de S3, lo procesa con un LLM, y escribe el resultado de vuelta a S3. Es el flujo completo: input → procesamiento → output — todo local, todo gratis. Este pipeline es el artefacto central del módulo y la base del proyecto final (cápsula 08).

Contexto: En las cápsulas anteriores aprendiste S3 local (cápsula 03) y Lambda local (cápsula 04) por separado. Ahora los combinas. Este patrón — almacenamiento + computación + LLM — es el building block de la mayoría de sistemas AI en producción. Dominarlo localmente te da confianza para implementarlo en AWS real (Módulo 5).

Arquitectura del Pipeline

El flujo completo

┌─────────────────────────────────────────────────┐
│                 LocalStack (localhost:4566)       │
│                                                   │
│   ┌──────────────┐     ┌──────────────────────┐  │
│   │  S3 Bucket   │     │    Lambda Function    │  │
│   │  ai-input    │────→│    ai-processor       │  │
│   │              │     │                        │  │
│   │ documents/   │     │  1. Lee de S3          │  │
│   │  doc.txt     │     │  2. Llama a OpenAI     │  │
│   └──────────────┘     │  3. Escribe a S3       │  │
│                         │                        │  │
│   ┌──────────────┐     └──────────────────────┘  │
│   │  S3 Bucket   │←────────────────────────────  │
│   │  ai-output   │                                │
│   │              │                                │
│   │ results/     │                                │
│   │  doc-result  │                                │
│   └──────────────┘                                │
└─────────────────────────────────────────────────┘

Los tres componentes

1. S3 Input (bucket: ai-input)
   └── Almacena documentos a procesar (textos, prompts, datos)

2. Lambda Processor (función: ai-processor)
   ├── Lee documento de S3 (boto3.get_object)
   ├── Procesa con LLM (OpenAI API)
   └── Escribe resultado a S3 (boto3.put_object)

3. S3 Output (bucket: ai-output)
   └── Almacena resultados del procesamiento

El Lambda Processor

Handler completo

# lambda/processor.py
import json
import os
import time
import boto3
from openai import OpenAI

AWS_ENDPOINT_URL = os.environ.get("AWS_ENDPOINT_URL", None)

openai_client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY", ""),
    timeout=50,
    max_retries=1,
)

MODEL_NAME = os.environ.get("MODEL_NAME", "gpt-4o-mini")
INPUT_BUCKET = os.environ.get("INPUT_BUCKET", "ai-input")
OUTPUT_BUCKET = os.environ.get("OUTPUT_BUCKET", "ai-output")
SYSTEM_PROMPT = os.environ.get(
    "SYSTEM_PROMPT",
    "Analiza el siguiente documento. Extrae: título, resumen (2-3 oraciones), "
    "y una lista de puntos clave. Responde en formato JSON."
)


def get_s3_client():
    """Crea cliente S3 — apunta a LocalStack si AWS_ENDPOINT_URL está configurado."""
    kwargs = {
        "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
    }
    if AWS_ENDPOINT_URL:
        kwargs["endpoint_url"] = AWS_ENDPOINT_URL
        kwargs["aws_access_key_id"] = "test"
        kwargs["aws_secret_access_key"] = "test"
    return boto3.client("s3", **kwargs)


def handler(event, context):
    start_time = time.time()
    s3 = get_s3_client()

    # 1. Obtener key del documento a procesar
    document_key = _get_document_key(event)
    if not document_key:
        return _response(400, {"error": "document_key is required"})

    # 2. Leer documento de S3
    try:
        response = s3.get_object(Bucket=INPUT_BUCKET, Key=document_key)
        document_text = response["Body"].read().decode("utf-8")
    except Exception as e:
        return _response(404, {
            "error": f"Cannot read document: {str(e)}",
            "bucket": INPUT_BUCKET,
            "key": document_key,
        })

    # 3. Procesar con LLM
    try:
        llm_response = openai_client.chat.completions.create(
            model=MODEL_NAME,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": document_text},
            ],
            max_tokens=800,
        )
        analysis = llm_response.choices[0].message.content
        tokens_used = llm_response.usage.total_tokens
    except Exception as e:
        return _response(502, {"error": f"LLM processing failed: {str(e)}"})

    # 4. Construir resultado
    result = {
        "source_document": document_key,
        "analysis": analysis,
        "model": MODEL_NAME,
        "tokens_used": tokens_used,
        "duration_ms": round((time.time() - start_time) * 1000),
        "processed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }

    # 5. Escribir resultado a S3
    doc_name = document_key.split("/")[-1].replace(".", "-")
    output_key = f"results/{doc_name}-analysis.json"

    try:
        s3.put_object(
            Bucket=OUTPUT_BUCKET,
            Key=output_key,
            Body=json.dumps(result, indent=2, ensure_ascii=False),
            ContentType="application/json",
        )
    except Exception as e:
        return _response(500, {"error": f"Cannot write result: {str(e)}"})

    return _response(200, {
        "status": "processed",
        "input": f"s3://{INPUT_BUCKET}/{document_key}",
        "output": f"s3://{OUTPUT_BUCKET}/{output_key}",
        "tokens_used": tokens_used,
        "duration_ms": result["duration_ms"],
    })


def _get_document_key(event):
    """Extrae document_key del evento — soporta invocación directa y API Gateway."""
    if "body" in event:
        try:
            body = json.loads(event["body"])
            return body.get("document_key", "")
        except (json.JSONDecodeError, TypeError):
            pass
    return event.get("document_key", "")


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

Lo importante del handler

Hay tres decisiones de diseño clave:

1. AWS_ENDPOINT_URL controla el target
   ├── Si está definido → usa LocalStack
   └── Si no está → usa AWS real (default de boto3)
   Este es el mecanismo de environment switching (cápsula 06)

2. El handler lee de S3 y escribe a S3
   ├── Input: get_object del bucket ai-input
   ├── Process: OpenAI API
   └── Output: put_object al bucket ai-output
   S3 es el bus de datos entre componentes

3. El resultado incluye metadata
   ├── source_document, model, tokens, duration
   └── Útil para auditoría, debugging, y cost tracking

Configurar el Pipeline

Crear los buckets

# Crear buckets de input y output
awslocal s3 mb s3://ai-input
awslocal s3 mb s3://ai-output

# Verificar
awslocal s3 ls
# ai-input
# ai-output

Subir un documento de prueba

# Crear un documento de texto
cat > data/documento-prueba.txt << 'EOF'
Informe Trimestral Q1 2026 — Departamento de Tecnología

Resumen Ejecutivo:
El departamento de tecnología completó la migración a arquitectura de microservicios
durante Q1 2026. Se redujeron los tiempos de deploy de 4 horas a 15 minutos.
El sistema de AI para clasificación de tickets redujo el tiempo de resolución
en un 40%.

Logros principales:
- Migración completa a Kubernetes (EKS)
- Pipeline CI/CD automatizado con GitHub Actions
- Sistema de clasificación de tickets con GPT-4o-mini
- Reducción de costes cloud en 25% via right-sizing

Próximos pasos:
- Implementar monitoring con Prometheus + Grafana
- Expandir el sistema AI a customer support
- Evaluar migración de base de datos a Aurora Serverless
EOF

# Subir a S3
awslocal s3 cp data/documento-prueba.txt s3://ai-input/documents/informe-q1.txt

# Verificar que está en S3
awslocal s3 ls s3://ai-input/documents/

Empaquetar y desplegar el Lambda processor

# Instalar dependencias
pip install openai boto3 -t lambda/package/ --quiet

# Copiar handler
cp lambda/processor.py lambda/package/

# Empaquetar
cd lambda/package
zip -r ../processor.zip . -q
cd ../..

# Desplegar
awslocal lambda create-function \
  --function-name ai-processor \
  --runtime python3.11 \
  --handler processor.handler \
  --zip-file fileb://lambda/processor.zip \
  --role arn:aws:iam::000000000000:role/lambda-role \
  --timeout 90 \
  --memory-size 768 \
  --environment "Variables={
    OPENAI_API_KEY=${OPENAI_API_KEY},
    MODEL_NAME=gpt-4o-mini,
    INPUT_BUCKET=ai-input,
    OUTPUT_BUCKET=ai-output,
    AWS_ENDPOINT_URL=http://host.docker.internal:4566,
    AWS_DEFAULT_REGION=us-east-1
  }"

La variable AWS_ENDPOINT_URL=http://host.docker.internal:4566 es necesaria porque Lambda en LocalStack corre en un container Docker separado. host.docker.internal resuelve al host desde dentro del container.

Si usas LAMBDA_EXECUTOR=local, el endpoint sería http://localhost:4566.

Ejecutar el Pipeline

Invocación manual

# Invocar el processor con el documento que subimos
awslocal lambda invoke \
  --function-name ai-processor \
  --payload '{"document_key": "documents/informe-q1.txt"}' \
  --cli-binary-format raw-in-base64-out \
  output.json

# Ver resultado de la invocación
cat output.json | python3 -m json.tool

Verificar el resultado en S3

# Listar resultados
awslocal s3 ls s3://ai-output/results/

# Descargar y ver el análisis
awslocal s3 cp s3://ai-output/results/informe-q1-txt-analysis.json - | python3 -m json.tool

Ejecutar desde Python

# run_pipeline.py
import boto3
import json

ENDPOINT = "http://localhost:4566"
CLIENT_KWARGS = {
    "endpoint_url": ENDPOINT,
    "aws_access_key_id": "test",
    "aws_secret_access_key": "test",
    "region_name": "us-east-1",
}

s3 = boto3.client("s3", **CLIENT_KWARGS)
lambda_client = boto3.client("lambda", **CLIENT_KWARGS)


def run_pipeline(document_key):
    """Ejecuta el pipeline completo: S3 → Lambda → S3."""
    print(f"Procesando: {document_key}")

    # Invocar Lambda
    response = lambda_client.invoke(
        FunctionName="ai-processor",
        InvocationType="RequestResponse",
        Payload=json.dumps({"document_key": document_key}),
    )

    result = json.loads(response["Payload"].read())
    body = json.loads(result["body"])

    if result["statusCode"] == 200:
        print(f"  Input:  {body['input']}")
        print(f"  Output: {body['output']}")
        print(f"  Tokens: {body['tokens_used']}")
        print(f"  Duration: {body['duration_ms']}ms")

        # Descargar resultado
        output_key = body["output"].split(f"ai-output/")[1]
        obj = s3.get_object(Bucket="ai-output", Key=output_key)
        analysis = json.loads(obj["Body"].read().decode("utf-8"))
        print(f"  Análisis: {analysis['analysis'][:150]}...")
    else:
        print(f"  ERROR: {body}")

    return body


# Ejecutar
result = run_pipeline("documents/informe-q1.txt")

Batch Processing: Múltiples Documentos

Subir varios documentos

# batch_upload.py
import boto3

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

documents = {
    "documents/email-cliente.txt": (
        "Asunto: Problema con la facturación\n\n"
        "Estimado equipo, llevo 3 meses con un cargo incorrecto en mi cuenta. "
        "El servicio Premium debería costar $49/mes pero me cobran $79. "
        "He contactado soporte 2 veces sin resolución. Necesito que corrijan "
        "el cobro y me reembolsen la diferencia. Gracias, María González."
    ),
    "documents/propuesta-tecnica.txt": (
        "Propuesta: Sistema de Recomendación para E-commerce\n\n"
        "Objetivo: Implementar un sistema de recomendación basado en "
        "embeddings que mejore el CTR en un 20%.\n"
        "Tecnología: FastAPI + Redis + OpenAI Embeddings\n"
        "Timeline: 6 semanas\n"
        "Presupuesto: $15,000 USD\n"
        "Equipo: 2 ML Engineers + 1 Backend Developer"
    ),
    "documents/reporte-bug.txt": (
        "Bug Report #4521\n"
        "Severidad: Alta\n"
        "Componente: API de pagos\n"
        "Descripción: Al procesar pagos con tarjetas internacionales, "
        "el sistema retorna error 500 intermitentemente. Afecta al 15% "
        "de las transacciones internacionales. Logs muestran timeout "
        "en la conexión con el gateway de pagos."
    ),
}

for key, content in documents.items():
    s3.put_object(Bucket="ai-input", Key=key, Body=content.encode("utf-8"))
    print(f"Subido: {key}")

Procesar batch

# batch_process.py
import boto3
import json
import time

ENDPOINT = "http://localhost:4566"
KWARGS = {
    "endpoint_url": ENDPOINT,
    "aws_access_key_id": "test",
    "aws_secret_access_key": "test",
    "region_name": "us-east-1",
}

s3 = boto3.client("s3", **KWARGS)
lambda_client = boto3.client("lambda", **KWARGS)


def process_all_documents():
    """Procesa todos los documentos en ai-input/documents/."""
    response = s3.list_objects_v2(Bucket="ai-input", Prefix="documents/")
    documents = [obj["Key"] for obj in response.get("Contents", [])]

    print(f"Documentos encontrados: {len(documents)}")
    results = []

    for doc_key in documents:
        print(f"\nProcesando: {doc_key}")
        start = time.time()

        resp = lambda_client.invoke(
            FunctionName="ai-processor",
            InvocationType="RequestResponse",
            Payload=json.dumps({"document_key": doc_key}),
        )

        result = json.loads(resp["Payload"].read())
        body = json.loads(result["body"])
        elapsed = round((time.time() - start) * 1000)

        status = "OK" if result["statusCode"] == 200 else "ERROR"
        tokens = body.get("tokens_used", 0)
        print(f"  {status} — {tokens} tokens — {elapsed}ms")

        results.append({
            "document": doc_key,
            "status": status,
            "tokens": tokens,
            "elapsed_ms": elapsed,
        })

    # Resumen
    print("\n" + "=" * 50)
    total_tokens = sum(r["tokens"] for r in results)
    total_time = sum(r["elapsed_ms"] for r in results)
    success = sum(1 for r in results if r["status"] == "OK")
    print(f"Procesados: {success}/{len(results)}")
    print(f"Total tokens: {total_tokens}")
    print(f"Total tiempo: {total_time}ms")
    print(f"Costo estimado (gpt-4o-mini): ~${total_tokens * 0.0000015:.4f}")


process_all_documents()

Verificar Resultados

Listar todos los resultados

# Ver todos los análisis generados
awslocal s3 ls s3://ai-output/results/ --recursive

# Descargar uno específico
awslocal s3 cp s3://ai-output/results/email-cliente-txt-analysis.json - | python3 -m json.tool

Script de verificación

# verify_results.py
import boto3
import json

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

# Listar inputs y outputs
inputs = s3.list_objects_v2(Bucket="ai-input", Prefix="documents/")
outputs = s3.list_objects_v2(Bucket="ai-output", Prefix="results/")

input_keys = [o["Key"] for o in inputs.get("Contents", [])]
output_keys = [o["Key"] for o in outputs.get("Contents", [])]

print(f"Documentos de entrada: {len(input_keys)}")
print(f"Resultados generados:  {len(output_keys)}")

for key in output_keys:
    obj = s3.get_object(Bucket="ai-output", Key=key)
    result = json.loads(obj["Body"].read().decode("utf-8"))
    print(f"\n--- {key} ---")
    print(f"  Fuente:   {result['source_document']}")
    print(f"  Modelo:   {result['model']}")
    print(f"  Tokens:   {result['tokens_used']}")
    print(f"  Duration: {result['duration_ms']}ms")
    print(f"  Análisis: {result['analysis'][:120]}...")

Ejercicios

Ejercicio 1: Pipeline con system prompt desde S3

Modifica el processor para que lea el system prompt de un archivo en S3 (ai-config/prompts/analyzer.txt) en lugar de tenerlo hardcoded. Si el archivo no existe, usa el prompt default.

Ver solución 
# En el handler, agregar función para leer prompt de S3:
CONFIG_BUCKET = os.environ.get("CONFIG_BUCKET", "ai-config")
PROMPT_KEY = os.environ.get("PROMPT_KEY", "prompts/analyzer.txt")

def get_system_prompt(s3):
    """Lee system prompt de S3, con fallback a default."""
    default = (
        "Analiza el siguiente documento. Extrae: título, resumen, "
        "y puntos clave. Responde en JSON."
    )
    try:
        response = s3.get_object(Bucket=CONFIG_BUCKET, Key=PROMPT_KEY)
        return response["Body"].read().decode("utf-8")
    except Exception:
        return default

# En handler(), reemplazar SYSTEM_PROMPT constante:
def handler(event, context):
    s3 = get_s3_client()
    system_prompt = get_system_prompt(s3)
    # ... usar system_prompt en la llamada a OpenAI
# Setup: crear el bucket de config y subir el prompt
awslocal s3 mb s3://ai-config
echo "Eres un analista de documentos. Extrae: categoría del documento, nivel de urgencia (alto/medio/bajo), y las 3 acciones requeridas. Responde en JSON." | \
  awslocal s3 cp - s3://ai-config/prompts/analyzer.txt

# Redesplegar y probar

Ejercicio 2: Pipeline con tracking de procesamiento

Agrega al pipeline un archivo de tracking: cada vez que se procesa un documento, agrega una línea a ai-output/tracking/log.jsonl (JSON Lines) con el document key, timestamp, tokens, y status.

Ver solución 
import datetime

def append_to_tracking(s3, document_key, tokens, status, duration_ms):
    """Agrega entrada al log de tracking en S3."""
    tracking_key = "tracking/log.jsonl"
    
    entry = json.dumps({
        "document": document_key,
        "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
        "tokens": tokens,
        "status": status,
        "duration_ms": duration_ms,
    })

    # Leer log existente (si hay)
    try:
        existing = s3.get_object(Bucket=OUTPUT_BUCKET, Key=tracking_key)
        current_log = existing["Body"].read().decode("utf-8")
    except Exception:
        current_log = ""

    # Agregar nueva entrada
    updated_log = current_log + entry + "\n"

    s3.put_object(
        Bucket=OUTPUT_BUCKET,
        Key=tracking_key,
        Body=updated_log.encode("utf-8"),
        ContentType="application/x-ndjson",
    )

# Llamar al final del handler:
# append_to_tracking(s3, document_key, tokens_used, "success", duration_ms)
# Después de procesar varios documentos:
awslocal s3 cp s3://ai-output/tracking/log.jsonl -
# {"document": "documents/informe-q1.txt", "timestamp": "...", "tokens": 342, ...}
# {"document": "documents/email-cliente.txt", "timestamp": "...", "tokens": 215, ...}

Ejercicio 3: Pipeline con verificación de documentos duplicados

Antes de procesar un documento, verifica si ya existe un resultado en ai-output. Si ya fue procesado, retorna el resultado existente sin re-procesar (evitando gasto innecesario de tokens).

Ver solución 
def check_existing_result(s3, document_key):
    """Verifica si un documento ya fue procesado."""
    doc_name = document_key.split("/")[-1].replace(".", "-")
    output_key = f"results/{doc_name}-analysis.json"

    try:
        response = s3.get_object(Bucket=OUTPUT_BUCKET, Key=output_key)
        existing = json.loads(response["Body"].read().decode("utf-8"))
        return output_key, existing
    except Exception:
        return None, None

# En handler(), antes de llamar al LLM:
def handler(event, context):
    s3 = get_s3_client()
    document_key = _get_document_key(event)

    # Verificar si ya existe resultado
    force = False
    if "body" in event:
        try:
            body = json.loads(event["body"])
            force = body.get("force", False)
        except Exception:
            pass

    if not force:
        existing_key, existing_result = check_existing_result(s3, document_key)
        if existing_result:
            return _response(200, {
                "status": "already_processed",
                "output": f"s3://{OUTPUT_BUCKET}/{existing_key}",
                "tokens_used": existing_result.get("tokens_used", 0),
                "message": "Resultado existente retornado (use force=true para re-procesar)",
            })

    # ... continuar con procesamiento normal
# Primera invocación: procesa
awslocal lambda invoke --function-name ai-processor \
  --payload '{"document_key": "documents/informe-q1.txt"}' \
  --cli-binary-format raw-in-base64-out output.json
# status: "processed"

# Segunda invocación: retorna existente
awslocal lambda invoke --function-name ai-processor \
  --payload '{"document_key": "documents/informe-q1.txt"}' \
  --cli-binary-format raw-in-base64-out output.json
# status: "already_processed"

# Forzar re-procesamiento:
awslocal lambda invoke --function-name ai-processor \
  --payload '{"body": "{\"document_key\": \"documents/informe-q1.txt\", \"force\": true}"}' \
  --cli-binary-format raw-in-base64-out output.json
# status: "processed"

Ejercicio 4: Script de pipeline end-to-end

Crea un script bash que ejecute el pipeline completo: crea buckets, sube un documento, despliega Lambda, invoca, descarga resultado, y muestra el análisis. Un solo comando para demo.

Ver solución 
#!/bin/bash
# scripts/demo-pipeline.sh
set -e

echo "=== Pipeline S3 + Lambda Demo ==="
echo ""

echo "1. Verificando LocalStack..."
curl -s http://localhost:4566/_localstack/health > /dev/null || { echo "LocalStack no disponible"; exit 1; }
echo "   OK"

echo "2. Creando buckets..."
awslocal s3 mb s3://ai-input 2>/dev/null || true
awslocal s3 mb s3://ai-output 2>/dev/null || true
echo "   Buckets: ai-input, ai-output"

echo "3. Subiendo documento de prueba..."
cat > /tmp/demo-doc.txt << 'CONTENT'
Propuesta de Proyecto: Chatbot para Soporte Técnico

El equipo propone implementar un chatbot basado en GPT-4o-mini para automatizar
el 60% de las consultas de soporte técnico. El chatbot usará RAG sobre la
documentación existente. Presupuesto: $8,000. Timeline: 4 semanas.
CONTENT
awslocal s3 cp /tmp/demo-doc.txt s3://ai-input/documents/demo.txt
echo "   Documento subido"

echo "4. Desplegando Lambda processor..."
if awslocal lambda get-function --function-name ai-processor > /dev/null 2>&1; then
  awslocal lambda update-function-code \
    --function-name ai-processor \
    --zip-file fileb://lambda/processor.zip > /dev/null
else
  awslocal lambda create-function \
    --function-name ai-processor \
    --runtime python3.11 \
    --handler processor.handler \
    --zip-file fileb://lambda/processor.zip \
    --role arn:aws:iam::000000000000:role/lambda-role \
    --timeout 90 --memory-size 768 \
    --environment "Variables={OPENAI_API_KEY=${OPENAI_API_KEY},MODEL_NAME=gpt-4o-mini,INPUT_BUCKET=ai-input,OUTPUT_BUCKET=ai-output,AWS_ENDPOINT_URL=http://host.docker.internal:4566}" > /dev/null
fi
echo "   Lambda desplegada"

echo "5. Ejecutando pipeline..."
awslocal lambda invoke \
  --function-name ai-processor \
  --payload '{"document_key": "documents/demo.txt"}' \
  --cli-binary-format raw-in-base64-out \
  /tmp/pipeline-result.json > /dev/null

STATUS=$(cat /tmp/pipeline-result.json | python3 -c "import json,sys; print(json.load(sys.stdin)['statusCode'])")

if [ "$STATUS" = "200" ]; then
  echo "   Pipeline ejecutado exitosamente"
  
  echo ""
  echo "6. Resultado del análisis:"
  echo "   ========================"
  awslocal s3 cp s3://ai-output/results/demo-txt-analysis.json - 2>/dev/null | \
    python3 -c "
import json, sys
data = json.load(sys.stdin)
print(f\"   Modelo: {data['model']}\")
print(f\"   Tokens: {data['tokens_used']}\")
print(f\"   Duración: {data['duration_ms']}ms\")
print(f\"   Análisis:\")
print(f\"   {data['analysis'][:300]}...\")
"
else
  echo "   ERROR — Status: $STATUS"
  cat /tmp/pipeline-result.json | python3 -m json.tool
fi

echo ""
echo "=== Pipeline completo ==="

Troubleshooting

"Lambda no puede conectar a S3 en LocalStack"

# Si LAMBDA_EXECUTOR=docker, Lambda corre en un container separado
# Necesita usar host.docker.internal para llegar a LocalStack

# Verificar el endpoint configurado:
awslocal lambda get-function-configuration \
  --function-name ai-processor \
  --query 'Environment.Variables.AWS_ENDPOINT_URL'

# Debe ser: http://host.docker.internal:4566
# NO: http://localhost:4566 (localhost dentro del container Lambda ≠ tu máquina)

# Si usas LAMBDA_EXECUTOR=local, usa http://localhost:4566

"El resultado en S3 está vacío o corrupto"

# Verifica que el body se serializa como string JSON:
s3.put_object(
    Bucket=bucket,
    Key=key,
    Body=json.dumps(result, ensure_ascii=False),  # string, no dict
    ContentType="application/json",
)

"Timeout al invocar Lambda"

# El timeout default de Lambda en LocalStack es 3s
# Para AI workloads necesitas más:
awslocal lambda update-function-configuration \
  --function-name ai-processor \
  --timeout 120

# El timeout del cliente Lambda también importa:
# boto3 tiene timeout de conexión de 60s por default

"Lambda procesa pero el resultado no aparece en S3"

# Verifica que los buckets existen:
awslocal s3 ls

# Verifica que no hay error silencioso en el handler
# Agrega logging:
import logging
logger = logging.getLogger()
logger.setLevel("DEBUG")

Resumen

  • El pipeline S3 → Lambda → S3 es el building block de sistemas AI en cloud: input, procesamiento con LLM, output.
  • AWS_ENDPOINT_URL en el handler controla si Lambda habla con LocalStack o AWS — sin cambiar una línea de código.
  • Batch processing aplica el pipeline a múltiples documentos secuencialmente, con tracking de tokens y costes.
  • Deduplicación evita re-procesar documentos ya analizados — ahorra tokens y dinero.
  • El endpoint de S3 desde Lambda depende del LAMBDA_EXECUTOR: host.docker.internal para Docker, localhost para local.
  • Todo este pipeline corre localmente, gratis — el mismo flujo que en AWS, pero sin coste.

Recursos Adicionales

  1. S3 Event Notifications — Trigger Lambda automáticamente cuando se sube un archivo a S3
  2. Lambda + S3 Tutorial (AWS) — Tutorial oficial de Lambda con S3
  3. LocalStack S3 + Lambda — S3 en LocalStack
  4. boto3 S3 Transfers — Upload/download eficiente
  5. JSON Lines Format — Formato de logging estructurado
  6. Lambda Environment Variables — Gestión de variables
← Anterior4. Lambda Local con LocalStackSiguiente →6. Environment Switching