Agentes de IA: Dominando 3 Patrones Esenciales (Reflexive). Parte 3 de 3

El código de estos patrones está disponible en Github. [Repo]

El Patrón Reflexivo (Reflexive) – Cuando no hay tiempo para pensar

En el Artículo 2 (ReAct), convertimos a nuestro agente en un "Investigador" meticuloso. El tipo se sentaba, analizaba, planeaba, buscaba en Google y luego respondía. Eso está genial para tareas complejas, pero tiene un bug enorme en el mundo real: es lento.

Piénsalo así: si tocas una estufa ardiendo, tu cerebro no se pone a debatir: "Mmm, detecto una temperatura de 200°C, esto causará desnaturalización de proteínas, sugiero mover el bíceps...". Jajaja, No. Tu médula espinal toma el mando y retira la mano antes de que siquiera sientas dolor.

El Patrón Reflexivo (a veces llamado Semantic Router) es exactamente eso: el "acto reflejo" de la Inteligencia Artificial.

El Caso de Uso: El Centinela de Seguridad

Para entender esto, imagina que montamos un sistema que lee los logs de tu servidor en tiempo real. Entre miles de visitas normales, de repente entra esto:
"GET /search?q=' OR 1=1;--"

Si usas un agente ReAct aquí, estás muerto. El agente pensaría:

"Vaya, un patrón inusual. Voy a usar la herramienta de búsqueda para entender qué es 'SQL Injection'. Ah, veo que es peligroso. Procedo a bloquear."

Para cuando terminó de "pensar" eso (10-15 segundos), tu base de datos ya está filtrada en la Dark Web.

Aquí es donde brilla el Agente Reflexivo:

1. Estímulo: Ve el string malicioso.
2. Reflejo: Ejecuta block_ip().
3. Tiempo: < 1 segundo. Sin dudas. Sin bucles.

Anatomía del Patrón: Cortando el cable rojo

La arquitectura aquí es radicalmente lineal.

A diferencia de ReAct, aquí eliminamos explícitamente la capacidad de razonar.

Veamos un ejemplo a continuación...

import os
import sys
import logging
import traceback
from typing import List
from dotenv import load_dotenv, find_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

# 1. Global Logging and Error Handling Configuration
LOG_DIR = os.path.join(os.path.dirname(__file__), "log")
LOG_FILE = os.path.join(LOG_DIR, "logs.txt")

if not os.path.exists(LOG_DIR):
    os.makedirs(LOG_DIR)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(LOG_FILE, encoding="utf-8"),
        logging.StreamHandler(sys.stdout)
    ]
)

logger = logging.getLogger(__name__)

def global_exception_handler(exctype, value, tb):
    """Captures unhandled exceptions and records them in the log."""
    error_msg = "".join(traceback.format_exception(exctype, value, tb))
    logger.error(f"Unhandled exception:\n{error_msg}")
    sys.__excepthook__(exctype, value, tb)

sys.excepthook = global_exception_handler

# 2. Environment Variables Loading
env_path = find_dotenv()
if env_path:
    load_dotenv(env_path)
    logger.info(f".env file loaded from: {env_path}")
else:
    logger.warning(".env file not found")

# 3. Action Tool Definitions (Simulated)
def block_ip(ip: str, reason: str) -> str:
    """
    Blocks a malicious IP address immediately.

    Args:
        ip (str): The IP address to block.
        reason (str): The reason for blocking (e.g., SQL Injection attempt).
    """
    msg = f"⛔ BLOCKING IP [{ip}]: {reason}"
    logger.warning(msg)
    return msg

def escalate_to_admin(log_id: str, severity: str) -> str:
    """
    Escalates a critical issue to a human administrator.

    Args:
        log_id (str): Unique identifier for the log or event.
        severity (str): Severity level (Critical, Urgent, etc.).
    """
    msg = f"⚠️ ADMIN ALERT: Log [{log_id}] - Severity [{severity}]"
    logger.critical(msg)
    return msg

def log_normal_traffic(source: str) -> str:
    """
    Registers legitimate and safe traffic.

    Args:
        source (str): Traffic source (IP or user).
    """
    msg = f"✅ Normal traffic from [{source}]"
    logger.info(msg)
    return msg

# 4. Agno Agent Configuration (Reflexive Pattern)
model_id = os.getenv("BASE_MODEL", "gpt-4o-mini")

agent = Agent(
    model=OpenAIChat(id=model_id),
    tools=[block_ip, escalate_to_admin, log_normal_traffic],
    instructions=[
        "You are an automated security system (Reflexive Pattern).",
        "Your job is to classify incoming logs and execute the corresponding tool IMMEDIATELY.",
        "Action rules:",
        "- If you detect SQL Injection, XSS, or known attacks -> Use 'block_ip'.",
        "- If you detect server errors (500), service downs, or timeouts -> Use 'escalate_to_admin'.",
        "- If traffic seems legitimate, successful login, or normal navigation -> Use 'log_normal_traffic'.",
        "DO NOT explain your decision. DO NOT greet. Just execute the tool directly in response to the stimulus.",
        "Process the log and terminate execution."
    ]
)

# 5. Flow Simulation (The Stream)
def main():
    logger.info("Starting Security Sentinel Agent (Reflexive)...")
    print("--- Security Sentinel - Reflexive Pattern ---")

    # Extended list of simulated logs
    logs = [
        "User 'admin' logged in successfully from 192.168.1.10",
        "GET /search?q=' OR 1=1;-- (SQL Injection attempt from 203.0.113.5)",
        "Service 'PaymentGateway' timed out after 3000ms (Error 500)",
        "User 'jdoe' updated profile picture",
        "<script>alert('hacked')</script> sent via form from 88.22.11.44",
        "DB_ERROR: Connection pool exhausted (LogID: 4452)",
        "Bot crawler identified: Googlebot/2.1 from 66.249.66.1",
        "Multiple failed login attempts for user 'sysadmin' from 45.33.22.11",
        "HealthCheck: All systems operational - latency 12ms",
        "Unauthorized access attempt to /etc/passwd from 172.16.0.50"
    ]

    print(f"\nProcessing {len(logs)} logs in the 'stream'...\n")

    for i, log_entry in enumerate(logs, 1):
        try:
            print(f"[{i}] Stimulus: {log_entry}")
            logger.info(f"Processing log {i}: {log_entry}")

            # In reflexive mode, we expect the agent to execute the tool immediately.
            # We use show_tool_calls=True to demonstrate the action.
            agent.print_response(log_entry, show_tool_calls=True)
            print("-" * 50)

        except Exception as e:
            logger.error(f"Error processing log {i}: {str(e)}")
            print(f"An error occurred in event {i}: {e}")

if __name__ == "__main__":
    main()

• Instrucciones Restrictivas (Negative Constraints): En el prompt del sistema (instructions), se prohíbe explícitamente al agente pensar o charlar ("DO NOT explain", "Just execute"). Se le programa solo para disparar herramientas.

• Mapeo Directo: No hay bucles de feedback. El flujo es lineal: Log entrante → Clasificación → Ejecución de Tool.

• Selección del Modelo: Se configura gpt-4o-mini. Al usar un modelo más pequeño y rápido, priorizamos la latencia y el volumen de procesamiento.

¿Por qué es tan Cool este patrón? (Pros)

• Velocidad Absurda (Latencia Mínima): Al no generar texto de "pensamiento", la respuesta es instantánea.
• Traga lo que le eches (High Throughput): Es perfecto para procesar miles de correos, chats en vivo o alertas de monitoreo por minuto.
• Es Predecible (Determinismo): Al quitarle la "creatividad", el agente se comporta casi como un script clásico. Ante X, siempre hace Y.
• Barato: Menos tokens de salida y modelos más pequeños = factura de API ridícula.

¿Dónde falla? (Contras)

• Memoria de Pez (Ceguera Contextual): El agente reacciona a lo que tiene delante ahora. Si un ataque está dividido en tres pasos a lo largo de 10 minutos, este agente no unirá los puntos.
• Es Cabezota (Rigidez): Si le llega algo ambiguo que no está en sus reglas, suele fallar o alucinar una categoría errónea porque le forzamos a elegir rápido.
• Caja Negra: A diferencia de ReAct, aquí no hay logs de "Thought" que te digan por qué bloqueó a un usuario. Simplemente lo hizo y ya.

Comparativa para Developers

Característica	ReAct (Nivel 2)	Reflexive (Nivel 3)
Analogía	Sherlock Holmes	Un Portero de Discoteca
Prioridad	Calidad y Razonamiento	Velocidad y Volumen
Bucle	Sí (Pensar-Actuar-Observar)	No (Disparar y Olvidar)
Modelo Ideal	Modelos Top (GPT-4o, Claude)	Modelos Ligeros (Mini, Haiku)

Consejos de Trinchera

Si vas a meter esto en producción, ten cuidado con estas tres cosas:

1. Cuidado con el "Ban Hammer": Como este agente dispara primero y pregunta después, vas a tener falsos positivos. Mi consejo: que la acción sea flag_for_review() (marcar para revisión humana) en vez de delete_account(), a menos que estés 100% seguro.

2. Sé borde con el Prompt (Negative Constraints): Tienes que usar restricciones negativas.
   Mal prompt: "Clasifica este texto."
   Buen prompt: "Clasifica este texto. NO expliques tu razón. NO generes texto introductorio. Solo devuelve el JSON."

3. La Arquitectura Híbrida: El patrón Reflexivo suele ser la "Infantería". Pones a este agente en primera línea para filtrar el 80% del ruido fácil. Si detecta algo que no entiende o es ambiguo, entonces (y solo entonces) escala el ticket a un agente ReAct (más listo y caro) para que investigue. Así tienes lo mejor de los dos mundos.

Happy Coding! 🤖