¿Cómo integrar LLM con Whatsapp? (Post N°5)
OpenAI ha sido una revolución desde que fue lanzado en 2022 popularmente al mercado, entregando de forma única y concreta la posibilidad de interactuar con modelos de inteligencia artificial como nunca antes. Ya han pasado más de dos años desde ese momento, y el camino ya no es solo interactuar con los modelos, sino integrarlos en los distintos canales y plataformas de las compañías, logrando eficientar procesos u exponer nuevos canales. Debido a esto, este blog tratará de cómo poder conectar modelos de LLM con una aplicación muy conocida por todos, que es WhatsApp, la cual puede ser una excelente herramienta para gestionar clientes o prospectos.
Para integrar WhatsApp con un modelo de LLM, es necesario realizar una serie de configuraciones en el portal de Meta. En este artículo, se documentarán etapas más relacionadas a código, no obstantes, para obtener información más detallada sobre la plataforma, tales como : guías paso a paso u otros, proporcionaré algunos enlaces y videos que explican esos procesos de mejor manera.
¿Cómo configurar WhatsApp para enviar mensajes usando su API?
Paso 1: Contruir tu APP en Meta for Developers
WhatsApp ofrece distintas formas de automatizar o gestionar flujos de conversaciones, desde contratar un partner de la plataforma, API On-premise de WhatsApp o la API de WhatsApp Cloud. En este articulo se implementó la API de WhatsApp Cloud, el cual es un servicio gratuito de Meta que ofrece 1000 conversaciones gratis al mes, cobrando un cargo adicional si se supera ese límite. Para poder usar este servicio, hay que seguir los siguientes pasos:
- 1-Developer Account: crear una cuenta de Developer en Meta. Meta for Developers
- 2-App en Meta: Crear una aplicación en tu cuenta de desarrollador.
- 3-Tipo de Cuenta: Seleccionar el tipo de aplicación y proporcionar la información básica de la misma.
- 4-Acceder a la App: Acceder a la aplicación creada en Meta.
Estos pasos están ampliamente documentados en diversos artículos y blogs. Por ello, voy a compartir varios recursos que explican detalladamente el proceso necesario para realizar las configuraciones. En mi caso, tras completar estas configuraciones, mi cuenta de Meta Developer quedó de la siguiente manera:
Para mayor información dejo el enlace Meta for Developers que documenta el proceso anterior.
Paso 2: Configurar Whatsapp en la APP de Meta
Dentro del sitio de la aplicación (de la imagen anterior), en la parte izquierda, encontrarás la sección “WhatsApp”. En este apartado, podrás enviar un mensaje de prueba a cualquier celular y añadir un número telefónico para que reciba los mensajes de tus clientes, permitiéndote así construir un asistente (bot) de mensajes. Al seleccionar "WhatsApp", presiona en "Configuración de la API", y encontrarás lo siguiente:
Inicialmente, Meta entrega un numero de prueba para prototipar envios de mensaje por API a cualquier celular, siendo una forma rápida para verificar la funcionalidad de Whatsapp. No obstante, esta funcionalidad es limitada, ya que solo sirve para enviar mensaje, no recibir respuesta y tiene un token de autorización que expira en menos de 24 horas, por lo que, no se puede usar en producción. El codigo de prueba de la plataforma tambien puede ser implementado en python de la siguiente manera:
import requests
url = "https://graph.facebook.com/v19.0/
{number_identification}/messages"
headers = {
"Authorization": "Bearer {access_token}",
"Content-Type": "application/json"
}
data = {
"messaging_product": "whatsapp",
"to": "56999319606",
"type": "template",
"template": {
"name": "hello_world",
"language": {
"code": "en_US"
}
}
}
response = requests.post(url, headers=headers, json=data)
El mensaje enviado es un texto asociado a un template por defecto de meta, pero este código contiene dos elementos importantes: N° destinatario y el token de acceso. El token de acceso para mensajes de prueba dura 24 horas, por lo que, para implementar soluciones reales, es necesario crear un token permanente.
Paso 3: ¿Cómo generar un token permanente?
Para crear un token permanente, es importante que la aplicación creada en Meta for Developers esté vinculada a una cuenta comercial. En este caso, mi app ya se encuentra asociada a una cuenta comercial. Si es así, las etapas para crear un token de acceso permanente son:
- Paso 1: En la sección de Meta, selecciona tu cuenta comercial en el menú desplegable de la izquierda y haz clic en el ícono de configuraciones.
- Paso 2: Haz clic en "Configuraciones del negocio". Dirígete a "Usuarios" > "Usuarios del sistema".
- Paso 3: Haz clic en el botón "Agregar" y crea un usuario del sistema de tipo administrador o empleado.
- Paso 4: En la sección "Usuario", selecciona el usuario recién creado y haz clic en los tres puntos a la derecha. Luego, haz clic en "Añadir activos".
- Paso 5: Se habilitará una ventana para añadir activos. En la pestaña "Aplicaciones", selecciona tu aplicación y activa la casilla de “Control total de administrador de la aplicación”.
- Paso 6: Para generar un token, en la sección "Usuarios" del menú principal, selecciona la opción “Usuarios del sistema” y, en la vista de información, haz clic en “Generar identificador”.
- Paso 7: En la nueva ventana, selecciona tu aplicación y habilita los siguientes tres campos: “business_management”, “whatsapp_business_messaging” y “whatsapp_business_management”.
- Paso 8: En la nueva pestaña se visualizará tu aplicación y el token permanente. Debes guardar el token generado, ya que esta es la llave para poder ejecutar la API
Para más detalles, dejo un video video sobre este proceso respectivamente.
Paso 5: ¿Cómo enviar un mensaje de Whatsapp usando python?
Después de obtener el token permanente, ya es posible realizar solicitudes constantes mediante la API de Whatsapp. En la sección anterior, se detallo un código por defecto de Meta para enviar mensajes. No obstante, ese tipo de mensajes no son el tipo de interacciones de la vida cotidiana, debido a que se necesitan mensajes personalizados según el tipo de conversación. La función para enviar mensajes de texto personalizado es la siguiente:
def text_message(access_token,
recipient_id, code_phone_from,
text_body):
url = 'https://graph.facebook.com/v19.0/{}/messages'.
format(code_phone_from)
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
payload = {
"messaging_product": "whatsapp",
"to": recipient_id,
"type": "text",
"text": {
"body": text_body}
}
response = requests.post(url, headers=headers,
data=json.dumps(payload))
return {
'status_code': response.status_code,
'response': response
}
La función tiene cuatro parámetros principales:
- 1-Access Token: Token de acceso que puede durar 24 horas o ser permanente.
- 2-recipientid: Número de teléfono del destinatario del mensaje con el código del país.
- 3-code_phone_from: Un código generado por Meta asociado al teléfono vinculado a la cuenta de la aplicación. Este código es unico por cada numero.
- 4-Text Body: Mensaje del usuario.
Paso 6: ¿Cómo configurar el webhook para recibir los mensajes (respuestas) de WhatsApp?
Despues de lo anterior, ahora es momento de estudiar cómo recibir las respuestas de los usuarios . Los webhooks son herramientas que permiten notificar cambios en el estado de un sistema, lo cual, en este caso, podría ser nuevos mensajes de un conversación con un usuario. Sabiendo esto, Meta ha construido una integracíon con webhook, el cual puede ser configurado para recibir los mensajes de los clientes y así programar sus respuestas. Dentro de la plataforma, los wedhook se crean de la siguiente manera:
- 1- Haz click en tu App, ingresando en la sección de Whatsapp.
- 2- En la sección de Whatssap de la barra lateral de la izquierda, haz click en “Configuraciones”.
- 3- Dentro de configuraciones, debería aparecer una sección llamada wedhook, con URL de devolución de llamado, Token de verificación y campos de webhook.
- 4- Haz click en "Editar", añadiendo la URL y el token de verificación. Si el endpoint no esta activo, Meta entregará un error como respuesta.
Paso 7: ¿Cómo genero un endpoint para configurar el webhook?
Como se reviso previamente, la configuración del token de verificación necesita una URL activa para ser validada por Meta. Para esto, es necesario construir dos endpoints: uno que permita activar el webhook (GET) y otro que reciba la notificación de Facebook (POST). En este caso, el código de un endpoint (GET) que puede ser utilizado para validar el token es
import os
import json
import logging
from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import PlainTextResponse, JSONResponse
from dotenv import load_dotenv
app = FastAPI()
load_dotenv()
TOKEN_VERIFICACION = os.getenv('TOKEN_VERIFICACION')
# Set up logging
logging.basicConfig(level=logging.INFO)
logging.info(f"TOKEN_VERIFICACION: {TOKEN_VERIFICACION}")
@app.get("/webhook")
async def get_webhook(req: Request):
hub_mode = req.query_params.get('hub.mode')
hub_challenge = req.query_params.get('hub.challenge')
hub_verify_token = req.query_params.get('hub.verify_token')
logging.info(f"hub_mode: {hub_mode}, hub_challenge: {hub_challenge},
hub_verify_token: {hub_verify_token}")
if hub_mode == 'subscribe' and hub_verify_token == TOKEN_VERIFICACION:
return PlainTextResponse(hub_challenge)
else:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
detail="Verification token mismatch")
Meta solicita tanto un endpoint (URL) y un token de verificación. Este token puede ser numero cualquiera, pero debe ser el mismo tanto en la interfaz de Meta, como en el código del endpoint. Si este endpoint se desplega dentro una Docker y deploya dentro de una maquina virtual u otra alternativa, Meta podra validar el endpoint, y así, configurar correctamente el webhook en la interfase de un aplicación en la plataforma.
Paso 8: ¿Cómo integrar OpenAI (u otros modelos de LLM) en nuestras respuestas?
Al despliegas el endpoint y configurarlo en la interfaz de Meta en la sección webhook, es tiempo de integrar mensajes de Whatsapp con modelos de lenguaje. Para lograr esto, se debe construir otro endpoint que acepte el método POST, ya que este recibirá y enviará las interacciones entre el asistente y los usuarios de Whatsapp. El endpoint podría de la siguiente manera:
@app.post("/webhook")
async def post_webhook(request: Request):
from whatsapp_apis import post_webhook
try:
# Add logging to see when the request starts processing
logging.info("Received a POST request")
# Read the raw request body
raw_body = await request.body()
logging.info(f"Raw request body: {raw_body}")
# Check if the body is empty
if not raw_body:
raise ValueError("Empty request body")
# Try to parse the JSON body
body = json.loads(raw_body)
logging.info(f"Incoming webhook body: {body}")
post_webhook_instance = post_webhook.PostWebhook(body)
result = post_webhook_instance.main()
logging.info(f"Result from PostWebhook: {result}")
return JSONResponse(content=result, status_code=status.HTTP_200_OK)
except ValueError as e:
logging.error(f"ValueError: {e}")
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid body: {e}")
except Exception as e:
logging.error(f"Exception: {e}")
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Internal Server Error: {e}")
Este endpoint es capaz de recibir el mensaje de Meta (mensaje de un usuario) y responderlo según su consulta. Dentro de este endpoint, se creó una función llamada 'PostWebhook', que contiene toda la estretegia de respuesta al usuario. Esta función es la siguiente:
class IntegrationWhatssap:
def __init__(self, body):
self.body = body
self.db_container_user = os.getenv('COSMOS_DB_CONTAINER_USER')
self.db_container_chat_history = os.getenv('COSMOS_DB_CONTAINER_CHAT_HISTORY')
self.access_token = os.getenv('ACCESS_TOKEN')
self.code_phone_from = os.getenv('CODE_PHONE_FROM')
self.cosmos_db = CosmoDataBase()
def main(self):
# Paso 1 - Detectar si el mensaje es un evento de WhatsApp (texto)
detect_body_response = WhatsappAPI(self.body).detect_body()
logging.info(f"detect_body_response: {detect_body_response}")
if detect_body_response['status'] == True:
# Paso 2 - Extraer la informacion del evento.
recipient_id = detect_body_response['from']
message_id_ = detect_body_response['id']
timestamp_ = detect_body_response['timestamp']
text_ = detect_body_response['text']['body']
type_ = detect_body_response['type']
metadata_ = detect_body_response['metadata']
json_response = {'from': recipient_id, 'message_id': message_id_, 'timestamp': timestamp_,
'text': text_, 'type': type_, 'metadata': metadata_, 'typeuser': 'user'}
# Paso 3 - Detectar si existe una conversacion anterior en Cosmos DB, sino crearla.
response_detection_cosmo = self.cosmos_db.get_save_cosmos_db_history_chat(self.db_container_chat_history,
json_response)
history_ = response_detection_cosmo['history']
# Paso 4 - Genearar una respuesta del modelo LLM.
response_llm_modelo = GroqModel(history_).main()['message']
current_timestamp = datetime.utcnow().isoformat()
json_response_assistant = {'from': recipient_id, 'message_id': message_id_, 'timestamp': current_timestamp,
'text': response_llm_modelo, 'type': type_, 'metadata': {}, 'typeuser': 'assistant'}
response_db_assitante = self.cosmos_db.get_save_cosmos_db_history_chat(self.db_container_chat_history,
json_response_assistant)
if response_db_assitante['replacted'] == False:
# Paso 5 -Enviar respuesta al usuario.
response_send_message = WhatsappAPI(self.body).text_message(self.access_token, recipient_id,
self.code_phone_from, response_llm_modelo)
return json.dumps({'status_code': 200, 'response': 'Message sent'})
else:
return json.dumps({'status_code': 200, 'response': 'Message already sent'})
return json.dumps({'status_code': 200, 'response': 'Event saved'}, ensure_ascii=False)
Este proceso consta de cinco etapas clave antes de enviar el mensaje al usuario. Estas etapas son:
- Detección del contenido de la respuesta del webhook: Se identifica y analiza el contenido recibido desde el webhook.
- Almacenamiento en la base de datos: La respuesta se guarda en una base de datos Cosmos DB para su registro y posterior consulta.
- Extracción del historial de conversación: Se recupera el historial de la conversación para mantener la coherencia y contexto del intercambio.
- Construcción del mensaje para el usuario: Se elabora un mensaje personalizado y relevante para el usuario basado en la información disponible.
- Envío del mensaje por API: El mensaje final se envía al usuario a través de la API correspondiente.
A continuación, se ofrece una breve explicación de cada una de estas etapas:
Etapa 1: Detección del contenido de la respuesta del webhook
La funcionalidad del Wedhook es alertar y enviar una notificación cada vez que hay un nuevo mensaje de un usuario al numero de WhatsApp configurado en Meta for Developers. Si hay un nuevo mensaje de texto, Meta envia una notificación en formato JSON de la siguiente manera:
{
"id": "6a05c44c-f2e2-40c0-8b79-04d1c96ea3da",
"response": {
"object": "whatsapp_business_account",
"entry": [
{
"id": "380441975145057",
"changes": [
{
"value": {
"messaging_product": "whatsapp",
"metadata": {
"display_phone_number": "#####",
"phone_number_id": "336050666261796"
},
"contacts": [
{
"profile": {
"name": "UserAccount"
},
"wa_id": "NumberAccount"
}
],
"messages": [
{
"from": "56999319606",
"id": "wamid.HBgLNTY5OTkzMTk2MDYVAgASGCA4Qz
ExQzUxNjJENTcxOEFBRkU5ODgwOEJFMzZDREYzQgA=",
"timestamp": "1719282436",
"text": {
"body": "Hola"
},
"type": "text",
"metadata": {
"display_phone_number": "56958381267",
"phone_number_id": "336050666261796"
},
"status": true
}
]
},
"field": "messages"
}
]
}
]
},
"_rid": "OcxCAPDc9z0CAAAAAAAAAA==",
"_self": "dbs/OcxCAA==/colls/OcxCAPDc9z0=/docs/OcxCAPDc9z0CA
AAAAAAAAA==/",
"_etag": "\"0000a024-0000-4d00-0000-667e37420000\"",
"_attachments": "attachments/",
"_ts": 1719547714
}
Dependiendo del mensaje, el body de la notificación de meta puede tener una estructura distinta, pero si es un texto libre, se deberia recibir un formato como el anterior. Este mensaje contiene elementos importantes, tales como:
- 1-Contact Information: Información de contacto del usuario de envio del mensaje.
- 2-recipientid: Un diccionario que contiene el número de teléfono, el timestamp del mensaje, el tipo de mensaje y el contenido del mensaje.
El webhook de Meta no siempre enviará mensaje de este tipo, por lo que es necesario detectar el mensaje del usuario, antes de responderlo. Para ello, se creó una función que detecta y extrae el contenido del mensaje recibido en el Webhook:
def detect_body(body):
try:
if 'entry' in body:
for entry in body['entry']:
if 'changes' in entry:
for change in entry['changes']:
if 'value' in change and 'messages' in
change['value']:
json_ = {}
json_= change['value']['messages'][0]
json_['metadata'] = change['value']['metadata']
json_['status'] = True
json_['type'] = change['value']
['messages'][0]['type']
return json_
except KeyError:
json_ = {}
json_['status'] = False
json_['type'] = False
json_['metadata'] = False
return json_
json_ = {}
json_['status'] = False
json_['type'] = False
json_['metadata'] = False
return json_
Etapa 2: Almacenamiento en la base de datos y extracción del historial de conversación
Como en cualquier conversación entre humanos, no se puede generar una interacción correcta sin el historial de la conversación. Los modelos de lenguaje también necesitan contexto (historia) para responder correctamente a las preguntas, por lo que es necesario recopilar los mensajes en cada iteración para luego volver a utilizarlo en la respuesta. Para este propósito, se creó una base de datos en CosmosDB en Azure que almacena la información de las conversaciones, junto con extraer los mensajes para responder al usuario en cada iteración (se puede usar cualquier base de datos, no es necesariamente obligatorio usar una base en Azure). La función es la siguiente:
def get_save_cosmos_db_history_chat(db_container,
body_response, cosmo_db_key, cosmo_db_url):
client = cosmos_client.CosmosClient(cosmo_db_url,
credential=cosmo_db_key, user_agent="CosmosDBPythonQuickstart",
user_agent_overwrite=True)
db = client.get_database_client(self.cosmo_db_name)
container = db.get_container_client(db_container)
from_ = body_response['from']
timestamp_str = body_response['timestamp']
text_ = body_response['text']
typeuser_ = body_response['typeuser']
message_id_= body_response['message_id']
type_= body_response['type']
# Ensure timestamp_str is a string
if not isinstance(timestamp_str, str):
timestamp_str = str(timestamp_str)
# Convert timestamp string to datetime object
try:
timestamp_ = datetime.fromisoformat(timestamp_str)
except ValueError:
timestamp_ = datetime.utcnow()
query = f"SELECT * FROM c WHERE c['from'] = '{from_}'"
items = list(container.query_items(query=query,
enable_cross_partition_query=True))
if items:
# Conversation exists, update it
item = items[0]
for entry in item['history']:
if entry['message_id_'] == message_id_ and
entry['role'] == 'assistant':
return {
'created': False,
'updated': False,
'history': item['history'],
'replacted': True
}
history_entry = {
'role': typeuser_,
'content': text_,
'timestamp': timestamp_.isoformat(),
'message_id_' : message_id_
}
item['history'].append(history_entry)
container.replace_item(item=item, body=item)
return {
'created': False,
'updated': True,
'history': item['history'],
'replacted': False
}
else:
new_conversation = {
'id': str(uuid.uuid4()),
'session': str(uuid.uuid4()),
'from': from_,
'created_at': datetime.utcnow().isoformat(),
'history': [{
'role': typeuser_,
'content': text_,
'timestamp': timestamp_.isoformat(),
'message_id_' : message_id_,
'type': type_
}]
}
container.create_item(body=new_conversation)
return {
'created': True,
'updated': False,
'history': new_conversation['history'],
'replacted': False
}
Esta función permite tanto guardar información de la conversación, como extraer su historial. Si no hay registros previos para ese número telefónico en la base de datos, se crea un nuevo registro para ese teléfono, almacenando su historial de conversaciones. La función se utiliza para almacenar tanto la información del usuario como la del asistente, diferenciándose únicamente por su rol en la conversación.
Etapa 3: Construcción del mensaje para el usuario
Hay muchos modelos de lenguaje disponibles, tanto de OpenAI como de otros proveedores. Uno que destaca por su rapidez en interacciones on-demand es Groq. Esta compañía ha optimizado la inferencia de los modelos de lenguaje, generarando respuestas en menos de un segundo. Además, su servicio ofrece una capa gratuita muy buena para MVPs, por lo que utilizaremos sus servicios en este caso (web). Para generar la respuesta del asistente, se debe incluir el historial de la conversación en el contexto del modelo, junto con un prompt de contexto. Este proceso se puede realizar de la siguiente manera:
class GroqModel:
def __init__(self, history):
self.history = history
self.api_groq = os.getenv('API_KEY_GROQ')
self.model_groq = 'llama3-8b-8192'
self.max_history = 10
def main(self):
try:
self.history = self.history[self.max_history:]
history = []
for entry in self.history:
json_ = {
'role': entry['role'],
'content': entry['content']
}
history.append(json_)
client = Groq(
api_key= self.api_groq,
)
history_system = [{"role": "system", "content": "You are a useful assistant. You always answer in Spanish. Be kind and respectful."}]
chat_completion = client.chat.completions.create(
messages=history_system + history,
model=self.model_groq,
)
return {
'message': chat_completion.choices[0].message.content
}
except Exception as e:
return {
'message': 'Disculpa, no puedo responder a eso en este momento.'
}
Este código es muy simple, el cual se debe incluir tanto el historial como su contexto. Groq utiliza modelos de código abierto, siendo el más actual, el modelo de lenguaje de Meta, Llama-3b. Para más información, el link de Groq es: Groq.
Etapa 4: Envío del mensaje por API
Finalmente, para enviar el mensaje creado por el LLM, se debe utilizar la función text_message de la clase WhatsAppApi del código (misma función revisada previamente). Esta función utiliza la respuesta generada por el LLM para enviarla a la conversación de WhatsApp al usuario. La función es la siguiente:
def text_message(access_token, recipient_id,
code_phone_from, text_body):
url = 'https://graph.facebook.com/v19.0/{}/messages'.
format(code_phone_from)
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
payload = {
"messaging_product": "whatsapp",
"to": recipient_id,
"type": "text",
"text": {
"body": text_body
}
}
response = requests.post(url, headers=headers,
data=json.dumps(payload))
return {
'status_code': response.status_code,
'response': response
}
Esta función es la responsable de enviar los mensajes del asistente al usuario, mediante la libreria requests.
Conclusiones
Hace un tiempo, los modelos de lenguaje dejaron de ser algo demasiado nuevo, por lo que cada vez es más importante trabajar en las integraciones con distintas plataformas para resolver problemas de negocios de las empresas. WhatsApp es una herramienta muy utilizada por una gran parte de personas en el mundo, y combinarla con el poder de la IA es relevante para generar interacciones más reales y resolver problemas de negocio.
El repositorio de este articulo es: Github