DEV Community: AWS Español

IAM Principal Cost Allocation para Amazon Bedrock (Novedad)

Hector Fernandez CloudparaTodo — Mon, 13 Apr 2026 07:30:00 +0000

Hace unos meses inicie una serie de posts sobre como gobernar el uso de IA en AWS. En la primera entrega de esta serie hablamos de cómo dar acceso gobernado a LLMs en AWS desde el día 0: IAM Policies, Guardrails, Inference Profiles y un mecanismo de corte de presupuesto por equipo. Si no la leíste, te recomiendo empezar por ahí.
Podcast: UNA API KEY para LLMs

Pero había un punto que me quedaba pendiente: no podíamos saber quién dentro de un equipo estaba generando el consumo. Cortábamos al equipo entero, y después había que armar una solución custom con Model Invocation Logging para identificar al responsable.

AWS acaba de resolver la parte de visibilidad de forma nativa.

¿Bloquear todo el acceso? Eso sigue siendo otro tema, y creo que de forma nativa sería complejo de implementar un proxy (por ahora).

¿Qué anunció AWS?

El 8 de Abril de 2026, Amazon Bedrock lanzó soporte para asignación de costos por IAM principal, hablando en criollo: por usuario o rol de IAM, directamente en Cost Explorer y en CUR 2.0 (Cost and Usage Report).

¡BIEN! Ahora podemos ver el costo desglosado de Bedrock en Cost Explorer sin muchas vueltas.
Anuncio oficial

Ahora, algo que debes tener claro desde ya: esta funcionalidad es de billing, no de enforcement. Los datos llegan a CUR 2.0 y Cost Explorer con 24-48 horas de latencia. Eso significa que puedes saber quién gastó cuánto, pero no puedes bloquear el acceso en tiempo real con esta data. Para eso, el Budget Cut Lambda de la Parte 1 sigue siendo necesario.

¿Qué cambia respecto a la Parte 1 que habíamos hablado?

En la primera publicación usamos Inference Profiles con tags (CostCenter, Team) para atribuir costos por equipo. Eso sigue siendo válido para agrupar por carga de trabajo. Pero ahora hay una capa adicional: la identidad del que hace la llamada.

Mecanismo	Granularidad	¿Qué resuelve?
Inference Profile + Resource Tags	Por equipo / carga de trabajo	"¿Cuánto gastó el equipo backend en Haiku?"
IAM Principal Cost Allocation (NUEVO)	Por usuario / rol	"¿Cuánto gastó `user@empresa.com` en todos los modelos?"

Juntando todo esto, tenemos: quién gastó cuánto, en qué modelo, para qué equipo. Pero recuerda: esta foto la ves con 24-48h de retraso. Es para análisis y chargeback, no para corte en tiempo real.

Pre-requisitos

Todo lo de la Parte 1 más (te invito a leerla)

Tags en tus IAM users/roles con atributos de negocio (team, business-unit, project)
Acceso a la consola de Billing and Cost Management (si es por Organizations, lo haces desde la cuenta management)
CUR 2.0 habilitado (el CUR legacy no soporta esto)

Paso 1: Taggear los IAM Principals

Si usas SSO, cada Permission Set genera un rol en la cuenta target con formato AWSReservedSSO_{PermissionSetName}_{hash}. Estos roles se pueden taggear.

"""
Taggea los roles SSO con atributos de negocio.
Estos tags son los que aparecerán en Cost Explorer y CUR 2.0.
"""
import boto3

iam = boto3.client("iam")

# Mapeo: role SSO -> tags de negocio
SSO_ROLES = {
    "AWSReservedSSO_BackendDev_a1b2c3d4": {
        "team": "backend",
        "business-unit": "BU-ENG-001",
        "department": "engineering",
        "environment": "development",
    },
    "AWSReservedSSO_FrontendDev_e5f6g7h8": {
        "team": "frontend",
        "business-unit": "BU-ENG-002",
        "department": "engineering",
        "environment": "development",
    },
    "AWSReservedSSO_DataTeam_i9j0k1l2": {
        "team": "data",
        "business-unit": "BU-DATA-001",
        "department": "data-science",
        "environment": "development",
    },
}

for role_name, tags_dict in SSO_ROLES.items():
    tags = [{"Key": k, "Value": v} for k, v in tags_dict.items()]

    iam.tag_role(RoleName=role_name, Tags=tags)
    print(f"✅ {role_name} taggeado con: {tags_dict}")

Si además tienen IAM Users (para casos legacy o service accounts), se taggean igual:

aws iam tag-user \
  --user-name "superuser-pipeline-sa" \
  --tags Key=team,Value=data Key=business-unit,Value=BU-DATA-001 Key=project,Value=recommendation-engine

Nota importante de AWS: Los tags solo aparecen para activación en la consola de Billing después de que el principal haya hecho al menos una llamada a Bedrock. Si taggeaste un rol pero nadie lo ha usado aún, no vas a verlo en Cost Allocation Tags.

Paso 2: Activar los tags como Cost Allocation Tags

Billing and Cost Management Console
    → Cost Organization 
        → Cost Allocation Tags (Etiquetas de asignación de costos)
            → Filtrar por "IAM principal type"
            → Seleccionar: team, business-unit, department
            → Click "Activate"

Después de activarlos, los tags tardan hasta 24 horas en estar disponibles para filtrado en Cost Explorer y CUR.

# Verificar qué tags están activos via CLI
aws ce list-cost-allocation-tags \
  --status Active \
  --tag-keys "team" "business-unit" "department" \
  --type "iamPrincipal"

Paso 3: Habilitar IAM Principal en CUR 2.0

Si hay un paso para que todo esto funcione, es este.

Activa la columna line_item_iam_principal en los reportes de costos.

Billing and Cost Management Console
    → Data Exports
        → Create export → Standard data export (CUR 2.0)
            → Additional export content:
                ✅ Include caller identity (IAM principal) allocation data
            → Destino: S3 bucket + Athena integration
        → Save

¿Qué genera esto? Cada línea del CUR 2.0 ahora incluye el ARN exacto del principal que hizo la llamada a Bedrock. Y los tags del principal aparecen con prefijo iamPrincipal/ para no colisionar con resource tags.

¿Cómo se ve la data?

En Cost Explorer: filtrar por equipo/usuario

Una vez activados los tags, Cost Explorer permite agrupar directamente:

Cost Explorer
    → Filtro: Service = "Amazon Bedrock"
    → Group by: Tag → "iamPrincipal/team"

Vas a ver algo como:

iamPrincipal/team    | Costo (USD)
---------------------|------------
backend              | $142.30
data                 | $89.50
frontend             | $23.10
(sin tag)            | $5.40

¿Y si quieres ver quién dentro del equipo backend está consumiendo más? Cambias el Group by:

Cost Explorer
    → Filtro: Service = "Amazon Bedrock"
    → Filtro: Tag "iamPrincipal/team" = "backend"
    → Group by: Tag → "iamPrincipal/business-unit"

BONUS

(Opcional) CUR 2.0 + Athena: análisis avanzado por usuario individual

Si Cost Explorer no te da suficiente granularidad y necesitas ver por usuario individual, puedes consultar CUR 2.0 directamente con Amazon Athena.

Sobre costos de Athena: Athena cobra $5 USD por TB escaneado (con el modo on-demand). Para CUR de organizaciones pequeñas/medianas esto suele ser centavos por consulta. Si quieres reducir costos, activa la compresión del CUR (formato Parquet) y particiona por mes. También existe el modo Provisioned Capacity para uso intensivo.

-- ¿Quiénes son los top 10 consumers de Bedrock este mes?
SELECT
    line_item_iam_principal AS iam_principal,
    tags['iamPrincipal/team'] AS team,
    tags['iamPrincipal/business-unit'] AS business_unit,
    line_item_product_code,
    SUM(line_item_unblended_cost) AS total_cost,
    SUM(line_item_usage_amount) AS total_usage
FROM cur_2_0.bedrock_usage
WHERE
    line_item_product_code = 'AmazonBedrock'
    AND EXTRACT(MONTH FROM billing_period) = EXTRACT(MONTH FROM CURRENT_DATE)
GROUP BY 1, 2, 3, 4
ORDER BY total_cost DESC
LIMIT 10;

Resultado:

iam_principal                                          | team     | business_unit | total_cost
-------------------------------------------------------+----------+---------------+-----------
arn:aws:sts::123456:assumed-role/BackendDev/javier@...  | backend  | BU-ENG-001    | $67.30
arn:aws:sts::123456:assumed-role/DataTeam/maria@...     | data     | BU-DATA-001   | $52.10
arn:aws:sts::123456:assumed-role/BackendDev/luis@...    | backend  | BU-ENG-001    | $41.20
...

Sin proxies, sin Lambdas, sin CloudTrail scraping: sabes exactamente que Javier del equipo backend gastó $67.30 este mes.

Recuerda que cada ejecución tiene un costo mínimo en Athena, pero para un CUR particionado en Parquet suele ser menos de $0.01 por consulta.

Recap ¿Cómo se complementa con la Parte 1?

Componente	Parte 1	Parte 2 (lo nuevo)
¿Quién puede usar qué modelo?	IAM Policy per-team	Sin cambios
¿Se protege PII?	Bedrock Guardrails	Sin cambios
¿Cuánto gastó cada equipo?	Inference Profile tags → Cost Explorer	IAM Principal tags → Cost Explorer (más granular)
¿Cuánto gastó cada usuario?	❌ No disponible	✅ `line_item_iam_principal` en CUR 2.0
¿Se corta al exceder presupuesto?	Budget Cut Lambda (per-team, ~10 min)	⚠️ CUR tiene 24h de delay. NO sirve para bloquear. El Budget Cut Lambda sigue siendo el único mecanismo de corte

Lo que sigue sin resolver nativamente

Seamos honestos: hay cosas que necesitan un proxy real, y para dar gobierno a la IA es lo mejor que puedes pensar.

Corte per-user en tiempo real: CUR tiene 24h de delay. Si necesitas cortar a un usuario específico en minutos, necesitas Model Invocation Logging + Lambda + iam:PutRolePolicy dirigido al session principal
Caching de respuestas: No hay caching nativo de Bedrock. Un proxy como LiteLLM puede cachear respuestas repetitivas y ahorrar costos
Multi-provider routing: Si quieres probar OpenAI y Anthropic directamente (no vía Bedrock), necesitas una capa de abstracción
Observabilidad semántica: Para ver el árbol de razonamiento de un agente, necesitas OTel + Langfuse. CloudTrail te dice "quién llamó", pero no "por qué razonó así"

La clave es que los tags del principal (quién) y los tags del recurso (qué) se complementan en CUR 2.0. No se pisan porque CUR los distingue con prefijos: iamPrincipal/team vs resourceTag/Team.

Conclusiones

Con este anuncio, el modelo de gobierno nativo que planteamos en la Parte 1 gana una pieza que faltaba: visibilidad de costos por usuario, sin añadir infraestructura, sin scraping de CloudTrail, sin Lambdas custom para atribución. Es una herramienta de análisis y chargeback, no de enforcement. El bloqueo en tiempo real sigue dependiendo del Budget Cut Lambda y CloudWatch.

Siempre pensemos en etapas:

Día 0: IAM Policies + Guardrails + Inference Profiles → acceso gobernado
Día 1: Budget Cut Lambda, protección contra gastos descontrolados
Ahora (Día 2): IAM Principal Cost Allocation, saber exactamente quién gasta qué

¿Falta mucho por hacer? Sí. Un proxy real (LiteLLM), observabilidad semántica (Langfuse), y caching siguen siendo evoluciones deseables si la organización escala. Pero la base está puesta, y es 100% nativa.

Lo bueno de todo esto: AWS sigue mejorando sus productos, pero nosotros como arquitectos debemos de saber identificar el uso para nuestras necesidades.

Hoy es cost allocation por principal. Mañana quizás sea throttling per-user nativo o guardrails a nivel de servicio que no requieran guardrailConfig en el código del dev.

Lo que conoces hoy en día en cloud no queda de un lado, todo ese conocimiento es MUY necesario para disponibilizar IA de forma responsable y sobre todo cuantificable.

¿Te gustaría que estemos en 📩 contacto?
Te espero en LinkedIn o desde el Podcast: Cloud para Todos

Héctor Fernández
AWS Community Builder

https://podcast.hectorfernandez.dev

Guardrails para Agentes de IA que se Autocorrigen en Lugar de Bloquear

Elizabeth Fuentes L — Tue, 07 Apr 2026 20:31:25 +0000

La mayoría de los guardrails para agentes hacen una sola cosa: bloquear. El agent choca con una regla, el flujo se detiene y el usuario tiene que intervenir. Agent Control añade una segunda opción: steer — el agent recibe instrucciones correctivas, se autocorrige y completa la tarea sin intervención humana.

Los guardrails para agentes hoy son binarios: permitir o denegar. Cuando un agent viola una política, la respuesta típica es bloquear la acción y mostrar un error. Esto funciona para restricciones estrictas (cumplimiento PCI, bloqueos regulatorios), pero genera fricción en reglas donde el agent podría resolver el problema por sí mismo: ajustar un parámetro, redactar datos sensibles o reformatear una salida.

Agent Control es un plano de control en tiempo de ejecución de código abierto que introduce steer controls junto con los bloqueos tradicionales. Los steer controls devuelven instrucciones correctivas mediante Guide() — el agent reintenta con la corrección aplicada y completa la tarea. Las reglas residen en un servidor, no en el código — se actualizan vía API o dashboard sin necesidad de redesplegar el agent.

Este artículo muestra cómo funciona Agent Control usando una demo de reservas construida con Strands Agents. Comparamos dos enfoques sobre el mismo escenario: hooks que bloquean vs Agent Control que corrige. Hooks y Agent Control son complementarios — usa hooks para bloqueos estrictos, steer para correcciones.

Resumen de la Serie

Este es un artículo adicional en la serie sobre cómo detener las alucinaciones de agentes de IA — añadido tras el lanzamiento de Agent Control:

RAG vs Graph-RAG — Los grafos de conocimiento previenen agregaciones alucinadas
Semantic Tool Selection — El filtrado vectorial reduce las elecciones incorrectas de herramientas
AI Agent Guardrails — Reglas simbólicas que el LLM no puede eludir
Bonus: Agent Control (este artículo) — Steer en lugar de block

El Problema: Bloquear Detiene el Flujo

Los Strands Hooks aplican reglas a nivel de tool. Cuando el agent llama a book_hotel(guests=15) y el máximo es 10, el hook establece cancel_tool y el agent recibe un mensaje de bloqueo:

class MaxGuestsHook(HookProvider):
    def check(self, event: BeforeToolCallEvent) -> None:
        guests = event.tool_use["input"].get("guests", 1)
        if guests > 10:
            event.cancel_tool = f"BLOCKED: {guests} guests exceeds maximum of 10"

El agent entonces le dice al usuario:

"El Grand Hotel tiene una capacidad máxima de 10 huéspedes. ¿Le gustaría ajustar el número de huéspedes?"

El flujo se detiene. El usuario debe responder. Para un asistente de reservas que maneja cientos de solicitudes, cada operación bloqueada es un punto de fricción.

La Solución: Steer en Lugar de Block

Agent Control es un plano de control en tiempo de ejecución de código abierto que evalúa las entradas y salidas del agent contra políticas gestionadas en el servidor. Se integra con Strands como un Plugin — el mismo punto de extensión que los Hooks, pero con dos diferencias clave:

Las reglas residen en un servidor — se modifican vía API o dashboard sin tocar el código del agent
Los steer controls devuelven Guide() en lugar de bloquear — el agent reintenta con instrucciones correctivas

Cómo Funciona Steer

Cuando el LLM genera una salida que menciona "15 guests", el AgentControlSteeringHandler la evalúa contra los controls definidos en el servidor:

El LLM genera: "I will book Grand Hotel for 15 guests from May 1 to May 3"
Agent Control evalúa la salida del LLM → el regex coincide con "15 guest"
El steer control se activa → devuelve Guide("reduce to 10, inform the user")
El LLM reintenta con la guía → llama a book_hotel(guests=10)
La reserva se completa → se informa al usuario sobre el ajuste

El agent responde:

"La capacidad máxima del Grand Hotel es de 10 huéspedes, así que he ajustado la reserva en consecuencia. ID de reserva: BK002."

Sin intervención del usuario. Sin reintentos manuales. El flujo se completó.

Implementación: Misma Consulta, Dos Enfoques

Las tools son idénticas — funciones de reserva limpias sin lógica de validación:

from strands import tool

@tool
def book_hotel(hotel: str, check_in: str, check_out: str, guests: int = 1) -> str:
    """Book a hotel room."""
    return f"SUCCESS: Booking {booking_id} — {hotel}, {guests} guests, {check_in} to {check_out}"

@tool
def process_payment(amount: float, booking_id: str) -> str:
    """Process payment for a booking."""
    return f"SUCCESS: Processed ${amount:.2f} for {booking_id}"

@tool
def confirm_booking(booking_id: str) -> str:
    """Confirm a booking."""
    return f"SUCCESS: Confirmed {booking_id}"

Las tools NO aplican la regla de máximo de huéspedes. Eso es responsabilidad de la capa de guardrails.

Test 1 — Hooks (Block)

from strands import Agent
from strands.hooks import HookProvider, HookRegistry, BeforeToolCallEvent

class MaxGuestsHook(HookProvider):
    def __init__(self):
        self.blocked = 0
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self.check)
    def check(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use["name"] != "book_hotel":
            return
        guests = event.tool_use["input"].get("guests", 1)
        if guests > 10:
            self.blocked += 1
            event.cancel_tool = f"BLOCKED: {guests} guests exceeds maximum of 10"

agent = Agent(model=MODEL, tools=[book_hotel, process_payment, confirm_booking],
              hooks=[MaxGuestsHook()])
agent("Book Grand Hotel for 15 guests from 2026-05-01 to 2026-05-03")

Resultado:

Hook blocked: 1 call(s)
Agent: "The Grand Hotel has a maximum capacity of 10 guests.
        Would you like to adjust?"
Outcome: BLOCKED — user must intervene

Test 2 — Agent Control (Steer)

Los controls se definen en el servidor de Agent Control — no en el código:

# setup_controls.py — run once
control = {
    "name": "steer-max-guests",
    "definition": {
        "scope": {"step_types": ["llm"], "stages": ["post"]},
        "selector": {"path": "output"},
        "evaluator": {"name": "regex", "config": {"pattern": r"(1[1-9]|[2-9]\d)\s*guest"}},
        "action": {
            "decision": "steer",
            "message": "Guest count exceeds maximum of 10",
            "steering_context": {
                "message": "Reduce the guest count to 10, retry the booking, "
                           "and inform the user that the maximum capacity is 10."
            }
        }
    }
}

El agent usa AgentControlPlugin + AgentControlSteeringHandler — ambos como Plugins de Strands:

from agent_control.integrations.strands import AgentControlPlugin, AgentControlSteeringHandler
import agent_control

agent_control.init(agent_name="booking-guardrails-demo")

plugin = AgentControlPlugin(agent_name="booking-guardrails-demo",
                            event_control_list=[BeforeToolCallEvent, AfterToolCallEvent])
steering = AgentControlSteeringHandler(agent_name="booking-guardrails-demo")

agent = Agent(model=MODEL, tools=[book_hotel, process_payment, confirm_booking],
              plugins=[plugin, steering])
agent("Book Grand Hotel for 15 guests from 2026-05-01 to 2026-05-03")

Resultado:

Steered: 1 time(s)
Agent: "The maximum capacity for the Grand Hotel is 10 guests,
        so I have adjusted the booking accordingly. Booking ID: BK002."
Outcome: SELF-CORRECTED — booking completed

Resultados

Misma consulta. Mismas tools. Mismo modelo. Solo cambia el guardrail.

Enfoque	Tiempo	Resultado
Test 1 — Hooks (`cancel_tool`)	~4s	BLOCKED — el agent pide al usuario que ajuste
Test 2 — Agent Control (steer)	~6s	Autocorregido — reserva completada con 10 huéspedes

Ambos aplican la misma regla. La diferencia es lo que ocurre cuando la regla se viola — no qué enfoque es "mejor". Los hooks son más rápidos y sencillos (Python puro, sin servidor). Agent Control añade latencia (steer → reintento) pero completa el flujo sin intervención del usuario. Elige según la regla, no según la tecnología.

Cuándo Usar Cada Enfoque

Enfoque	Mejor para
Hooks (block)	Reglas que DEBEN bloquear estrictamente — sin alternativa posible (p. ej., pago antes de confirmación, cumplimiento PCI)
Agent Control (steer)	Reglas donde el agent PUEDE autocorregirse — ajustar parámetros, redactar PII, corregir formato
Agent Control (deny)	Igual que hooks pero gestionado en un servidor — cambiar reglas sin redesplegar código

Son complementarios, no competidores. Los hooks son más sencillos (Python puro, sin servidor, sin sobrecarga de latencia). Agent Control es más flexible (gestionado en servidor, steer + deny, actualizaciones en tiempo de ejecución sin redespliegue). Muchos sistemas en producción usan ambos:

Hooks para reglas de cumplimiento que nunca deben eludirse — verificación de pagos, bloqueos regulatorios, PII en parámetros de tools
Agent Control (deny) para los mismos bloqueos estrictos pero gestionados centralmente en múltiples agents — actualización vía dashboard, sin redespliegue
Agent Control (steer) para reglas flexibles donde la autocorrección es preferible — ajustes de capacidad, redacción de PII en salidas, formato de fechas

Dos Formas de Definir Controls

Modo	Mejor para
Server (esta demo)	Equipos, producción, gestión vía dashboard — los controls residen en el servidor de Agent Control
Local YAML	Prototipado rápido — controls definidos en `controls.yaml`, sin necesidad de servidor

Consulta la documentación de Agent Control para más detalles sobre ambos modos.

Por Qué Strands lo Hace Sencillo

Tanto Hooks como Agent Control se integran con un solo cambio de línea:

# Hooks — block violations (existing Strands API):
agent = Agent(tools=[...], hooks=[MaxGuestsHook()])

# Agent Control — steer violations (plugin API):
agent = Agent(tools=[...], plugins=[AgentControlPlugin(...), AgentControlSteeringHandler(...)])

Sin orquestación personalizada. Sin lógica de reintentos. Strands gestiona el ciclo de vida — los hooks interceptan antes de las llamadas a tools, el steering evalúa después de la salida del modelo, y Guide() activa el reintento automático con instrucciones correctivas.

Strands Hooks Documentation
Strands Steering Documentation
Agent Control Plugin for Strands

Puntos Clave

Los hooks bloquean violaciones — efectivo pero detiene el flujo y requiere intervención del usuario
Agent Control redirige las violaciones — el agent se autocorrige y completa la tarea
Los steer controls devuelven Guide() con instrucciones correctivas — el LLM reintenta con la corrección aplicada
Los controls residen en un servidor — se actualizan las reglas vía API o dashboard sin tocar el código del agent
Ambos enfoques aplican la misma regla (máximo 10 huéspedes) — la diferencia es lo que ocurre cuando la regla se viola
Hooks para bloqueos estrictos, Agent Control para autocorrección — usa ambos cuando sea necesario

Ejecútalo Tú Mismo

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations/05-agent-control-demo

# Start Agent Control server (see setup instructions)
# https://github.com/agentcontrol/agent-control

# Install and run
uv venv && uv pip install -r requirements.txt
uv run setup_controls.py
uv run test_hooks_vs_control.py

Puedes cambiar a cualquier proveedor compatible con Strands — consulta Strands Model Providers para la configuración.

Referencias

Investigación

ATA: Autonomous Trustworthy Agents (2024) — Patrones de fallo en guardrails
Enhancing LLMs through Neuro-Symbolic Integration — Razonamiento neural + simbólico

Strands Agents

Strands Agents with Agent Control — Anuncio en blog
Agent Control Plugin — Documentación de integración
Strands Hooks — BeforeToolCallEvent, cancel_tool
Strands Steering — Guide, Proceed, SteeringHandler
Strands Model Providers — Cambiar a Amazon Bedrock, Anthropic, Ollama

Agent Control

Agent Control GitHub — Código abierto, Apache 2.0
Agent Control Docs — Configuración del servidor y referencia de API

Código

Code Repository

Guardrails para Agentes de IA: Reglas Que los LLM No Pueden Evadir

Elizabeth Fuentes L — Tue, 31 Mar 2026 18:49:56 +0000

Los agentes de IA pueden alucinar el éxito de una operación incluso cuando violan reglas de negocio. Confirman reservas sin verificación de pago, aceptan parámetros inválidos como 15 huéspedes cuando el máximo es 10, o ignoran prerrequisitos obligatorios.

El problema central: los agentes pueden confirmar reservas de hotel a pesar de que el pago nunca fue verificado, violar restricciones de capacidad, o saltarse pasos de validación obligatorios. El prompt engineering por sí solo no puede prevenir estos errores.

Este post demuestra cómo la validación neurosimbólica — combinando razonamiento del LLM con reglas simbólicas deterministas aplicadas a nivel de framework — bloquea operaciones inválidas antes de que se ejecuten.

Esta demo utiliza Strands Agents. Patrones similares pueden aplicarse en LangGraph, AutoGen u otros frameworks de agentes.

Resumen de la Serie

Esta es la Parte 3 de una serie sobre cómo detener las hallucinations en agentes de IA:

RAG vs Graph-RAG — Knowledge graphs para prevenir hallucinations
Semantic Tool Selection — Filtrado de tools basado en vectores
AI Agent Guardrails (este post) — Reglas simbólicas que los LLM no pueden evadir
Runtime Guardrails — Redirigir en lugar de bloquear
Multi-Agent Validation — Detección de hallucinations basada en equipos

El Problema: Los Prompts Son Sugerencias, No Restricciones

La investigación de ATA: Autonomous Trustworthy Agents (2024) identifica tres patrones de hallucination:

Errores de parámetros: El agent llama a book_hotel(guests=15) a pesar de que el docstring indica un máximo de 10
Errores de completitud: El agent ejecuta reservas sin la verificación de pago requerida
Comportamiento de evasión de tools: El agent confirma éxito sin llamar a los tools de validación obligatorios

Causa raíz: Los prompts son texto que el LLM interpreta. Las reglas de negocio embebidas en docstrings se convierten en sugerencias — el modelo decide si las sigue en cada llamada.

La Solución: Validación Neurosimbólica con Strands Hooks

Los hooks a nivel de framework interceptan las llamadas a tools antes de la ejecución. Usando BeforeToolCallEvent en Strands Agents, puedes:

Validar reglas simbólicas de forma determinista
Cancelar la ejecución del tool si alguna regla falla
Enviar un mensaje de cancelación directamente al LLM (no puede ser anulado)

Diferencia arquitectónica clave:

Prompts: Entrada del LLM (interpretable, anulable)
Hooks: Interceptores del framework (deterministas, obligatorios)

Prerrequisitos

Experiencia con Python y uso de tools en agentes LLM
Familiaridad con Strands Agents

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations/04-neurosymbolic-demo
uv venv && uv pip install -r requirements.txt

Implementación: Reglas, Hook y Dos Agentes

Paso 1: Definir Reglas Simbólicas

from dataclasses import dataclass
from typing import Callable

@dataclass
class Rule:
    name: str
    condition: Callable[[dict], bool]
    message: str

BOOKING_RULES = [
    Rule("max_guests",   lambda ctx: ctx.get("guests", 1) <= 10,     "Maximum 10 guests per booking"),
    Rule("valid_dates",  lambda ctx: ctx["check_in"] < ctx["check_out"], "Check-in must be before check-out"),
]

CONFIRMATION_RULES = [
    Rule("payment_before_confirm", lambda ctx: ctx.get("payment_verified", False),
         "Payment must be verified before confirmation"),
]

Las reglas son funciones Python simples — deterministas, testeables y auditables de forma independiente a cualquier agent.

Paso 2: Crear el Hook de Validación

from strands.hooks import HookProvider, HookRegistry, BeforeToolCallEvent

class NeurosymbolicHook(HookProvider):

    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self.validate)

    def validate(self, event: BeforeToolCallEvent) -> None:
        tool_name = event.tool_use["name"]
        if tool_name not in self.rules:
            return
        context = self._build_context(tool_name, event.tool_use["input"])
        passed, violations = validate(self.rules[tool_name], context)
        if not passed:
            event.cancel_tool = f"BLOCKED: {', '.join(violations)}"

Cuando se establece event.cancel_tool, Strands reemplaza el resultado del tool con ese mensaje antes de que el LLM lo vea. El tool nunca se ejecuta.

Paso 3: Definir Tools Limpios

from strands import tool

@tool
def book_hotel(hotel: str, check_in: str, check_out: str, guests: int = 1) -> str:
    """Book a hotel room."""
    return f"SUCCESS: Booked {hotel} for {guests} guests, {check_in} to {check_out}"

@tool
def process_payment(amount: float, booking_id: str) -> str:
    """Process payment for a booking."""
    return f"SUCCESS: Processed ${amount:.2f} for {booking_id}"

@tool
def confirm_booking(booking_id: str) -> str:
    """Confirm a booking."""
    return f"SUCCESS: Confirmed {booking_id}"

Los tools contienen solo lógica de negocio — sin validación mezclada.

Paso 4: Crear Ambos Agentes

from strands import Agent
# Using OpenAI-compatible interface via Strands SDK (not direct OpenAI usage)
from strands.models.openai import OpenAIModel

MODEL = OpenAIModel(model_id="gpt-4o-mini")

# Baseline: no hook, no validation
baseline_agent = Agent(tools=[book_hotel, process_payment, confirm_booking], model=MODEL)

# Guarded: hook intercepts every tool call
hook = NeurosymbolicHook(STATE)
guarded_agent = Agent(tools=[book_hotel, process_payment, confirm_booking], hooks=[hook], model=MODEL)

Mismos tools, mismo modelo, mismos prompts. La única diferencia: hooks=[hook]

Resultados de las Pruebas

Prueba 1: Confirmar Reserva Sin Pago

Consulta: "Confirm booking BK001"

Agent	Resultado
Baseline	❌ Ejecuta — pago nunca verificado, docstring ignorado
Guarded	✅ Bloqueado — el hook evalúa `CONFIRMATION_RULES`, encuentra `payment_verified=False`, cancela

El LLM recibe: BLOCKED: Payment must be verified before confirmation

Prueba 2: Reservar Hotel Excediendo el Límite de Huéspedes

Consulta: "Book Grand Hotel for 15 people from 2026-03-20 to 2026-03-25"

Agent	Resultado
Baseline	❌ Ejecuta con 15 huéspedes — máximo en docstring ignorado
Guarded	✅ Bloqueado — `max_guests_check(15 <= 10)` falla, ejecución cancelada

La validación ocurre antes de la ejecución — no hay reserva que revertir.

Prueba 3: Reserva Válida

Consulta: "Book Grand Hotel for 5 guests from 2026-03-20 to 2026-03-25"

Agent	Resultado
Baseline	✅ Ejecuta
Guarded	✅ Ejecuta — todas las reglas pasan, el hook no agrega fricción

Resumen de Escenarios

Escenario	Baseline	Guarded
Confirmar sin pago	❌ Ejecuta — hallucination	✅ Bloqueado antes de la ejecución
Reservar 15 huéspedes (máx. 10)	❌ Ejecuta — regla violada	✅ Bloqueado antes de la ejecución
Reserva válida (5 huéspedes)	✅ Ejecuta	✅ Ejecuta

Punto Clave: Dónde Se Aplica la Validación

Prompt Engineering (Insuficiente)

system_prompt = """
IMPORTANT: Never confirm bookings without payment verification.
CRITICAL: Maximum 10 guests per booking.
"""
# The LLM reads this as context. It can hallucinate compliance.

❌ El LLM decide si sigue estas instrucciones en cada llamada.

Aplicación Basada en Hooks (Suficiente)

def validate(self, event: BeforeToolCallEvent) -> None:
    passed, violations = validate(self.rules[tool_name], context)
    if not passed:
        event.cancel_tool = f"BLOCKED: {', '.join(violations)}"
        # Tool never executes. LLM receives the cancellation.
        # There is no path to override this.

✅ El hook se ejecuta fuera del LLM. La decisión no depende del LLM.

Diferencia arquitectónica: Los prompts son entrada del LLM (interpretable). Los hooks son interceptores del framework (deterministas).

Consideraciones para Producción

Ventajas

Restricciones verificables: Las reglas son código, no instrucciones
Centralizadas: Un solo hook valida todos los tools
Testeables de forma independiente: Las reglas funcionan fuera de cualquier agent o contexto de LLM
Auditables: Las violaciones de reglas producen eventos explícitos con nombre del tool, parámetros y razón

Desafíos

Las reglas deben definirse explícitamente para cada operación protegida
No maneja lógica difusa o probabilística — las reglas son booleanas
Los casos límite requieren manejo explícito en las condiciones
Las reglas necesitan mantenimiento a medida que evoluciona la lógica de negocio

Mejores Prácticas

Define hooks para operaciones críticas: reservas, pagos, cancelaciones
Registra todas las violaciones de reglas con nombre del tool, parámetros y razón
Testea las reglas de forma independiente y exhaustiva
Combina con semantic tool selection (Parte 2) y multi-agent validation (Parte 5) para protección en capas

Siguiente Paso

Los guardrails simbólicos bloquean violaciones de reglas a nivel de tool. Pero bloquear detiene el flujo de trabajo — el usuario debe intervenir.

Parte 4: Runtime Guardrails muestra cómo Agent Control redirige a los agentes para que se autocorrijan en lugar de bloquearlos, completando flujos de trabajo sin intervención humana.

Conclusiones Clave

Los prompts son sugerencias: El LLM interpreta docstrings; puede alucinar cumplimiento con cualquier instrucción
Los hooks son aplicación forzada: BeforeToolCallEvent intercepta antes de la ejecución a nivel de framework — el LLM no puede anular un tool cancelado
Todas las operaciones inválidas bloqueadas: Cero cambios en tools, cero cambios en prompts, un hook agregado
Separación limpia: Los tools manejan operaciones; los hooks manejan la aplicación de restricciones
Auditable por diseño: Las violaciones de reglas son condiciones Python explícitas — testeables, registrables, trazables
Un hook, todos los tools: Un solo NeurosymbolicHook valida cada llamada a tool de forma centralizada

Ejecútalo Tú Mismo

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations/04-neurosymbolic-demo
uv venv && uv pip install -r requirements.txt
uv run test_neurosymbolic_hooks.py

Puedes cambiar a cualquier proveedor soportado por Strands — consulta Strands Model Providers para la configuración.

Referencias

Investigación

Strands Agents

Código

Repositorio de Código

Gracias!

🇻🇪🇨🇱 Dev.to Linkedin GitHub Twitter Instagram Youtube
Linktr

Elizabeth Fuentes LFollow

I help developers build production-ready AI applications through hands-on tutorials and open-source projects.

Reduce errores y costos de tokens en agentes con seleccion semantica de herramientas

Elizabeth Fuentes L — Mon, 23 Mar 2026 16:25:05 +0000

Cuando los agentes de IA tienen muchas herramientas similares, a menudo seleccionan la incorrecta y consumen tokens excesivos al procesar todas las descripciones de herramientas.

Este articulo demuestra como la seleccion semantica de herramientas filtra tools antes del procesamiento del agente, mejorando la precision y reduciendo costos de tokens. La demostracion utiliza Strands Agents y FAISS para filtrar 29 herramientas y quedarse con las 3 mas relevantes.

Esta demo utiliza Strands Agents. Patrones similares se pueden aplicar en LangGraph, AutoGen u otros frameworks de agentes.

Vision general de la serie

Esta es la Parte 2 de una serie sobre como detener las hallucinations en agentes de IA:

RAG vs Graph-RAG — Knowledge graphs para prevenir hallucinations
Seleccion semantica de herramientas (este articulo) — Filtrado de tools basado en vectores
Validacion multi-agente — Deteccion de hallucinations basada en equipos
AI Agent Guardrails — Aplicacion de razonamiento simbolico
Runtime Guardrails — Controles autocorrectivos

Configuracion

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations/02-semantic-tools-demo
uv venv && uv pip install -r requirements.txt

El problema dual: errores + desperdicio de tokens

Escenario del problema

Un agente de viajes con 29 herramientas similares (search_hotels, search_flights, search_hotel_reviews, etc.) recibe la consulta: "How much does Hotel Marriott cost?"

El agente puede seleccionar get_hotel_details() en lugar de get_hotel_pricing() — una seleccion incorrecta de tool. Este error consume 4,500 tokens procesando las descripciones de las 29 herramientas.

Causas raiz

Nombres de herramientas similares causan confusion
Las herramientas genericas se usan en exceso
Mas herramientas aumentan la probabilidad de hallucination

Contexto de investigacion

"Las hallucinations en la seleccion de herramientas aumentan con la cantidad de tools. Sistemas en produccion reportan una reduccion del 89% en tokens con seleccion semantica de herramientas."

La solucion: seleccion semantica de herramientas con FAISS

Modos de falla de agentes a escala

La investigacion identifica cinco modos criticos de falla:

Errores de seleccion de funcion — Llamar a tools inexistentes
Errores de parametros — Argumentos mal formados
Errores de completitud — Parametros requeridos faltantes
Comportamiento de bypass de tools — Generar respuestas en lugar de llamar herramientas
Desbordamiento de contexto — Desperdicio de tokens al procesar todas las descripciones

Analisis costo-beneficio

Cada llamada al LLM envia las descripciones de las 29 herramientas
En un flujo de 50 pasos: 29 tools x 50 llamadas
Genera un desperdicio significativo de tokens y retrasos en el procesamiento

La investigacion muestra hasta un 86.4% de precision previniendo hallucinations en la seleccion de herramientas en sistemas de produccion.

Demostracion: tres enfoques

Prueba 1: Enfoque tradicional (las 29 herramientas)

from strands import Agent
# Using OpenAI-compatible interface via Strands SDK (not direct OpenAI usage)
from strands.models.openai import OpenAIModel
from enhanced_tools import ALL_TOOLS

for query, expected in TESTS:
    agent = Agent(tools=ALL_TOOLS, system_prompt=PROMPT, model=MODEL)
    tools, tokens = run_and_capture_with_tokens(agent, query)

Resultado: ~1,557 tokens promedio por consulta con precision variable.

Prueba 2: Enfoque semantico (top-3 herramientas filtradas)

El agente recibe unicamente las 3 herramientas mas relevantes por consulta.

Construccion del indice FAISS

import faiss
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

def build_index(tools):
    """Build FAISS index from tool docstrings"""
    texts = [f"{t.__name__}: {t.__doc__}" for t in tools]
    embeddings = model.encode(texts)

    index = faiss.IndexFlatL2(embeddings.shape[1])
    index.add(embeddings.astype('float32'))
    return index

def search_tools(query: str, top_k: int = 3):
    """Find most relevant tools using FAISS"""
    emb = model.encode([query])
    _, indices = index.search(emb.astype('float32'), top_k)
    return [tools[i] for i in indices[0]]

El modelo all-MiniLM-L6-v2 es ligero (22M parametros, 384 dimensiones), optimizado para similitud semantica y funciona de manera eficiente en CPUs.

Filtrado en tiempo de ejecucion

from strands import Agent
# Using OpenAI-compatible interface via Strands SDK (not direct OpenAI usage)
from strands.models.openai import OpenAIModel

for query, expected in TESTS:
    selected = search_tools(query, top_k=3)
    selected_names = [t.__name__ for t in selected]

    agent = Agent(tools=selected, system_prompt=PROMPT, model=MODEL)
    tools, tokens = run_and_capture_with_tokens(agent, query)

Resultado: ~275 tokens promedio por consulta — una reduccion desde 1,557 tokens.

Dato clave: el agente nunca procesa las otras 26 herramientas. Estas tools permanecen en el sistema pero nunca entran en la ventana de contexto del agente.

Prueba 3: Semantico + memoria (agente unico)

Para conversaciones multi-turno en produccion, manteniendo la memoria mientras se intercambian herramientas dinamicamente.

Intercambio dinamico de herramientas

def swap_tools(agent, new_tools):
    """Swap tools in a live agent without losing conversation memory."""
    reg = agent.tool_registry
    reg.registry.clear()
    reg.dynamic_tools.clear()
    for t in new_tools:
        reg.register_tool(t)

Implementacion completa

from strands import Agent
# Using OpenAI-compatible interface via Strands SDK (not direct OpenAI usage)
from strands.models.openai import OpenAIModel
from enhanced_tools import ALL_TOOLS
from registry import build_index, search_tools, swap_tools

initial_tools = search_tools(TESTS[0][0], top_k=3)
memory_agent = Agent(tools=initial_tools, system_prompt=PROMPT, model=MODEL)

for query, expected in TESTS:
    selected = search_tools(query, top_k=3)
    swap_tools(memory_agent, selected)
    tools, tokens = run_and_capture_with_tokens(memory_agent, query)

Resultado: Mas tokens que el enfoque solo semantico (debido al contexto acumulado), pero significativamente menos comparado con el metodo tradicional, manteniendo el historial completo de la conversacion.

Por que funciona

Strands llama a tool_registry.get_all_tools_config() en cada ciclo del event loop, detectando automaticamente los cambios en tiempo de ejecucion.

Ventajas:

Cero perdida de conversacion: agent.messages se preserva entre intercambios de tools
No requiere recrear el agente
Flexibilidad en tiempo de ejecucion para necesidades dinamicas de herramientas
Listo para produccion en conversaciones largas

Contexto de investigacion y rendimiento en el mundo real

La demo controlada logro precision perfecta en 13 consultas. Sin embargo, los sistemas de produccion reales con cientos de herramientas y consultas ambiguas muestran resultados diferentes.

"La investigacion muestra que en sistemas de produccion con cientos de herramientas, la seleccion semantica de tools alcanza hasta un 86.4% de precision en la deteccion y prevencion de hallucinations en la seleccion de herramientas."

Esto reduce significativamente los errores comparado con los enfoques tradicionales (que caen por debajo del 50% de precision con mas de 100 tools), pero sigue siendo un desafio en dominios con semantica de herramientas superpuesta.

Siguiente paso

La seleccion semantica de herramientas reduce los errores de seleccion de tools y los costos de tokens. Sin embargo, los agentes aun pueden generar hallucinations sobre el exito de operaciones (por ejemplo, confirmar reservaciones sin procesar pagos o ignorar reglas de negocio).

Parte 3: Validacion multi-agente demuestra como equipos especializados de agentes (Executor -> Validator -> Critic) detectan hallucinations antes de que lleguen a los usuarios.

Esta capacidad esta disponible de forma nativa a traves de Amazon Bedrock Agentcore Gateway.

Puntos clave

Problema dual: Errores en la seleccion de herramientas Y desperdicio de tokens
Reduccion significativa de errores: Menos herramientas = menos selecciones incorrectas
89% de ahorro en tokens: Reduciendo de 29 a 3 herramientas por llamada
Implementacion sencilla: ~20 lineas de codigo con FAISS
Integracion con Strands: El decorador @tool y la carga dinamica permiten el filtrado semantico con intercambio de herramientas en tiempo de ejecucion

Ejecutalo tu mismo

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations/02-semantic-tools-demo
uv venv && uv pip install -r requirements.txt
uv run test_semantic_tool_selection.py

Puedes cambiar a cualquier proveedor soportado por Strands — consulta Strands Model Providers para la configuracion.

Referencias

Investigacion

Internal Representations as Indicators of Hallucinations
Solving Context Window Overflow — Reduccion de tokens 7x
Semantic Tool Selection in Practice — Reduccion del 89%

Strands Agents

Codigo

Repositorio de codigo

Gracias!

🇻🇪🇨🇱 Dev.to Linkedin GitHub Twitter Instagram Youtube

EKS Auto Mode: Kubernetes sin drama😝

Rubén Rodríguez — Mon, 23 Mar 2026 08:07:05 +0000

Llevábamos tiempo queriendo dedicarle un rato a EKS Auto Mode, pero entre proyectos, eventos y vacaciones no conseguíamos encontrar ese hueco. Hoy lo hemos encontrado.

Y es que una de las conversaciones recurrentes con compañeros, clientes y jefes es precisamente cómo la tecnología está evolucionando para automatizar capas de infraestructura que antes requerían expertise y horas de configuración.
Nuestra respuesta es siempre la misma: la tecnología no elimina el conocimiento. Donde antes necesitabas configurar Karpenter, gestionar AMIs, parchear nodos y mantener add-ons... ahora puedes focalizarte en arquitectura, seguridad y en construir cosas que aporten valor de otra manera.

¿Qué es EKS Auto Mode?

Podríamos explicarlo de muchas maneras pero ya nos conocéis, nos gusta ser directos:

AWS gestiona por ti el autoscaling de nodos (con Karpenter), networking de pods, load balancing, DNS, almacenamiento EBS, parches del SO y rotación de nodos. Tú te centras en tus cargas de trabajo.

¿Qué ventajas nos da esto?

AWS gestiona los add-ons core. No los ves, no los tocas, no los parcheas.
Karpenter gestionado por AWS, sin instalación ni configuración.
NodePools y NodeClasses disponibles para tener flexibilidad sobre tus nodos.
Optimización de costes gracias a la consolidación automática de Karpenter.

¡Al lio!

Para el test hemos utilizado el módulo oficial de la community eks-auto-mode, modificando la llamada para adaptarla a este caso. Lo importante del módulo es que con "compute_config.enabled = true" activamos Auto Mode completo.

Parece broma pero sí.🥶Es así de fácil.🥶

¿Qué crea Auto Mode automáticamente?

Una vez desplegado el cluster, veamos como se gestiona Karpenter:

kubectl get nodeclass NAME ROLE READY AGE default awtwins-cluster-eks-auto-20260322210213399200000003 True 55m

kubectl get nodepools NAME NODECLASS NODES READY AGE general-purpose default 0 True 56m

Auto Mode crea automáticamente el NodeClass y el NodePool correspondiente.
Puedes añadir configuración extra de almacenamiento, restricciones de instancias, límites de capacidad o tags.

El test de autoscaling

Para validar que todo funciona usamos el deployment de pause que usa AWS en sus ejemplos oficiales de Karpenter y que ya hemos utilizado anteriormente. No hace nada, solo consume recursos.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: inflate
spec:
  replicas: 5
  selector:
    matchLabels:
      app: inflate
  template:
    metadata:
      labels:
        app: inflate
    spec:
      containers:
        - name: inflate
          image: public.ecr.aws/eks-distro/kubernetes/pause:3.7
          resources:
            requests:
              cpu: "1"
              memory: "1Gi"

Aplicamos el manifest y vemos que tenemos los pods creandose correctamente:

Vamos a EC2 mediante la consola y podremos ver como Karpenter está creando el nodo necesario para desplegar este tipo de carga.

Para validar que el desescalado también funciona sin problema procedemos a escalar el deployment a 0 replicas:

kubectl scale deployment inflate --replicas=0

El resultado habla por sí solo: nodos provisionados en segundos, pods corriendo, y consolidación automática al bajar la carga. Sin tocar nada más.

En este post solo estamos validando el autoscaling, pero el mismo principio aplica para todo lo que mencionamos antes como load balancing, networking, DNS, almacenamiento, parches de nodos... Todo eso se gestiona por AWS sin que tengas que instalar, configurar ni actualizar nada. Como ya sabéis, somos muy fans de Karpenter, pero la gracia de Auto Mode es que Karpenter es solo una pieza de todo lo que te quitas de encima.

Adjuntamos una imagen descriptiva como comparación:

¿Cuánto cuesta?

El modelo de precios de EKS Auto Mode tiene tres componentes importantes:

Precio del cluster igual que en EKS estándar.
Precio de las EC2 desplegadas.
Precio de Auto Mode, aprox un 12% extra sobre el coste de cada instancia.

Nuestra opinión

Como hemos hablado de estos temas en diferentes foros con muchos compañeros, llega el momento más esperado...🥁

EKS Auto Mode no es solo una feature más. Es un cambio en cómo operar Kubernetes en AWS. Pasamos de gestionar un cluster a consumirlo. Y eso, bien entendido, desgasta menos y minimiza problemas.

Además, como ya vimos en nuestro anterior post de awtwins, si le sumamos features como ArgoCD Capability la vida nos vuelve a sonreír con un cluster magnífico con mucha menos gestión que antes. Los que ya nos conocéis sabéis que para nosotros este tipo de soluciones siempre son un SÍ... si lo puedes pagar, claro💰😄

Como siempre, esperamos vuestros comentarios y feedback👇

RAG vs GraphRAG: Cuando los Agentes Alucinan Respuestas

Elizabeth Fuentes L — Wed, 18 Mar 2026 01:05:06 +0000

Un agent que alucina durante la ejecucion es catastrofico: podria fabricar parametros de API, inventar confirmaciones de exito tras fallos, o ejecutar acciones basadas en creencias falsas.

Este articulo explora como el Retrieval-Augmented Generation (RAG) tradicional provoca que los AI agents alucinen estadisticas y agregaciones, en contraste con GraphRAG usando knowledge graphs de Neo4j. Demostramos las diferencias a traves de un agent de reservas de viajes que compara RAG (FAISS) frente a GraphRAG sobre 300 documentos FAQ de hoteles.

Esta demo utiliza Strands Agents. Patrones similares se pueden aplicar en LangGraph, AutoGen u otros frameworks de agents.

Vision General de la Serie

Esta es la Parte 1 de una serie sobre como detener las hallucinations en AI agents:

RAG vs Graph-RAG (este articulo) — Los knowledge graphs previenen agregaciones alucinadas
Semantic Tool Selection — El filtrado vectorial reduce las elecciones incorrectas de tools
Multi-Agent Validation — Los swarms detectan hallucinations antes de que lleguen a los usuarios
AI Agent Guardrails — Reglas simbolicas que el LLM no puede eludir
Runtime Guardrails — Dirigir en lugar de bloquear

El Problema: Tres Tipos de Hallucinations en RAG

La investigacion (MetaRAG: Metamorphic Testing for Hallucination Detection) identifica tres tipos de hallucination en sistemas RAG:

Estadisticas fabricadas — Los sistemas RAG producen numeros estimados en lugar de valores calculados. El LLM adivina agregaciones a partir de unos pocos documentos recuperados en vez de calcularlas sobre el conjunto completo de datos.
Recuperacion incompleta — La busqueda vectorial recupera los top-k fragmentos mas similares, pero los datos necesarios para una respuesta pueden estar dispersos en muchos documentos. El LLM llena los vacios con fabricaciones que suenan plausibles.
Fabricacion fuera de dominio — Cuando no existen datos relevantes, RAG devuelve los resultados de apariencia mas similar de todos modos. El LLM trata estas coincidencias parciales como contexto valido y fabrica una respuesta con confianza.

La Solucion: GraphRAG con Knowledge Graphs

Los knowledge graphs resuelven estos problemas a nivel arquitectonico:

Las agregaciones son calculadas por el motor de base de datos (AVG, COUNT, SUM) — no adivinadas por el LLM
Las relaciones son aristas explicitas en el grafo — no inferidas por similitud de texto
Los datos faltantes devuelven resultados vacios — el LLM recibe "no se encontraron resultados" en lugar de coincidencias parciales de las cuales alucinar

La diferencia clave: RAG siempre devuelve algo. GraphRAG devuelve resultados vacios cuando los datos no existen.

Requisitos Previos

Python 3.9+
Neo4j ejecutandose localmente con el plugin APOC
Clave de API de OpenAI (o cualquier proveedor de modelos compatible con Strands)

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations/01-faq-graphrag-demo
uv venv && uv pip install -r requirements.txt

Implementacion

Agent 1: RAG Tradicional (FAISS)

from strands import Agent, tool

@tool
def search_faqs(query: str) -> str:
    """Search hotel FAQs using vector similarity."""
    query_embedding = embed_model.encode([query])
    distances, indices = index.search(query_embedding.astype('float32'), 3)
    return "\n".join([documents[idx]['text'][:500] for idx in indices[0]])

rag_agent = Agent(tools=[search_faqs], model=model)

El agent RAG recupera los 3 documentos mas similares y permite que el LLM los resuma. No puede agregar sobre el conjunto completo de datos, recorrer relaciones ni reconocer cuando los datos estan ausentes.

Agent 2: GraphRAG (Neo4j + Cypher)

@tool
def query_knowledge_graph(cypher_query: str) -> str:
    """Execute a Cypher query against the hotel knowledge graph.

    Node labels: Hotel, Room, Amenity, Policy, Service
    Relationships: (Hotel)-[:HAS_ROOM]->(Room), (Hotel)-[:OFFERS_AMENITY]->(Amenity),
                   (Hotel)-[:HAS_POLICY]->(Policy), (Hotel)-[:PROVIDES_SERVICE]->(Service)

    Properties:
    - Hotel: name, location, rating, price_range
    - Room: type, capacity, price_per_night
    - Amenity: name, category
    """
    with driver.session() as session:
        result = session.run(cypher_query)
        records = list(result)
        if not records:
            return "No results found."
        return f"Found {len(records)} results:\n" + "\n".join(str(dict(r.items())) for r in records[:15])

graph_agent = Agent(tools=[query_knowledge_graph], model=model)

El schema en el docstring habilita Text2Cypher — el LLM genera consultas Cypher precisas en lugar de adivinar a partir de fragmentos de texto.

Resultados: 4 Escenarios de Prueba

Prueba 1: Agregacion — "What is the average hotel rating in Paris?"

Enfoque	Resultado	Por que
RAG	4.7 (de solo 2 documentos)	Recupero 2 documentos similares y promedio esas calificaciones
GraphRAG	4.7 (a nivel de base de datos) ✅	`MATCH (h:Hotel {location:'Paris'}) RETURN avg(h.rating)` — calculado sobre todos los hoteles

RAG tuvo suerte aqui — mismo numero, pero de 2 documentos en vez del conjunto completo de datos. Con datos diferentes, esto divergiria.

Prueba 2: Conteo — "How many hotels have a swimming pool?"

Enfoque	Resultado	Por que
RAG	"No tengo los datos" ❌	Los 3 documentos principales no mencionaban piscinas; no pudo contar entre 300 documentos
GraphRAG	"133 hoteles" ✅	`MATCH (h)-[:OFFERS_AMENITY]->(a {name:'pool'}) RETURN count(h)`

Prueba 3: Multi-hop — "What room types does the highest-rated hotel offer?"

Enfoque	Resultado	Por que
RAG	Incompleto ⚠️	Encontro informacion del hotel pero no pudo recorrer hasta los datos de habitaciones
GraphRAG	Completo con datos de habitaciones ✅	`MATCH (h:Hotel) WITH h ORDER BY h.rating DESC LIMIT 1 MATCH (h)-[:HAS_ROOM]->(r) RETURN r`

Prueba 4: Fuera de dominio — "Find hotels in Antarctica"

Enfoque	Resultado	Por que
RAG	Fabrico "Estaciones de Investigacion" como hoteles ❌	Devolvio documentos de apariencia similar sobre alojamiento en clima frio
GraphRAG	"No se encontraron hoteles" ✅	`MATCH (h:Hotel {location:'Antarctica'}) RETURN h` — resultado vacio

Cuando Usar Cada Enfoque

Criterio	RAG	GraphRAG
Estructura de datos	Texto no estructurado, documentos	Estructurado, con relaciones
Tipo de consulta	Similitud semantica, busqueda difusa	Agregaciones precisas, recorridos
Datos faltantes	Devuelve resultados similares (riesgo de hallucination)	Devuelve vacio (fallo honesto)
Complejidad de configuracion	Menor (embeddings + vector store)	Mayor (knowledge graph + schema)
Ideal para	Busqueda de contenido, Q&A sobre documentos	Analitica, razonamiento multi-hop, verificacion

Conclusiones Clave

RAG siempre devuelve algo — incluso cuando los datos no existen, creando oportunidades de hallucination
GraphRAG devuelve resultados vacios cuando los datos estan ausentes — forzando respuestas honestas de "No lo se"
Las agregaciones deben ser calculadas por la base de datos, no adivinadas por el LLM
El schema en el docstring del tool habilita Text2Cypher — consultas precisas en lugar de similitud de texto
Ambos enfoques tienen su lugar — usa RAG para busqueda semantica, GraphRAG para recuperacion precisa de datos

Siguiente Paso

GraphRAG previene estadisticas y agregaciones alucinadas. Pero los agents aun pueden seleccionar el tool incorrecto o desperdiciar tokens procesando descripciones de tools irrelevantes.

Parte 2: Semantic Tool Selection muestra como el filtrado vectorial reduce los errores de seleccion de tools y los costos de tokens.

Pruebalo Tu Mismo

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations/01-faq-graphrag-demo
uv venv && uv pip install -r requirements.txt
# Start Neo4j with APOC plugin first
uv run test_graphrag_hallucinations.py

Puedes cambiar a cualquier proveedor compatible con Strands — consulta Strands Model Providers para la configuracion.

Referencias

Investigacion

Strands Agents

Codigo

Repositorio de Codigo

Gracias!

🇻🇪🇨🇱 Dev.to Linkedin GitHub Twitter Instagram Youtube

Detén las Hallucinations en Agentes de IA: 4 Técnicas Esenciales

Elizabeth Fuentes L — Wed, 18 Mar 2026 00:13:33 +0000

4 técnicas para detener las hallucinations en AI agents: Graph-RAG para recuperación precisa de datos, selección semántica de tools para elegir la herramienta correcta, guardrails neurosimbólicos para aplicar reglas de negocio, y validación multi-agent para detección de errores.

Objetivos de Aprendizaje

Cómo Graph-RAG previene hallucinations estadísticas con datos estructurados
Por qué la selección semántica de tools mejora la precisión y reduce costos de tokens
Cómo los guardrails neurosimbólicos con Strands Agents bloquean operaciones inválidas
Cómo la validación multi-agent detecta hallucinations antes de que lleguen a los usuarios

Esta demo utiliza Strands Agents. Patrones similares pueden aplicarse en LangGraph, AutoGen u otros frameworks de agents.

Repositorio y Prerequisitos

Clonar:

git clone https://github.com/aws-samples/sample-why-agents-fail
cd stop-ai-agent-hallucinations

Requisitos:

Python 3.9+
Acceso a un LLM (Amazon Bedrock, OpenAI, Anthropic u Ollama)
Credenciales de AWS configuradas (si se usa Bedrock)
Comprensión básica de AI agents y tool calling
Bibliotecas clave: Strands Agents, Neo4j, FAISS, SentenceTransformers

Técnica 1: Graph-RAG para Recuperación Precisa de Datos

Problema: El RAG tradicional recupera fragmentos de texto en lugar de ejecutar cálculos, lo que produce tres tipos de hallucination:

Estadísticas fabricadas a partir de agregaciones adivinadas
Recuperación incompleta cuando los datos están dispersos
Fabricación fuera de dominio cuando no existen datos relevantes

Solución: Los knowledge graphs proporcionan datos estructurados y verificables, donde las agregaciones son calculadas por la base de datos en lugar de ser adivinadas por los LLMs.

Código de Comparación

from strands import Agent, tool

# RAG Agent — vector similarity search
@tool
def search_faqs(query: str) -> str:
    """Search hotel FAQs using vector similarity."""
    query_embedding = embed_model.encode([query])
    distances, indices = index.search(query_embedding.astype('float32'), 3)
    return "\n".join([documents[idx]['text'][:500] for idx in indices[0]])

# Graph-RAG Agent — Cypher queries on knowledge graph
@tool
def query_knowledge_graph(cypher_query: str) -> str:
    """Execute a Cypher query against the hotel knowledge graph.
    Node labels: Hotel, Room, Amenity, Policy, Service
    Relationships: (Hotel)-[:HAS_ROOM]->(Room), (Hotel)-[:OFFERS_AMENITY]->(Amenity)
    """
    with driver.session() as session:
        result = session.run(cypher_query)
        records = list(result)
        if not records:
            return "No results found."
        return f"Found {len(records)} results:\n" + "\n".join(str(dict(r.items())) for r in records[:15])

rag_agent = Agent(tools=[search_faqs], model=model)
graph_agent = Agent(tools=[query_knowledge_graph], model=model)

Resultados de la Demo

Tipo de Consulta	RAG	Graph-RAG
Agregación: "¿Calificación promedio en París?"	⚠️ Calcula solo con 2 documentos	✅ AVG() nativo sobre todos
Conteo: "¿Hoteles con piscina?"	❌ "No tengo los datos"	✅ Preciso: 133
Multi-hop: "¿Tipos de habitación del mejor hotel?"	❌ No puede recorrer relaciones	✅ Recorrido Hotel → Room
Fuera de dominio: "Hoteles en la Antártida"	❌ Fabrica respuestas	✅ Honesto: "No hay hoteles"

Hallazgo Clave: Graph-RAG reduce las hallucinations porque devuelve resultados vacíos en lugar de respuestas fabricadas cuando los datos no están disponibles.

Técnica 2: Selección Semántica de Tools para Elegir la Herramienta Correcta

Problema: "Las hallucinations en tool-calling aumentan con la cantidad de tools." Cuando los agents tienen muchas tools similares, exhiben errores de selección de función, errores de idoneidad, errores de parámetros, errores de completitud y comportamiento de evasión de tools.

Problema Dual:

❌ Riesgo de hallucination: Más tools = más selecciones inapropiadas
❌ Desperdicio de tokens: 31 tools ≈ 4,500 tokens por consulta

Solución: Filtrado Semántico de Tools

Filtra las tools antes de que el agent las vea usando similitud vectorial.

from sentence_transformers import SentenceTransformer
import faiss

# Build index once
model = SentenceTransformer('all-MiniLM-L6-v2')
tool_embeddings = model.encode([tool.description for tool in ALL_TOOLS])
index = faiss.IndexFlatL2(384)
index.add(tool_embeddings)

# Filter per query
query_embedding = model.encode([query])
distances, indices = index.search(query_embedding, k=3)
relevant_tools = [ALL_TOOLS[i] for i in indices[0]]

Intercambio Dinámico de Tools con Strands Agents

from strands import Agent

# Create agent once
agent = Agent(tools=[...], model=model)

# Swap tools per query without losing conversation history
for query in conversation:
    relevant_tools = search_tools(query, top_k=3)
    agent.tool_registry.registry.clear()
    for tool in relevant_tools:
        agent.tool_registry.register_tool(tool)

    response = agent(query)  # agent.messages preserved

Resultados en 29 Consultas de Viaje

Las pruebas muestran mejoras significativas:

86.4% de reducción en errores de selección de tools
89% de reducción en costos de tokens
Precisión mantenida con menos tools visibles

Técnica 3: Guardrails Neurosimbólicos para AI Agents

Problema: Investigaciones demuestran que los agents generan hallucinations cuando las reglas de negocio se expresan únicamente en prompts de lenguaje natural. El prompt engineering no puede aplicar restricciones porque los prompts son sugerencias, no reglas ejecutables.

Solución: Usar hooks de Strands Agents para interceptar llamadas a tools antes de su ejecución y aplicar reglas simbólicas a nivel del framework.

Implementación

from strands import Agent, tool
from strands.hooks import HookProvider, HookRegistry, BeforeToolCallEvent

# Define symbolic rules
BOOKING_RULES = [
    Rule(
        name="max_guests",
        condition=lambda ctx: ctx.get("guests", 1) <= 10,
        message="Maximum 10 guests per booking"
    ),
]

# Create validation hook
class NeurosymbolicHook(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self.validate)

    def validate(self, event: BeforeToolCallEvent) -> None:
        ctx = {"guests": event.tool_use["input"].get("guests", 1)}
        passed, violations = validate(BOOKING_RULES, ctx)

        if not passed:
            event.cancel_tool = f"BLOCKED: {', '.join(violations)}"

# Clean tool (no validation logic)
@tool
def book_hotel(hotel: str, guests: int = 1) -> str:
    """Book a hotel room."""
    return f"SUCCESS: Booked {hotel} for {guests} guests"

# Attach hook to agent
hook = NeurosymbolicHook()
agent = Agent(tools=[book_hotel], hooks=[hook])

# Test
query = "Book hotel for 15 guests"
result = agent(query)  # ✅ Hook blocks before tool executes

Por Qué los Hooks de Strands Destacan

API sencilla: Implementar HookProvider y registrar callbacks
Validación centralizada: Un solo hook valida todas las tools
Tools limpias: Sin lógica de validación mezclada con lógica de negocio
Tipado seguro: Objetos de eventos fuertemente tipados
El LLM no puede evadir: Las reglas se aplican antes de la ejecución de la tool

Efectividad

Escenario	Prompt Engineering	Neurosimbólico con Hooks
Parámetros Inválidos	❌ Acepta	✅ Bloquea
Prerequisitos Faltantes	⚠️ A veces detecta	✅ Siempre bloquea
Evasión de Reglas	❌ Posible	✅ Imposible

Técnica 4: Validación Multi-Agent para Detección de Errores

Problema: Los agents individuales operan en aislamiento. Cuando generan hallucinations, no existe un mecanismo para detectar errores antes de que lleguen a los usuarios.

Solución: Múltiples agents especializados se validan mutuamente mediante debate estructurado, utilizando Strands Swarm para transferencias autónomas con contexto compartido.

Implementación

from strands import Agent
from strands.multiagent import Swarm

# Define specialized agents
executor = Agent(
    name="executor",
    tools=ALL_TOOLS,
    system_prompt="Execute requests, then hand off to validator"
)

validator = Agent(
    name="validator",
    system_prompt="Check for hallucinations. Say VALID or HALLUCINATION"
)

critic = Agent(
    name="critic",
    system_prompt="Final review. Say APPROVED or REJECTED"
)

# Create swarm - agents hand off autonomously
swarm = Swarm(
    [executor, validator, critic],
    entry_point=executor,
    max_handoffs=5
)

result = swarm("Book grand_hotel for John")

Comparación de Rendimiento

Enfoque	Detección de Hallucinations	Precisión	Latencia
Agent Individual	❌ Ninguna	⚠️ Fabrica	✅ Rápido
Multi-Agent	✅ Detecta errores	✅ Valida	⚠️ Más lento

Ejemplo: Cuando el executor intenta reservar "the_ritz_paris" (no existe), el validator detecta el hotel inválido y el critic devuelve FAILED en lugar de fabricar una alternativa.

Combinando Técnicas para Producción

Estas técnicas se complementan entre sí:

Graph-RAG asegura la precisión de los datos
Selección semántica de tools reduce errores de tools y costos de tokens
Reglas neurosimbólicas aplican restricciones de negocio
Validación multi-agent detecta las hallucinations restantes

Por Qué Strands Agents Python SDK

Gestión dinámica de tools sin perder el estado de la conversación
Multi-agent nativo con Swarm manejando transferencias
Validación a nivel de tool ejecutándose antes de que el LLM vea los resultados
Flexibilidad de modelos (Bedrock, OpenAI, Anthropic, Ollama)
Listo para producción en despliegues AWS

Conclusiones Clave

Las hallucinations son inevitables — enfócate en detección y mitigación, no en eliminación
Graph-RAG para precisión cuando necesitas cálculos exactos o relaciones
Filtrado semántico esencial con más de 10 tools similares
Las reglas simbólicas aplican cumplimiento de negocio donde el prompt engineering falla
La validación multi-agent detecta errores que los agents individuales no captan
Strands Agents permite despliegues en producción con tools dinámicas y soporte nativo multi-agent

Referencias

Gracias!

🇻🇪🇨🇱 Dev.to Linkedin GitHub Twitter Instagram Youtube
Linktr

Elizabeth Fuentes LFollow

I help developers build production-ready AI applications through hands-on tutorials and open-source projects.

🤔AWS me está quitando el trabajo...😵

Alex Rodríguez — Wed, 04 Mar 2026 12:36:21 +0000

Cómo desplegar ArgoCD en EKS con SSO sin gestionar un solo pod

Después de bastante tiempo entre eventos, organizaciones y mucho trabajo, volvemos por aquí para hablar de uno de los últimos “avances” que AWS ha presentado.

Con todo el tema de la IA, parece que solo estemos hablando de modelos, agentes y prompts. Pero nada más lejos de la realidad, seguimos gestionando muchísima infraestructura, y AWS sigue lanzando cosas muy interesantes para simplificar esa parte. (menos mal, algún anuncio no relacionado con IA también nos gusta escuchar).

Sin intentar que esto tuviese algún sentido, hemos terminado cerrando una rueda interesante alrededor de EKS, ArgoCD y IAM Identity Center en nuestros posts de #awstwins. La verdad es que en este post he tirado un poco de clickbait, a lo influencer, pero en realidad volvemos al barro y con lo de siempre: Kubernetes y automatización.

En concreto, vamos a hablar de EKS Capabilities.

En posts anteriores como el de acontinuación instalábamos y gestionábamos Argo CD nosotros. Hoy toca darle otra vuelta. Que para eso estamos, ¿no?

argocd&terraform

¿Qué es EKS Capabilities?

EKS Capabilities es una nueva feature de AWS que permite utilizar herramientas del ecosistema Kubernetes gestionadas directamente por AWS.

Es decir,no tienes que instalar nada, mantener nada ni preocuparte de por qué algo está caído.

Básicamente nuestro estilo de vida, no?

Actualmente hay 3 capabilities disponibles:

ArgoCD
De este ya hemos hablado mucho en diferentes posts y es el que vamos a usar puesto que cerramos el circulo. Incluiremos la parte del SSO.
KRO (Kubernetes Resource Orchestrator)
Permite crear recursos de forma declarativa agrupando varios recursos en uno solo.
ACK (AWS Controllers for Kubernetes)
Permite gestionar recursos de AWS directamente desde Kubernetes.

¿La parte interesante de todo esto?

Estas capabilities operan en el control plane de EKS, no en tus nodos ni en tu cuenta.
Resultado: menos componentes que mantener y menos cosas que se rompan.

¿Qué hemos desplegado?

Para ir rápido y evitar mantener algo "custom", hemos hecho lo de siempre: usar módulos de la community.

En este caso:

Una VPC
Un cluster EKS
Karpenter para el autoscaling de nodos (no entraremos en detalle aquí)

Si queréis ver más sobre Karpenter, mi hermano ya publicó un post sobre esto, que siempre podéis dejar un Like!💕:

karpenter

Volvamos a "Capabilities" que es lo que nos interesa. Tal y como decíamos, para esta PoC, podéis encontrar el módulo de EKS Capabilities aquí con los ejemplos, ya sabéis "Copy & Paste" o si no, siempre podéis tirar de IA para un código rápido.

terraform-aws-modules

SSO con IAM Identity Center

Una de las partes más interesantes de usar EKS Capabilities con ArgoCD es que el SSO no requiere ninguna integración adicional.

No hay que tocar:

ConfigMap de ArgoCD
argocd-cm
configuración de OIDC

Simplemente le dices a AWS qué grupos de IAM Identity Center corresponden a qué roles dentro de ArgoCD, y a funcionar.

Ejemplo en Terraform:

module "argocd_eks_capability" {
  source = "terraform-aws-modules/eks/aws//modules/capability"

  type         = "ARGOCD"
  cluster_name = module.eks.cluster_name

  configuration = {
    argo_cd = {
      aws_idc = {
        idc_instance_arn = one(data.aws_ssoadmin_instances.this.arns)
      }

      namespace = "argocd"

      rbac_role_mapping = [{
        role = "ADMIN"

        identity = [{
          id   = data.aws_identitystore_group.aws_administrator.group_id
          type = "SSO_GROUP"
        }]
      }]
    }
  }
}

En mi caso, para ir rápido con la PoC:

He usado mi grupo Admin
He desplegado el cluster en la misma cuenta donde tengo IAM Identity Center

Si queréis hacerlo cross-account, tendrás que crear un Role que permita a Terraform leer los grupos de Identity Center.

Dicho esto, vamos a la práctica de cómo quedaría esta parte.

Una vez desplegado nuestro cluster con todos los requisitos, podemos observar que tenemos argocd funcionando en nuestra pestaña de Capabilities.

Nos conectamos a nuestro endpoint (Argo API endpoint).

Como hemos comentado, la integración con SSO es super rápida, solo indica qué roles quieres usar y el mapping y listo!

Introducimos nuestras credenciales de AWS.

Y ya tenemos argo cd funcionando.

Lo mejor de todo

Cuando lances el código:

Se desplegará ArgoCD como capability
Tendrás SSO funcionando
Podrás loguearte directamente
Y empezar a desplegar tus aplicaciones

Y ahora viene lo mejor:

kubectl get pods -A

No verás ni un solo pod de ArgoCD.

Porque ArgoCD está ejecutándose en el control plane de EKS, no dentro de tu cluster.

Llevamos varios posts hablando de ArgoCD, cual es el mejor diseño, como instalarlo, como configurarlo, como gestionarlo y de repente llega un update de AWS y nos dice y si lo hago por ti? Encima con SSO de serie que más podemos pedir? 😁

Cabe remarcar como siempre que el objetivo es simplemente una PoC, para PROD deberíamos de revisar el tema de dominio privado, clusters adicionales, AppProjects por equipo... pero de eso ya hemos hablado en varias otros posts que podéis recuperar desde #AWTwinS.

Como siempre, cualquier feedback es bienvenido! o por aquí, por LinkedIn o donde podáis encontrarnos! Nos vemos por los eventos!

Crea tu asistente personal en los Meta Glasses.

Fernando Silva T — Fri, 27 Feb 2026 22:56:03 +0000

Fernando Silva TFollow

AWS Community Builder | Estudiante de Ingeniería Informática en Duoc UC. Apasionado por AWS, desarrollo Full-Stack y proyectos open source. Siempre buscando aprender y colaborar.

Implementación de un asistente de IA con entrada de voz y visión para lentes inteligentes usando Android, Amazon Bedrock y Claude Sonnet 4.5, sin dependencia de la IA nativa de Meta ni restricciones geográficas.

Este proyecto aborda cuatro problemas técnicos concretos:

Wake word detection sin librerías externas
Conversión de frames del SDK de Meta a formatos compatibles con Bedrock
Uso de la Converse API con input multimodal (imagen + texto)
Coordinación de flujos asíncronos en Jetpack Compose

Stack: Kotlin · Jetpack Compose · Meta Wearables DAT SDK · Amazon Bedrock

Nivel: Intermedio

Repositorio: https://github.com/fernandosilvot/aws-bedrock-meta-glasses

Introducción

Todo comenzó con una conversación casual. Elizabeth Fuentes me comentó sobre el Meta Wearables DAT SDK — una API que Meta había liberado para desarrolladores que permitía integrar funcionalidades personalizadas a los Meta Smart Glasses, más allá de la IA nativa de Meta.

La idea me fascinó de inmediato: ¿Y si pudiera conectar estos lentes con Amazon Bedrock y usar Claude Sonnet 4.5 en lugar de la IA de Meta? ¿Podría crear un asistente que no solo escuchara, sino que también viera lo que yo veo?

El resultado es Meta-Rock, un asistente activado por voz que captura el frame actual de la cámara, lo envía junto al texto de la consulta y reproduce la respuesta mediante TTS.

El problema y la visión

Los Meta Ray-Ban Smart Glasses vienen con IA integrada de Meta, pero tienen una limitación importante: solo está disponible en ciertos países. En Chile, donde vivo, no puedo acceder a esa funcionalidad nativa.

Además, incluso si estuviera disponible, quería explorar las posibilidades de integrar modelos de IA más avanzados y personalizables. ¿Qué podría lograr si conectara estos lentes directamente con Amazon Bedrock y Claude Sonnet 4.5?

Objetivo:

Construir un asistente hands-free que combine audio y visión, delegando toda la inferencia a Bedrock.

Arquitectura

Flujo de datos

Pipeline resumido:

Wake word → SpeechRecognizer
Captura de frame desde los lentes
Conversión I420 → JPEG
Request multimodal a Bedrock
Respuesta → Text-to-Speech

Stack tecnológico

Capa	Tecnología	Justificación
UI	Jetpack Compose	Estado reactivo
Voz	Android SpeechRecognizer	API nativa
Video	Meta Wearables DAT SDK	Acceso directo a cámara
IA	Amazon Bedrock	Inferencia multimodal
Modelo	Claude Sonnet 4.5	Razonamiento visual sólido

Demostración en video

Meta-Rock en ejecución

El video muestra el flujo completo:

Activación por wake word
Captura del frame desde los lentes
Envío multimodal a Bedrock
Respuesta hablada

📽️ Video de demostración:

Desafíos técnicos y decisiones de diseño

1. Wake word detection sin librerías externas

Decisión: usar SpeechRecognizer en escucha pasiva.

Motivación: evitar dependencias comerciales y mantener el stack 100% nativo.

Trade-offs:

Falsos positivos en ambientes ruidosos
Mayor consumo de batería
Latencia aproximada de 500 ms

class SpeechManager(
    private val context: Context,
    private val onWakeWord: () -> Unit,
    private val onTranscript: (String) -> Unit
) {
    private val speechRecognizer = SpeechRecognizer.createSpeechRecognizer(context)

    fun startPassiveListening() {
        val intent = Intent(RecognizerIntent.ACTION_RECOGNIZE_SPEECH).apply {
            putExtra(RecognizerIntent.EXTRA_LANGUAGE_MODEL, 
                    RecognizerIntent.LANGUAGE_MODEL_FREE_FORM)
            putExtra(RecognizerIntent.EXTRA_PARTIAL_RESULTS, true)
        }
        speechRecognizer.startListening(intent)
    }

    private fun checkForWakeWord(text: String) {
        val normalized = text.lowercase().trim()
        if (normalized.contains("hey friday") || normalized.contains("oye viernes")) {
            onWakeWord()
        }
    }
}

Aceptable para prototipo funcional y pruebas reales.

2. Envío constante de contexto visual

Decisión clave: enviar siempre el frame de la cámara junto al texto.

Motivación técnica:

Clasificadores por palabras clave son frágiles
Consultas ambiguas requieren visión (“¿y esto?”)
El modelo puede ignorar la imagen si no es relevante

private fun onTranscriptReady(transcript: String) {
    Log.d(TAG, "Transcript: $transcript")
    val frame = currentFrameProvider?.invoke() // Siempre captura el frame

    _uiState.update { 
        it.copy(
            state = NovaState.PROCESSING, 
            transcript = transcript, 
            sentWithImage = frame != null
        ) 
    }

    processingJob = viewModelScope.launch(Dispatchers.IO) {
        val systemPrompt = Prompts.system(language.locale.language)
        val response = if (frame != null) {
            bedrockClient.askWithImage(transcript, frame, systemPrompt)
        } else {
            bedrockClient.askTextOnly(transcript, systemPrompt)
        }
        // ...
    }
}

Costo: mayor consumo de tokens y ancho de banda

Beneficio: lógica simple y UX consistente

3. Conversión de formatos de video

El SDK de Meta entrega frames en formato I420 (YUV).

Bedrock requiere imágenes codificadas (JPEG).

Pipeline implementado:

I420 → NV21 → Bitmap → JPEG (85%)

private fun i420ToNV21(i420: ByteArray, width: Int, height: Int): ByteArray {
    val frameSize = width * height
    val nv21 = ByteArray(frameSize * 3 / 2)

    // Copiar Y
    System.arraycopy(i420, 0, nv21, 0, frameSize)

    // Intercalar U y V para NV21
    var uvIndex = frameSize
    var uIndex = frameSize
    var vIndex = frameSize + frameSize / 4

    for (i in 0 until frameSize / 4) {
        nv21[uvIndex++] = i420[vIndex++]
        nv21[uvIndex++] = i420[uIndex++]
    }

    return nv21
}

private fun bitmapToBytes(bitmap: Bitmap): ByteArray {
    val stream = ByteArrayOutputStream()
    bitmap.compress(Bitmap.CompressFormat.JPEG, 85, stream)
    return stream.toByteArray()
}

El 85% de compresión mantiene legibilidad de texto con menor latencia.

4. Integración con Bedrock (Converse API)

Se utiliza una única llamada con input multimodal:

Texto del usuario
Imagen capturada
System prompt dependiente del idioma

class BedrockClient {
    private val client = BedrockRuntimeClient {
        region = "us-east-1"
        credentialsProvider = StaticCredentialsProvider(
            accessKeyId,
            secretAccessKey
        )
    }

    suspend fun askWithImage(prompt: String, frame: Bitmap, systemPrompt: String): String {
        val imageBytes = bitmapToBytes(frame)
        val request = ConverseRequest {
            modelId = "us.anthropic.claude-sonnet-4-5-20250929-v1:0"
            system = listOf(SystemContentBlock.Text(systemPrompt))
            messages = listOf(
                Message {
                    role = ConversationRole.User
                    content = listOf(
                        ContentBlock.Image(
                            ImageBlock {
                                format = ImageFormat.Jpeg
                                source = ImageSource.Bytes(imageBytes)
                            }
                        ),
                        ContentBlock.Text(prompt)
                    )
                }
            )
            inferenceConfig {
                maxTokens = 300
                temperature = 0.7F
            }
        }
        return client.converse(request).output!!.asMessage().content.first().asText()
    }
}

La inferencia se realiza directamente desde Android usando el SDK oficial de AWS para Kotlin.

5. Gestión de estado en la UI

Se definen estados explícitos para coordinar audio, video e inferencia:

enum class NovaState {
    IDLE,           // Esperando "Hey Viernes"
    LISTENING,      // Escuchando pregunta
    PROCESSING,     // Enviando a Bedrock
    RESPONDING      // Mostrando/leyendo respuesta
}

data class NovaUiState(
    val state: NovaState = NovaState.IDLE,
    val transcript: String? = null,
    val response: String? = null,
    val sentWithImage: Boolean = false
)

Esto reduce errores visuales y facilita el debugging.

Casos de uso validados

Identificación visual (objetos, colores, texto)
Lectura de pantallas
Consultas generales con o sin contexto visual

El modelo decide dinámicamente si usar la imagen.

Resultados

Latencia end-to-end: 2–3 segundos
Precisión wake word: ~95% en ambientes silenciosos
Consumo de batería: ~15% por hora
Tamaño del APK: 12.4 MB

Lecciones técnicas

Optimizar demasiado temprano complica la arquitectura
JPEG al 85% es un buen punto de equilibrio
Wake word sin SDK dedicado tiene límites claros
Claude maneja bien inputs multimodales ruidosos

Próximos pasos

Clasificador local ligero para decidir envío de imagen
Optimización de consumo energético
Nuevos flujos de análisis visual continuo

Conclusión

Construir Meta-Rock fue un desafío técnico fascinante que surgió de una necesidad real: acceder a funcionalidades de IA en los Meta Ray-Ban que no están disponibles en mi región.

Pero más allá de resolver esa limitación, este proyecto me permitió explorar las posibilidades de integrar wearables con modelos de IA de última generación. ¿Qué más se puede lograr cuando tienes acceso directo a la cámara de unos lentes inteligentes y a modelos como Claude Sonnet 4.5?

Lo más emocionante es que esto es solo el comienzo. Planeo seguir mejorando Meta-Rock, explorando nuevas funcionalidades y documentando cada avance en futuros blogs. Desde análisis de escenas en tiempo real hasta traducción visual instantánea, las posibilidades son infinitas.

Este proyecto me enseñó que las limitaciones geográficas o de plataforma no tienen por qué detenerte. Con las herramientas adecuadas (AWS, SDKs abiertos, creatividad), puedes construir soluciones que no solo resuelven tus necesidades, sino que abren puertas a experiencias completamente nuevas.

Recursos

Repositorio GitHub: aws-bedrock-meta-glasses
Repo original de Meta: CameraAccess Sample
Meta Wearables SDK: Documentación oficial
Amazon Bedrock: Converse API
Claude Sonnet 4.5: Modelo en Bedrock

Agradecimientos

Este proyecto no hubiera sido posible sin:

Elizabeth Fuentes por compartir el descubrimiento del Meta Wearables DAT SDK y motivarme a explorar esta tecnología
El equipo de Meta por abrir el SDK de los Ray-Ban a la comunidad de desarrolladores
AWS por hacer Bedrock accesible y proporcionar modelos de IA de clase mundial
La comunidad de AWS Community Builders por el apoyo constante y el intercambio de conocimientos

¿Tienes preguntas o ideas para mejorar Meta-Rock? Déjalas en los comentarios o contáctame en fernandosilvot.cl

Made with ❤️ in Chile 🇨🇱

⚙️ Control PID con AWS Lambda Durable Functions: Simulando un PID y un reactor ⚗️ con estado persistente ☁️

olcortesb — Tue, 10 Feb 2026 09:04:24 +0000

Introducción

Recordando conceptos básicos de mis años de Ing. Electrónica y cómo ha evolucionado, pensaba en cómo sería la implementación de un controlador PID completamente cloud, al menos a modo práctico, en la operativa posiblemente no sea aplicable a todos los sistemas de control. Los controladores PID (Proporcional-Integral-Derivativo) tradicionalmente requieren ejecución continua, estado persistente y ciclos de control precisos. ¿Cómo implementar esto en un entorno serverless donde las funciones son efímeras por naturaleza?

En este artículo presento una prueba de concepto que combina control PID con AWS Lambda Durable Functions, demostrando cómo crear sistemas de control de larga duración sin "pagar" por tiempo de espera entre iteraciones. La implementación simula un reactor con control de temperatura, utilizando las nuevas capacidades de Lambda para workflows con estado persistente.

🎯 ¿Qué es un Controlador PID?

Un controlador PID es un mecanismo de control por retroalimentación ampliamente utilizado en sistemas industriales. Calcula una señal de control basándose en tres elementos:

Proporcional (P): Responde proporcionalmente al error actual entre el setpoint (SP) y el valor del proceso (PV). Cuanto mayor es el error, mayor es la acción correctiva
Integral (I): Corrige el error acumulado en el tiempo, eliminando el error residual en estado estacionario
Derivativo (D): Anticipa cambios futuros basándose en la tasa de cambio del error, proporcionando amortiguación al sistema

La fórmula básica del PID es:

CV(t) = Kp·e(t) + Ki·∫e(t)dt + Kd·de(t)/dt

Donde:

CV: Control Variable o salida del controlador normalmente en porcentaje (0-100%)
e(t): Error = SP - PV (Setpoint - Process Variable)
SP: Setpoint o valor deseado
PV: Process Variable o valor actual medido
Kp, Ki, Kd: Constantes de ajuste o ganancias del controlador

Tipos de respuesta:

La relación entre las constantes y su aplicación sobre distintos sistemas puede producir tres tipos de respuestas conocidas:

Subamortiguada (underdamped): Oscila antes de estabilizarse (overshoot + oscilaciones)
Críticamente amortiguada (critically damped): Converge lo más rápido posible sin overshoot
Sobreamortiguada (overdamped): Converge lentamente sin overshoot

🚀 ¿Qué son AWS Lambda Durable Functions?

AWS Lambda Durable Functions es una característica nativa que permite crear workflows de larga duración con estado persistente. A diferencia de las funciones Lambda tradicionales, las Durable Functions pueden:

Suspenderse sin costo: context.wait() pausa la ejecución sin cargos de cómputo
Mantener estado: Checkpointing automático en cada step
Recuperarse de fallos: Reinicia desde el último checkpoint exitoso
Ejecutar por horas/días: Sin mantener la Lambda activa continuamente

Referencia oficial: AWS Lambda Durable Functions

Primitivas Core del SDK

from aws_durable_execution_sdk_python import (
    DurableContext,
    durable_execution,
    durable_step,
)
from aws_durable_execution_sdk_python.config import Duration

@durable_step
def my_step(step_context, arg):
    # Lógica con checkpointing automático
    return result

@durable_execution
def lambda_handler(event, context: DurableContext):
    result = context.step(my_step(arg))
    context.wait(Duration.from_seconds(60))  # SIN COSTO
    return result

🏗️ Arquitectura de la POC

Bueno, si has llegado leyendo hasta aquí ya es hora de jugarnos, construir e ir a la sustancia. Vamos a implementar una POC que nos permita simular en Lambda Durable Functions tanto un PID como un reactor, persistiendo sus estados internamente y gestionando el flujo con las primitivas del SDK.

El sistema sería el que se muestra en la imagen a continuación.

Bien, ahora, si tuviéramos que simular esto con la versión tradicional de Lambda, necesitaríamos un par de SQS al menos para gestionar el estado y la invocación de las lambdas. La implementación que proponemos sigue este flujo:

Componentes principales:

API Gateway: Recibe el setpoint deseado
PID Controller Lambda: Ejecuta el loop de control con estado persistente
Reactor Simulator Lambda: Simula el comportamiento físico del reactor
CloudWatch Metrics: Almacena métricas para visualización

🛠️ Implementación

Lambda PID Controller

El controlador implementa el algoritmo PID completo usando Durable Functions. El código completo está disponible en el repositorio:

📁 Código fuente: terraform/src/pid_controller/app.py

Componentes principales:

@durable_step calculate_pid(): Implementa el algoritmo PID en forma paralela discreta
@durable_step invoke_reactor(): Invoca la Lambda del reactor para obtener la nueva temperatura
@durable_step publish_metrics(): Publica métricas a CloudWatch (SetpointTemperature, ActualTemperature, TemperatureError)
@durable_execution lambda_handler(): Orquesta el loop de control con context.step() y context.wait()

Ecuación PID implementada:

error = setpoint - current_temp
integral += error * SAMPLE_TIME
derivative = (error - last_error) / SAMPLE_TIME
cv = KP * error + KI * integral + KD * derivative
cv = max(0, min(100, cv))  # Limitar entre 0-100

Lambda Reactor Simulator

El simulador implementa un modelo físico simplificado del reactor. El código completo está disponible en el repositorio:

📁 Código fuente: terraform/src/reactor_simulator/app.py

Componentes principales:

@durable_step simulate_reactor_step(): Simula el comportamiento térmico del reactor
@durable_execution lambda_handler(): Recibe el control value y retorna la nueva temperatura

Física del reactor implementada:

# Enfriamiento natural (Ley de enfriamiento de Newton)
cooling = (current_temp - AMBIENT_TEMP) * COOLING_RATE

# Calentamiento por control value (0-100)
heating = control_value * HEATING_EFFICIENCY

# Cambio de temperatura con inercia térmica
temp_change = heating - cooling
new_temp = current_temp + temp_change * (1 - THERMAL_INERTIA)

🚀 Despliegue con Terraform

Inicialmente el despliegue lo realizaría con AWS SAM, sin embargo encontré algunos issues con la integración del SDK con AWS SAM. La alternativa era agregarle una lambda layer, pero no quería probar Lambda Durable fuera de la configuración más básica, por lo tanto me mudé a Terraform para el despliegue.

Configuración de Lambda Durable

En Terraform, las funciones durable requieren configuración específica:

resource "aws_lambda_function" "pid_controller" {
  filename         = data.archive_file.pid_controller.output_path
  function_name    = "${var.project_name}-pid-controller"
  role             = aws_iam_role.pid_controller.arn
  handler          = "app.lambda_handler"
  runtime          = "python3.13"
  timeout          = 900
  memory_size      = 128
  publish          = true

  durable_config {
    execution_timeout = 3600  # 1 hora
    retention_period  = 7     # 7 días
  }

  logging_config {
    log_format = "JSON"
    log_group  = aws_cloudwatch_log_group.pid_controller.name
  }

  environment {
    variables = {
      REACTOR_FUNCTION_NAME = "${aws_lambda_function.reactor_simulator.function_name}:prod"
      KP                    = "0.50"
      KI                    = "0.0004"
      KD                    = "0.20"
      SAMPLE_TIME           = "60"
      MAX_ITERATIONS        = "40"
    }
  }
}

IAM Policy para Durable Functions

resource "aws_iam_role_policy_attachment" "pid_controller_durable" {
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicDurableExecutionRolePolicy"
  role       = aws_iam_role.pid_controller.name
}

🧪 Pruebas y Simulación

Cómo lo probamos: dejamos un valor esperado para que el PID use como referencia realizando un CURL al API Gateway y empezamos a ver cómo se comporta el sistema.

Iniciar Control PID

curl -X POST https://xxxxx.execute-api.us-east-1.amazonaws.com/prod/setpoint \
  -H "Content-Type: application/json" \
  -d '{"setpoint": 75.0}'

Respuesta:

{
  "message": "PID control completed - 40 iterations",
  "final_temperature": 74.82,
  "setpoint": 75.0,
  "iterations": 40
}

¿Cómo se ven los Steps?

Una nueva pestaña (1) nos mostrará cómo se ven las ejecuciones de toda la Lambda Durable Function, indicando qué versión (2) está corriendo, indicando un ID (3) y el estado de la ejecución (4).

La relación de las máquinas de estado de las dos lambdas

Una de las features más interesantes es que podemos aprovechar la persistencia del estado evitando colocar SQS, DynamoDB, S3 dependiendo del caso, y ahora podemos llamar a otra función directamente gracias a la gestión de la máquina de estado.

En la imagen arriba, 1 y 2 muestran las interacciones entre los steps y los waits, que indicamos antes en el presente artículo. Lo interesante es que se ve claramente el invoke_reactor que para nuestra simulación es invocar a otra Lambda Durable Function que permitirá simular el estado del reactor, guardando la temperatura y la inercia térmica del mismo.

Ya dentro de la Lambda del reactor podemos ver cómo se ejecutan los pasos que recalculan el comportamiento del reactor respecto a la temperatura, simulando el comportamiento real de un reactor.

La respuesta final:

Para visualizar de manera aproximada cómo se estudian las gráficas de los sistemas PID, creé un dashboard que pueden consultar en el código fuente:

Código en Terraform del dashboard

Los valores configurados son para obtener una respuesta críticamente amortiguada que sería el objetivo a perseguir para un sistema de control tipo PID. Los valores los encontré realizando una simulación en Python que muestro a continuación.

Simulación Local

Sí, esto podría ser otro post, pero he disfrutado tanto haciendo este artículo que lo dejo por aquí. Como quería refrescar cómo funcionaba el PID, me creé un proyecto que es básicamente un simulador Python para probar diferentes configuraciones de PID. En base a esta simulación encontré los valores que colocamos al sistema simulado en la POC.

cd simulation

# Configurar parámetros en .env
cat > .env << EOF
KP=0.50
KI=0.0004
KD=0.20
SETPOINT=75.0
SAMPLE_TIME=30
MAX_ITERATIONS=40
THERMAL_INERTIA=0.18
EOF

# Ejecutar simulación
python simulate_pid.py

Resultados de Simulación

El simulador genera gráficas mostrando el comportamiento del sistema con diferentes configuraciones de PID. Todas las simulaciones utilizan:

Setpoint: 75°C
Temperatura inicial: 20°C
Sample time: 30 segundos
Iteraciones: 40
Inercia térmica: 0.18

Configuración 1: Kp=0.20, Ki=0.0001, Kd=0.30 (Sobreamortiguada)

Características observadas:

Convergencia lenta y suave sin overshoot
Tiempo de estabilización: ~20 minutos
Temperatura final: ~74.5°C
Error final: ~0.5°C
Control Value máximo: ~15%
Comportamiento: Sistema muy conservador, ideal cuando no se permiten sobrepasadas

Configuración 2: Kp=0.50, Ki=0.0004, Kd=0.20 (Críticamente amortiguada)

Características observadas:

Convergencia rápida sin overshoot significativo
Tiempo de estabilización: ~15 minutos
Temperatura final: ~74.8°C
Error final: ~0.2°C
Control Value máximo: ~30%
Comportamiento: Balance óptimo entre velocidad y estabilidad

Configuración 3: Kp=1.2, Ki=0.002, Kd=0.1 (Subamortiguada)

Características observadas:

Convergencia muy rápida con overshoot
Overshoot máximo: ~2°C (alcanza ~77°C)
Oscilaciones visibles antes de estabilizar
Tiempo de estabilización: ~12 minutos
Temperatura final: ~75.0°C
Error final: ~0.05°C
Control Value máximo: ~70%
Comportamiento: Sistema agresivo, respuesta rápida pero con oscilaciones

📊 Verificación de Resultados

CloudWatch Logs

Los logs muestran cada iteración del control:

{
  "timestamp": "2026-01-15T10:30:45.123Z",
  "level": "INFO",
  "message": "Iteration 15: Setpoint=75.00, Temp=72.34, CV=45.67"
}

CloudWatch Metrics

Las métricas publicadas permiten visualizar:

SetpointTemperature: Temperatura deseada (constante)
ActualTemperature: Temperatura real del reactor
TemperatureError: Error absoluto

Para visualizar en CloudWatch:

Ir a CloudWatch Console → Metrics
Buscar namespace "PIDControl-v2"
Crear dashboard con las tres métricas

Logs del Reactor

{
  "timestamp": "2026-01-15T10:30:45.456Z",
  "level": "INFO",
  "message": "Reactor: CV=45.67, Temp=72.34, Ambient=20.0"
}

💡 Ventajas de Lambda Durable Functions

✅ Sin Costo Durante Esperas

El uso de context.wait() es una de las características más poderosas de Lambda Durable Functions. Según la documentación oficial de AWS, cuando una función durable está en estado de espera usando context.wait(), no se cobra por tiempo de cómputo.

# Antes: Lambda ejecutándose = $$$
time.sleep(60)  # Pagas por 60 segundos de ejecución

# Ahora: Lambda suspendida = $0
context.wait(Duration.from_seconds(60))  # Sin cargo durante la espera

Ejemplo de ahorro de costos para el caso de nuestro control PID con SAMPLE_TIME de 60 segundos y 40 iteraciones:

Sin Durable: 40 minutos de ejecución continua = 2,400 segundos facturables
Con Durable: ~5 segundos de ejecución real (solo cómputo activo) = 5 segundos facturables
Ahorro: ~99.8% en costos de cómputo

Referencia: How Lambda Durable Functions work - AWS Documentation

Checkpointing Automático

Cada context.step() guarda el estado:

# Si falla aquí, reinicia desde el último step exitoso
pid_result = context.step(calculate_pid(...))  # ✓ Checkpoint
reactor_data = context.step(invoke_reactor(...))  # ✓ Checkpoint
context.step(publish_metrics(...))  # ✓ Checkpoint

Sin Infraestructura Extra

Comparación con arquitecturas tradicionales:

Sin Durable Functions:

DynamoDB para estado
SQS para coordinación
EventBridge para scheduling
Step Functions para workflow

Con Durable Functions:

Solo Lambda (estado incluido)

Código Más Simple

El código es más legible y mantenible:

# Loop natural con estado persistente
for iteration in range(MAX_ITERATIONS):
    pid_result = context.step(calculate_pid(...))
    reactor_data = context.step(invoke_reactor(...))
    context.step(publish_metrics(...))
    context.wait(Duration.from_seconds(SAMPLE_TIME))

📋 Casos de Uso Prácticos

Según el blog oficial de AWS y la documentación técnica, Lambda Durable Functions es especialmente útil para escenarios donde se requieren workflows de larga duración con estado persistente. Los casos de uso documentados incluyen: workflows de aprobación humana donde los procesos esperan aprobaciones sin mantener recursos activos; procesamiento de datos por lotes como ETL jobs que esperan entre etapas; orquestación de microservicios para coordinar múltiples servicios con estado; monitoreo y polling para sistemas que verifican condiciones periódicamente; y workflows de IA y ML para pipelines de entrenamiento y procesamiento de modelos. Estos patrones aprovechan la capacidad de context.wait() para suspender la ejecución sin cargos de cómputo, haciendo viable económicamente mantener workflows activos durante períodos prolongados.

Referencias: AWS Blog - Lambda Durable Functions | Use Cases Documentation

⚠️ Consideraciones Importantes

Límites del Servicio

Según la documentación oficial de AWS Lambda Durable Functions:

Execution timeout: Sin límite de tiempo total de ejecución (la función puede ejecutarse indefinidamente usando context.wait())
Individual function timeout: Máximo 15 minutos por invocación individual de función
Retention period: Máximo 1 año (365 días) para estado persistente
Payload size: Máximo 256 KB por step
Maximum concurrent executions: Sujeto a las cuotas de concurrencia de Lambda de la cuenta

Nota importante: Aunque no hay límite en el tiempo total de ejecución durable, cada invocación individual de la función Lambda está limitada a 15 minutos. El estado se persiste automáticamente entre invocaciones.

Mejores Prácticas

Usar steps granulares: Cada step debe ser una unidad lógica de trabajo
Limitar payload: Pasar solo datos necesarios entre steps
Implementar timeouts: Configurar timeouts apropiados para cada función
Monitorear métricas: Usar CloudWatch para detectar problemas
Manejar errores: Implementar retry logic en steps críticos

Cuándo NO usar Durable Functions

Latencia ultra-baja: Si necesitas respuesta en milisegundos
Alto throughput: Para miles de ejecuciones concurrentes
Workflows complejos: Step Functions puede ser mejor opción
Estado muy grande: Si el estado excede 256 KB

📊 Conclusiones (Para nuestro caso de prueba PID)

Esta POC intente validar que es posible implementar un sistema de control PID en arquitecturas serverless usando AWS Lambda Durable Functions. y lo que podemos identificar como concluciones, comparado con el uso de AWS lambda tradicional sin 'Durable Function' seria:

Economía: context.wait() elimina costos de tiempo de espera entre iteraciones
Simplicidad: Código más limpio sin necesidad de DynamoDB, SQS o Step Functions
Resiliencia: Checkpointing automático permite recuperación ante fallos
Flexibilidad: Fácil ajuste de parámetros PID y configuración del reactor
Escalabilidad: Cada control PID es independiente y puede escalar horizontalmente

Limitaciones identificadas:

Latencia: No apto para control en tiempo real (< 1 segundo)
Cold starts: Primera invocación puede tardar 2-3 segundos
Debugging: Más complejo que funciones Lambda tradicionales

Finalmente el código completo está disponible en GitHub

🔗 Referencias y Enlaces Útiles

Gracias por leer.

¡Saludos!

Oscar Cortés

Agregando reactividad ⚡ a nuestras arquitecturas con AWS CloudWatch Subscription Filter

olcortesb — Wed, 21 Jan 2026 09:04:40 +0000

Introducción

Uno de los principales desafíos al implementar sistemas basados en eventos es lograr la reactividad necesaria para que respondan automáticamente ante eventos específicos. Cuando integramos sistemas tradicionales de request-response con arquitecturas gestionadas por eventos, siempre surge la necesidad de agregar esta capacidad reactiva.

En esta búsqueda constante de soluciones robustas para agregar reactividad a los sistemas, hoy presento una prueba de concepto utilizando una de esas features que están "escondidas" dentro de los servicios de AWS. Exploraremos AWS CloudWatch Subscription Filter como una herramienta poderosa para agregar reactividad automática a nuestros sistemas.

La intención de esta POC (Proof of Concept) es utilizar AWS CloudWatch Subscription Filter para procesar logs automáticamente y ejecutar acciones basadas en el contenido de los mismos. Crearemos una arquitectura serverless que captura logs específicos y los procesa en tiempo real.

🚀 ¿Qué es AWS CloudWatch Subscription Filter?

AWS CloudWatch Subscription Filter es una feature que permite filtrar y procesar logs en "tiempo real" desde CloudWatch Logs. Cuando los logs coinciden con un patrón específico, el filtro puede enviar automáticamente esos datos a destinos como Lambda, Kinesis Data Streams, o Amazon Data Firehose. En este post vamos a explorar la integración con AWS Lambda específicamente.

AWS CloudWatch Logs Subscription Filters

🏗️ Arquitectura de la POC

Nuestra implementación sigue este flujo:

API Gateway → Recibe POST requests
Lambda Controller → Procesa el POST y loggea siempre
CloudWatch Subscription Filter → Captura logs con enable=true automáticamente
Lambda Processor → Procesa logs capturados y guarda en DynamoDB

🛠️ Implementación

Lambda Controller

La Lambda Controller recibe los POST requests y genera logs específicos que serán capturados por el subscription filter. Aquí es importante anotar que esta lambda podría ser cualquier log de nuestro sistema tradicional o reactivo que tenga la capacidad de escribir logs en CloudWatch.

import json
from datetime import datetime

def handler(event, context):
    try:
        body = json.loads(event['body'])

        # Log específico para el subscription filter
        if body.get('enable', False):
            print(f"✅ ORIGINAL_POST: {json.dumps(body)}")
            response_message = "POST logged - subscription filter will process it"
        else:
            print(f"❌ POST received but enable=false: {json.dumps(body)}")
            response_message = "POST received but enable=false, not processed"

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': response_message,
                'received_data': body,
                'timestamp': datetime.utcnow().isoformat()
            })
        }
    except Exception as e:
        print(f"Error processing request: {str(e)}")
        return {'statusCode': 500, 'body': json.dumps({'error': str(e)})}

⚙️ CloudWatch Subscription Filter

El filtro se configura para capturar únicamente logs que contengan el patrón ✅ ORIGINAL_POST. Es importante destacar que este patrón lo agrego en la lambda cuando el post cumple las condiciones de enable:true. Destaco la manera cómo se configura el filtro; aunque hay límites en el número de filtros que se pueden configurar por log group (Ref), la forma de agregar filtros es muy simple.

# Ejemplo en Terraform de cómo configurar un subscription filter
resource "aws_cloudwatch_log_subscription_filter" "posts_filter" {
  name            = "${var.project_name}-posts-filter"
  log_group_name  = aws_cloudwatch_log_group.controller_lambda_logs.name
  filter_pattern  = "✅ ORIGINAL_POST"
  destination_arn = aws_lambda_function.processor_lambda.arn
}

Lambda Processor

La Lambda Processor recibe los logs filtrados, los descomprime y procesa el contenido. En este caso, a diferencia de la lambda controller, sí necesitamos esta lambda para que procese el log y se active ante el evento del subscription filter.

import json
import boto3
import gzip
import base64
import uuid
from datetime import datetime

def handler(event, context):
    try:
        table = boto3.resource('dynamodb').Table(os.environ['DYNAMODB_TABLE'])

        # Descomprimir datos de CloudWatch Logs
        # Este punto es importante: CloudWatch envía los datos codificados 
        # en base64, hay que decodificarlos primero antes de trabajarlos
        compressed_payload = base64.b64decode(event['awslogs']['data'])
        uncompressed_payload = gzip.decompress(compressed_payload)
        log_data = json.loads(uncompressed_payload)

        # Procesar cada evento de log
        for log_event in log_data['logEvents']:
            message = log_event['message']

            if '✅ ORIGINAL_POST:' in message:
                # Extraer JSON del POST original
                json_start = message.find('{')
                if json_start != -1:
                    original_post = json.loads(message[json_start:])

                    # Guardar en DynamoDB
                    item = {
                        'id': str(uuid.uuid4()),
                        'original_post': original_post,
                        'timestamp': datetime.utcnow().isoformat(),
                        'processed_at': datetime.utcnow().isoformat()
                    }

                    table.put_item(Item=item)
                    print(f"Saved to DynamoDB: {item['id']}")

        return {'statusCode': 200, 'body': 'Successfully processed logs'}
    except Exception as e:
        print(f"Error processing logs: {str(e)}")
        return {'statusCode': 500, 'body': f'Error: {str(e)}'}

🚀 Despliegue con Terraform

Estructura del proyecto

aws-cloudwatch-subscription-filter/
├── src/
│   ├── controller_lambda.py
│   └── processor_lambda.py
├──  main.tf
├──  api_gateway.tf
├──  lambda_functions.tf
├──  cloudwatch_subscription_filter.tf
├──  dynamodb_table.tf
├──  iam_roles.tf
├──  build_lambdas.sh
└──  README.md

Comandos de despliegue

# 1. Construir las Lambdas (zip y dependencias)
./build_lambdas.sh

# 2. Inicializar Terraform (credenciales previamente configuradas)
terraform init

# 3. Planificar el despliegue
terraform plan

# 4. Aplicar la infraestructura
terraform apply

# 5. Luego de las pruebas
terraform destroy

🧪 Pruebas y Validación

POST que NO será procesado

Es importante tomar los outputs del Terraform y reemplazarlos en {your-api-gateway-url}

curl -X POST https://your-api-gateway-url/posts \
  -H "Content-Type: application/json" \
  -d '{
    "enable": false,
    "message": "Este mensaje no será procesado",
    "data": "algunos datos"
  }'

POST que SÍ será procesado

curl -X POST https://your-api-gateway-url/posts \
  -H "Content-Type: application/json" \
  -d '{
    "enable": true,
    "message": "Este mensaje será procesado y guardado en DynamoDB",
    "data": "algunos datos importantes"
  }'

📊 Verificación de Resultados

1. CloudWatch Logs

Revisar los logs de la Lambda Controller para confirmar que se generan correctamente:


❌ POST received but enable=false: {"enable": false, "message": "Este mensaje no ser\u00e1 procesado", "data": "algunos datos"}
END RequestId: 
REPORT RequestId: 
Duration: 1.81 ms Billed Duration: 87 ms Memory Size: 128 MB Max Memory Used: 32 MB Init Duration: 84.44 ms

START RequestId:  Version: $LATEST
✅ ORIGINAL_POST: {"enable": true, "message": "Este mensaje ser\u00e1 procesado y guardado en DynamoDB", "data": "algunos datos importantes"}
END RequestId: 
REPORT RequestId:  
Duration: 1.46 ms Billed Duration: 2 ms 
Memory Size: 128 MB Max Memory Used: 32 MB

2. DynamoDB

Verificar que los datos se guardaron en la tabla:

aws dynamodb scan --table-name cloudwatch-subscription-filter-posts --limit 1 --region us-east-1

// Response
{
    "Items": [
        {
            "original_post": {
                "M": {
                    "message": {
                        "S": "Este mensaje será procesado y guardado en DynamoDB"
                    },
                    "data": {
                        "S": "algunos datos importantes"
                    },
                    "enable": {
                        "BOOL": true
                    }
                }
            },
            "log_timestamp": {
                "N": "1768222052525"
            },
            "id": {
                "S": "51991cf6-23a5-463b-a8d1-dc2f047ab13c"
            },
            "processed_at": {
                "S": "2026-01-12T12:47:42.465810"
            },
            "timestamp": {
                "S": "2026-01-12T12:47:42.465790"
            }
        }
    ],
    "Count": 1,
    "ScannedCount": 1,
    "LastEvaluatedKey": {
        "id": {
            "S": "51991cf6-23a5-463b-a8d1-dc2f047ab13c"
        }
    }
}

3. CloudWatch Subscription Filter

Confirmar que el filtro se creó y está activo en la consola de AWS.

🔧 Características Técnicas

Filtro de Patrones

Patrón: ✅ ORIGINAL_POST (He probado íconos porque me parece un punto crítico)
Sensible a mayúsculas: Sí
Procesamiento: máximo 30 segundos hasta que se persiste el registro en DynamoDB.

Compresión de Datos

CloudWatch Logs envía los datos comprimidos con gzip y codificados en base64:

# Proceso de descompresión
compressed_payload = base64.b64decode(event['awslogs']['data'])
uncompressed_payload = gzip.decompress(compressed_payload)
log_data = json.loads(uncompressed_payload)

Estructura de Datos

Los logs procesados incluyen:

ID único generado con UUID
POST original completo
Timestamps de creación y procesamiento
Metadatos del log event

💡 Casos de Uso Prácticos (Siempre como referencia, pero hay muchos más...)

1. Monitoreo de Errores

# En la Lambda Controller
if error_occurred:
    print(f"🚨 ERROR_ALERT: {json.dumps(error_details)}")

2. Auditoría de Transacciones

# Para transacciones críticas
if transaction_type == "payment":
    print(f"💰 PAYMENT_LOG: {json.dumps(transaction_data)}")

3. Análisis de Comportamiento

# Para eventos de usuario
if user_action in ["login", "purchase", "signup"]:
    print(f"👤 USER_EVENT: {json.dumps(user_data)}")

📋 Conclusiones

Aproveché esta POC para validar el funcionamiento de AWS CloudWatch Subscription Filter identificando:

Automatización del procesamiento de logs basado en patrones
Integración despierta la lambda processor con una latencia razonable
Arquitectura serverless escalable y costo-eficiente. Importante anotar que no se cobra por detectar los eventos sino por los datos enviados a la lambda o el Kinesis.
Filtrado inteligente que procesa solo logs relevantes y se puede tener un control atómico de los eventos que se necesitan.

Esta implementación sirve como base para casos de uso más complejos como monitoreo de errores, auditoría de transacciones, análisis de comportamiento de usuarios, y alertas en tiempo real.

El código completo está disponible en GitHub y listo para desplegar con Terraform, proporcionando una base sólida para proyectos que requieran procesamiento automático de logs en AWS.

⚠️ Consideraciones Importantes

Límites del Servicio

Máximo 2 subscription filters por log group
Filtros de hasta 1024 caracteres
Rate limiting en destinos Lambda

Mejores Prácticas

Usar patrones específicos para evitar ruido
Implementar retry logic en el processor
Monitorear métricas de CloudWatch
Configurar alertas para errores

🔗 Referencias y Enlaces Útiles

Gracias por leer.

¡Saludos!

Oscar Cortés

AWS DocumentDB Streams: Configuración y caso de uso

olcortesb — Wed, 07 Jan 2026 15:29:39 +0000

Entre finales del 2025 y principios del presente año, con el equipo de trabajo de ACKStorm nos enfrentamos a un desafío interesante. Necesitábamos diseñar una arquitectura serverless para una aplicación que necesitaba un manejo de un alto volumen de datos. En el proceso de diseño se identifico que el uso de una DynamoDB no era una buena opción. Con este escenario analizamos el costo de las distintas bases de datos que tenemos en AWS donde DocumentDB se presentó como la mejor opción.

Una de las mejores características, al menos desde mi parecer, que tiene DynamoDB es el DynamoDB Streams, feature que permite disparar eventos en base a las modificaciones que se realicen sobre la base e integrarlas con el ecosistema de eventos de AWS(lambda, eventbridge, etc). Bien, DocumentDB tiene streams heredados de MongoDB pero no es directamente integrable con el ecosistema. En este post pretendo mostrar cómo configurar DocumentDB streams e integrarlo al flujo de trabajo de AWS.

¿Qué es AWS DocumentDB Stream?

AWS DocumentDB Streams es una funcionalidad que permite capturar y procesar cambios en los datos de una base de datos DocumentDB en tiempo real. Esta característica es especialmente útil para aplicaciones que requieren sincronización de datos, auditoría o integración con otros servicios.

Referencia oficial: AWS DocumentDB Change Streams

Características principales:

Captura de cambios en tiempo real: Detecta inserciones, actualizaciones y eliminaciones
Compatibilidad con MongoDB: Utiliza la API de Change Streams de MongoDB 3.6+
Integración nativa: Se puede conectar directamente con servicios AWS como Lambda, Kinesis y EventBridge
Filtrado granular: Permite filtrar eventos por colección, operación o campos específicos
Durabilidad: Los eventos se almacenan de forma persistente durante un período configurable

Documentación técnica: Working with Change Streams

POC: Arquitectura propuesta

Como muestra la figura, la arquitectura propuesta se basa en una lambda que permite escribir los documentos tomando lo que el API Gateway reciba como un POST desde el exterior, luego eso se envía al DocumentDB y del otro lado del dominio de la aplicación está el Event Bridge que está evaluando cada tiempo (1m para esta POC) si se insertó un nuevo documento.

EL codigo:

Todo el codigo utilizada para esta POC esta ene este repositorio Link

Configuración de DocumentDB Streams

A diferencia de DynamoDB Streams, en DocumentDB es necesario configurarla. Para configurarla se puede hacer a nivel global de la base de datos o a nivel de colecciones; vamos a configurar a nivel de colecciones. También se puede configurar por la línea de comandos o a través de SDK, este segundo será nuestra opción.

def lambda_handler(event, context):
    docdb_uri = os.environ['DOCDB_URI']
    client = None

    try:
        print(f"Connecting to DocumentDB...")

        # Conectar a DocumentDB con timeouts agresivos
        client = pymongo.MongoClient(
            docdb_uri,
            ssl=True,
            tlsCAFile='global-bundle.pem',
            retryWrites=False,
            connectTimeoutMS=5000,  # 5 segundos
            serverSelectionTimeoutMS=5000,  # 5 segundos
            socketTimeoutMS=10000,  # 10 segundos
            maxPoolSize=1,  # Conexión única
            waitQueueTimeoutMS=2000  # 2 segundos
        )

        # Verificar conexión rápidamente
        client.admin.command('ping')
        print("Connected successfully")

        db = client['demo_db']
        collection = db['users']

        # Habilitar change streams si no están habilitados
        try:
            client.admin.command("modifyChangeStreams", database="demo_db", collection="users", enable=True)
            print("Change streams enabled for collection")
        except Exception as e:
            print(f"Change streams already enabled or error: {e}")
        ....

Prueba de funcionamiento

Una vez configurado el DocumentDB Stream, nos queda realizar pruebas.

Primero realizamos un POST al API Gateway para agregar un nuevo documento a la DocumentDB.

~/workspace/aws-documentdb-streams/lambda/writer (main)$ curl -X POST https://umncz7md48.execute-api.us-east-1.amazonaws.com/dev/write   -H "Content-Type: application/json"   -d '{
    "user_id": "user_008",
    "name": "Elena Morales", 
    "email": "elena@example.com"
  }'
# {"message": "Document inserted successfully", "document_id": "695e6e66a68ca00398aeb3ca"}

Verificamos el log del lambda-writer:

Registro del Log completo:

~/workspace/aws-documentdb-streams/lambda/writer (main)$ aws logs tail /aws/lambda/docdb-streams-demo-writer --since 1m --region us-east-1
## 2026/01/07/[$LATEST] INIT_START Runtime Version: python:3.11.v109      Runtime Version ARN: arn:aws:lambda:us-east-1::runtime:49f733259c7ce7e0deee75ff91c6afe35c7d58b04ed300f32701216263b4590c
## 2026/01/07/[$LATEST] START RequestId: 717faf29-0efd-470e-925f-12434921dc45 Version: $LATEST
## 2026/01/07/[$LATEST] Connecting to DocumentDB...
## 2026/01/07/[$LATEST] /var/task/lambda_function.py:14: UserWarning: You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb
## 2026/01/07/[$LATEST] client = pymongo.MongoClient(
## 2026/01/07/[$LATEST] Connected successfully
## 2026/01/07/[$LATEST] Change streams enabled for collection
## 026/01/07/[$LATEST] Inserting document: {'user_id': 'user_008', 'name': 'Elena Morales', 'email': 'elena@example.com', 'timestamp': datetime.datetime(2026, 1, 7, 14, 32, 6, 64071), 'action': 'user_created'}
## 2026/01/07/[$LATEST] Document inserted with ID: 695e6e66a68ca00398aeb3ca
## 2026/01/07/[$LATEST] END RequestId: 717faf29-0efd-470e-925f-12434921dc45
## 2026/01/07/[$LATEST] REPORT RequestId: 717faf29-0efd-470e-925f-12434921dc45    Duration: 1026.52 ms    Billed Duration: 1470 msMemory Size: 128 MB     Max Memory Used: 58 MB  Init Duration: 443.42 ms

Registro específico del momento en que el documento se inserta:

Inserting document: {'user_id': 'user_008', 'name': 'Elena Morales', 'email': 'elena@example.com', 'timestamp': datetime.datetime(2026, 1, 7, 14, 32, 6, 64071), 'action': 'user_created'}

Luego verificamos el log del lambda-processor:

Registro del log completo:

~/workspace/aws-documentdb-streams/lambda/processor (main)$ aws logs tail /aws/lambda/docdb-streams-demo-stream-processor --since 1m --region us-east-1
## 2026/01/07/[$LATEST] INIT_START Runtime Version: python:3.11.v109 Runtime Version ARN: arn:aws:lambda:us-east-1::runtime:49f733259c7ce7e0deee75ff91c6afe35c7d58b04ed300f32701216263b4590c
## 2026/01/07/[$LATEST] START RequestId: ba850df0-2e96-4c01-81cd-9e2109db3587 Version: $LATEST
## 2026/01/07/[$LATEST] /var/task/lambda_function.py:17: UserWarning: You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb
## 2026/01/07/[$LATEST] client = pymongo.MongoClient(docdb_uri,
## 2026/01/07/[$LATEST] Starting change stream listener...
## 2026/01/07/[$LATEST] Change detected: insert on document 695e6e66a68ca00398aeb3ca
## 2026/01/07/[$LATEST] New document: {"_id": "695e6e66a68ca00398aeb3ca", "user_id": "user_008", "name": "Elena Morales", "email": "elena@example.com", "timestamp": "2026-01-07 14:32:06.064071", "action": "user_created"}
## 2026/01/07/[$LATEST] Change detected: insert on document 695e6e66a68ca00398aeb3ca
## 2026/01/07/[$LATEST] New document: {"_id": "695e6e66a68ca00398aeb3ca", "user_id": "user_008", "name": "Elena Morales", "email": "elena@example.com", "timestamp": "2026-01-07 14:32:06.064071", "action": "user_created"}

Registro específico del momento en que el stream detecta el nuevo documento:

Change detected: insert on document 695e6e66a68ca00398aeb3ca
New document: {"_id": "695e6e66a68ca00398aeb3ca", "user_id": "user_008", "name": "Elena Morales", "email": "elena@example.com", "timestamp": "2026-01-07 14:32:06.064071", "action": "user_created"}

Conclusiones

La implementación de AWS DocumentDB Streams es una solución viable para capturar y procesar cambios en tiempo real en aplicaciones serverless de alto volumen cuando DynamoDB no es una opción. Sin embargo, hay que tener en cuenta:

Ventajas identificadas:

✅ Integración nativa con el ecosistema AWS: Aunque requiere configuración adicional comparado con DynamoDB Streams, la integración con Lambda y EventBridge funciona de manera fluida. La desventaja en este punto es que el sistema en general no es reactivo por naturaleza; necesitamos EventBridge para poder detectar el evento. Sin embargo, una vez detectado, lo podemos integrar con el resto del ecosistema.

✅ Flexibilidad de configuración: La posibilidad de habilitar streams a nivel de colección específica permite un control granular sobre qué cambios capturar.

✅ Compatibilidad con MongoDB: Al utilizar la API estándar de Change Streams de MongoDB, facilita la migración de aplicaciones existentes.

Consideraciones importantes:

⚠️ Configuración manual requerida: A diferencia de DynamoDB, DocumentDB Streams requiere habilitación explícita por colección.

⚠️ Gestión de conexiones: Es crucial implementar timeouts y pooling de conexiones adecuados para evitar problemas de rendimiento.

⚠️ Monitoreo: Se recomienda implementar alertas para detectar fallos en el procesamiento de streams.

Casos de uso recomendados (en el papel, pueden ser muchos más):

Sincronización de datos entre microservicios
Auditoría en tiempo real de cambios en documentos críticos
Triggers de notificaciones basados en cambios de estado
Replicación selectiva de datos a sistemas analíticos

Importante antes de implementar esta solución en un ambiente productivo:

Implementar manejo de errores robusto con DLQ (Dead Letter Queues)
Configurar métricas personalizadas en CloudWatch
Evaluar el uso de Kinesis Data Streams para mayor throughput
Implementar filtros más específicos para optimizar el procesamiento

En resumen, DocumentDB Streams ofrece una alternativa sólida a DynamoDB Streams cuando se requiere flexibilidad de esquemas, volumen alto de datos y compatibilidad con MongoDB, aunque con un overhead de configuración adicional que debe considerarse en el diseño de la arquitectura.

Gracias por leer.

¡Saludos!

Oscar Cortés