¿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:
- Indexación previa (Off-line): Un script en Python escanea la carpeta de manuales (
.pdfy.docx), extrae texto estructurado, tablas y aplica OCR con Tesseract a capturas de pantalla. El texto se vectoriza connomic-embed-texty se guarda en ChromaDB. - 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.
- Bot XMPP (
slixmpp): Se autentica en Openfire como un usuario de soporte (ej.asistente@xmpp.empresa.local). - 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. - 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
.docxdentro 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:
- Conectarse de forma asíncrona al servidor Openfire.
- Emitir el paquete
XEP-0085para mostrar «Escribiendo…» de inmediato en Spark. - Consultar la base vectorial sin bloquear el hilo principal.
- 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.
Deja un comentario