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

Prompt Management y Flows con versiones, regresión y alias

Qué vas a construir

Un clasificador de tickets de soporte gobernado como un artefacto de software:

1. Un prompt gestionado en Amazon Bedrock Prompt Management, con variables, publicado como versión 1.
2. Una prueba de regresión con un conjunto de oro que calcula la exactitud y la publica como métrica de CloudWatch por versión.
3. Una versión 2 mejorada con few-shot, comparada con la 1 antes de «aprobarla».
4. Un Amazon Bedrock Flow (la guía los llama Prompt Flows) con rama condicional que reutiliza el prompt gestionado, desplegado con versión y alias, y un rollback cambiando el alias.
5. La auditoría de los cambios en CloudTrail.

flowchart LR
  PM["Prompt Management: DRAFT"] --> V1["Versión 1"]
  PM --> V2["Versión 2 (few-shot)"]
  V1 --> REG["Regresión: exactitud por versión en CloudWatch"]
  V2 --> REG
  V2 --> FLOW["Flow: Input, Condition, Prompt, Outputs"]
  FLOW --> FV["Versión del Flow"]
  FV --> ALIAS["Alias prod"]
  APP["InvokeFlow"] --> ALIAS
  CT["CloudTrail"] -.-> PM

Antes de empezar

• Región: eu-central-1. Modelo: Amazon Nova Micro mediante el perfil EU eu.amazon.nova-micro-v1:0.
• CloudShell en eu-central-1 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 ~/lab07 && cd ~/lab07

Paso 1: crea el prompt con variables

Plantilla CHAT con system prompt, un mensaje de usuario con dos variables (categorias y ticket) y parámetros deterministas. Las variables se escriben entre dobles llaves dentro del texto:

cat > variantes-v1.json <<'EOF'
[
  {
    "name": "v1",
    "modelId": "eu.amazon.nova-micro-v1:0",
    "templateType": "CHAT",
    "templateConfiguration": {
      "chat": {
        "system": [{ "text": "Eres un clasificador de tickets de soporte de una operadora de telecomunicaciones. Responde solo con una categoría, en minúsculas y sin tildes." }],
        "messages": [{ "role": "user", "content": [{ "text": "Categorías posibles: {{categorias}}\nTicket: {{ticket}}\nCategoría:" }] }],
        "inputVariables": [{ "name": "categorias" }, { "name": "ticket" }]
      }
    },
    "inferenceConfiguration": { "text": { "maxTokens": 10, "temperature": 0 } }
  }
]
EOF

export PROMPT_ID=$(aws bedrock-agent create-prompt \
  --name lab07-clasificador \
  --description "Clasificador de tickets (lab 07)" \
  --default-variant v1 \
  --variants file://variantes-v1.json \
  --region $AWS_REGION --query id --output text)
echo "PROMPT_ID=$PROMPT_ID"

Si quieres verlo en la consola: Amazon Bedrock → Prompt management → lab07-clasificador → Edit in prompt builder. Rellena las variables de prueba y pulsa Run.

Paso 2: publica la versión 1 e invócala desde Converse

El borrador (DRAFT) se puede seguir editando; la versión es una instantánea inmutable, que es lo que debe usar producción.

export PROMPT_V1=$(aws bedrock-agent create-prompt-version --prompt-identifier $PROMPT_ID \
  --description "Versión inicial zero-shot" --region $AWS_REGION --query arn --output text)
echo "PROMPT_V1=$PROMPT_V1"   # termina en :1

Guarda invocar.py:

import os, sys, boto3
brt = boto3.client("bedrock-runtime", region_name=os.environ["AWS_REGION"])
CATEGORIAS = "facturacion, averia, portabilidad, baja, otros"

def clasificar(prompt_arn, ticket):
    r = brt.converse(
        modelId=prompt_arn,
        promptVariables={"categorias": {"text": CATEGORIAS}, "ticket": {"text": ticket}},
    )
    return r["output"]["message"]["content"][0]["text"].strip().lower().rstrip(".")

if __name__ == "__main__":
    print(clasificar(os.environ["PROMPT_V1"], "Me han cobrado dos veces la factura de agosto."))
    # Prueba la restricción: con un prompt gestionado no se puede enviar inferenceConfig
    try:
        brt.converse(modelId=os.environ["PROMPT_V1"],
                     promptVariables={"categorias": {"text": CATEGORIAS}, "ticket": {"text": "hola"}},
                     inferenceConfig={"temperature": 0.9})
    except brt.exceptions.ValidationException as e:
        print("ValidationException esperada:", str(e)[:120])

python3 invocar.py

Comprueba dos lecciones de examen: el modelId es el ARN de la versión del prompt y las variables van en promptVariables; y si intentas enviar inferenceConfig (o system, toolConfig, additionalModelRequestFields), la llamada falla, porque esos campos los fija el prompt gestionado. Tu identidad necesita bedrock:RenderPrompt sobre el prompt.

Paso 3: prueba de regresión con conjunto de oro y métrica en CloudWatch

Guarda regresion.py. Ejecuta el conjunto de oro contra una versión, valida que la salida sea una categoría permitida y publica la exactitud en CloudWatch con la dimensión PromptVersion:

import os, sys, boto3
from invocar import clasificar

PERMITIDAS = {"facturacion", "averia", "portabilidad", "baja", "otros"}
ORO = [
    ("Me habéis cobrado 15 euros de más este mes.", "facturacion"),
    ("No tengo línea desde ayer por la tarde.", "averia"),
    ("Quiero llevarme mi número a otra compañía.", "portabilidad"),
    ("Deseo cancelar mi contrato cuanto antes.", "baja"),
    ("¿Cuál es el horario de vuestra tienda de Sevilla?", "otros"),
    ("El router parpadea en rojo y no hay internet.", "averia"),
    ("La factura viene a nombre de mi antiguo piso.", "facturacion"),
    ("Me mudo al extranjero y no voy a seguir con vosotros.", "baja"),
]

def evaluar(prompt_arn, etiqueta_version):
    aciertos = formato_ok = 0
    for ticket, esperada in ORO:
        salida = clasificar(prompt_arn, ticket)
        formato_ok += salida in PERMITIDAS
        aciertos += salida == esperada
        print(f"  {'OK ' if salida == esperada else 'MAL'} esperada={esperada:<12} obtenida={salida}")
    exactitud = aciertos / len(ORO)
    boto3.client("cloudwatch", region_name=os.environ["AWS_REGION"]).put_metric_data(
        Namespace="Lab07/Prompts",
        MetricData=[
            {"MetricName": "Exactitud", "Value": exactitud, "Unit": "None",
             "Dimensions": [{"Name": "PromptVersion", "Value": etiqueta_version}]},
            {"MetricName": "FormatoValido", "Value": formato_ok / len(ORO), "Unit": "None",
             "Dimensions": [{"Name": "PromptVersion", "Value": etiqueta_version}]},
        ],
    )
    print(f"Versión {etiqueta_version}: exactitud={exactitud:.2f} formato={formato_ok}/{len(ORO)}")
    return exactitud

if __name__ == "__main__":
    evaluar(sys.argv[1], sys.argv[2])

python3 regresion.py "$PROMPT_V1" v1

Esto es la skill 1.6.4 en pequeño: un validador (en producción, una Lambda), un conjunto de casos (en producción, cientos, orquestados con un estado Map de Step Functions) y una métrica de regresión en CloudWatch sobre la que puedes poner una alarma.

Paso 4: mejora el prompt (few-shot) y publica la versión 2

Actualiza el borrador con ejemplos few-shot y reglas de desempate. update-prompt sustituye la definición completa del borrador:

cat > variantes-v2.json <<'EOF'
[
  {
    "name": "v1",
    "modelId": "eu.amazon.nova-micro-v1:0",
    "templateType": "CHAT",
    "templateConfiguration": {
      "chat": {
        "system": [{ "text": "Eres un clasificador de tickets de soporte de una operadora de telecomunicaciones. Responde solo con una categoría de la lista, en minúsculas y sin tildes. Si el cliente quiere irse a otra compañía conservando el número, es portabilidad; si quiere dejar de ser cliente sin más, es baja; problemas técnicos de línea, router o cobertura son averia." }],
        "messages": [
          { "role": "user", "content": [{ "text": "Categorías posibles: facturacion, averia, portabilidad, baja, otros\nTicket: El móvil no tiene cobertura en casa.\nCategoría:" }] },
          { "role": "assistant", "content": [{ "text": "averia" }] },
          { "role": "user", "content": [{ "text": "Categorías posibles: facturacion, averia, portabilidad, baja, otros\nTicket: Me paso a otra operadora con mi mismo número.\nCategoría:" }] },
          { "role": "assistant", "content": [{ "text": "portabilidad" }] },
          { "role": "user", "content": [{ "text": "Categorías posibles: {{categorias}}\nTicket: {{ticket}}\nCategoría:" }] }
        ],
        "inputVariables": [{ "name": "categorias" }, { "name": "ticket" }]
      }
    },
    "inferenceConfiguration": { "text": { "maxTokens": 10, "temperature": 0 } }
  }
]
EOF

aws bedrock-agent update-prompt --prompt-identifier $PROMPT_ID --name lab07-clasificador \
  --default-variant v1 --variants file://variantes-v2.json --region $AWS_REGION >/dev/null

export PROMPT_V2=$(aws bedrock-agent create-prompt-version --prompt-identifier $PROMPT_ID \
  --description "Few-shot y reglas de desempate" --region $AWS_REGION --query arn --output text)
echo "PROMPT_V2=$PROMPT_V2"   # termina en :2

python3 regresion.py "$PROMPT_V2" v2
python3 regresion.py "$PROMPT_V1" v1   # la versión 1 sigue intacta: las versiones son inmutables

Abre CloudWatch → Metrics → All metrics → Lab07/Prompts → PromptVersion y compara Exactitud de v1 y v2. En un flujo de gobierno real, este paso lo haría CodeBuild dentro de CodePipeline, con una aprobación manual antes de publicar el ARN de la nueva versión en AWS AppConfig.

Paso 5: construye el Flow en la consola

Vas a crear este flujo: si el cliente es premium, se clasifica el ticket con el prompt gestionado (versión 2) y se devuelve la categoría; si no, se devuelve un mensaje fijo de cola estándar.

flowchart LR
  IN["Flow input (Object)"] --> C{"Condition: plan == premium"}
  C -- "premium" --> P["Prompt (Prompt Management, versión 2)"]
  P --> O1["Output: categoria"]
  C -- "default" --> O2["Output: cola_estandar"]

1. Consola de Amazon Bedrock → Flows → Create flow. Nombre lab07-flow. En Service role, elige Create and use a new service role (la consola añadirá los permisos de cada nodo, incluido bedrock:RenderPrompt).
2. En el Flow builder, selecciona el nodo Flow input y cambia su tipo de salida a Object. El flujo recibirá un JSON como el del paso 7, con los campos ticket y plan.
3. Arrastra un nodo Condition. En Inputs define una entrada plan de tipo String con la expresión $.data.plan y conéctala a la salida del nodo de entrada. En Conditions, añade una condición llamada premium con la expresión plan == "premium". Deja la condición por defecto.
4. Configura el nodo Prompt que trae el flujo por defecto (o arrastra uno): elige Use a prompt from your Prompt management, selecciona lab07-clasificador y la versión 2. Aparecerán sus entradas categorias y ticket:
   • ticket: tipo String, expresión $.data.ticket, conectada a la salida del nodo de entrada.
   • categorias: tipo String, expresión $.data.categorias, también desde el nodo de entrada (enviaremos las categorías en la petición; así el prompt es reutilizable).
5. Conecta la condición premium al nodo Prompt, y la salida modelCompletion del Prompt al nodo Flow output (renómbralo a categoria).
6. Añade un segundo Flow output llamado cola_estandar, con entrada de tipo String y expresión $.data.ticket, conectado a la salida del nodo de entrada. Conecta la condición default a este nodo.
7. Pulsa Save (la consola prepara el flujo) y pruébalo en el panel Test flow con los dos JSON del paso 7.

Paso 6: versión, alias e invocación desde código

export FLOW_ID=$(aws bedrock-agent list-flows --region $AWS_REGION \
  --query "flowSummaries[?name=='lab07-flow'].id" --output text)

aws bedrock-agent create-flow-version --flow-identifier $FLOW_ID --region $AWS_REGION \
  --query version --output text          # 1

export ALIAS_ID=$(aws bedrock-agent create-flow-alias --flow-identifier $FLOW_ID --name prod \
  --routing-configuration '[{"flowVersion":"1"}]' --region $AWS_REGION --query id --output text)
echo "FLOW_ID=$FLOW_ID ALIAS_ID=$ALIAS_ID"

Paso 7: invoca el alias

Guarda flow.py:

import os, boto3
art = boto3.client("bedrock-agent-runtime", region_name=os.environ["AWS_REGION"])

def invocar(ticket, plan):
    doc = {"ticket": ticket, "plan": plan, "categorias": "facturacion, averia, portabilidad, baja, otros"}
    r = art.invoke_flow(
        flowIdentifier=os.environ["FLOW_ID"], flowAliasIdentifier=os.environ["ALIAS_ID"],
        inputs=[{"nodeName": "FlowInputNode", "nodeOutputName": "document", "content": {"document": doc}}],
    )
    for ev in r["responseStream"]:
        if "flowOutputEvent" in ev:
            print(f"[{plan}] {ev['flowOutputEvent']['nodeName']}: {ev['flowOutputEvent']['content']['document']}")
        if "flowCompletionEvent" in ev:
            print("  fin:", ev["flowCompletionEvent"]["completionReason"])

invocar("El router no enciende desde esta mañana.", "premium")
invocar("El router no enciende desde esta mañana.", "basico")

python3 flow.py

Si tu nodo de entrada tiene otro nombre en la consola (por defecto es FlowInputNode), cámbialo en nodeName.

Paso 8: rollback con el alias

Simula una versión defectuosa: en la consola edita el nodo Prompt para que use la versión 1 del prompt, guarda y crea la versión 2 del Flow. Apunta el alias a ella y, después, vuelve atrás sin tocar el código de la aplicación:

aws bedrock-agent create-flow-version --flow-identifier $FLOW_ID --region $AWS_REGION --query version --output text   # 2
aws bedrock-agent update-flow-alias --flow-identifier $FLOW_ID --alias-identifier $ALIAS_ID --name prod \
  --routing-configuration '[{"flowVersion":"2"}]' --region $AWS_REGION
python3 flow.py
# Rollback: el alias vuelve a la versión 1
aws bedrock-agent update-flow-alias --flow-identifier $FLOW_ID --alias-identifier $ALIAS_ID --name prod \
  --routing-configuration '[{"flowVersion":"1"}]' --region $AWS_REGION

Paso 9: auditoría con CloudTrail

Las llamadas de gestión a Prompt Management y Flows quedan en el historial de eventos de CloudTrail (skill 1.6.3):

for EV in CreatePrompt UpdatePrompt CreatePromptVersion CreateFlowAlias UpdateFlowAlias; do
  aws cloudtrail lookup-events --region $AWS_REGION --max-results 3 \
    --lookup-attributes AttributeKey=EventName,AttributeValue=$EV \
    --query 'Events[].[EventTime,EventName,Username]' --output text
done

Los eventos pueden tardar unos minutos en aparecer. Para registrar además el contenido de prompts y respuestas, activarías el model invocation logging de Bedrock hacia CloudWatch Logs o S3 (módulo 08).

Comprueba que funciona

• invocar.py clasifica el ticket y muestra la ValidationException esperada al enviar inferenceConfig.
• CloudWatch muestra Exactitud para v1 y v2 en Lab07/Prompts.
• flow.py devuelve una categoría para premium y el ticket en cola_estandar para basico.
• Tras el rollback, el alias prod apunta a la versión 1 del Flow.
• CloudTrail lista los eventos de creación y actualización.

Limpieza

Orden: alias → versiones del Flow → Flow → prompt → rol creado por la consola → ficheros locales.

aws bedrock-agent delete-flow-alias --flow-identifier $FLOW_ID --alias-identifier $ALIAS_ID --region $AWS_REGION
for V in 1 2; do
  aws bedrock-agent delete-flow-version --flow-identifier $FLOW_ID --flow-version $V --region $AWS_REGION 2>/dev/null
done
aws bedrock-agent delete-flow --flow-identifier $FLOW_ID --region $AWS_REGION
aws bedrock-agent delete-prompt --prompt-identifier $PROMPT_ID --region $AWS_REGION

# Rol de servicio que creó la consola para el Flow (su nombre empieza por AmazonBedrockExecutionRoleForFlows)
aws iam list-roles --query "Roles[?starts_with(RoleName,'AmazonBedrockExecutionRoleForFlows')].RoleName" --output text

Para cada rol listado que corresponda a lab07-flow, borra sus políticas y el rol (en la consola de IAM es más cómodo: Roles → el rol → Delete). Por último, rm -rf ~/lab07. La métrica Lab07/Prompts de CloudWatch no se puede borrar: deja de cobrarse al no recibir datos y caduca sola.

Preguntas para pensar como arquitecto