Módulo 4: LocalStack — AWS Local Development

8. Proyecto: LocalStack AI Pipeline

Siguiente →

Descripción del proyecto

Este es el proyecto integrador del Módulo 4. Vas a construir un pipeline AI completo corriendo enteramente en LocalStack dentro de Docker Compose: un bucket S3 recibe documentos, una Lambda los procesa con un LLM, y el resultado se escribe a otro bucket S3. Todo orquestado con un script Python que demuestra el flujo end-to-end. Es la culminación de todo lo que aprendiste en este módulo: LocalStack setup, S3 local, Lambda local, pipeline, environment switching, y debugging.

Por qué importa: Este pipeline es el artefacto central de la Phase 2. En el Módulo 5, trabajarás con los mismos servicios (S3, Lambda) pero en contexto AWS real. En el Módulo 6, el código de este pipeline se abstrae para correr contra LocalStack O AWS con el mismo codebase — y la transición será trivial porque el environment switching ya está implementado. En el Módulo 8 (Proyecto Integrador), LocalStack sigue siendo tu entorno de desarrollo local.

Objetivo del proyecto

Producir un LocalStack AI Pipeline funcional que:

  1. Corre enteramente en LocalStack dentro de Docker Compose
  2. Tiene dos buckets S3: ai-input (documentos) y ai-output (resultados)
  3. Tiene una Lambda ai-document-processor que lee de S3, procesa con GPT-4o-mini, y escribe a S3
  4. Soporta environment switching: el mismo código funciona contra LocalStack o AWS
  5. Tiene scripts de setup, ejecución, y verificación
  6. Incluye un script de diagnóstico que verifica el health de todo el pipeline
  7. Se inicializa automáticamente con init scripts de LocalStack

Recap del Módulo

CápsulaConceptoLo usas en el proyecto
02LocalStack en Docker ComposeCompose file con LocalStack como servicio
03S3 local con boto3Buckets de input/output, upload/download
04Lambda en LocalStackDeploy y ejecución de la función processor
05Pipeline S3 + LambdaFlujo completo input → process → output
06Environment switchingAWS_ENDPOINT_URL controla el target
07Debugging LocalStackScripts de diagnóstico y recovery

Especificaciones Técnicas

Arquitectura

┌─────────────────────────────────────────────────────────────┐
│                     Docker Compose                           │
│                                                               │
│  ┌──────────┐   ┌──────────┐   ┌──────────────────────────┐ │
│  │  FastAPI  │   │  Redis   │   │      LocalStack          │ │
│  │  (api)    │   │ (cache)  │   │                          │ │
│  │  :8000    │   │  :6379   │   │  S3: ai-input            │ │
│  └──────────┘   └──────────┘   │  S3: ai-output           │ │
│                                 │  Lambda: ai-doc-processor │ │
│                                 │  :4566                    │ │
│                                 └──────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

Pipeline Flow:
  Upload doc → S3 (ai-input)
       ↓
  Invoke Lambda (ai-document-processor)
       ↓
  Lambda reads from S3 → GPT-4o-mini → writes to S3
       ↓
  Download result ← S3 (ai-output)

Archivos requeridos

localstack-ai-pipeline/
├── api/
│   ├── main.py                  # FastAPI con endpoints del M2
│   ├── config.py                # Settings con Pydantic
│   ├── aws_clients.py           # Factory de clientes AWS
│   ├── requirements.txt         # Dependencias
│   └── Dockerfile               # Container de la API
├── lambda/
│   ├── processor.py             # Lambda handler
│   ├── requirements.txt         # Dependencias Lambda (openai, boto3)
│   └── package/                 # Directorio de packaging
├── scripts/
│   ├── setup.sh                 # Setup completo del pipeline
│   ├── deploy-lambda.sh         # Deploy/update de Lambda
│   ├── run-pipeline.py          # Ejecuta el pipeline end-to-end
│   ├── diagnose.sh              # Diagnóstico del sistema
│   └── demo.sh                  # Demo completa en un comando
├── init-scripts/
│   └── setup.sh                 # Init script de LocalStack (crea buckets)
├── data/
│   └── sample-documents/        # Documentos de prueba
├── docker-compose.yml           # Compose completo
├── .env                         # Variables (OPENAI_API_KEY)
├── .env.example                 # Template
└── .gitignore                   # Ignora .env, package/, *.zip

Código Completo

docker-compose.yml

services:
  api:
    build:
      context: ./api
    ports:
      - "${API_PORT:-8000}:8000"
    env_file:
      - .env
    environment:
      - REDIS_URL=redis://cache:6379
      - AWS_ENDPOINT_URL=http://localstack:4566
      - AWS_ACCESS_KEY_ID=test
      - AWS_SECRET_ACCESS_KEY=test
      - AWS_DEFAULT_REGION=us-east-1
    depends_on:
      cache:
        condition: service_healthy
      localstack:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 15s
    restart: unless-stopped

  cache:
    image: redis:7-alpine
    command: redis-server --appendonly yes --maxmemory 128mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
    restart: unless-stopped

  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=s3,lambda
      - DEBUG=0
      - LAMBDA_EXECUTOR=docker
      - DOCKER_HOST=unix:///var/run/docker.sock
    volumes:
      - localstack_data:/var/lib/localstack
      - /var/run/docker.sock:/var/run/docker.sock
      - ./init-scripts:/etc/localstack/init/ready.d
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 20s
    restart: unless-stopped

volumes:
  redis_data:
  localstack_data:

init-scripts/setup.sh

#!/bin/bash
echo "=== LocalStack Init: Creando recursos ==="

awslocal s3 mb s3://ai-input 2>/dev/null || true
awslocal s3 mb s3://ai-output 2>/dev/null || true

echo "Buckets creados:"
awslocal s3 ls

echo "=== LocalStack Init: Completo ==="

api/aws_clients.py

import boto3
import os
import logging

logger = logging.getLogger(__name__)

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


def _base_kwargs():
    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"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
        kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")
    return kwargs


def get_s3_client():
    client = boto3.client("s3", **_base_kwargs())
    target = f"LocalStack ({AWS_ENDPOINT_URL})" if AWS_ENDPOINT_URL else "AWS"
    logger.debug(f"S3 client → {target}")
    return client


def get_lambda_client():
    client = boto3.client("lambda", **_base_kwargs())
    target = f"LocalStack ({AWS_ENDPOINT_URL})" if AWS_ENDPOINT_URL else "AWS"
    logger.debug(f"Lambda client → {target}")
    return client


def is_local():
    return AWS_ENDPOINT_URL is not None

api/config.py

from pydantic_settings import BaseSettings
from pydantic import field_validator
from typing import Optional


class Settings(BaseSettings):
    openai_api_key: str
    redis_url: str = "redis://cache:6379"
    environment: str = "development"
    log_level: str = "debug"
    cache_ttl: int = 3600
    model_name: str = "gpt-4o-mini"
    max_tokens: int = 500
    aws_endpoint_url: Optional[str] = None
    s3_input_bucket: str = "ai-input"
    s3_output_bucket: str = "ai-output"

    @field_validator("openai_api_key")
    @classmethod
    def validate_key(cls, v):
        if not v or "replace" in v.lower():
            raise ValueError("Set a real OPENAI_API_KEY in .env")
        return v

    @property
    def is_local(self) -> bool:
        return self.aws_endpoint_url is not None

    class Config:
        env_file = ".env"


settings = Settings()

api/main.py

import hashlib
import json
import logging
import time
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from openai import OpenAI
import redis

from config import settings
from aws_clients import get_s3_client, get_lambda_client, is_local

logging.basicConfig(level=getattr(logging, settings.log_level.upper()))
logger = logging.getLogger(__name__)

app = FastAPI(title="AI Pipeline API", version="1.0.0")
openai_client = OpenAI(api_key=settings.openai_api_key, timeout=30.0)
cache = redis.Redis.from_url(settings.redis_url, decode_responses=True)


class AskRequest(BaseModel):
    prompt: str
    max_tokens: int = 500
    use_cache: bool = True


class ProcessRequest(BaseModel):
    document_key: str
    force: bool = False


@app.get("/health")
def health():
    checks = {"api": "up"}

    try:
        cache.ping()
        checks["redis"] = "up"
    except Exception:
        checks["redis"] = "down"

    try:
        s3 = get_s3_client()
        s3.list_buckets()
        checks["localstack_s3"] = "up"
    except Exception:
        checks["localstack_s3"] = "down"

    overall = "healthy" if all(v == "up" for v in checks.values()) else "degraded"
    status_code = 200 if overall == "healthy" else 503

    return JSONResponse(
        status_code=status_code,
        content={
            "status": overall,
            "environment": "localstack" if is_local() else "aws",
            "services": checks,
        },
    )


@app.post("/ask")
def ask(request: AskRequest):
    cache_key = f"ask:{hashlib.md5(f'{request.prompt}:{request.max_tokens}'.encode()).hexdigest()}"

    if request.use_cache:
        try:
            cached = cache.get(cache_key)
            if cached:
                return json.loads(cached) | {"cached": True}
        except Exception:
            pass

    try:
        response = openai_client.chat.completions.create(
            model=settings.model_name,
            messages=[{"role": "user", "content": request.prompt}],
            max_tokens=request.max_tokens,
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"LLM error: {str(e)}")

    result = {
        "answer": response.choices[0].message.content,
        "model": settings.model_name,
        "tokens_used": response.usage.total_tokens,
        "cached": False,
    }

    try:
        cache.setex(cache_key, settings.cache_ttl, json.dumps(result))
    except Exception:
        pass

    return result


@app.post("/process")
def process_document(request: ProcessRequest):
    """Invoca Lambda para procesar un documento de S3."""
    lambda_client = get_lambda_client()

    try:
        response = lambda_client.invoke(
            FunctionName="ai-document-processor",
            InvocationType="RequestResponse",
            Payload=json.dumps({
                "document_key": request.document_key,
                "force": request.force,
            }),
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Lambda invocation failed: {str(e)}")

    result = json.loads(response["Payload"].read())
    body = json.loads(result.get("body", "{}"))

    if result.get("statusCode", 500) != 200:
        raise HTTPException(status_code=result["statusCode"], detail=body)

    return body


@app.get("/documents")
def list_documents():
    """Lista documentos pendientes en S3 input."""
    s3 = get_s3_client()
    response = s3.list_objects_v2(Bucket=settings.s3_input_bucket, Prefix="documents/")
    return {
        "bucket": settings.s3_input_bucket,
        "documents": [
            {"key": obj["Key"], "size": obj["Size"]}
            for obj in response.get("Contents", [])
        ],
    }


@app.get("/results")
def list_results():
    """Lista resultados procesados en S3 output."""
    s3 = get_s3_client()
    response = s3.list_objects_v2(Bucket=settings.s3_output_bucket, Prefix="results/")
    return {
        "bucket": settings.s3_output_bucket,
        "results": [
            {"key": obj["Key"], "size": obj["Size"]}
            for obj in response.get("Contents", [])
        ],
    }

api/requirements.txt

fastapi==0.115.0
uvicorn==0.30.0
openai>=1.0.0
redis==5.0.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
boto3>=1.34.0

api/Dockerfile

FROM python:3.11-slim

RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*

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

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

lambda/processor.py

import json
import os
import time
import boto3
from openai import OpenAI

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

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")


def get_s3_client():
    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()

    document_key = event.get("document_key", "")
    if not document_key:
        if "body" in event:
            try:
                body = json.loads(event["body"])
                document_key = body.get("document_key", "")
            except (json.JSONDecodeError, TypeError):
                pass

    if not document_key:
        return _response(400, {"error": "document_key is required"})

    force = event.get("force", False)

    # Check if already processed
    doc_name = document_key.split("/")[-1].replace(".", "-")
    output_key = f"results/{doc_name}-analysis.json"

    if not force:
        try:
            existing = s3.get_object(Bucket=OUTPUT_BUCKET, Key=output_key)
            result = json.loads(existing["Body"].read().decode("utf-8"))
            return _response(200, {
                "status": "already_processed",
                "output": f"s3://{OUTPUT_BUCKET}/{output_key}",
                "tokens_used": result.get("tokens_used", 0),
            })
        except Exception:
            pass

    # Read document
    try:
        doc = s3.get_object(Bucket=INPUT_BUCKET, Key=document_key)
        document_text = doc["Body"].read().decode("utf-8")
    except Exception as e:
        return _response(404, {
            "error": f"Document not found: {str(e)}",
            "bucket": INPUT_BUCKET,
            "key": document_key,
        })

    # Process with LLM
    system_prompt = (
        "Analiza el siguiente documento de negocio. Responde en JSON con estos campos:\n"
        "- titulo: título o asunto del documento\n"
        "- categoria: tipo de documento (informe, email, propuesta, reporte, otro)\n"
        "- resumen: resumen en 2-3 oraciones\n"
        "- puntos_clave: lista de 3-5 puntos importantes\n"
        "- urgencia: alta, media, o baja\n"
        "- acciones_sugeridas: lista de 1-3 acciones recomendadas"
    )

    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,
            response_format={"type": "json_object"},
        )
        analysis = llm_response.choices[0].message.content
        tokens_used = llm_response.usage.total_tokens
    except Exception as e:
        return _response(502, {"error": f"LLM failed: {str(e)}"})

    duration_ms = round((time.time() - start_time) * 1000)

    # Build result
    result = {
        "source_document": document_key,
        "analysis": json.loads(analysis),
        "metadata": {
            "model": MODEL_NAME,
            "tokens_used": tokens_used,
            "duration_ms": duration_ms,
            "processed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "environment": "localstack" if AWS_ENDPOINT_URL else "aws",
        },
    }

    # Write to S3
    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}",
        "analysis": result["analysis"],
        "tokens_used": tokens_used,
        "duration_ms": duration_ms,
    })


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

lambda/requirements.txt

openai>=1.0.0
boto3>=1.34.0

.env.example

OPENAI_API_KEY=sk-proj-replace-with-your-key
ENVIRONMENT=development
LOG_LEVEL=debug
CACHE_TTL=3600
MODEL_NAME=gpt-4o-mini
MAX_TOKENS=500
API_PORT=8000

.gitignore

.env
lambda/package/
lambda/*.zip
data/output/
__pycache__/
*.pyc

Scripts de Operación

scripts/setup.sh

#!/bin/bash
set -e

echo "=== Setup: LocalStack AI Pipeline ==="

echo "1. Copiando .env..."
if [ ! -f .env ]; then
  cp .env.example .env
  echo "   Creado .env desde .env.example"
  echo "   IMPORTANTE: edita .env con tu OPENAI_API_KEY real"
else
  echo "   .env ya existe"
fi

echo "2. Creando directorio de init scripts..."
mkdir -p init-scripts
chmod +x init-scripts/setup.sh 2>/dev/null || true

echo "3. Empaquetando Lambda..."
mkdir -p lambda/package
pip install -r lambda/requirements.txt -t lambda/package/ --quiet
cp lambda/processor.py lambda/package/
cd lambda/package
zip -r ../processor.zip . -q
cd ../..
echo "   Package: $(ls -lh lambda/processor.zip | awk '{print $5}')"

echo "4. Levantando Docker Compose..."
docker compose up -d --build

echo "5. Esperando health checks..."
for i in {1..30}; do
  ALL_HEALTHY=true
  for service in api cache localstack; do
    STATUS=$(docker compose ps $service --format json 2>/dev/null | python3 -c "import json,sys; d=json.load(sys.stdin); print(d.get('Health',''))" 2>/dev/null || echo "")
    if [ "$STATUS" != "healthy" ]; then
      ALL_HEALTHY=false
    fi
  done
  if [ "$ALL_HEALTHY" = "true" ]; then
    echo "   Todos los servicios healthy"
    break
  fi
  echo "   Esperando... ($i/30)"
  sleep 3
done

echo "6. Desplegando Lambda a LocalStack..."
source .env 2>/dev/null || true

awslocal lambda create-function \
  --function-name ai-document-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
  }" 2>/dev/null || \
awslocal lambda update-function-code \
  --function-name ai-document-processor \
  --zip-file fileb://lambda/processor.zip > /dev/null

echo "7. Subiendo documentos de prueba..."
mkdir -p data/sample-documents

cat > data/sample-documents/informe-q1.txt << 'EOF'
Informe Trimestral Q1 2026 — Departamento de Tecnología

Resumen: Completamos la migración a microservicios, reduciendo deploy times de 4h a 15min.
El sistema AI de clasificación de tickets redujo tiempo de resolución en 40%.

Logros: Migración a Kubernetes, CI/CD automatizado, sistema AI de clasificación.
Próximos pasos: Monitoring con Prometheus, expandir AI a customer support.
EOF

cat > data/sample-documents/email-soporte.txt << 'EOF'
De: cliente@empresa.com
Asunto: Problema urgente con facturación

Llevo 3 meses con un cargo incorrecto. El servicio Premium cuesta $49/mes pero
me cobran $79. He contactado soporte 2 veces sin resolución. Necesito corrección
y reembolso de la diferencia. Si no se resuelve esta semana, cancelaré el servicio.
EOF

cat > data/sample-documents/propuesta-tecnica.txt << 'EOF'
Propuesta: Sistema de Recomendación para E-commerce

Objetivo: Mejorar CTR en 20% con recomendaciones basadas en embeddings.
Stack: FastAPI + Redis + OpenAI Embeddings.
Timeline: 6 semanas. Presupuesto: $15,000.
Equipo: 2 ML Engineers + 1 Backend Developer.
ROI estimado: $50,000/año en aumento de ventas.
EOF

for f in data/sample-documents/*.txt; do
  FILENAME=$(basename "$f")
  awslocal s3 cp "$f" "s3://ai-input/documents/$FILENAME"
done

echo "   Documentos subidos a S3"

echo ""
echo "=== Setup completo ==="
echo ""
echo "Servicios:"
docker compose ps --format "table {{.Name}}\t{{.Status}}\t{{.Ports}}"
echo ""
echo "S3 Buckets:"
awslocal s3 ls
echo ""
echo "Lambda Functions:"
awslocal lambda list-functions --query 'Functions[].FunctionName' --output text
echo ""
echo "Documentos en S3:"
awslocal s3 ls s3://ai-input/documents/
echo ""
echo "Próximo paso: python scripts/run-pipeline.py"

scripts/run-pipeline.py

#!/usr/bin/env python3
"""Ejecuta el pipeline AI completo contra LocalStack."""
import boto3
import json
import time
import sys

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 list_pending_documents():
    response = s3.list_objects_v2(Bucket="ai-input", Prefix="documents/")
    return [obj["Key"] for obj in response.get("Contents", [])]


def process_document(document_key, force=False):
    payload = {"document_key": document_key, "force": force}

    start = time.time()
    response = lambda_client.invoke(
        FunctionName="ai-document-processor",
        InvocationType="RequestResponse",
        Payload=json.dumps(payload),
    )
    elapsed = round((time.time() - start) * 1000)

    result = json.loads(response["Payload"].read())
    body = json.loads(result.get("body", "{}"))

    return {
        "status_code": result.get("statusCode", 500),
        "body": body,
        "invoke_ms": elapsed,
    }


def download_result(output_path):
    parts = output_path.replace("s3://ai-output/", "")
    response = s3.get_object(Bucket="ai-output", Key=parts)
    return json.loads(response["Body"].read().decode("utf-8"))


def main():
    force = "--force" in sys.argv

    print("=" * 60)
    print("  LocalStack AI Pipeline — Document Processor")
    print("=" * 60)
    print(f"\nEntorno: LocalStack ({ENDPOINT})")
    print(f"Force re-process: {force}\n")

    documents = list_pending_documents()
    print(f"Documentos encontrados: {len(documents)}\n")

    if not documents:
        print("No hay documentos en ai-input/documents/")
        print("Sube documentos con: awslocal s3 cp archivo.txt s3://ai-input/documents/")
        return

    results = []
    for i, doc_key in enumerate(documents, 1):
        doc_name = doc_key.split("/")[-1]
        print(f"[{i}/{len(documents)}] Procesando: {doc_name}")

        result = process_document(doc_key, force=force)
        status = result["body"].get("status", "unknown")
        tokens = result["body"].get("tokens_used", 0)

        if result["status_code"] == 200:
            print(f"    Status: {status}")
            print(f"    Tokens: {tokens}")
            print(f"    Tiempo: {result['invoke_ms']}ms")

            if "analysis" in result["body"]:
                analysis = result["body"]["analysis"]
                if isinstance(analysis, dict):
                    print(f"    Categoría: {analysis.get('categoria', 'N/A')}")
                    print(f"    Urgencia: {analysis.get('urgencia', 'N/A')}")
        else:
            print(f"    ERROR: {result['body']}")

        results.append(result)
        print()

    # Summary
    print("=" * 60)
    print("  RESUMEN")
    print("=" * 60)

    success = sum(1 for r in results if r["status_code"] == 200)
    total_tokens = sum(r["body"].get("tokens_used", 0) for r in results)
    total_time = sum(r["invoke_ms"] for r in results)

    print(f"\n  Documentos procesados: {success}/{len(results)}")
    print(f"  Total tokens:         {total_tokens}")
    print(f"  Tiempo total:         {total_time}ms")
    print(f"  Costo estimado:       ~${total_tokens * 0.0000015:.4f} (gpt-4o-mini)")
    print(f"  Costo LocalStack:     $0.00")
    print()

    # List results in S3
    print("Resultados en S3:")
    response = s3.list_objects_v2(Bucket="ai-output", Prefix="results/")
    for obj in response.get("Contents", []):
        print(f"  s3://ai-output/{obj['Key']} ({obj['Size']} bytes)")


if __name__ == "__main__":
    main()

scripts/diagnose.sh

#!/bin/bash
echo "=== DIAGNÓSTICO: LocalStack AI Pipeline ==="
echo ""

echo "1. Docker Compose"
docker compose ps --format "table {{.Name}}\t{{.Status}}"
echo ""

echo "2. LocalStack Health"
curl -s http://localhost:4566/_localstack/health | python3 -m json.tool 2>/dev/null || echo "   No disponible"
echo ""

echo "3. API Health"
curl -s http://localhost:8000/health | python3 -m json.tool 2>/dev/null || echo "   No disponible"
echo ""

echo "4. S3 Buckets"
awslocal s3 ls 2>/dev/null || echo "   S3 no disponible"
echo ""

echo "5. S3 Input Documents"
awslocal s3 ls s3://ai-input/documents/ 2>/dev/null || echo "   Bucket vacío o no existe"
echo ""

echo "6. S3 Output Results"
awslocal s3 ls s3://ai-output/results/ 2>/dev/null || echo "   Sin resultados"
echo ""

echo "7. Lambda Functions"
awslocal lambda list-functions --query 'Functions[].{Name:FunctionName,Runtime:Runtime,Memory:MemorySize,Timeout:Timeout}' --output table 2>/dev/null || echo "   Lambda no disponible"
echo ""

echo "8. Errores Recientes"
ERRORS=$(docker compose logs localstack --tail 50 2>&1 | grep -i "error\|exception" | tail -3)
if [ -n "$ERRORS" ]; then
  echo "$ERRORS"
else
  echo "   Sin errores recientes"
fi
echo ""

echo "9. Recursos"
docker stats --no-stream --format "table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}" 2>/dev/null | grep -E "NAME|api|cache|localstack"
echo ""

echo "=== FIN DIAGNÓSTICO ==="

scripts/demo.sh

#!/bin/bash
set -e

echo "╔════════════════════════════════════════════╗"
echo "║  LocalStack AI Pipeline — Demo Completa    ║"
echo "╚════════════════════════════════════════════╝"
echo ""

# Verificar que todo está corriendo
echo "Verificando servicios..."
curl -s http://localhost:8000/health > /dev/null 2>&1 || { echo "API no disponible. Ejecuta: bash scripts/setup.sh"; exit 1; }
curl -s http://localhost:4566/_localstack/health > /dev/null 2>&1 || { echo "LocalStack no disponible"; exit 1; }
echo "Servicios OK"
echo ""

# Listar documentos
echo "Documentos disponibles:"
curl -s http://localhost:8000/documents | python3 -c "
import json, sys
data = json.load(sys.stdin)
for doc in data['documents']:
    print(f\"  - {doc['key']} ({doc['size']} bytes)\")
"
echo ""

# Procesar un documento via la API
echo "Procesando documento via API..."
RESULT=$(curl -s -X POST http://localhost:8000/process \
  -H "Content-Type: application/json" \
  -d '{"document_key": "documents/email-soporte.txt", "force": true}')

echo "$RESULT" | python3 -c "
import json, sys
data = json.load(sys.stdin)
print(f\"  Status: {data.get('status', 'unknown')}\")
print(f\"  Tokens: {data.get('tokens_used', 0)}\")
print(f\"  Duración: {data.get('duration_ms', 0)}ms\")
if 'analysis' in data:
    a = data['analysis']
    print(f\"  Categoría: {a.get('categoria', 'N/A')}\")
    print(f\"  Urgencia: {a.get('urgencia', 'N/A')}\")
    print(f\"  Resumen: {a.get('resumen', 'N/A')[:100]}...\")
"
echo ""

# Listar resultados
echo "Resultados en S3:"
curl -s http://localhost:8000/results | python3 -c "
import json, sys
data = json.load(sys.stdin)
for r in data['results']:
    print(f\"  - {r['key']} ({r['size']} bytes)\")
"
echo ""

echo "Pipeline ejecutado exitosamente"
echo "Costo AWS: \$0.00 (LocalStack)"

Paso a Paso para Construir

1. Crear estructura (2 min)

mkdir -p localstack-ai-pipeline/{api,lambda/package,scripts,init-scripts,data/sample-documents}
cd localstack-ai-pipeline

2. Crear archivos (15 min)

Copia los archivos de las secciones anteriores en sus ubicaciones correspondientes.

3. Configurar environment (2 min)

cp .env.example .env
# Edita .env con tu OPENAI_API_KEY real

4. Hacer scripts ejecutables (1 min)

chmod +x scripts/*.sh
chmod +x init-scripts/setup.sh

5. Setup completo (5 min)

bash scripts/setup.sh

6. Ejecutar pipeline (3 min)

python scripts/run-pipeline.py

7. Demo interactiva (2 min)

bash scripts/demo.sh

8. Diagnóstico (1 min)

bash scripts/diagnose.sh

Checklist de Completitud

Infraestructura

  • docker compose up levanta api, cache, y localstack sin errores
  • Los tres servicios reportan healthy en docker compose ps
  • Init scripts crean los buckets automáticamente al arrancar
  • Lambda ai-document-processor está desplegada en LocalStack

Pipeline funcional

  • Documentos de prueba están en s3://ai-input/documents/
  • python scripts/run-pipeline.py procesa todos los documentos
  • Los resultados aparecen en s3://ai-output/results/
  • Los análisis incluyen: título, categoría, resumen, puntos clave, urgencia
  • Re-ejecución sin --force retorna "already_processed"
  • Re-ejecución con --force re-procesa correctamente

API

  • GET /health retorna status de api, redis, y localstack
  • POST /process invoca Lambda y retorna resultado
  • GET /documents lista documentos en S3 input
  • GET /results lista resultados en S3 output
  • POST /ask funciona con cache (Redis)

Environment switching

  • El código usa AWS_ENDPOINT_URL para determinar el target
  • Sin AWS_ENDPOINT_URL, boto3 apuntaría a AWS real
  • No hay if/else de entorno en el código de negocio
  • Los nombres de buckets son configurables via variables de entorno

Scripts y operación

  • scripts/setup.sh configura todo el sistema desde cero
  • scripts/run-pipeline.py ejecuta el pipeline end-to-end
  • scripts/diagnose.sh reporta el estado de todos los componentes
  • scripts/demo.sh demuestra el pipeline en un comando
  • .gitignore excluye .env, package/, y *.zip

Calidad

  • Lambda retorna errores claros (400 para input inválido, 502 para LLM failure)
  • Los resultados incluyen metadata (modelo, tokens, duración, entorno)
  • Deduplicación funciona (no re-procesa sin force=true)
  • El sistema se recupera de reinicios (init scripts + deploy scripts)

Troubleshooting del Proyecto

"setup.sh falla en 'Desplegando Lambda'"

# Verifica que LocalStack está healthy
docker compose ps localstack

# Verifica que el zip existe y tiene contenido
ls -lh lambda/processor.zip
unzip -l lambda/processor.zip | head -5

# Verifica que OPENAI_API_KEY está en .env
grep OPENAI_API_KEY .env

"run-pipeline.py retorna 502 para todos los documentos"

# El LLM no puede conectar — verifica la API key
awslocal lambda get-function-configuration \
  --function-name ai-document-processor \
  --query 'Environment.Variables.OPENAI_API_KEY'
# Si es vacío o placeholder, actualiza:
source .env
awslocal lambda update-function-configuration \
  --function-name ai-document-processor \
  --environment "Variables={OPENAI_API_KEY=${OPENAI_API_KEY},...}"

"Lambda no puede leer de S3"

# Verificar el endpoint URL de Lambda
awslocal lambda get-function-configuration \
  --function-name ai-document-processor \
  --query 'Environment.Variables.AWS_ENDPOINT_URL'

# Si LAMBDA_EXECUTOR=docker: debe ser http://host.docker.internal:4566
# Si LAMBDA_EXECUTOR=local: debe ser http://localhost:4566

"Los resultados tienen análisis vacíos"

# Verificar que response_format=json_object funciona
# Algunos modelos no soportan json_object mode
# Prueba sin él o usa un modelo que lo soporte

Conexión con la Guía

Lo que construiste

LocalStack AI Pipeline
├── Docker Compose (3 servicios: api, cache, localstack)
├── S3 Storage (ai-input → documents, ai-output → results)
├── Lambda Processor (lee S3 → GPT-4o-mini → escribe S3)
├── FastAPI Gateway (endpoints para operar el pipeline)
├── Environment Switching (AWS_ENDPOINT_URL)
├── Init Scripts (setup automático)
└── Operación Scripts (setup, run, diagnose, demo)

Lo que sigue

Módulo 5 (AWS Services for AI):
├── Mismo pipeline, ahora con profundidad en S3 y Lambda reales
├── Integración con SageMaker basics
└── Tus scripts de LocalStack siguen siendo tu entorno de dev

Módulo 6 (Cloud Migration Patterns):
├── El environment switching de este módulo es la base
├── Tu código ya funciona en LocalStack — ahora lo llevas a AWS
├── Migración = cambiar AWS_ENDPOINT_URL + configurar IAM
└── Confianza: "ya probé todo localmente"

Módulo 8 (Proyecto Integrador):
├── LocalStack sigue siendo tu entorno de desarrollo
├── El pipeline de este módulo se integra con el sistema final
└── Desarrollas en LocalStack, despliegas a producción

Resumen

  • Construiste un pipeline AI completo en LocalStack dentro de Docker Compose (api, cache, localstack).
  • Integraste S3 (buckets ai-input y ai-output) con Lambda para procesamiento de documentos con GPT-4o-mini.
  • La Lambda lee documentos de S3, los analiza con el LLM y escribe resultados estructurados a S3.
  • Implementaste environment switching con AWS_ENDPOINT_URL: el mismo código funciona contra LocalStack o AWS.
  • Configuraste init scripts de LocalStack para crear buckets automáticamente al arrancar.
  • Integraste el pipeline con FastAPI, Redis y scripts de operación (setup, run-pipeline, diagnose, demo).

Recursos para el Proyecto

  1. LocalStack Documentation — Documentación oficial completa
  2. LocalStack Docker Compose — Setup con Compose
  3. boto3 S3 Reference — API completa de S3
  4. boto3 Lambda Reference — API completa de Lambda
  5. LocalStack Init Hooks — Scripts de inicialización
  6. OpenAI JSON Mode — response_format=json_object
  7. Docker Compose Override — Compose por entorno
  8. FastAPI + boto3 Best Practices — Settings y configuración
← Anterior7. Debugging LocalStackSiguiente →1. Introducción: AWS Services for AI (S3, Lambda, SageMaker Basics)