8. Proyecto: S3+Lambda AI Service

Descripción del proyecto

Este es el proyecto integrador del Módulo 5. Vas a construir un servicio AI completo que usa S3 como data layer y Lambda como compute layer. El servicio almacena assets de AI en S3 (prompt templates versionados, documentos RAG, configuración), usa Lambda para leer esos assets, construir prompts contextualizados, invocar un LLM, y persistir las respuestas para auditoría. Incluye IAM con least privilege, S3 event triggers, y estimación de costes documentada.

Por qué importa: Este servicio integra todo lo que aprendiste en las cápsulas del módulo: S3 como storage para AI (C02), Lambda para inferencia (C03), integración event-driven (C04), decision framework SageMaker vs Lambda (C05), IAM least privilege (C06), y cost estimation (C07). Es un artefacto funcional que demuestra que sabes construir un servicio AI serverless en AWS. En el Módulo 6, abstraerás este código para que funcione contra LocalStack Y AWS con el mismo codebase — la base de tu migration path.

---

Objetivo del proyecto

Producir un S3+Lambda AI Service funcional que:

1. Almacena prompt templates versionados en S3 (prompts/{name}/{version}/system.txt)
2. Almacena documentos RAG en S3 (documents/{collection}/{doc_id}.json)
3. Recibe requests via API Gateway (POST /process) con documento y prompt a usar
4. Lambda lee el prompt template de S3, construye el prompt con el documento, invoca el LLM
5. Persiste las respuestas en S3 (responses/{date}/{request_id}.json) para auditoría
6. Procesa documentos automáticamente via S3 trigger (upload a documents/inbox/)
7. Tiene un endpoint de health (GET /health) que verifica conectividad S3 y configuración
8. IAM configurado con least privilege (Lambda solo accede a prefijos necesarios)
9. Cost estimation documentada para tres escenarios de uso

---

Recap del Módulo

Cápsula	Concepto	Lo usas en el proyecto
02	S3 para AI Assets	Almacenar prompts, documentos, responses en S3
03	Lambda para Inferencia	Handler con retry, structured logging, error handling
04	Integración S3+Lambda	S3 trigger → Lambda procesa → escribe a S3
05	SageMaker Basics	Justificación de por qué usas Lambda (no SageMaker)
06	IAM Permisos Mínimos	Roles y policies least privilege
07	Cost Estimation	Estimación de costes para el servicio

---

Especificaciones Técnicas

Arquitectura

                       ┌──────────────────────────────────┐
  POST /process ──────→│                                  │
  GET /health ────────→│  API Gateway (HTTP API v2)       │
                       │  CORS + Throttling               │
                       └──────────┬───────────────────────┘
                                  │
                       ┌──────────┴───────────────────────┐
                       │   Lambda: ai-service-process      │
                       │   768MB / 120s / arm64            │
                       │                                   │
                       │   1. Parse request                │
                       │   2. Read prompt template (S3)    │
                       │   3. Read/receive document        │
                       │   4. Build contextual prompt      │
                       │   5. Invoke LLM (with retry)     │
                       │   6. Save response to S3          │
                       │   7. Return result                │
                       └──────────┬───────────────────────┘
                                  │
                  ┌───────────────┼───────────────┐
                  │               │               │
           ┌──────┴──────┐  ┌────┴───────┐  ┌────┴──────┐
           │  S3 Bucket  │  │  OpenAI    │  │ CloudWatch│
           │  ai-service │  │  API       │  │ Logs      │
           │  -assets    │  │            │  │           │
           └─────────────┘  └────────────┘  └───────────┘

                    S3 Bucket Structure:
                    ai-service-assets-{account}/
                    ├── prompts/
                    │   ├── summarizer/v1/system.txt
                    │   ├── summarizer/v2/system.txt
                    │   └── classifier/v1/system.txt
                    ├── documents/
                    │   ├── inbox/          ← S3 trigger
                    │   └── knowledge-base/
                    ├── responses/
                    │   └── 2026/03/08/
                    └── config/
                        └── service-config.json

Endpoints requeridos

POST /process  → Procesa un documento con un prompt template
GET  /health   → Status del servicio, conectividad S3, configuración

Request/Response format

// POST /process — Request
{
  "prompt_name": "summarizer",
  "prompt_version": "v1",
  "document": {
    "id": "doc-001",
    "title": "Introducción a Serverless",
    "content": "Serverless computing permite ejecutar código..."
  },
  "max_tokens": 500
}

// POST /process — Response (200)
{
  "request_id": "req-a1b2c3d4",
  "answer": "Los puntos clave del documento son...",
  "model": "gpt-4o-mini",
  "tokens_used": 342,
  "duration_ms": 3245,
  "prompt_used": "summarizer/v1",
  "response_key": "responses/2026/03/08/req-a1b2c3d4.json",
  "cold_start": false
}

// POST /process — Response (400)
{
  "error": "prompt_name is required"
}

// POST /process — Response (404)
{
  "error": "Prompt template 'analyzer/v3' not found in S3"
}

// GET /health — Response (200)
{
  "status": "healthy",
  "checks": {
    "lambda": "up",
    "s3_bucket": "accessible",
    "openai_key": "configured"
  },
  "config": {
    "bucket": "ai-service-assets-dev",
    "model": "gpt-4o-mini",
    "memory_mb": "768",
    "region": "us-east-1"
  },
  "prompt_templates": ["summarizer/v1", "summarizer/v2", "classifier/v1"]
}

Configuración Lambda

Parámetro	Process Function	Health Function
Runtime	Python 3.11	Python 3.11
Architecture	arm64	arm64
Memory	768 MB	128 MB
Timeout	120s	10s
Handler	process_handler.handler	health_handler.handler

---

Código Completo

src/process_handler.py

import json
import logging
import os
import time
import uuid
from datetime import datetime

import boto3
from openai import OpenAI, APITimeoutError, RateLimitError, APIConnectionError

logger = logging.getLogger()
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

BUCKET = os.environ.get("AI_BUCKET", "ai-service-assets-dev")
MODEL = os.environ.get("MODEL_NAME", "gpt-4o-mini")
MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3"))

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
openai_client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY", ""),
    max_retries=0,
)

CORS_HEADERS = {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": os.environ.get("ALLOWED_ORIGIN", "*"),
    "Access-Control-Allow-Methods": "POST, GET, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type",
}

IS_COLD_START = True
RETRYABLE = (APITimeoutError, RateLimitError, APIConnectionError)


def _response(status_code: int, body) -> dict:
    return {
        "statusCode": status_code,
        "headers": CORS_HEADERS,
        "body": json.dumps(body) if isinstance(body, dict) else body,
    }


def get_prompt_template(prompt_name: str, prompt_version: str) -> str:
    """Lee un prompt template de S3."""
    key = f"prompts/{prompt_name}/{prompt_version}/system.txt"
    try:
        response = s3.get_object(Bucket=BUCKET, Key=key)
        return response["Body"].read().decode("utf-8")
    except s3.exceptions.NoSuchKey:
        raise FileNotFoundError(f"Prompt template '{prompt_name}/{prompt_version}' not found in S3")


def get_document_from_s3(collection: str, doc_id: str) -> dict:
    """Lee un documento de S3."""
    key = f"documents/{collection}/{doc_id}.json"
    try:
        response = s3.get_object(Bucket=BUCKET, Key=key)
        return json.loads(response["Body"].read().decode("utf-8"))
    except s3.exceptions.NoSuchKey:
        raise FileNotFoundError(f"Document '{collection}/{doc_id}' not found in S3")


def save_response(request_id: str, data: dict) -> str:
    """Guarda la respuesta de inferencia en S3."""
    now = datetime.utcnow()
    key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"
    payload = {
        "request_id": request_id,
        "timestamp": now.isoformat(),
        **data,
    }
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        ContentType="application/json",
    )
    return key


def invoke_llm(messages: list, max_tokens: int, context) -> dict:
    """Invoca LLM con retry y backoff exponencial."""
    last_error = None

    for attempt in range(1, MAX_RETRIES + 1):
        remaining_ms = context.get_remaining_time_in_millis()
        if remaining_ms < 15000:
            raise TimeoutError(f"Insufficient time: {remaining_ms}ms")

        try:
            start = time.time()
            response = openai_client.chat.completions.create(
                model=MODEL,
                messages=messages,
                max_tokens=max_tokens,
                timeout=(remaining_ms / 1000) - 10,
            )
            duration_ms = int((time.time() - start) * 1000)

            return {
                "content": response.choices[0].message.content,
                "model": response.model,
                "tokens_used": response.usage.total_tokens,
                "input_tokens": response.usage.prompt_tokens,
                "output_tokens": response.usage.completion_tokens,
                "llm_duration_ms": duration_ms,
                "attempt": attempt,
            }

        except RETRYABLE as e:
            last_error = e
            if attempt < MAX_RETRIES:
                delay = 1.0 * (2 ** (attempt - 1))
                logger.warning(json.dumps({
                    "event": "llm_retry",
                    "attempt": attempt,
                    "error": str(e),
                    "delay_s": delay,
                }))
                time.sleep(delay)
        except Exception as e:
            raise

    raise last_error


def process_request(body: dict, context) -> dict:
    """Procesa un request de inferencia completo."""
    prompt_name = body.get("prompt_name", "").strip()
    prompt_version = body.get("prompt_version", "v1").strip()
    document = body.get("document", {})
    max_tokens = min(body.get("max_tokens", 500), 2000)

    if not prompt_name:
        raise ValueError("prompt_name is required")
    if not document or not document.get("content"):
        raise ValueError("document with 'content' field is required")

    prompt_template = get_prompt_template(prompt_name, prompt_version)

    doc_title = document.get("title", "Untitled")
    doc_content = document.get("content", "")

    full_prompt = f"{prompt_template}\n\nTítulo: {doc_title}\n\nContenido:\n{doc_content}"

    messages = [{"role": "user", "content": full_prompt}]
    llm_result = invoke_llm(messages, max_tokens, context)

    return {
        "answer": llm_result["content"],
        "model": llm_result["model"],
        "tokens_used": llm_result["tokens_used"],
        "prompt_used": f"{prompt_name}/{prompt_version}",
        "document_title": doc_title,
        "llm_duration_ms": llm_result["llm_duration_ms"],
        "retries": llm_result["attempt"] - 1,
    }


def process_s3_event(record: dict, context) -> dict:
    """Procesa un documento triggereado por S3 event."""
    from urllib.parse import unquote_plus

    source_bucket = record["s3"]["bucket"]["name"]
    source_key = unquote_plus(record["s3"]["object"]["key"])

    doc_response = s3.get_object(Bucket=source_bucket, Key=source_key)
    document = json.loads(doc_response["Body"].read().decode("utf-8"))

    prompt_name = document.get("prompt_name", os.environ.get("DEFAULT_PROMPT", "summarizer"))
    prompt_version = document.get("prompt_version", "v1")

    body = {
        "prompt_name": prompt_name,
        "prompt_version": prompt_version,
        "document": document,
        "max_tokens": document.get("max_tokens", 500),
    }

    return process_request(body, context)


def handler(event, context):
    """Handler principal — maneja API Gateway y S3 triggers."""
    global IS_COLD_START
    was_cold = IS_COLD_START
    IS_COLD_START = False

    start_time = time.time()
    request_id = f"req-{uuid.uuid4().hex[:8]}"

    # S3 trigger
    if "Records" in event and event["Records"][0].get("eventSource") == "aws:s3":
        results = []
        for record in event["Records"]:
            try:
                result = process_s3_event(record, context)
                response_key = save_response(request_id, result)
                result["response_key"] = response_key
                result["request_id"] = request_id
                results.append({"status": "success", **result})
            except Exception as e:
                logger.error(json.dumps({"event": "s3_trigger_error", "error": str(e)}))
                results.append({"status": "error", "error": str(e)})
        return {"processed": len(results), "results": results}

    # API Gateway
    method = event.get("requestContext", {}).get("http", {}).get("method", "")
    if method == "OPTIONS":
        return _response(200, "")

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

    try:
        result = process_request(body, context)
    except ValueError as e:
        return _response(400, {"error": str(e)})
    except FileNotFoundError as e:
        return _response(404, {"error": str(e)})
    except TimeoutError as e:
        return _response(408, {"error": str(e)})
    except RETRYABLE as e:
        return _response(502, {"error": f"LLM unavailable: {str(e)}"})
    except Exception as e:
        logger.error(json.dumps({"event": "process_error", "error": str(e)}))
        return _response(500, {"error": f"Internal error: {str(e)}"})

    total_ms = int((time.time() - start_time) * 1000)

    response_key = save_response(request_id, result)

    logger.info(json.dumps({
        "event": "process_success",
        "request_id": request_id,
        "prompt": result["prompt_used"],
        "tokens": result["tokens_used"],
        "duration_ms": total_ms,
        "cold_start": was_cold,
    }))

    return _response(200, {
        "request_id": request_id,
        "answer": result["answer"],
        "model": result["model"],
        "tokens_used": result["tokens_used"],
        "duration_ms": total_ms,
        "prompt_used": result["prompt_used"],
        "response_key": response_key,
        "cold_start": was_cold,
    })

src/health_handler.py

import json
import os
import time

import boto3

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = os.environ.get("AI_BUCKET", "ai-service-assets-dev")


def handler(event, context):
    start = time.time()
    checks = {"lambda": "up"}

    # Check S3 bucket
    try:
        s3.head_bucket(Bucket=BUCKET)
        checks["s3_bucket"] = "accessible"
    except Exception:
        checks["s3_bucket"] = "unreachable"

    # Check OpenAI key
    openai_key = os.environ.get("OPENAI_API_KEY", "")
    if not openai_key:
        checks["openai_key"] = "missing"
    elif not openai_key.startswith("sk-"):
        checks["openai_key"] = "invalid_format"
    else:
        checks["openai_key"] = "configured"

    # List available prompt templates
    templates = []
    try:
        response = s3.list_objects_v2(Bucket=BUCKET, Prefix="prompts/", Delimiter="/")
        for prefix in response.get("CommonPrefixes", []):
            name = prefix["Prefix"].replace("prompts/", "").rstrip("/")
            versions_resp = s3.list_objects_v2(
                Bucket=BUCKET, Prefix=f"prompts/{name}/", Delimiter="/"
            )
            for v in versions_resp.get("CommonPrefixes", []):
                version = v["Prefix"].split("/")[-2]
                templates.append(f"{name}/{version}")
    except Exception:
        pass

    all_healthy = all(
        v in ("up", "accessible", "configured") for v in checks.values()
    )
    status = "healthy" if all_healthy else "degraded"
    status_code = 200 if all_healthy else 503

    return {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": os.environ.get("ALLOWED_ORIGIN", "*"),
        },
        "body": json.dumps({
            "status": status,
            "checks": checks,
            "config": {
                "bucket": BUCKET,
                "model": os.environ.get("MODEL_NAME", "gpt-4o-mini"),
                "memory_mb": os.environ.get("AWS_LAMBDA_FUNCTION_MEMORY_SIZE", "unknown"),
                "region": os.environ.get("AWS_REGION", "unknown"),
            },
            "prompt_templates": sorted(templates),
            "latency_ms": round((time.time() - start) * 1000, 1),
        }),
    }

src/requirements.txt

openai>=1.0.0
boto3>=1.28.0

src/seed_data.py

"""Script para poblar S3 con datos iniciales del servicio."""
import boto3
import json
import os

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = os.environ.get("AI_BUCKET", "ai-service-assets-dev")


def seed():
    """Crea bucket y sube datos iniciales."""
    try:
        s3.create_bucket(Bucket=BUCKET)
        print(f"Bucket created: {BUCKET}")
    except Exception:
        print(f"Bucket already exists: {BUCKET}")

    # Prompt templates
    prompts = {
        "summarizer/v1": (
            "Genera un resumen conciso del siguiente documento.\n"
            "Máximo 3 párrafos. Incluye los puntos más importantes.\n"
            "Usa lenguaje claro y directo."
        ),
        "summarizer/v2": (
            "Genera un resumen del siguiente documento en formato de bullets.\n"
            "- Máximo 5 bullets\n"
            "- Cada bullet debe ser una oración completa\n"
            "- Incluye datos numéricos si existen\n"
            "- Prioriza información actionable"
        ),
        "classifier/v1": (
            "Clasifica el siguiente documento en una de estas categorías:\n"
            "- technical: documentación técnica, guías, tutoriales\n"
            "- business: reportes, propuestas, análisis de negocio\n"
            "- support: tickets, preguntas frecuentes, soporte\n"
            "- general: otros\n\n"
            "Responde SOLO con un JSON: {\"category\": \"...\", \"confidence\": 0.95, \"reason\": \"...\"}"
        ),
    }

    for path, content in prompts.items():
        key = f"prompts/{path}/system.txt"
        s3.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=content.encode("utf-8"),
            ContentType="text/plain",
        )
        print(f"  Uploaded: {key}")

    # Sample documents
    documents = [
        {
            "id": "doc-001",
            "title": "Guía de Instalación del SDK",
            "content": (
                "Para instalar el SDK de la plataforma, sigue estos pasos: "
                "1. Ejecuta pip install platform-sdk en tu terminal. "
                "2. Configura las credenciales con platform configure --key TU_API_KEY. "
                "3. Verifica la instalación con platform status. "
                "El SDK soporta Python 3.9+ y requiere al menos 100MB de espacio libre."
            ),
        },
        {
            "id": "doc-002",
            "title": "Reporte Q1 2026",
            "content": (
                "En Q1 2026, la plataforma alcanzó 50,000 usuarios activos mensuales, "
                "un crecimiento del 35% respecto a Q4 2025. Los ingresos mensuales recurrentes "
                "llegaron a $125,000. El churn rate se redujo de 8% a 5.2%. "
                "Los principales drivers de crecimiento fueron: lanzamiento del plan Enterprise, "
                "integración con Slack, y el programa de referidos."
            ),
        },
        {
            "id": "doc-003",
            "title": "FAQ: Problemas de Conexión",
            "content": (
                "P: No puedo conectar al API. R: Verifica que tu API key es válida y no ha expirado. "
                "P: Recibo error 429. R: Estás excediendo el rate limit. Reduce la frecuencia de requests. "
                "P: Los responses son muy lentos. R: Verifica tu conexión a internet y la latencia al endpoint."
            ),
        },
    ]

    for doc in documents:
        key = f"documents/knowledge-base/{doc['id']}.json"
        s3.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=json.dumps(doc, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        print(f"  Uploaded: {key}")

    # Service config
    config = {
        "default_prompt": "summarizer",
        "default_version": "v1",
        "max_tokens": 500,
        "available_prompts": list(prompts.keys()),
    }
    s3.put_object(
        Bucket=BUCKET,
        Key="config/service-config.json",
        Body=json.dumps(config).encode("utf-8"),
        ContentType="application/json",
    )
    print("  Uploaded: config/service-config.json")

    print(f"\nSeed complete. Bucket: {BUCKET}")


if __name__ == "__main__":
    seed()

template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: S3+Lambda AI Service — Module 5 Project

Parameters:
  OpenAiApiKey:
    Type: String
    NoEcho: true
  BucketName:
    Type: String
    Default: ai-service-assets-dev
  ModelName:
    Type: String
    Default: gpt-4o-mini
  AllowedOrigin:
    Type: String
    Default: "*"

Globals:
  Function:
    Runtime: python3.11
    Architectures: [arm64]

Resources:
  AIBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Ref BucketName
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256

  ServiceApi:
    Type: AWS::Serverless::HttpApi
    Properties:
      StageName: prod
      CorsConfiguration:
        AllowOrigins: [!Ref AllowedOrigin]
        AllowMethods: [POST, GET, OPTIONS]
        AllowHeaders: [Content-Type]
        MaxAge: 3600

  ProcessFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ai-service-process
      Handler: process_handler.handler
      CodeUri: ./src/
      MemorySize: 768
      Timeout: 120
      Environment:
        Variables:
          OPENAI_API_KEY: !Ref OpenAiApiKey
          AI_BUCKET: !Ref BucketName
          MODEL_NAME: !Ref ModelName
          ALLOWED_ORIGIN: !Ref AllowedOrigin
          MAX_RETRIES: "3"
          DEFAULT_PROMPT: summarizer
          LOG_LEVEL: INFO
      Policies:
        - S3ReadPolicy:
            BucketName: !Ref BucketName
        - Statement:
            - Sid: WriteResponses
              Effect: Allow
              Action: [s3:PutObject]
              Resource: !Sub "arn:aws:s3:::${BucketName}/responses/*"
      Events:
        ProcessRoute:
          Type: HttpApi
          Properties:
            ApiId: !Ref ServiceApi
            Path: /process
            Method: POST
        S3Trigger:
          Type: S3
          Properties:
            Bucket: !Ref AIBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: documents/inbox/
                  - Name: suffix
                    Value: .json

  HealthFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ai-service-health
      Handler: health_handler.handler
      CodeUri: ./src/
      MemorySize: 128
      Timeout: 10
      Environment:
        Variables:
          OPENAI_API_KEY: !Ref OpenAiApiKey
          AI_BUCKET: !Ref BucketName
          MODEL_NAME: !Ref ModelName
          ALLOWED_ORIGIN: !Ref AllowedOrigin
      Policies:
        - S3ReadPolicy:
            BucketName: !Ref BucketName
      Events:
        HealthRoute:
          Type: HttpApi
          Properties:
            ApiId: !Ref ServiceApi
            Path: /health
            Method: GET

Outputs:
  ApiUrl:
    Value: !Sub "https://${ServiceApi}.execute-api.${AWS::Region}.amazonaws.com/prod"
  ProcessEndpoint:
    Value: !Sub "https://${ServiceApi}.execute-api.${AWS::Region}.amazonaws.com/prod/process"
  HealthEndpoint:
    Value: !Sub "https://${ServiceApi}.execute-api.${AWS::Region}.amazonaws.com/prod/health"
  BucketName:
    Value: !Ref AIBucket

env.json

{
  "ProcessFunction": {
    "OPENAI_API_KEY": "sk-proj-your-key-here",
    "AI_BUCKET": "ai-service-assets-dev",
    "MODEL_NAME": "gpt-4o-mini",
    "ALLOWED_ORIGIN": "*",
    "MAX_RETRIES": "3",
    "DEFAULT_PROMPT": "summarizer",
    "LOG_LEVEL": "DEBUG"
  },
  "HealthFunction": {
    "OPENAI_API_KEY": "sk-proj-your-key-here",
    "AI_BUCKET": "ai-service-assets-dev",
    "MODEL_NAME": "gpt-4o-mini",
    "ALLOWED_ORIGIN": "*"
  }
}

events/process.json

{
  "version": "2.0",
  "routeKey": "POST /process",
  "headers": {"content-type": "application/json"},
  "requestContext": {
    "http": {"method": "POST", "path": "/prod/process"}
  },
  "body": "{\"prompt_name\": \"summarizer\", \"prompt_version\": \"v1\", \"document\": {\"id\": \"doc-001\", \"title\": \"Guía de Instalación\", \"content\": \"Para instalar el SDK, ejecuta pip install platform-sdk. Configura con platform configure. Verifica con platform status.\"}, \"max_tokens\": 300}"
}

---

Paso a Paso para Construir

1. Crear estructura (2 min)

mkdir -p s3-lambda-ai-service/src
mkdir -p s3-lambda-ai-service/events
mkdir -p s3-lambda-ai-service/tests
cd s3-lambda-ai-service

2. Copiar archivos (5 min)

Copia todos los archivos de la sección anterior en sus ubicaciones correspondientes.

3. Seed data en S3 (3 min)

# Con LocalStack
export AWS_ENDPOINT_URL=http://localhost:4566
export AWS_ACCESS_KEY_ID=test
export AWS_SECRET_ACCESS_KEY=test
export AI_BUCKET=ai-service-assets-dev

python src/seed_data.py

4. Test local con SAM (5 min)

sam build

# Test health
sam local invoke HealthFunction \
  --event events/health.json \
  --env-vars env.json

# Test process
sam local invoke ProcessFunction \
  --event events/process.json \
  --env-vars env.json

# API completa
sam local start-api --env-vars env.json

# En otra terminal:
curl http://127.0.0.1:3000/health

curl -X POST http://127.0.0.1:3000/process \
  -H "Content-Type: application/json" \
  -d '{
    "prompt_name": "summarizer",
    "prompt_version": "v1",
    "document": {
      "id": "test-001",
      "title": "Test Document",
      "content": "This is a test document about serverless computing."
    }
  }'

5. Test S3 trigger (5 min)

# Sube un documento al inbox (triggerear Lambda)
aws --endpoint-url=http://localhost:4566 s3 cp \
  <(echo '{"id": "trigger-test", "title": "Trigger Test", "content": "Testing S3 trigger pipeline."}') \
  s3://ai-service-assets-dev/documents/inbox/trigger-test.json

# Verifica que Lambda procesó el documento
aws --endpoint-url=http://localhost:4566 s3 ls \
  s3://ai-service-assets-dev/responses/ --recursive

6. Verificar responses en S3 (3 min)

# Listar responses
aws --endpoint-url=http://localhost:4566 s3 ls \
  s3://ai-service-assets-dev/responses/ --recursive

# Leer una response
aws --endpoint-url=http://localhost:4566 s3 cp \
  s3://ai-service-assets-dev/responses/2026/03/08/req-abc12345.json \
  - | python -m json.tool

7. Deploy (opcional, si tienes cuenta AWS)

sam deploy --guided
# Stack name: ai-service
# Region: us-east-1
# OpenAiApiKey: sk-proj-...
# BucketName: ai-service-assets-{tu-account-id}
# Confirm changes: y

---

Cost Estimation para Este Servicio

Escenario: Desarrollo/Testing

Parámetros: 100 invocaciones/día, 768MB, 6s avg

S3:
├── Storage: 1GB × $0.023 = $0.02
├── PUTs: 3,000/mes × $0.005/1K = $0.015
├── GETs: 10,000/mes × $0.0004/1K = $0.004
└── Subtotal S3: $0.04

Lambda:
├── 3,000 inv/mes → Free tier: $0.00
├── 13,500 GB-s → Free tier: $0.00
└── Subtotal Lambda: $0.00

OpenAI (gpt-4o-mini):
├── Input: 3K × 200 tokens × $0.15/1M = $0.09
├── Output: 3K × 400 tokens × $0.60/1M = $0.72
└── Subtotal OpenAI: $0.81

══════════════════════
TOTAL: ~$0.85/mes
══════════════════════

Escenario: Producción Ligera

Parámetros: 5,000 invocaciones/día, 768MB, 8s avg

S3:
├── Storage: 10GB × $0.023 = $0.23
├── PUTs: 150K/mes × $0.005/1K = $0.75
├── GETs: 500K/mes × $0.0004/1K = $0.20
└── Subtotal S3: $1.18

Lambda:
├── 150K inv/mes → Free tier: $0.00
├── 900K GB-s → Billable: 500K × $0.0000133334 = $6.67
└── Subtotal Lambda: $6.67

OpenAI (gpt-4o-mini):
├── Input: 150K × 300 tokens × $0.15/1M = $6.75
├── Output: 150K × 500 tokens × $0.60/1M = $45.00
└── Subtotal OpenAI: $51.75

══════════════════════════
TOTAL: ~$59.60/mes
├── AWS infra: $7.85 (13%)
├── OpenAI API: $51.75 (87%)
══════════════════════════

Escenario: Producción Alta

Parámetros: 50,000 invocaciones/día, 768MB, 8s avg

S3:
├── Storage: 50GB × $0.023 = $1.15
├── PUTs: 1.5M/mes × $0.005/1K = $7.50
├── GETs: 5M/mes × $0.0004/1K = $2.00
└── Subtotal S3: $10.65

Lambda:
├── 1.5M inv/mes → Billable: 500K × $0.0000002 = $0.10
├── 9M GB-s → Billable: 8.6M × $0.0000133334 = $114.67
└── Subtotal Lambda: $114.77

OpenAI (gpt-4o-mini):
├── Input: 1.5M × 300 tokens × $0.15/1M = $67.50
├── Output: 1.5M × 500 tokens × $0.60/1M = $450.00
└── Subtotal OpenAI: $517.50

══════════════════════════
TOTAL: ~$642.92/mes
├── AWS infra: $125.42 (20%)
├── OpenAI API: $517.50 (80%)
══════════════════════════

---

Checklist de Completitud

Código y estructura

• src/process_handler.py implementa POST /process con S3 read/write
• src/health_handler.py implementa GET /health con checks de S3 y config
• src/seed_data.py pobla S3 con prompt templates y documentos de ejemplo
• src/requirements.txt contiene openai>=1.0.0 y boto3>=1.28.0
• template.yaml define ambas funciones, S3 bucket, API Gateway, y S3 trigger
• env.json existe con variables locales (NO en Git)
• events/process.json existe para testing con SAM

Funcionalidad

• POST /process recibe prompt_name + document y retorna respuesta del LLM
• POST /process lee prompt template de S3 dinámicamente
• POST /process guarda response en S3 (responses/{date}/{request_id}.json)
• POST /process retorna metadata: model, tokens, duration, prompt_used, response_key
• GET /health verifica S3, OpenAI key, lista prompt templates disponibles
• S3 trigger procesa documentos subidos a documents/inbox/*.json
• Errores manejados: prompt no encontrado (404), JSON inválido (400), LLM error (502)
• Retry con backoff exponencial para errores transitorios del LLM
• CORS headers en todos los responses

IAM y Seguridad

• Process Function: S3 read en prompts/*, documents/*, config/*
• Process Function: S3 write solo en responses/*
• Health Function: S3 read only
• Block Public Access habilitado en el bucket
• Encryption at rest (AES256) habilitada
• API keys en environment variables (no hardcodeadas)
• env.json en .gitignore

Testing

• seed_data.py ejecuta sin errores (crea bucket y datos)
• sam local invoke funciona para ambas funciones
• sam local start-api levanta y responde en /process y /health
• S3 trigger procesa documento y guarda response
• (Opcional) Deploy a AWS funciona sin errores

Cost Estimation

• Estimación documentada para escenario de desarrollo (~$1/mes)
• Estimación documentada para producción ligera (~$60/mes)
• Estimación documentada para producción alta (~$643/mes)
• Identificado que OpenAI API es el componente dominante de coste

---

Conexión con la Guía

Lo que construiste

S3+Lambda AI Service
├── Data Layer (S3)
│   ├── Prompt templates versionados
│   ├── Documentos RAG
│   ├── Responses para auditoría
│   └── Configuración del servicio
├── Compute Layer (Lambda)
│   ├── Process handler (POST /process)
│   ├── Health handler (GET /health)
│   ├── S3 trigger handler
│   ├── Retry con backoff
│   └── Structured logging
├── API Layer (API Gateway)
│   ├── CORS configurado
│   └── Throttling
├── Security (IAM)
│   ├── Least privilege policies
│   └── S3 encryption
└── Operations
    ├── Cost estimation por escenario
    └── Seed data script

Lo que sigue en los próximos módulos

Módulo 6 (Cloud Migration Patterns):
├── Este servicio funciona contra LocalStack y AWS
├── Abstraerás el endpoint_url con environment variables
├── Mismo código, diferente infraestructura
├── Testing de migration path: LocalStack → AWS
└── El S3+Lambda Service del M5 es el "target" de la migración

Módulo 7 (Alternative Platforms):
├── Comparas el coste de este servicio en AWS vs Render/Railway/Fly.io
├── Decision matrix: ¿cuándo AWS es mejor opción?
└── El cost estimation del M5 te da la baseline

Módulo 8 (Proyecto Integrador):
├── Este servicio puede ser parte de tu sistema final
├── O puedes elegir otra plataforma basándote en M7
├── La decision matrix del M1 + los costes del M5 informan la decisión
└── Tu servicio se despliega en producción

---

Troubleshooting del Proyecto

"seed_data.py falla con NoSuchBucket"

# Verifica que LocalStack está corriendo
curl http://localhost:4566/_localstack/health

# Verifica el endpoint URL
echo $AWS_ENDPOINT_URL
# Debe ser: http://localhost:4566

# El script crea el bucket, pero si hay error de conexión:
aws --endpoint-url=http://localhost:4566 s3 mb s3://ai-service-assets-dev

"Process handler retorna 404 para prompt template"

# Verifica que seed_data.py se ejecutó correctamente
aws --endpoint-url=http://localhost:4566 s3 ls \
  s3://ai-service-assets-dev/prompts/ --recursive

# Si no hay prompts, ejecuta seed_data.py de nuevo
python src/seed_data.py

"S3 trigger no se activa"

# En LocalStack, verifica la notification configuration
aws --endpoint-url=http://localhost:4566 \
  s3api get-bucket-notification-configuration \
  --bucket ai-service-assets-dev

# Verifica que Lambda tiene permiso
aws --endpoint-url=http://localhost:4566 \
  lambda get-policy --function-name ai-service-process

"Response no se guarda en S3 después de POST /process"

# Verifica que Lambda tiene permiso de escritura en responses/
# En el template.yaml, confirma:
# - Statement:
#     - Sid: WriteResponses
#       Effect: Allow
#       Action: [s3:PutObject]
#       Resource: !Sub "arn:aws:s3:::${BucketName}/responses/*"

---

Resumen

• Construiste un servicio AI completo en AWS real con S3 como capa de datos y Lambda como compute layer.
• Almacenaste prompt templates versionados en S3 y documentos RAG; Lambda los lee, construye prompts y persiste respuestas.
• Configuraste S3 event triggers para procesar automáticamente documentos subidos a documents/inbox/.
• Definiste IAM con least privilege: Lambda solo accede a los prefijos S3 necesarios (prompts, documents, responses).
• Desplegaste con SAM: bucket S3 encriptado, API Gateway, funciones Process y Health, y S3 trigger.
• Documentaste la estimación de costes para desarrollo ($0.85/mes), producción ligera ($60/mes) y alta (~$643/mes).

---

Recursos para el Proyecto

1. AWS SAM CLI Reference — Referencia SAM CLI
2. S3 Event Notifications — Triggers S3
3. Lambda Python Handler — Handler reference
4. boto3 S3 Client — S3 SDK reference
5. OpenAI Python SDK — SDK oficial
6. IAM Policy Reference — Policies
7. AWS Pricing Calculator — Calculadora de costes
8. LocalStack SAM Integration — SAM con LocalStack

---