Cómo guardar logs de mensajes de watsonx Assistant utilizando Webhooks

En los proyectos donde utilizamos watsonx Assistant como asistente conversacional, es crucial contar con un mecanismo para registrar los mensajes intercambiados entre los usuarios y el asistente. Esto permite analizar conversaciones, mejorar respuestas y auditar interacciones. Una forma eficiente de lograrlo es mediante el uso de webhooks, los cuales pueden extender la funcionalidad del asistente y conectarlo con sistemas externos.

En este artículo, explicaré cómo crear un microservicio para guardar logs de los mensajes enviados al asistente utilizando el tipo de webhook Log. Este microservicio almacenará los datos en una base de datos MongoDB y estará listo para integrarse con watsonx Assistant.

¿Qué son los webhooks en watsonx Assistant?

Un webhook es un mecanismo que permite realizar llamadas desde el asistente hacia un servicio externo durante una conversación. El webhook de tipo Log es particularmente útil para registrar cada mensaje y respuesta generados en el asistente, sin necesidad de condiciones adicionales.

El flujo típico de un webhook Log es el siguiente:

1. El usuario envía un mensaje al asistente.
2. El asistente procesa el mensaje y genera una respuesta.
3. El webhook se activa, enviando el mensaje y su respuesta al microservicio configurado.
4. El microservicio almacena la información, permitiendo su análisis posterior.

Creando el microservicio para guardar logs

El siguiente código muestra un ejemplo de microservicio desarrollado con FastAPI. Este microservicio se encarga de recibir los mensajes del webhook Log, validarlos y almacenarlos en una base de datos MongoDB.

Para implementar el microservicio que registra los logs de mensajes del asistente, debemos seguir los siguientes pasos:

Paso 1: Crear el ambiente virtual

Ejecuta los siguientes comandos en tu terminal:

1
python -m venv venv
2
source venv/bin/activate

Paso 2: Crear el archivo requirements.txt

Este archivo contendrá las dependencias del proyecto. Crea un archivo llamado requirements.txt con el siguiente contenido:

1
fastapi[standard]
2
pymongo
3
pydantic-settings

Paso 3: Instalar las dependencias

Ejecuta el siguiente comando para instalar las librerías necesarias:

1
pip install -r requirements.txt

Con esto, el ambiente virtual estará configurado y las dependencias listas para usar.

Paso 4: Crear el archivo de configuración

Es necesario configurar las credenciales y parámetros de conexión a MongoDB. Para esto, crearemos un archivo llamado config.py con el siguiente contenido:

1
from pydantic_settings import BaseSettings
2

3
class Settings(BaseSettings):
4
    mongo_username: str
5
    mongo_password: str
6
    mongo_host: str
7
    mongo_db: str
8
    mongo_auth_source: str
9
    mongo_replica_set: str
10
    mongo_tls: bool
11
    mongo_tls_ca_file: str
12

13
    class Config:
14
        env_file = ".env"
15

16
settings = Settings()

Además, debemos crear un archivo .env en el mismo directorio con las variables de entorno necesarias, como este ejemplo:

1
mongo_username=tu_usuario
2
mongo_password=tu_contraseña
3
mongo_host=tu_host
4
mongo_db=tu_base_de_datos
5
mongo_auth_source=admin
6
mongo_replica_set=tu_replica
7
mongo_tls=True
8
mongo_tls_ca_file=/path/to/ca.pem

Esto permitirá cargar la configuración de manera centralizada y segura.

Paso 5: Crear los modelos de datos

Ahora definiremos los modelos necesarios para validar y estructurar los datos recibidos del webhook. Crea un archivo llamado models.py con el siguiente contenido:

1
from pydantic import BaseModel, Field
2
from typing import Dict, Any, Optional
3

4
class Event(BaseModel):
5
    name: str
6

7
class PayloadResponse(BaseModel):
8
    output: Dict[str, Any]
9
    internal: Dict[str, Any]
10
    user_id: str
11
    context: Dict[str, Any]
12

13
class Payload(BaseModel):
14
    log_id: str
15
    request: Dict[str, Any]
16
    response: PayloadResponse
17
    assistant_id: str
18
    request_timestamp: str
19
    response_timestamp: str
20
    language: str
21
    skill_id: str
22
    user_id: str
23
    environment: str
24
    channel: str
25
    customer_id: str
26

27
class LogMessage(BaseModel):
28
    event: Optional[Event]
29
    payload: Optional[Payload]

Estos modelos permiten estructurar los datos enviados por watsonx Assistant, garantizando que sean válidos y estén completos antes de ser procesados.

Paso 6: Implementar el microservicio

Con la configuración y los modelos listos, podemos implementar el microservicio. Este se encargará de recibir los logs, validarlos y almacenarlos en MongoDB. Aquí está el código completo del microservicio en main.py:

1
from fastapi import FastAPI, HTTPException, status
2
from pymongo import MongoClient
3
from pymongo.errors import PyMongoError
4
from .models import LogMessage
5
from .config import settings
6
from datetime import datetime
7

8
app = FastAPI()
9

10
connection_string = f"mongodb://{settings.mongo_username}:{settings.mongo_password}@{settings.mongo_host}/{settings.mongo_db}?authSource={settings.mongo_auth_source}&replicaSet={settings.mongo_replica_set}"
11

12
tls_options = {
13
    'tls': settings.mongo_tls,
14
    'tlsCAFile': settings.mongo_tls_ca_file
15
}
16

17
client = MongoClient(connection_string, **tls_options)
18
client.admin.command('ping')
19

20
db = client["logs"]
21

22
@app.post("/logs/message")
23
async def create_log(log: LogMessage):
24
    collection = db["messages"]
25

26
    if not log.event.name == "message_logged":
27
        raise HTTPException(
28
            status_code=status.HTTP_400_BAD_REQUEST,
29
            detail="El evento enviado no es del tipo message_logged."
30
        )
31

32
    log_dict = log.payload.model_dump()
33
    log_dict["received_at"] = datetime.utcnow()
34

35
    collection.insert_one(log_dict)
36

37
    return { "status": "success" }
38
    except PyMongoError as e:
39
        raise HTTPException(
40
            status_code=500,
41
            detail=f"Error al insertar el log: {str(e)}"
42
        )
43
    except Exception as e:
44
        raise HTTPException(
45
            status_code=500,
46
            detail=f"Error inesperado al procesar el log: {str(e)}"
47
        )

Integrando el microservicio con Watsonx Assistant

Una vez que el microservicio está configurado y ejecutándose, el siguiente paso es integrarlo con watsonx Assistant mediante el uso de webhooks. Aquí te explico cómo hacerlo paso a paso:

Conclusión

Utilizar webhooks de tipo Log es una excelente estrategia para centralizar y analizar las interacciones entre usuarios y el asistente. Con este enfoque, podrás almacenar de manera segura y eficiente todos los mensajes en una base de datos, habilitando capacidades avanzadas como análisis de datos, auditorías o visualización de métricas.