Módulo 4: Bedrock Guardrails And Defense In Depth
6. Manos a la obra: el validador del esquema de salida
Descripción
La lección 4 lo demostró con un ejemplo concreto: una respuesta puede pasar las seis políticas de Bedrock Guardrails sin ninguna objeción y, aun así, no tener la forma que Shipments necesita. Esta lección construye la pieza que cierra esa brecha exacta: guardrails/post_invoke_checks.py, un validador determinista que confirma, campo por campo, que una respuesta —real o representativa— coincide con ShipmentFields, el contrato de cinco campos que parse_manifest() (Módulo 1, lección 3) ya produce para el camino determinista, antes de que cualquier código intente escribirla en Shipments. Corrió de verdad para escribir esta lección; el bloque "Qué esperar" de pytest es salida literal, dieciséis casos fijos.
Conexión con el módulo
Donde la lección 5 protege la entrada de extract-shipment-manifest-fields, esta lección protege su salida — el mismo patrón de dos capas que la lección 1 de este módulo ya anticipó en su mapa. Ninguna de las dos depende de la otra; ninguna de las dos depende de que Bedrock exista.
Analogía: el control de calidad al final de la línea, no en la puerta de entrada
Un almacén de logística serio no confía en que la mercancía que entra por la puerta esté correctamente empacada — tiene, además, un control de calidad al final de la línea de empaque, justo antes de que una caja salga hacia el camión: ¿la caja tiene todos los artículos que la guía de envío dice que debería tener? ¿Falta algo? ¿Sobra algo que no debería estar ahí? Ese control no le importa si la mercancía adentro es peligrosa o no —eso ya se revisó en un punto anterior—; le importa exclusivamente si la caja, tal como está, coincide con lo que el sistema espera recibir. post_invoke_checks.py es exactamente ese control de calidad final, aplicado a una respuesta de Bedrock en vez de a una caja física: no vuelve a preguntar "¿esto es seguro?" (eso ya lo evaluó el guardrail gestionado en la lección 3) — pregunta, exclusivamente, "¿esto tiene la forma exacta que Shipments necesita recibir?".
Paso 1 — El script completo
guardrails/post_invoke_checks.py, en la raíz de andes-cargo-infra/:
#!/usr/bin/env python3
"""post_invoke_checks.py -- a deterministic, local schema validator that runs
AFTER extract-shipment-manifest-fields gets a response back from Bedrock (or,
in this $0 lab, a representative dict standing in for one -- see Module 4,
lesson 7), and BEFORE that response is ever written to Shipments.
This is defense in depth on the OUTPUT side (pre_invoke_checks.py is the input
side): Bedrock Guardrails' contextual grounding policy (bedrock.tf, Module 4,
lesson 3) can tell you a response is grounded in the manifest text -- it says
nothing about whether that response has the exact five fields Shipments
requires. A perfectly grounded, perfectly ungrounded-free response that is
missing weightKg is still not writable to Shipments. That gap is this script's
entire job (Module 4, lesson 4).
ShipmentFields is the name this guide gives to that contract: exactly the five
fields parse_manifest() already produces for the deterministic path (Module 1,
lesson 3) -- shipmentId, originCountry, destinationCountry, carrier, weightKg
-- all non-empty strings, because write_shipment_record() expects the LLM path
to hand it the same shape the deterministic path always has.
Never uses random or datetime.now(). Given the same candidate dict, this script
always produces the same PASS/FAIL result. Run the test suite with:
pytest guardrails/test_post_invoke_checks.py -v
"""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import dataclass, field
# The exact contract Shipments requires -- the same five keys parse_manifest()
# (Module 1, lesson 3) produces from a well-formed clave=valor manifest. The
# LLM path must match this shape exactly, or write_shipment_record() would
# receive a dict it was never built to handle.
SHIPMENT_FIELDS_SCHEMA = (
"shipmentId",
"originCountry",
"destinationCountry",
"carrier",
"weightKg",
)
@dataclass(frozen=True)
class ValidationResult:
candidate: dict
missing_fields: tuple[str, ...] = field(default_factory=tuple)
empty_fields: tuple[str, ...] = field(default_factory=tuple)
unexpected_fields: tuple[str, ...] = field(default_factory=tuple)
weight_not_numeric: bool = False
@property
def is_valid(self) -> bool:
return not (
self.missing_fields
or self.empty_fields
or self.unexpected_fields
or self.weight_not_numeric
)
@property
def errors(self) -> tuple[str, ...]:
errors: list[str] = []
if self.missing_fields:
errors.append(f"missing required field(s): {', '.join(self.missing_fields)}")
if self.empty_fields:
errors.append(f"empty value for field(s): {', '.join(self.empty_fields)}")
if self.unexpected_fields:
errors.append(f"unexpected field(s) not in ShipmentFields: {', '.join(self.unexpected_fields)}")
if self.weight_not_numeric:
errors.append("weightKg is not a numeric string")
return tuple(errors)
def _is_numeric_string(value: object) -> bool:
if not isinstance(value, str):
return False
try:
float(value)
except ValueError:
return False
return True
def validate_shipment_fields(candidate: dict) -> ValidationResult:
"""Validate that `candidate` matches ShipmentFields exactly: all five
required keys present, every value a non-empty string, weightKg parseable
as a number, and no extra keys the deterministic path never produces."""
required = set(SHIPMENT_FIELDS_SCHEMA)
present = set(candidate.keys())
missing = tuple(sorted(required - present))
unexpected = tuple(sorted(present - required))
empty = tuple(
k
for k in SHIPMENT_FIELDS_SCHEMA
if k in candidate and (not isinstance(candidate[k], str) or candidate[k].strip() == "")
)
weight_not_numeric = "weightKg" in candidate and "weightKg" not in empty and not _is_numeric_string(candidate["weightKg"])
return ValidationResult(
candidate=candidate,
missing_fields=missing,
empty_fields=empty,
unexpected_fields=unexpected,
weight_not_numeric=weight_not_numeric,
)
def format_report(result: ValidationResult) -> str:
status = "PASS" if result.is_valid else "FAIL"
lines = [f"post_invoke_checks: {status}"]
if result.is_valid:
lines.append(" All five ShipmentFields present, non-empty, weightKg numeric.")
else:
for err in result.errors:
lines.append(f" - {err}")
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Deterministic ShipmentFields schema check, before a candidate dict is written to Shipments."
)
parser.add_argument("--json", required=True, dest="json_text", help="candidate fields as a JSON object")
args = parser.parse_args(argv)
try:
candidate = json.loads(args.json_text)
except json.JSONDecodeError as exc:
print(f"error: --json is not valid JSON: {exc}", file=sys.stderr)
return 2
if not isinstance(candidate, dict):
print("error: --json must decode to a JSON object", file=sys.stderr)
return 2
result = validate_shipment_fields(candidate)
print(format_report(result))
return 0 if result.is_valid else 1
if __name__ == "__main__":
raise SystemExit(main())Cuatro tipos de fallo, cada uno una categoría distinta de "esta caja no coincide con la guía de envío": campos faltantes (missing_fields), valores vacíos (empty_fields — una cadena vacía o solo espacios cuenta como si el campo no existiera), campos inesperados (unexpected_fields — algo que ShipmentFields nunca esperó recibir), y un caso específico de tipo (weight_not_numeric — weightKg tiene que poder interpretarse como número, aunque se almacene como cadena, exactamente como parse_manifest() ya lo entrega). El código de salida de main() sí distingue PASS de FAIL (0 o 1) — a diferencia de pre_invoke_checks.py de la lección 5, este chequeo sí decide si algo continúa: una respuesta que falla aquí nunca debería llegar a Shipments.
Paso 2 — Corriendo el validador, de verdad
Un candidato representativo que sí coincide con ShipmentFields — el envío 4471 del Módulo 1, lección 3, ahora en la forma que una extracción exitosa produciría:
python3 guardrails/post_invoke_checks.py --json '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress", "weightKg": "120"}'Qué esperar (literal — corrido de verdad, mismo entorno de esta guía; el JSON de entrada es un candidato representativo hecho a mano, nunca la salida de una invocación real, ver lección 7):
post_invoke_checks: PASS
All five ShipmentFields present, non-empty, weightKg numeric.
Ahora, exactamente el caso de la Brecha 2 de la lección 4 — el mismo candidato, sin weightKg:
python3 guardrails/post_invoke_checks.py --json '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress"}'Qué esperar (literal):
post_invoke_checks: FAIL
- missing required field(s): weightKg
Código de salida 1 — este es, exactamente, el candidato que la lección 4 demostró que las seis políticas de Bedrock Guardrails aprobarían sin ninguna objeción. post_invoke_checks.py es la única capa de todo este módulo que lo rechaza, y lo hace con el mismo determinismo que cualquier otro chequeo de esta guía: el mismo JSON, siempre el mismo resultado.
Paso 3 — El suite de pytest, dieciséis casos fijos
guardrails/test_post_invoke_checks.py:
"""pytest suite for post_invoke_checks.py -- fixed, deterministic candidate
dicts only. No random, no datetime.now(): the same candidate must always
produce the same PASS/FAIL result, on any machine. Run with:
pytest guardrails/test_post_invoke_checks.py -v
Every candidate here is a REPRESENTATIVE dict -- never the output of a real
Bedrock invocation (Module 4, lesson 7 explains exactly why none was invoked).
These are hand-built fixtures shaped like what a real response would look
like, used only to exercise this script's own logic.
"""
import pytest
from post_invoke_checks import SHIPMENT_FIELDS_SCHEMA, format_report, main, validate_shipment_fields
VALID_SHIPMENT_4471 = {
"shipmentId": "4471",
"originCountry": "Peru",
"destinationCountry": "Chile",
"carrier": "AndesExpress",
"weightKg": "120",
}
def test_valid_candidate_passes():
result = validate_shipment_fields(VALID_SHIPMENT_4471)
assert result.is_valid is True
assert result.errors == ()
def test_missing_weight_kg_fails():
candidate = {k: v for k, v in VALID_SHIPMENT_4471.items() if k != "weightKg"}
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.missing_fields == ("weightKg",)
def test_missing_multiple_fields_reports_all_of_them():
candidate = {"shipmentId": "4472", "carrier": "AndesExpress"}
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.missing_fields == ("destinationCountry", "originCountry", "weightKg")
def test_empty_string_value_fails():
candidate = dict(VALID_SHIPMENT_4471, carrier="")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.empty_fields == ("carrier",)
def test_whitespace_only_value_counts_as_empty():
candidate = dict(VALID_SHIPMENT_4471, originCountry=" ")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.empty_fields == ("originCountry",)
def test_non_numeric_weight_fails():
candidate = dict(VALID_SHIPMENT_4471, weightKg="approximately 120kg")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.weight_not_numeric is True
def test_numeric_weight_as_plain_number_string_passes():
candidate = dict(VALID_SHIPMENT_4471, weightKg="45.5")
result = validate_shipment_fields(candidate)
assert result.is_valid is True
def test_unexpected_extra_field_fails():
"""A representative case of exactly what Module 4, lesson 4 warns about:
a response that IS grounded and DOES have the five required fields, but
also invents a sixth field (e.g. a hallucinated confidence score in the
wrong shape) that Shipments was never built to receive."""
candidate = dict(VALID_SHIPMENT_4471, confidenceScore="very high")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.unexpected_fields == ("confidenceScore",)
def test_empty_dict_reports_all_five_missing():
result = validate_shipment_fields({})
assert result.is_valid is False
assert set(result.missing_fields) == set(SHIPMENT_FIELDS_SCHEMA)
def test_schema_has_exactly_five_fields():
assert len(SHIPMENT_FIELDS_SCHEMA) == 5
def test_report_format_for_pass():
result = validate_shipment_fields(VALID_SHIPMENT_4471)
report = format_report(result)
assert "PASS" in report
def test_report_format_for_fail_lists_each_error():
candidate = {"shipmentId": "4471"}
result = validate_shipment_fields(candidate)
report = format_report(result)
assert "FAIL" in report
assert "missing required field(s)" in report
def test_cli_end_to_end_pass(capsys):
exit_code = main(["--json", '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress", "weightKg": "120"}'])
captured = capsys.readouterr()
assert exit_code == 0
assert "PASS" in captured.out
def test_cli_end_to_end_fail_missing_field(capsys):
exit_code = main(["--json", '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress"}'])
captured = capsys.readouterr()
assert exit_code == 1
assert "FAIL" in captured.out
def test_cli_rejects_invalid_json(capsys):
exit_code = main(["--json", "{not valid json"])
captured = capsys.readouterr()
assert exit_code == 2
assert "error:" in captured.err
def test_cli_rejects_non_object_json(capsys):
exit_code = main(["--json", "[1, 2, 3]"])
captured = capsys.readouterr()
assert exit_code == 2
assert "error:" in captured.errDieciséis casos: el candidato válido, un campo faltante, múltiples campos faltantes, un valor vacío, un valor de solo espacios, un weightKg no numérico, un weightKg numérico válido, un campo inesperado (la Brecha 2 exacta de la lección 4), un diccionario completamente vacío, la forma del esquema en sí, el formato del reporte en sus dos estados, el flujo completo de la interfaz de línea de comandos en sus dos casos, y el rechazo de JSON inválido o de un JSON que no decodifica a un objeto.
cd guardrails/
pytest test_post_invoke_checks.py -v -p no:randomlyQué esperar (literal — corrido de verdad):
============================= test session starts ==============================
platform darwin -- Python 3.14.0, pytest-9.1.1, pluggy-1.6.0
rootdir: guardrails/
collected 16 items
test_post_invoke_checks.py::test_valid_candidate_passes PASSED [ 6%]
test_post_invoke_checks.py::test_missing_weight_kg_fails PASSED [ 12%]
test_post_invoke_checks.py::test_missing_multiple_fields_reports_all_of_them PASSED [ 18%]
test_post_invoke_checks.py::test_empty_string_value_fails PASSED [ 25%]
test_post_invoke_checks.py::test_whitespace_only_value_counts_as_empty PASSED [ 31%]
test_post_invoke_checks.py::test_non_numeric_weight_fails PASSED [ 37%]
test_post_invoke_checks.py::test_numeric_weight_as_plain_number_string_passes PASSED [ 43%]
test_post_invoke_checks.py::test_unexpected_extra_field_fails PASSED [ 50%]
test_post_invoke_checks.py::test_empty_dict_reports_all_five_missing PASSED [ 56%]
test_post_invoke_checks.py::test_schema_has_exactly_five_fields PASSED [ 62%]
test_post_invoke_checks.py::test_report_format_for_pass PASSED [ 68%]
test_post_invoke_checks.py::test_report_format_for_fail_lists_each_error PASSED [ 75%]
test_post_invoke_checks.py::test_cli_end_to_end_pass PASSED [ 81%]
test_post_invoke_checks.py::test_cli_end_to_end_fail_missing_field PASSED [ 87%]
test_post_invoke_checks.py::test_cli_rejects_invalid_json PASSED [ 93%]
test_post_invoke_checks.py::test_cli_rejects_non_object_json PASSED [100%]
============================== 16 passed in 0.02s ==============================
Dieciséis de dieciséis — incluido test_unexpected_extra_field_fails, el caso que hace explícito, en código ejecutable, exactamente el mismo argumento que la lección 4 desarrolló en prosa: una respuesta con confidenceScore agregado, un campo que ni parse_manifest() ni Shipments esperaron jamás, falla este chequeo aunque hubiera pasado las seis políticas de Bedrock Guardrails sin ningún problema.
Errores comunes
Aceptar weightKg como número de Python (int/float) en vez de cadena, y sorprenderse cuando _is_numeric_string() lo rechaza (de suponer el tipo equivocado). Qué pasa: alguien construye un candidato con "weightKg": 120 (sin comillas, un entero de Python/JSON) en vez de "weightKg": "120" (una cadena). Cómo detectarlo: _is_numeric_string() empieza con if not isinstance(value, str): return False — un entero real nunca pasa esa primera verificación, así que el candidato se marca como weight_not_numeric. Cómo corregirlo: ShipmentFields, tal como esta guía lo define, exige que los cinco campos sean cadenas —el mismo tipo que parse_manifest() produce para todos sus valores, incluido el peso, en el Módulo 1, lección 3—; si tu extracción representativa produce un weightKg como número JSON nativo, conviértelo a cadena antes de pasarlo a este validador, exactamente como tendría que hacerlo el código real de extract-shipment-manifest-fields.
Confundir un campo con valor null/None con un campo faltante (de no distinguir "ausente" de "presente pero vacío"). Qué pasa: alguien construye un candidato con "carrier": null en JSON, esperando que se reporte como missing_fields, y se sorprende al ver que aparece, en cambio, en una categoría distinta o pasa sin marcarse. Cómo detectarlo: revisa la lógica de empty_fields en el Paso 1 — la condición es not isinstance(candidate[k], str) or candidate[k].strip() == ""; un valor None (que en Python no es una cadena) cae en la primera parte de esa condición (not isinstance(...)), así que sí se marca como empty_fields, no como missing_fields. Cómo corregirlo: la distinción importa para el mensaje de error, no solo por precisión académica — missing_fields significa "la clave nunca estuvo en el diccionario", empty_fields significa "la clave está, pero su valor no sirve" (vacío, solo espacios, o de un tipo que no es cadena) — dos causas raíz distintas que un ingeniero depurando una extracción fallida necesita distinguir.
Escribir un test nuevo que solo verifica is_valid, sin verificar el contenido específico de errors o de los campos de fallo (de una aserción demasiado débil). Qué pasa: alguien agrega un caso de prueba nuevo con assert result.is_valid is False, sin confirmar además qué campo específico causó el fallo. Cómo detectarlo: si tu test pasaría igual de "verde" ante cualquier tipo de fallo —un campo faltante, uno vacío, uno inesperado—, sin distinguir cuál. Cómo corregirlo: sigue el patrón de los dieciséis casos del Paso 3 — cada uno que espera un fallo verifica, además de is_valid is False, el campo exacto en missing_fields/empty_fields/unexpected_fields/weight_not_numeric que debería estar poblado. Un test que solo confirma "falló, de alguna forma" no protege contra el día en que la lógica interna cambie y el chequeo empiece a fallar por la razón equivocada, silenciosamente.
Ejercicios
Ejercicio 1 — Construye, tú mismo, un candidato JSON que falle por DOS razones distintas a la vez (por ejemplo, un campo faltante y otro con valor vacío), y predice qué mostraría format_report(). Verifica tu predicción corriendo el script.
Ver solución
Por ejemplo: {"shipmentId": "4475", "originCountry": "", "carrier": "AndesExpress", "weightKg": "60"} (falta destinationCountry, y originCountry está vacío). format_report() mostraría dos líneas de error, una por cada categoría poblada: - missing required field(s): destinationCountry y - empty value for field(s): originCountry — el diseño de errors como una tupla que recorre las cuatro categorías posibles, agregando una línea por cada una que tenga contenido, es precisamente lo que permite reportar varios problemas a la vez, en vez de detenerse en el primero que encuentra.
Ejercicio 2 — Explica por qué este validador rechaza un campo INESPERADO (unexpected_fields), en vez de simplemente ignorarlo y aceptar el resto de los campos como válidos. ¿Qué riesgo real evita esta decisión, más allá de "los datos extra no tienen sentido"?
Ver solución
Ignorar campos extra silenciosamente sería una decisión razonable en algunos contextos, pero no en este: write_shipment_record() (heredado del Módulo 1, lección 3) espera un diccionario con una forma específica, y un campo adicional inesperado —por ejemplo, un confidenceScore que el modelo decidió agregar por su cuenta— podría, dependiendo de cómo esté escrito ese código heredado, causar un comportamiento no anticipado si алgún día ese código cambiara para usar **candidate de forma menos cuidadosa, o si un campo extra se propagara sin querer a un log o a otra tabla. Rechazar explícitamente, en vez de ignorar en silencio, obliga a que cualquier campo nuevo que el modelo empiece a producir sea una decisión consciente de actualizar SHIPMENT_FIELDS_SCHEMA, no un accidente que se cuela sin que nadie lo note — la misma filosofía de "nunca adivinar, nunca aceptar en silencio" que bedrock_cost_estimate.py (Módulo 2, lección 7) ya aplicó al rechazar un modelo sin precio conocido.
Ejercicio 3 — Predice qué pasaría si, en el Módulo 5, alguien intentara usar este mismo script para validar un candidato con un SEXTO campo que Andes Cargo decidiera agregar en el futuro (por ejemplo, packageCount). ¿Necesitarías cambiar post_invoke_checks.py, SHIPMENT_FIELDS_SCHEMA, o ambos?
Ver solución
Solo SHIPMENT_FIELDS_SCHEMA necesitaría cambiar —agregar "packageCount" a la tupla—; ninguna otra línea de post_invoke_checks.py tendría que tocarse. La lógica de validate_shipment_fields() entera está escrita en términos de SHIPMENT_FIELDS_SCHEMA como una fuente única de verdad (required = set(SHIPMENT_FIELDS_SCHEMA)), nunca con los nombres de los cinco campos escritos directamente dentro de la función — el mismo principio de diseño que MODEL_PRICING_USD_PER_MILLION_TOKENS ya aplicó en el Módulo 2, lección 7: una tabla o esquema central, referenciado por el código, en vez de valores repetidos en múltiples lugares que alguien tendría que recordar mantener sincronizados.
Resumen y siguiente paso
Esta lección construyó post_invoke_checks.py, el validador determinista que cierra la Brecha 2 exacta de la lección 4: confirmaste, con el mismo candidato hipotético de esa lección (sin weightKg), que este script sí lo rechaza —FAIL, código de salida 1— donde las seis políticas de Bedrock Guardrails no encontrarían nada que objetar. Corriste el suite completo de dieciséis casos con pytest, incluido el caso explícito de un campo inesperado, y confirmaste que SHIPMENT_FIELDS_SCHEMA, como fuente única de verdad, es lo único que cambiaría si el contrato de Andes Cargo alguna vez creciera.
Antes de avanzar deberías poder: explicar la diferencia entre missing_fields y empty_fields, con un ejemplo de cada uno; construir un candidato que falle por dos razones simultáneas y predecir el reporte exacto; y explicar por qué un campo inesperado se rechaza en vez de ignorarse en silencio.
Con las dos mitades de la defensa en profundidad propia construidas —entrada (lección 5) y salida (esta lección)—, la lección 7 marca, con la misma honestidad de todo este ecosistema, el límite exacto que ningún chequeo propio ni ningún terraform plan puede cruzar: el bloqueo real de Bedrock Guardrails ante un intento genuino de ataque, que solo una invocación real —fuera del alcance de este laboratorio $0— podría confirmar.
Recursos
- Python Docs —
dataclasses— referencia deValidationResult, el tipo que estructura el resultado de este validador. - Python Docs —
json— referencia del manejo dejson.JSONDecodeErroren la interfaz de línea de comandos. - Este módulo, lección 4 (
04-why-a-managed-guardrail-is-not-enough-alone.md) — la Brecha 2, fuente directa del caso de pruebatest_unexpected_extra_field_failsy del ejemplo deweightKgfaltante corrido en el Paso 2. - Módulo 1, lección 3 de esta guía (
03-andes-cargos-ai-workload-when-the-deterministic-parser-is-not-enough.md) — la fuente deparse_manifest()y de los cinco campos exactos queSHIPMENT_FIELDS_SCHEMAcodifica.