El desafío del arrranque frío en sistemas de conocimiento local

En el panorama actual de sistemas de preguntas y respuestas impulsados por modelos de lenguaje, la velocidad de respuesta inicial es un factor crítico para la experiencia del usuario. Langchain-Chatchat, un marco de código abierto para consultas sobre bases de conocimiento privadas, enfrenta típicamente demoras significativas durante su primer arranque. Esto se debe a una cadena de procesamiento intensivo que incluye la lectura de documentos, el análisis de contenido, la segmentación en fragmentos, la generación de incrustaciones vectoriales y la construcción de índices de búsqueda. Estas operaciones, ejecutadas sincrónicamente al inicio, pueden prolongarse varios minutos, especialmente con grandes volúmenes de datos.

Para mitigar este problema, se propone una arquitectura de caché multicapa que almacena los resultados intermedios de cada etapa, evitando reprocesos innecesarios. Esta aproximación transforma un arranque frío en una carga quasi-caliente, mejorando drásticamente el rendimiento.

Estrategia de caché para el análisis de documentos

La fase inicial consiste en extraer texto de archivos en formatos variados (PDF, DOCX, TXT, etc.). Sin caché, esta operación se repite cada vez que el sistema se reinicia, consumiendo recursos de CPU e I/O. La solución implica persistir el texto analizado, identificando los archivos mediante un hash único de su contenido para detectar cambios.

Implementación en Python con variaciones en nombres de variables y lógica:

import hashlib
import os
import pickle

def calcular_fingerprint_archivo(ruta: str) -> str:
    """Genera un hash SHA-256 del archivo basado en su contenido."""
    sha = hashlib.sha256()
    with open(ruta, "rb") as f:
        while bloque := f.read(8192):
            sha.update(bloque)
    return sha.hexdigest()

def guardar_texto_parseado(ruta_archivo: str, contenido: str, dir_cache: str = "./cache_textos"):
    """Almacena el texto extraído en un archivo de caché."""
    os.makedirs(dir_cache, exist_ok=True)
    fingerprint = calcular_fingerprint_archivo(ruta_archivo)
    archivo_cache = os.path.join(dir_cache, f"{fingerprint}.dat")
    with open(archivo_cache, 'wb') as f:
        pickle.dump((ruta_archivo, contenido), f)

def recuperar_texto_cache(ruta_archivo: str, dir_cache: str = "./cache_textos") -> str | None:
    """Intenta cargar el texto desde la caché si existe y es válido."""
    if not os.path.exists(dir_cache):
        return None
    fingerprint = calcular_fingerprint_archivo(ruta_archivo)
    archivo_cache = os.path.join(dir_cache, f"{fingerprint}.dat")
    if os.path.isfile(archivo_cache):
        with open(archivo_cache, 'rb') as f:
            ruta_guardada, texto = pickle.load(f)
            if ruta_guardada == ruta_archivo:
                return texto
    return None

Este mecanismo reduce el consumo de CPU en análisis documental en más del 70% en pruebas con conjuntos de archivos extensos.

Caché para incrustaciones vectoriales

La conversión de fragmentos de texto en vectores numéricos mediante modelos de incrustación (como BGE o Sentence-BERT) es el paso más costoso computacionalmente. Almacenar los vectores generados evita recalcularlos para textos idénticos, utilizando hashes del contenido como claves de caché.

Código optimizado con cambios estructurales:

import numpy as np
import json
import os
import hashlib
from sentence_transformers import SentenceTransformer

modelo_inc = SentenceTransformer("bge-small-zh-v1.5")
RUTA_CACHE = "./cache_vectores"

def incrustar_con_cache(textos: list[str]) -> np.ndarray:
    vectores = []
    archivo_indice = os.path.join(RUTA_CACHE, "indice_hashes.json")
    os.makedirs(RUTA_CACHE, exist_ok=True)
    
    indice = {}
    if os.path.isfile(archivo_indice):
        with open(archivo_indice, 'r', encoding='utf-8') as f:
            indice = json.load(f)
    
    for texto in textos:
        hash_texto = hashlib.sha256(texto.encode('utf-8')).hexdigest()
        archivo_vec = os.path.join(RUTA_CACHE, f"{hash_texto}.vec")
        
        if hash_texto in indice and os.path.isfile(archivo_vec):
            vector = np.fromfile(archivo_vec, dtype=np.float32)
        else:
            vector = modelo_inc.encode([texto])[0].astype(np.float32)
            vector.tofile(archivo_vec)
            indice[hash_texto] = 1
            with open(archivo_indice, 'w', encoding='utf-8') as f:
                json.dump(indice, f)
        
        vectores.append(vector)
    
    return np.array(vectores)

En pruebas con 500 documentos, esta caché redujo el tiempo de carga de incrustaciones de 183 a 27 segundos, una mejora del 85%.

Caché de índices para bases vectoriales

FAISS, la biblioteca para búsqueda de vecinos cercanos, permite serializar y deserializar índices completos. Guardar el índice en disco tras su construcción evita reconstruirlo en reinicios subsiguientes, lo cual es crucial para índices avanzados como HNSW o IVF-PQ.

Funciones reescritas para manejo de índices:

import faiss
import numpy as np
import os

def crear_y_persistir_indice(vectores: np.ndarray, ubicacion: str):
    """Crea un índice FAISS y lo guarda en disco."""
    dim = vectores.shape[1]
    idx = faiss.IndexHNSWFlat(dim, 40)
    idx.add(vectores.astype('float32'))
    faiss.write_index(idx, ubicacion)

def cargar_indice_existente(ubicacion: str) -> faiss.Index | None:
    """Carga un índice desde archivo si existe."""
    return faiss.read_index(ubicacion) if os.path.isfile(ubicacion) else None

La caché de índices reduce los tiempos de carga de decenas de segundos a menos de un segundo, completando el conjunto de optimizaciones.

Integración de las capas de caché

Las tres capas trabajan secuencialmente en la tubería de procesamiento: primero se verifica la caché de texto, luego la de vectores, y finalmente la del índice. Este enfoque en cascada permite omitir etapas completas cuando los datos están disponibles, logrando arranques casi instantáneos después de la inicialización primera.

Recomendaciones prácticas para implementación:

• Implementar un mecanismo de invalidación basado en fechas de modificación de archivos.
• Monitorear el uso de espacio en disco y establecer políticas de limpieza automática.
• Asegurar permisos de archivo adecuados para proteger datos sensibles en caché.
• Registrar métricas de tasa de aciertos y tiempos de cada fase para ajustes finos.

Un esquema de configuración sugerido podría incluir habilitaciones por capa y rutas de almacenamiento específicas.