Registro de actividad para agentes de IA

IRC-Mania Die Seite rund um Sicherheit und Programmierung

Internet, Sicherheit , Internet Relay Chat

Gesponsert

Qué cubre este artículo sobre registro de agentes de IA

• Cómo configurar el registro de actividad para agentes de IA.
• Qué eventos deberías registrar: llamadas a herramientas, respuestas del modelo, errores, decisiones.
• Cómo implementar logging estructurado, audit trails y cumplimiento normativo.
• Ejemplos prácticos de registro con Python, logs en JSON y análisis de logs.
• Buenas prácticas para seguridad, rendimiento y retención de datos.

Introducción: Registro de actividad para agentes de IA explicado

Los agentes de IA toman decisiones, invocan herramientas, procesan datos. Cuando algo falla, necesitas poder reconstruir qué ocurrió exactamente. El registro de actividad es la captura de todas las acciones del agente para que puedas analizar errores, demostrar cumplimiento normativo y mejorar la calidad continuamente.

Este artículo está dirigido a desarrolladores que crean agentes de IA y quieren implementar registro de actividad. Deberías entender qué son los agentes de IA y cómo ejecutarlos localmente con Ollama. Los fundamentos de programación en Python los encontrarás en IRC-Coding.de.

¿Por qué necesitas registro de actividad para agentes de IA?

Imagina que tu agente eliminó un correo que no debería haber eliminado. ¿Por qué lo hizo? ¿Qué herramientas invocó? ¿Qué decisión tomó el modelo y por qué? Sin registro de actividad, es imposible saberlo. Con registro, ves cada paso, cada decisión, cada invocación.

El registro de actividad es especialmente crítico cuando los agentes ejecutan acciones sensibles, cuando necesitas demostrar cumplimiento normativo (GDPR, ISO 27001) o cuando quieres mejorar sistemáticamente el comportamiento del agente.

Registro de actividad para agentes de IA en resumen

El registro de actividad es la captura de todas las acciones del agente en logs estructurados. Cada invocación de herramienta, cada respuesta del modelo, cada decisión, cada error se registra con marca de tiempo, contexto y resultado. Los logs se pueden analizar después para identificar errores, demostrar cumplimiento y mejorar calidad.

La idea central es simple: sin logs no hay trazabilidad, sin trazabilidad no hay confianza.

A quién está dirigido este artículo

• Desarrolladores que crean agentes de IA e implementan registro de actividad.
• Administradores de sistemas que mantienen agentes y analizan errores.
• Responsables de cumplimiento que necesitan audit trails.
• Equipos que quieren mejorar sistemáticamente la calidad del agente.

Se requieren conocimientos previos en Python, agentes de IA y Ollama.

Términos clave sobre registro de actividad

• Registro de actividad - Captura de eventos. Útil para: trazabilidad.
• Audit Log - Registro de todas las acciones relevantes para seguridad. Útil para: cumplimiento normativo.
• Logging estructurado - Logs en formato JSON. Útil para: procesamiento automatizado.
• Log Level - Prioridad del log (DEBUG, INFO, WARNING, ERROR). Útil para: filtrado.
• Invocación de herramienta - El agente llama a una herramienta. Útil para: crítico en auditoría.
• Agentes de IA - Programas que resuelven tareas de forma autónoma. Útil para: qué se registra.
• Ollama - Servidor local de modelos. Útil para: backend cuyas llamadas se registran.
• Function Calling - Respuestas estructuradas de IA. Útil para: qué se registra.
• Cumplimiento normativo - Adherencia a normas y regulaciones. Útil para: razón de los audit logs.

Qué deberías registrar

Llamadas al modelo

Cada invocación del modelo de lenguaje debe registrarse:

import logging
import json
from datetime import datetime

logger = logging.getLogger("agent")
logger.setLevel(logging.INFO)

def log_model_call(model, messages, response, duration_ms):
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "type": "model_call",
        "model": model,
        "input_messages": len(messages),
        "input_tokens": count_tokens(messages),
        "output_tokens": count_tokens(response),
        "duration_ms": duration_ms,
        "status": "success"
    }
    logger.info(json.dumps(log_entry))

Invocaciones de herramientas

Cada invocación de herramienta es crítica para auditoría:

def log_tool_call(tool_name, parameters, result, duration_ms, status):
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "type": "tool_call",
        "tool": tool_name,
        "parameters": parameters,
        "result_summary": str(result)[:200],
        "duration_ms": duration_ms,
        "status": status
    }
    logger.info(json.dumps(log_entry))

Decisiones

Cada decisión tomada por el agente:

def log_decision(decision_type, reasoning, action):
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "type": "decision",
        "decision_type": decision_type,
        "reasoning": reasoning,
        "action": action
    }
    logger.info(json.dumps(log_entry))

Errores

Cada error con su contexto:

def log_error(error_type, message, context):
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "type": "error",
        "error_type": error_type,
        "message": message,
        "context": context
    }
    logger.error(json.dumps(log_entry))

Logging estructurado con Python

Configuración

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage()
        }
        if hasattr(record, "extra"):
            log_entry.update(record.extra)
        return json.dumps(log_entry)

# Configurar logger
logger = logging.getLogger("agent")
logger.setLevel(logging.DEBUG)

# File Handler
file_handler = logging.FileHandler("/var/log/agent.log")
file_handler.setFormatter(JSONFormatter())
logger.addHandler(file_handler)

# Console Handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(JSONFormatter())
logger.addHandler(console_handler)

Uso

# Registrar llamada al modelo
logger.info("Modelo invocado", extra={
    "type": "model_call",
    "model": "llama3.1",
    "duration_ms": 1234
})

# Registrar invocación de herramienta
logger.info("Herramienta invocada", extra={
    "type": "tool_call",
    "tool": "send_email",
    "parameters": {"to": "user@example.com"}
})

# Registrar error
logger.error("Herramienta falló", extra={
    "type": "error",
    "tool": "send_email",
    "error": "SMTP timeout"
})

Implementación de Audit Trail

Un audit trail es una forma especial de registro que captura todas las acciones relevantes para seguridad.

class AuditLogger:
    def __init__(self, log_file="/var/log/agent_audit.log"):
        self.logger = logging.getLogger("audit")
        self.logger.setLevel(logging.INFO)
        handler = logging.FileHandler(log_file)
        handler.setFormatter(JSONFormatter())
        self.logger.addHandler(handler)

    def log_action(self, agent_id, action, target, parameters, result, user_id=None):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": agent_id,
            "user_id": user_id,
            "action": action,
            "target": target,
            "parameters": parameters,
            "result": result,
            "status": "success" if result.get("success") else "failed"
        }
        self.logger.info(json.dumps(entry))

# Uso
audit = AuditLogger()
audit.log_action(
    agent_id="email_agent_1",
    action="send_email",
    target="user@example.com",
    parameters={"subject": "Re: Support-Anfrage"},
    result={"success": True, "message_id": "abc123"},
    user_id="user_42"
)

Ejemplo práctico: Agente completo con registro

class LoggedAgent:
    def __init__(self, model="llama3.1"):
        self.model = model
        self.logger = logging.getLogger("agent")
        self.audit = AuditLogger()

    def call_model(self, messages):
        start = time.time()
        try:
            response = requests.post(
                "http://localhost:11434/api/chat",
                json={"model": self.model, "messages": messages, "stream": False}
            ).json()
            duration = (time.time() - start) * 1000

            self.logger.info("Modell aufgerufen", extra={
                "type": "model_call",
                "model": self.model,
                "duration_ms": duration,
                "input_messages": len(messages),
                "status": "success"
            })
            return response
        except Exception as e:
            duration = (time.time() - start) * 1000
            self.logger.error("Modell-Aufruf fehlgeschlagen", extra={
                "type": "error",
                "model": self.model,
                "duration_ms": duration,
                "error": str(e)
            })
            raise

    def call_tool(self, tool_name, parameters):
        start = time.time()
        try:
            result = execute_tool(tool_name, parameters)
            duration = (time.time() - start) * 1000

            self.logger.info("Tool aufgerufen", extra={
                "type": "tool_call",
                "tool": tool_name,
                "duration_ms": duration,
                "status": "success"
            })

            self.audit.log_action(
                agent_id=self.agent_id,
                action=tool_name,
                target=parameters.get("target", "unknown"),
                parameters=parameters,
                result={"success": True, "data": str(result)[:200]}
            )
            return result
        except Exception as e:
            duration = (time.time() - start) * 1000
            self.logger.error("Tool fehlgeschlagen", extra={
                "type": "error",
                "tool": tool_name,
                "duration_ms": duration,
                "error": str(e)
            })
            self.audit.log_action(
                agent_id=self.agent_id,
                action=tool_name,
                target=parameters.get("target", "unknown"),
                parameters=parameters,
                result={"success": False, "error": str(e)}
            )
            raise

Análisis de registros

Búsqueda de logs

# Todos los errores
jq 'select(.level == "ERROR")' /var/log/agent.log

# Todos los llamados a herramientas
jq 'select(.type == "tool_call")' /var/log/agent.log

# Llamados a herramientas fallidos
jq 'select(.type == "tool_call" and .status == "failed")' /var/log/agent.log

# Llamados al modelo lentos (>5 segundos)
jq 'select(.type == "model_call" and .duration_ms > 5000)' /var/log/agent.log

Análisis con Python

import json

def analyze_logs(log_file):
    with open(log_file) as f:
        logs = [json.loads(line) for line in f]

    # Estadísticas
    total_calls = len([l for l in logs if l.get("type") == "model_call"])
    failed_tools = len([l for l in logs if l.get("type") == "tool_call" and l.get("status") == "failed"])
    errors = len([l for l in logs if l.get("level") == "ERROR"])

    # Duración promedio del modelo
    model_durations = [l["duration_ms"] for l in logs if l.get("type") == "model_call"]
    avg_duration = sum(model_durations) / len(model_durations) if model_durations else 0

    return {
        "total_model_calls": total_calls,
        "failed_tool_calls": failed_tools,
        "total_errors": errors,
        "avg_model_duration_ms": avg_duration
    }

Rotación de registros

Los registros pueden volverse muy grandes. Utiliza la rotación de logs:

import logging.handlers

handler = logging.handlers.RotatingFileHandler(
    "/var/log/agent.log",
    maxBytes=100*1024*1024,  # 100 MB
    backupCount=5
)
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

Consideraciones de seguridad

• Sin datos sensibles en logs: No registres contraseñas, claves API ni datos sensibles de usuarios.
• Protección de logs: Protege los registros contra manipulación (solo agregar, no sobrescribir).
• Retención: Define períodos de retención explícitos (por ejemplo, 90 días para Audit Logs).
• Control de acceso: Solo personas autorizadas pueden leer los registros.
• Anonimización: Anonimiza datos personales en los registros.
• Referencias: Audit Logging, Protección de datos.

Trampas comunes

• Demasiados logs: Los logs DEBUG en producción generan archivos enormes. Usa niveles de log.
• Logs sin estructura: Los logs desestructurados son difíciles de analizar. Usa JSON.
• Datos sensibles en logs: Las contraseñas o claves API en los registros son un riesgo de seguridad.
• Sin rotación de logs: Los registros crecen sin límite. Usa RotatingFileHandler.
• Sin logs de error: Si solo se registran éxitos, no puedes analizar fallos.
• Sin marcas de tiempo: Sin timestamps no puedes correlacionar eventos.

Enlaces y recursos sobre registro

• Fundamentos de agentes IA - Qué son los agentes IA.
• Agentes IA con Ollama - Ejecuta agentes localmente.
• Audit Logging - Seguridad y trazabilidad.
• Protección de datos - RGPD e IA.
• Aprobación manual - Human-in-the-Loop.
• Configurar guardrails - Guardrails para agentes.
• Automatización de correo - Ejemplo práctico con registro.

Puntos clave:

• El registro es la documentación de todas las acciones del agente.
• Registra: llamadas al modelo, llamadas a herramientas, decisiones, errores.
• Usa logs estructurados (JSON) para análisis automatizados.
• Audit-trails para cumplimiento y seguridad.
• Sin datos sensibles en logs, utiliza rotación de registros.

FAQ: Registro para agentes IA - Preguntas frecuentes

Fuentes y lecturas recomendadas

• Python Logging - Documentación oficial de logging en Python.
• jq - Procesador JSON para logs.
• ELK Stack - Plataforma de análisis de logs.
• Grafana Loki - Agregación de logs.

Premium VPS Hosting

Schnelle und sichere VPS Server

Gesponsert