Lab práctico · Semana 4: Ingeniería de prompts, gobierno y APIs de los modelos

API Converse, streaming, tool use y resiliencia con API Gateway

Qué vas a construir

Vas a trabajar las APIs de inferencia como lo harías en producción:

1. Converse frente a InvokeModel con el mismo modelo, leyendo usage, stopReason y latencia.
2. ConverseStream: medir el tiempo hasta el primer token frente a la respuesta completa.
3. Tool use con un bucle robusto (validación de argumentos, errores devueltos al modelo, límite de iteraciones).
4. Errores y resiliencia: ValidationException, truncado por max_tokens, reintentos con backoff exponencial y jitter, modo adaptive de boto3 y fallback de modelo.
5. Una API REST de API Gateway con response streaming delante de una Lambda en Node.js que llama a ConverseStream.

flowchart LR
  subgraph CS["CloudShell (Python)"]
    A["converse_vs_invoke.py"]
    B["stream.py"]
    C["herramientas.py"]
    D["resiliencia.py"]
  end
  CURL["curl -N"] --> APIGW["API Gateway REST (responseTransferMode STREAM)"]
  APIGW --> L["Lambda Node.js (streamifyResponse)"]
  L --> BR["Bedrock ConverseStream (eu.amazon.nova-micro-v1:0)"]
  A --> BR
  B --> BR
  C --> BR
  D --> BR

Antes de empezar

• Región eu-central-1; modelos Nova Micro (eu.amazon.nova-micro-v1:0) y Nova Lite (eu.amazon.nova-lite-v1:0) vía perfil de inferencia EU.
• CloudShell con boto3 actualizado:

pip3 install --user --upgrade boto3 botocore
export AWS_REGION=eu-central-1
export CUENTA=$(aws sts get-caller-identity --query Account --output text)
mkdir -p ~/lab08 && cd ~/lab08

Paso 1: Converse frente a InvokeModel

Guarda converse_vs_invoke.py:

import os, json, boto3
brt = boto3.client("bedrock-runtime", region_name=os.environ["AWS_REGION"])
MODELO = "eu.amazon.nova-micro-v1:0"
SYSTEM = [{"text": "Eres un profesor de IA. Responde en español en dos frases."}]
MENSAJES = [{"role": "user", "content": [{"text": "¿Qué diferencia hay entre temperatura y top-p?"}]}]

# 1) Converse: formato común a todos los modelos de mensajes
r = brt.converse(modelId=MODELO, system=SYSTEM, messages=MENSAJES,
                 inferenceConfig={"maxTokens": 200, "temperature": 0.2},
                 requestMetadata={"app": "lab08", "caso": "comparativa"})
print("CONVERSE:", r["output"]["message"]["content"][0]["text"])
print("  stopReason:", r["stopReason"], "| usage:", r["usage"], "| latencia:", r["metrics"]["latencyMs"], "ms\n")

# 2) InvokeModel: cuerpo nativo de Amazon Nova (cambia por proveedor)
body = {"system": SYSTEM, "messages": MENSAJES, "inferenceConfig": {"maxTokens": 200, "temperature": 0.2}}
r2 = brt.invoke_model(modelId=MODELO, body=json.dumps(body))
nativo = json.loads(r2["body"].read())
print("INVOKEMODEL (respuesta nativa):", json.dumps(nativo, ensure_ascii=False)[:400])

python3 converse_vs_invoke.py

Observa: con Converse, cambiar a otro proveedor es cambiar modelId; con InvokeModel tendrías que reescribir el cuerpo y el parseo. requestMetadata sirve para filtrar las invocaciones en los logs de invocación.

Paso 2: streaming y tiempo hasta el primer token

Guarda stream.py:

import os, time, boto3
brt = boto3.client("bedrock-runtime", region_name=os.environ["AWS_REGION"])
MODELO = "eu.amazon.nova-micro-v1:0"
MSG = [{"role": "user", "content": [{"text": "Explica en 8 frases qué es RAG y cuándo usarlo."}]}]
CFG = {"maxTokens": 500, "temperature": 0.3}

t0 = time.perf_counter()
brt.converse(modelId=MODELO, messages=MSG, inferenceConfig=CFG)
print(f"Sin streaming: primera palabra visible a los {time.perf_counter() - t0:.2f} s (todo de golpe)\n")

t0 = time.perf_counter(); primero = None
resp = brt.converse_stream(modelId=MODELO, messages=MSG, inferenceConfig=CFG)
for ev in resp["stream"]:
    if "contentBlockDelta" in ev:
        if primero is None:
            primero = time.perf_counter() - t0
        print(ev["contentBlockDelta"]["delta"].get("text", ""), end="", flush=True)
    elif "messageStop" in ev:
        print("\n\nstopReason:", ev["messageStop"]["stopReason"])
    elif "metadata" in ev:
        print("usage:", ev["metadata"]["usage"], "| métricas:", ev["metadata"].get("metrics"))
print(f"Con streaming: primer token a los {primero:.2f} s; total {time.perf_counter() - t0:.2f} s")

python3 stream.py

El tiempo total es parecido, pero con streaming el usuario empieza a leer casi al instante: eso es lo que mejora la experiencia (skill 2.4.2).

Paso 3: tool use con un bucle robusto

Guarda herramientas.py. Dos herramientas simuladas; el bucle valida argumentos, devuelve errores al modelo como toolResult con status: error y corta tras 5 iteraciones:

import os, re, json, boto3
brt = boto3.client("bedrock-runtime", region_name=os.environ["AWS_REGION"])
MODELO = "eu.amazon.nova-micro-v1:0"

PEDIDOS = {"P-1001": {"estado": "en reparto", "importe": 59.90}, "P-2002": {"estado": "entregado", "importe": 120.00}}

def estado_pedido(numero):
    if not re.fullmatch(r"P-\d{4}", numero or ""):
        raise ValueError("Formato de número inválido; debe ser P-NNNN")
    if numero not in PEDIDOS:
        raise KeyError(f"No existe el pedido {numero}")
    return {"numero": numero, "estado": PEDIDOS[numero]["estado"]}

def calcular_reembolso(numero, porcentaje):
    if not 0 < float(porcentaje) <= 100:
        raise ValueError("El porcentaje debe estar entre 1 y 100")
    return {"numero": numero, "reembolso_eur": round(PEDIDOS[numero]["importe"] * float(porcentaje) / 100, 2)}

FUNCIONES = {"estado_pedido": estado_pedido, "calcular_reembolso": calcular_reembolso}
TOOLS = {"tools": [
    {"toolSpec": {"name": "estado_pedido", "description": "Devuelve el estado de un pedido.",
        "inputSchema": {"json": {"type": "object", "properties": {
            "numero": {"type": "string", "description": "Número de pedido con formato P-NNNN"}},
            "required": ["numero"]}}}},
    {"toolSpec": {"name": "calcular_reembolso", "description": "Calcula el reembolso de un pedido a partir de un porcentaje.",
        "inputSchema": {"json": {"type": "object", "properties": {
            "numero": {"type": "string"}, "porcentaje": {"type": "number", "description": "Entre 1 y 100"}},
            "required": ["numero", "porcentaje"]}}}},
]}

def chat(pregunta, max_iter=5):
    mensajes = [{"role": "user", "content": [{"text": pregunta}]}]
    for _ in range(max_iter):
        r = brt.converse(modelId=MODELO, messages=mensajes, toolConfig=TOOLS,
                         system=[{"text": "Eres un agente de atención al cliente. Usa las herramientas; no inventes datos."}],
                         inferenceConfig={"maxTokens": 400, "temperature": 0})
        msg = r["output"]["message"]; mensajes.append(msg)
        if r["stopReason"] != "tool_use":
            return next((b["text"] for b in msg["content"] if "text" in b), "")
        resultados = []
        for b in msg["content"]:
            if "toolUse" not in b:
                continue
            tu = b["toolUse"]
            try:
                salida = FUNCIONES[tu["name"]](**tu["input"])
                resultados.append({"toolResult": {"toolUseId": tu["toolUseId"], "content": [{"json": salida}], "status": "success"}})
                print(f"  -> {tu['name']}({tu['input']}) = {salida}")
            except Exception as e:
                resultados.append({"toolResult": {"toolUseId": tu["toolUseId"], "content": [{"text": str(e)}], "status": "error"}})
                print(f"  -> {tu['name']}({tu['input']}) ERROR: {e}")
        mensajes.append({"role": "user", "content": resultados})
    return "He alcanzado el límite de pasos; te paso con un agente humano."

print(chat("¿Cómo va mi pedido P-1001?"), "\n")
print(chat("Quiero que me devolváis el 25 % del pedido P-2002. ¿Cuánto sería?"), "\n")
print(chat("¿Y el pedido P-9999?"))

python3 herramientas.py

Fíjate en el tercer caso: la herramienta lanza un error, el modelo lo recibe como toolResult con status: error y responde al usuario con sentido en vez de inventarse un estado. Validar argumentos, devolver errores estructurados y limitar iteraciones es la base de las integraciones fiables (2.1.6, 2.4.3).

Paso 4: errores, truncado y reintentos

Guarda resiliencia.py:

import os, time, random, boto3
from botocore.config import Config
from botocore.exceptions import ClientError

REGION = os.environ["AWS_REGION"]
# Cliente con reintentos adaptativos del SDK, timeouts y keep-alive
brt = boto3.client("bedrock-runtime", region_name=REGION, config=Config(
    retries={"mode": "adaptive", "total_max_attempts": 6}, read_timeout=120, connect_timeout=5, tcp_keepalive=True))
MSG = [{"role": "user", "content": [{"text": "Enumera 20 buenas prácticas de seguridad en la nube, una por línea."}]}]

def codigo(e):
    return e.response["Error"]["Code"]

# 1) ValidationException: parámetro fuera de rango (Nova Micro admite hasta 5K tokens de salida)
try:
    brt.converse(modelId="eu.amazon.nova-micro-v1:0", messages=MSG, inferenceConfig={"maxTokens": 100000})
except ClientError as e:
    print("1)", codigo(e), "->", e.response["Error"]["Message"][:120])

# 2) Modelo inexistente: error de cliente, no se debe reintentar
try:
    brt.converse(modelId="eu.amazon.nova-inexistente-v1:0", messages=MSG)
except ClientError as e:
    print("2)", codigo(e), "->", e.response["Error"]["Message"][:120])

# 3) Truncado: stopReason = max_tokens
r = brt.converse(modelId="eu.amazon.nova-micro-v1:0", messages=MSG, inferenceConfig={"maxTokens": 30})
print("3) stopReason:", r["stopReason"], "| salida cortada:", r["output"]["message"]["content"][0]["text"][-40:])

# 4) Backoff exponencial con jitter completo (por si reintentas en tu propia capa)
REINTENTABLES = {"ThrottlingException", "ServiceUnavailableException", "InternalServerException", "ModelNotReadyException"}

def con_reintentos(fn, intentos=5, base=0.5, tope=8.0):
    for i in range(intentos):
        try:
            return fn()
        except ClientError as e:
            if codigo(e) not in REINTENTABLES or i == intentos - 1:
                raise
            espera = random.uniform(0, min(tope, base * 2 ** i))
            print(f"   {codigo(e)}: reintento {i + 1} en {espera:.2f} s")
            time.sleep(espera)

fallos = {"n": 0}
def llamada_inestable():  # simula dos ThrottlingException antes de funcionar
    if fallos["n"] < 2:
        fallos["n"] += 1
        raise ClientError({"Error": {"Code": "ThrottlingException", "Message": "Too many requests"}}, "Converse")
    return "ok"
print("4)", con_reintentos(llamada_inestable))

# 5) Fallback de modelo: si el primero falla con un error no transitorio, probar el siguiente
CADENA = ["eu.amazon.nova-inexistente-v1:0", "eu.amazon.nova-lite-v1:0", "eu.amazon.nova-micro-v1:0"]
def converse_con_fallback(mensajes):
    for modelo in CADENA:
        try:
            r = brt.converse(modelId=modelo, messages=mensajes, inferenceConfig={"maxTokens": 100})
            return modelo, r["output"]["message"]["content"][0]["text"]
        except ClientError as e:
            print(f"   {modelo} falló con {codigo(e)}; paso al siguiente")
    return None, "Servicio no disponible: respuesta degradada."
print("5)", converse_con_fallback([{"role": "user", "content": [{"text": "Di hola en una palabra."}]}]))

python3 resiliencia.py

Qué debes llevarte: los errores 400/403/404 no se reintentan (se corrigen); los 429/500/503 sí, con backoff exponencial y jitter. El modo adaptive añade además limitación de tasa en el cliente. La lista de modelos de fallback, en producción, iría en AWS AppConfig para cambiarla sin desplegar.

import os, boto3
from botocore.exceptions import ClientError
brt = boto3.client("bedrock-runtime", region_name=os.environ["AWS_REGION"])
texto = "palabra " * 150000
try:
    brt.converse(modelId="eu.amazon.nova-micro-v1:0",
                 messages=[{"role": "user", "content": [{"text": texto + "\nResume lo anterior."}]}],
                 inferenceConfig={"maxTokens": 50})
except ClientError as e:
    print(e.response["Error"]["Code"], "->", e.response["Error"]["Message"][:200])

Paso 5: rol de la Lambda de streaming

La Lambda necesita escribir logs e invocar el modelo en streaming. Con un perfil de inferencia, el permiso debe cubrir el perfil y el modelo en las regiones de destino:

cat > confianza-lambda.json <<'EOF'
{ "Version": "2012-10-17", "Statement": [{ "Effect": "Allow",
  "Principal": { "Service": "lambda.amazonaws.com" }, "Action": "sts:AssumeRole" }] }
EOF

cat > permisos-lambda.json <<EOF
{ "Version": "2012-10-17", "Statement": [
  { "Effect": "Allow", "Action": ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream"],
    "Resource": [
      "arn:aws:bedrock:$AWS_REGION:$CUENTA:inference-profile/eu.amazon.nova-micro-v1:0",
      "arn:aws:bedrock:*::foundation-model/amazon.nova-micro-v1:0"
    ] }
] }
EOF

aws iam create-role --role-name lab08-lambda-role --assume-role-policy-document file://confianza-lambda.json
aws iam attach-role-policy --role-name lab08-lambda-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
aws iam put-role-policy --role-name lab08-lambda-role --policy-name lab08-bedrock --policy-document file://permisos-lambda.json
export ROL_LAMBDA_ARN=$(aws iam get-role --role-name lab08-lambda-role --query Role.Arn --output text)
sleep 10

Paso 6: Lambda en Node.js con response streaming

El streaming de respuestas de Lambda hacia API Gateway usa el decorador awslambda.streamifyResponse de los runtimes de Node.js. HttpResponseStream.from envía primero el código de estado y las cabeceras; después se escribe el cuerpo por trozos a medida que llegan los eventos de ConverseStream.

mkdir -p funcion && cat > funcion/index.mjs <<'EOF'
import { BedrockRuntimeClient, ConverseStreamCommand } from "@aws-sdk/client-bedrock-runtime";

const client = new BedrockRuntimeClient({ region: process.env.AWS_REGION });

export const handler = awslambda.streamifyResponse(async (event, responseStream) => {
  let pregunta = "";
  try { pregunta = (JSON.parse(event.body || "{}").pregunta || "").slice(0, 2000); } catch { pregunta = ""; }

  responseStream = awslambda.HttpResponseStream.from(responseStream, {
    statusCode: pregunta ? 200 : 400,
    headers: { "Content-Type": "text/plain; charset=utf-8" },
  });
  if (!pregunta) {
    responseStream.write("Falta el campo 'pregunta' en el cuerpo JSON.\n");
    responseStream.end();
    return;
  }

  try {
    const resp = await client.send(new ConverseStreamCommand({
      modelId: process.env.MODEL_ID,
      system: [{ text: "Responde en español de España, de forma clara y breve." }],
      messages: [{ role: "user", content: [{ text: pregunta }] }],
      inferenceConfig: { maxTokens: 500, temperature: 0.3 },
    }));
    for await (const ev of resp.stream) {
      const texto = ev.contentBlockDelta?.delta?.text;
      if (texto) responseStream.write(texto);
    }
  } catch (e) {
    console.error(e);
    responseStream.write(`\n[error: ${e.name}]\n`);
  }
  responseStream.end();
});
EOF

(cd funcion && zip -q ../funcion.zip index.mjs)

aws lambda create-function --function-name lab08-stream \
  --runtime nodejs22.x --handler index.handler \
  --role $ROL_LAMBDA_ARN --zip-file fileb://funcion.zip \
  --timeout 120 --memory-size 256 \
  --environment "Variables={MODEL_ID=eu.amazon.nova-micro-v1:0}" \
  --region $AWS_REGION --query FunctionArn --output text
aws lambda wait function-active-v2 --function-name lab08-stream --region $AWS_REGION
export LAMBDA_ARN=$(aws lambda get-function --function-name lab08-stream --region $AWS_REGION \
  --query Configuration.FunctionArn --output text)

Paso 7: API REST de API Gateway con transferencia en streaming

La integración usa el URI de Lambda terminado en /response-streaming-invocations y responseTransferMode: STREAM. Con streaming, el timeout de integración puede llegar a 15 minutos (aquí usamos 90 s); sin streaming, el valor por defecto de una integración REST es de 29 s.

cat > api.yaml <<EOF
openapi: "3.0.1"
info:
  title: "lab08-chat-streaming"
  version: "1.0"
paths:
  /chat:
    post:
      x-amazon-apigateway-integration:
        type: "aws_proxy"
        httpMethod: "POST"
        uri: "arn:aws:apigateway:$AWS_REGION:lambda:path/2021-11-15/functions/$LAMBDA_ARN/response-streaming-invocations"
        responseTransferMode: "STREAM"
        timeoutInMillis: 90000
EOF

export API_ID=$(aws apigateway import-rest-api --body fileb://api.yaml \
  --parameters endpointConfigurationTypes=REGIONAL --region $AWS_REGION --query id --output text)

aws lambda add-permission --function-name lab08-stream --statement-id apigw-invoke \
  --action lambda:InvokeFunction --principal apigateway.amazonaws.com \
  --source-arn "arn:aws:execute-api:$AWS_REGION:$CUENTA:$API_ID/*/POST/chat" --region $AWS_REGION

aws apigateway create-deployment --rest-api-id $API_ID --stage-name prod --region $AWS_REGION >/dev/null
export URL="https://$API_ID.execute-api.$AWS_REGION.amazonaws.com/prod/chat"
echo $URL

Paso 8: prueba el streaming

curl -N desactiva el búfer para que veas llegar el texto por trozos:

curl -N -X POST "$URL" -H "Content-Type: application/json" \
  -d '{"pregunta":"Explica en 6 frases cómo funciona el backoff exponencial con jitter."}'
echo
curl -s -X POST "$URL" -H "Content-Type: application/json" -d '{}' -w "\nHTTP %{http_code}\n"

La primera petición muestra el texto apareciendo poco a poco; la segunda devuelve 400 con el mensaje de validación. En un despliegue real añadirías un request validator con un modelo JSON Schema en API Gateway (para rechazar peticiones mal formadas antes de invocar la Lambda), throttling por etapa o usage plans con API keys, AWS WAF y X-Ray activado en la etapa y en la Lambda para trazar la llamada de extremo a extremo.

Comprueba que funciona

• converse_vs_invoke.py muestra usage, stopReason y latencia, y la respuesta nativa de InvokeModel.
• stream.py muestra un primer token mucho antes que el total.
• herramientas.py llama a las herramientas y gestiona el pedido inexistente sin inventar.
• resiliencia.py muestra ValidationException, el truncado por max_tokens, dos reintentos simulados y el fallback.
• curl -N recibe la respuesta en streaming a través de API Gateway.

Limpieza

Orden: API → permiso y función Lambda → logs → rol → ficheros.

aws apigateway delete-rest-api --rest-api-id $API_ID --region $AWS_REGION
aws lambda delete-function --function-name lab08-stream --region $AWS_REGION
aws logs delete-log-group --log-group-name /aws/lambda/lab08-stream --region $AWS_REGION 2>/dev/null
aws iam delete-role-policy --role-name lab08-lambda-role --policy-name lab08-bedrock
aws iam detach-role-policy --role-name lab08-lambda-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
aws iam delete-role --role-name lab08-lambda-role
rm -rf ~/lab08

aws apigateway get-rest-apis --region $AWS_REGION --query "items[?name=='lab08-chat-streaming'].id"

Preguntas para pensar como arquitecto