¿Cuántas horas a la semana invierte el área de soporte o consultoría en responder las mismas preguntas sobre rutas de menús, códigos de error, tablas de base de datos o procedimientos de SAP Business One?

Para solucionar esto sin comprometer la privacidad corporativa ni pagar costosas suscripciones de APIs en la nube, implementamos un asistente de soporte 100% local. El bot se conecta directamente como un usuario al servidor corporativo de mensajería instantánea Openfire / Spark, lee manuales en PDF y Word (.docx) mediante RAG (Retrieval-Augmented Generation), extrae texto de diagramas o capturas de pantalla con OCR, muestra el indicador animado de «Escribiendo…» y entrega respuestas precisas con enlaces directos a las capturas del sistema.

A continuación, te mostramos la arquitectura y el paso a paso para construirlo en Ubuntu Server.

1. Arquitectura de la Solución

El sistema opera completamente dentro de la red corporativa bajo el siguiente flujo:

  1. Indexación previa (Off-line): Un script en Python escanea la carpeta de manuales (.pdf y .docx), extrae texto estructurado, tablas y aplica OCR con Tesseract a capturas de pantalla. El texto se vectoriza con nomic-embed-text y se guarda en ChromaDB.
  2. Servidor HTTP de imágenes: Las capturas de pantalla extraídas quedan disponibles en un microservidor web local para que los usuarios puedan abrirlas en alta resolución con un solo clic.
  3. Bot XMPP (slixmpp): Se autentica en Openfire como un usuario de soporte (ej. asistente@xmpp.empresa.local).
  4. Respuesta contextualizada (LLM): Al recibir un mensaje, avisa a Spark que está escribiendo (XEP-0085), recupera los fragmentos del manual más relevantes y consulta a Qwen 2.5:7b acelerado por GPU local.
  5. Auditoría (SQLite): Cada consulta, tipo de respuesta (manual interno vs. estándar de SAP) y usuario queda registrada para auditoría y mejora continua de la documentación.

2. Preparación del Servidor y Dependencias

En tu máquina con Ubuntu Server, instala las herramientas del sistema, el motor de OCR y las librerías de Python necesarias:

Bash

# 1. Herramientas del sistema y OCR en español
sudo apt update && sudo apt install -y tesseract-ocr tesseract-ocr-spa libgl1 python3-pip sqlite3

# 2. Librerías de Python requeridas
pip install --break-system-packages slixmpp chromadb pymupdf python-docx pytesseract pillow ollama

# 3. Modelos locales con Ollama
ollama pull nomic-embed-text
ollama pull qwen2.5:7b

# 4. Estructura de directorios
mkdir -p ~/sap_bot/manuales
mkdir -p ~/sap_bot/static_imgs
cd ~/sap_bot

Nota: Coloca tus guías operativas, diccionarios de datos y manuales en formato .pdf o .docx dentro de la carpeta ~/sap_bot/manuales/.

3. Servidor Web Local para Capturas de Pantalla

Para que el bot pueda adjuntar enlaces a las capturas de los manuales y los usuarios de Spark puedan verlas en el navegador sin fricción, creamos un servicio ligero con Python:

Bash

sudo tee /etc/systemd/system/sap-imgs.service << EOF
[Unit]
Description=Servidor HTTP para Capturas de Manuales SAP
After=network.target

[Service]
Type=simple
User=$USER
WorkingDirectory=$HOME/sap_bot/static_imgs
ExecStart=/usr/bin/python3 -m http.server 8080 --bind 0.0.0.0
Restart=always

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload
sudo systemctl enable --now sap-imgs.service

4. Script de Indexación Inteligente con OCR (indexar_manuales.py)

Este script se encarga de analizar los manuales, separar el texto en fragmentos con solapamiento (chunking), extraer tablas, guardar capturas de pantalla y procesar imágenes que contengan texto con Tesseract.

Crea el archivo ~/sap_bot/indexar_manuales.py:

Python

#!/usr/bin/env python3
"""
Indexador de Documentación SAP B1 (PDF y DOCX) con OCR y ChromaDB.
"""
import os
import io
import re
import fitz  # PyMuPDF
import docx  # python-docx
import pytesseract
from PIL import Image
import chromadb
import ollama

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CARPETA_MANUALES = os.path.join(BASE_DIR, "manuales")
CARPETA_IMGS = os.path.join(BASE_DIR, "static_imgs")
DB_PATH = os.path.join(BASE_DIR, "chroma_sap_db")

CHUNK_SIZE = 900
CHUNK_OVERLAP = 150
MODELO_EMBED = "nomic-embed-text"

def limpiar_nombre(nombre: str) -> str:
    return re.sub(r'[^a-zA-Z0-9_\-\.]', '_', nombre)

def fragmentar_texto(texto: str, tamano: int = CHUNK_SIZE, solapamiento: int = CHUNK_OVERLAP) -> list[str]:
    fragmentos = []
    inicio = 0
    while inicio < len(texto):
        fin = inicio + tamano
        fragmentos.append(texto[inicio:fin])
        inicio += tamano - solapamiento
    return fragmentos

def obtener_vector(texto: str) -> list[float]:
    res = ollama.embeddings(model=MODELO_EMBED, prompt=texto)
    return res["embedding"]

def procesar_pdf(ruta_pdf: str, archivo: str, coleccion, contador_inicio: int) -> int:
    doc = fitz.open(ruta_pdf)
    contador = contador_inicio
    nombre_base = limpiar_nombre(os.path.splitext(archivo)[0])

    for num_pagina in range(len(doc)):
        pagina = doc[num_pagina]
        texto = pagina.get_text().strip()
        imagenes = pagina.get_images(full=True)
        imagenes_guardadas = []

        if imagenes:
            for idx_img, img_info in enumerate(imagenes):
                try:
                    xref = img_info[0]
                    img_dict = doc.extract_image(xref)
                    img_pil = Image.open(io.BytesIO(img_dict["image"]))
                    
                    # Filtrar viñetas o elementos menores a 120px
                    if img_pil.width > 120 and img_pil.height > 120:
                        nombre_img = f"{nombre_base}_p{num_pagina + 1}_img{idx_img}.png"
                        img_pil.save(os.path.join(CARPETA_IMGS, nombre_img))
                        imagenes_guardadas.append(nombre_img)

                        if len(texto) < 120:
                            ocr = pytesseract.image_to_string(img_pil, lang="spa")
                            if ocr.strip():
                                texto += f"\n[Texto en captura {nombre_img}]:\n" + ocr.strip()
                except Exception:
                    continue

        if len(texto.strip()) < 30 and not imagenes_guardadas:
            continue

        bloques = fragmentar_texto(texto)
        str_imgs = ",".join(imagenes_guardadas)

        for idx, bloque in enumerate(bloques):
            try:
                vector = obtener_vector(bloque)
                chunk_id = f"{archivo}_p{num_pagina + 1}_b{idx}_{contador}"
                coleccion.add(
                    ids=[chunk_id],
                    embeddings=[vector],
                    documents=[bloque],
                    metadatas=[{
                        "archivo": archivo,
                        "pagina": str(num_pagina + 1),
                        "imagenes": str_imgs
                    }]
                )
                contador += 1
            except Exception as err:
                print(f"[-] Error en vector: {err}")

    return contador

def procesar_docx(ruta_docx: str, archivo: str, coleccion, contador_inicio: int) -> int:
    doc = docx.Document(ruta_docx)
    contador = contador_inicio
    nombre_base = limpiar_nombre(os.path.splitext(archivo)[0])
    contenido = []
    imagenes_guardadas = []

    for p in doc.paragraphs:
        if p.text.strip():
            contenido.append(p.text.strip())

    for t_idx, tabla in enumerate(doc.tables):
        filas = [" | ".join(c.text.strip() for c in f.cells if c.text.strip()) for f in tabla.rows]
        if filas:
            contenido.append(f"\n[Tabla {t_idx + 1}]:\n" + "\n".join(filas))

    idx_img = 0
    for part in doc.part.related_parts.values():
        if getattr(part, "content_type", "").startswith("image/"):
            try:
                img_pil = Image.open(io.BytesIO(part.blob))
                if img_pil.width > 120 and img_pil.height > 120:
                    nombre_img = f"{nombre_base}_word_img{idx_img}.png"
                    img_pil.save(os.path.join(CARPETA_IMGS, nombre_img))
                    imagenes_guardadas.append(nombre_img)
                    idx_img += 1

                    ocr = pytesseract.image_to_string(img_pil, lang="spa")
                    if ocr.strip():
                        contenido.append(f"\n[Texto en captura {nombre_img}]:\n" + ocr.strip())
            except Exception:
                continue

    texto_total = "\n\n".join(contenido).strip()
    if len(texto_total) < 30 and not imagenes_guardadas:
        return contador

    bloques = fragmentar_texto(texto_total)
    str_imgs = ",".join(imagenes_guardadas)

    for idx, bloque in enumerate(bloques):
        try:
            vector = obtener_vector(bloque)
            chunk_id = f"{archivo}_sec_{idx}_{contador}"
            coleccion.add(
                ids=[chunk_id],
                embeddings=[vector],
                documents=[bloque],
                metadatas=[{
                    "archivo": archivo,
                    "pagina": f"Sección {idx + 1}",
                    "imagenes": str_imgs
                }]
            )
            contador += 1
        except Exception as err:
            print(f"[-] Error en vector Word: {err}")

    return contador

def ejecutar_indexacion():
    os.makedirs(CARPETA_MANUALES, exist_ok=True)
    os.makedirs(CARPETA_IMGS, exist_ok=True)
    client = chromadb.PersistentClient(path=DB_PATH)

    formatos = (".pdf", ".docx")
    archivos = [f for f in os.listdir(CARPETA_MANUALES) if f.lower().endswith(formatos)]

    if not archivos:
        print("[!] No hay archivos para indexar. Creando base de datos vacía...")
        client.get_or_create_collection(name="manuales_sap")
        return

    try:
        client.delete_collection(name="manuales_sap")
    except Exception:
        pass
    coleccion = client.create_collection(name="manuales_sap")

    contador = 0
    for archivo in archivos:
        ruta = os.path.join(CARPETA_MANUALES, archivo)
        print(f"[+] Indexando: {archivo}")
        if archivo.lower().endswith(".pdf"):
            contador = procesar_pdf(ruta, archivo, coleccion, contador)
        elif archivo.lower().endswith(".docx"):
            contador = procesar_docx(ruta, archivo, coleccion, contador)

    print(f"\n[✓] Indexación finalizada. Total de fragmentos: {contador}")

if __name__ == "__main__":
    ejecutar_indexacion()

5. El Bot XMPP para Spark (bot_spark.py)

Este script se encarga de:

  1. Conectarse de forma asíncrona al servidor Openfire.
  2. Emitir el paquete XEP-0085 para mostrar «Escribiendo…» de inmediato en Spark.
  3. Consultar la base vectorial sin bloquear el hilo principal.
  4. Generar respuestas en GPU y registrar la interacción en SQLite.

Crea el archivo ~/sap_bot/bot_spark.py:

Python

#!/usr/bin/env python3
"""
Bot Asistente SAP B1 para Spark (XMPP + Ollama + ChromaDB + SQLite).
"""
import os
import ssl
import asyncio
import sqlite3
from datetime import datetime
import slixmpp
import chromadb
import ollama

# ================= CONFIGURACIÓN DEL SERVIDOR =================
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(BASE_DIR, "chroma_sap_db")
DB_LOGS_PATH = os.path.join(BASE_DIR, "historial_conversaciones.db")

# Datos genéricos de conexión Openfire / Spark
JID_BOT = "asistente_sap@xmpp.empresa.local"
PASSWORD_BOT = "PasswordSeguro123"
HOST_SERVIDOR = "192.168.1.50"
PUERTO_XMPP = 5222

# Enlace base para servir las capturas de pantalla
URL_BASE_IMGS = f"http://{HOST_SERVIDOR}:8080"

MODELO_LLM = "qwen2.5:7b"
MODELO_EMBED = "nomic-embed-text"
# ==============================================================

def inicializar_bd_logs():
    with sqlite3.connect(DB_LOGS_PATH) as conn:
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS conversaciones (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                fecha_hora TEXT,
                usuario_spark TEXT,
                pregunta TEXT,
                tipo_fuente TEXT,
                respuesta TEXT
            )
        """)
        conn.commit()

def registrar_log(usuario: str, pregunta: str, tipo_fuente: str, respuesta: str):
    try:
        with sqlite3.connect(DB_LOGS_PATH) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO conversaciones (fecha_hora, usuario_spark, pregunta, tipo_fuente, respuesta)
                VALUES (?, ?, ?, ?, ?)
            """, (
                datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                usuario,
                pregunta,
                tipo_fuente,
                respuesta
            ))
            conn.commit()
    except Exception as err:
        print(f"[-] Error en log SQLite: {err}")

class SparkAIBot(slixmpp.ClientXMPP):
    def __init__(self, jid: str, password: str):
        super().__init__(jid, password)
        self.add_event_handler("session_start", self.on_start)
        self.add_event_handler("message", self.on_message)

        # Aceptar certificados locales/autofirmados
        self.ssl_context = ssl.create_default_context()
        self.ssl_context.check_hostname = False
        self.ssl_context.verify_mode = ssl.CERT_NONE

        self.chroma_client = chromadb.PersistentClient(path=DB_PATH)

    async def on_start(self, event):
        self.send_presence(pstatus="Asistente SAP B1 En Línea")
        await self.get_roster()
        print(f"\n[✓] Asistente conectado a Openfire como: {self.boundjid.bare}")

    def notificar_escribiendo(self, destinatario: str):
        try:
            msg = self.make_message(mto=destinatario, mtype='chat')
            msg['chat_state'] = 'composing'
            msg.send()
        except Exception:
            pass

    def notificar_inactivo(self, destinatario: str):
        try:
            msg = self.make_message(mto=destinatario, mtype='chat')
            msg['chat_state'] = 'active'
            msg.send()
        except Exception:
            pass

    async def on_message(self, msg):
        if msg["type"] not in ("chat", "normal") or not msg["body"].strip():
            return

        pregunta = msg["body"].strip()
        remitente = msg["from"].bare
        print(f"[+] Consulta de [{remitente}]: {pregunta}")

        self.notificar_escribiendo(remitente)
        respuesta = await asyncio.to_thread(self.procesar_consulta, remitente, pregunta)
        msg.reply(respuesta).send()
        self.notificar_inactivo(remitente)

    def procesar_consulta(self, remitente: str, pregunta: str) -> str:
        try:
            coleccion = self.chroma_client.get_or_create_collection(name="manuales_sap")
            total_manuales = coleccion.count()
            fragmentos = []
            referencias = []
            imagenes_adjuntas = []

            if total_manuales > 0:
                res_emb = ollama.embeddings(model=MODELO_EMBED, prompt=pregunta)
                num_a_buscar = min(3, total_manuales)
                coincidencias = coleccion.query(query_embeddings=[res_emb["embedding"]], n_results=num_a_buscar)

                if coincidencias and coincidencias["documents"] and coincidencias["documents"][0]:
                    for doc_txt, meta in zip(coincidencias["documents"][0], coincidencias["metadatas"][0]):
                        fragmentos.append(doc_txt)
                        referencias.append(f"{meta['archivo']} (Pág. {meta['pagina']})")
                        imgs_str = meta.get("imagenes", "")
                        if imgs_str:
                            for img in imgs_str.split(","):
                                if img.strip():
                                    imagenes_adjuntas.append(img.strip())

            contexto = "\n---\n".join(fragmentos) if fragmentos else "NO_HAY_MANUALES_DISPONIBLES"

            prompt_sistema = (
                "Eres un consultor senior experto en SAP Business One de la empresa.\n"
                "Tu objetivo es resolver dudas operativas, funcionales y técnicas sobre SAP B1, SQL/HANA y flujos del sistema.\n\n"
                "REGLAS:\n"
                "1. Si la DOCUMENTACIÓN OFICIAL contiene la respuesta, responde con base en ella con total precisión.\n"
                "2. Si la DOCUMENTACIÓN OFICIAL indica 'NO_HAY_MANUALES_DISPONIBLES' o no contiene la respuesta, responde usando tu conocimiento estándar de SAP B1, pero inicia con:\n"
                "   '⚠️ *Nota: Esta respuesta se basa en el estándar general de SAP B1 y no en un manual interno documentado.*'\n"
                "3. Si la pregunta no tiene relación con temas de trabajo o sistemas de la empresa, recházala amablemente.\n\n"
                f"DOCUMENTACIÓN OFICIAL:\n{contexto}"
            )

            res_chat = ollama.chat(
                model=MODELO_LLM,
                messages=[
                    {"role": "system", "content": prompt_sistema},
                    {"role": "user", "content": pregunta}
                ],
                options={"temperature": 0.15, "num_ctx": 2048}
            )

            cuerpo = res_chat["message"]["content"].strip()

            if "⚠️ *Nota:" in cuerpo:
                tipo_fuente = "ESTANDAR_SAP"
            elif referencias and contexto != "NO_HAY_MANUALES_DISPONIBLES":
                tipo_fuente = "MANUAL_EMPRESA"
                refs_unicas = list(dict.fromkeys(referencias))
                cuerpo += f"\n\n📖 *Referencia:* {', '.join(refs_unicas)}"

                if imagenes_adjuntas:
                    imgs_unicas = list(dict.fromkeys(imagenes_adjuntas))[:2]
                    cuerpo += "\n\n📸 *Capturas del manual:*"
                    for img_name in imgs_unicas:
                        cuerpo += f"\n• {URL_BASE_IMGS}/{img_name}"
            else:
                tipo_fuente = "GENERAL_O_RECHAZO"

            registrar_log(remitente, pregunta, tipo_fuente, cuerpo)
            return cuerpo

        except Exception as err:
            error_msg = f"Error al procesar la consulta: {str(err)}"
            registrar_log(remitente, pregunta, "ERROR", error_msg)
            return error_msg

def main():
    inicializar_bd_logs()
    bot = SparkAIBot(JID_BOT, PASSWORD_BOT)
    bot.register_plugin("xep_0030")
    bot.register_plugin("xep_0199")
    bot.register_plugin("xep_0085")  # Soporte de estado 'Escribiendo...'

    print(f"[*] Conectando a Openfire en {HOST_SERVIDOR}:{PUERTO_XMPP}...")
    bot.connect(host=HOST_SERVIDOR, port=PUERTO_XMPP)

    loop = getattr(bot, "loop", None) or asyncio.get_event_loop()
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        bot.disconnect()

if __name__ == "__main__":
    main()

6. Ejecución como Servicio Continuo (systemd)

Para que el asistente se inicie automáticamente tras cualquier reinicio del servidor:

Bash

sudo tee /etc/systemd/system/sap-bot.service << EOF
[Unit]
Description=Servicio Bot Asistente SAP B1 para Spark
After=network.target ollama.service

[Service]
Type=simple
User=$USER
WorkingDirectory=$HOME/sap_bot
ExecStart=/usr/bin/python3 $HOME/sap_bot/bot_spark.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload
sudo systemctl enable --now sap-bot.service

7. Ventajas y Resultados Obtenidos

  • Privacidad total: Ningún dato, consulta contable o captura de pantalla sale de la red local.
  • Cero costo operativo recurrente: No se depende de tokens ni de APIs de terceros.
  • Experiencia de usuario nativa: El usuario no requiere abrir una página web extra ni loguearse; simplemente abre una ventana de chat en su cliente Spark cotidiano.
  • Mantenimiento simplificado: Para actualizar la base de conocimientos, basta con soltar nuevos manuales en la carpeta y ejecutar python3 indexar_manuales.py. El bot detecta la colección actualizada de forma dinámica sin necesidad de reiniciar el servicio.