En esta guía, te mostraré cómo construir un asistente de búsqueda inteligente que comprende tanto texto como imágenes, utilizando el modelo GME (Generating Multimodal Embeddings) basado en Qwen2-VL-2B.

1. Fundamentos del Sistema

1.1 ¿Qué es GME?

El modelo GME es un sistema de embedding multimodal que procesa tres tipos de入口ada:

• Texto plano: artículos, descripciones técnicas, resúmenes
• Imágenes aisladas: gráficos, capturas de pantalla, fotografías
• Pares imagen-texto: figuras conlegendaindividual

La magia reside en que todos estos elementos se convierten en vectores numéricos en un espacio compartido. Cuando buscas algo, el sistema calcula la similitud entre vectores para encontrar coincidencias.

1.2 Ventajas de Qwen2-VL-2B

Este modelo visual ofrece características destacables:

Resolución dinámica: Procesa imágenes de cualquier dimensionesin perder calidad interpretativa.

Comprensión de documentos: Especialmente efectivo con gráficos complejos, ecuaciones y diagramas técnicos.

Espacio unificado: Textos e imágenes cohabitan en el mismo espacio vectorial, habilitando búsquedas cruzadas: texto→imagen, imagen→texto, imagen→imagen.

2. Preparación del Entorno

2.1 Requisitos del Sistema

Antes de comenzar, asegurate de tener:

• Python 3.8+
• 8GB mínimo de RAM
• 3GB de espacio en disco para modelos
• Conexión a internet para descarga inicial

2.2 Instalación de Dependencias

Ejecuta estos comandos en tu terminal:

mkdir asistente_busqueda
cd asistente_busqueda

python -m venv env
# Windows
env\Scripts\activate
# Linux/Mac
source env/bin/activate

pip install sentence-transformers gradio torch torchvision pillow pymupdf

2.3 Verificación del Modelo

# verificar_modelo.py
from sentence_transformers import SentenceTransformer

print("Cargando modelo GME...")

modelo = SentenceTransformer('Alibaba-NLP/gte-multimodel-qwen2-vl-2b')

# Prueba de codificación
texto_prueba = "Redes neuronales convolucionales para análisis de imágenes"
vector = modelo.encode(texto_prueba)

print(f"Dimensión del vector: {vector.shape}")
print("Modelo operativo correctamente")

3. Arquitectura del Sistema

3.1 Módulo de Procesamiento de Documentos

# procesador_docs.py
import os
import fitz
import json
from PIL import Image
import io
from typing import List, Dict

class ProcesadorDocumentos:
    def __init__(self, modelo):
        self.modelo = modelo
        
    def extraer_pdf(self, ruta_pdf: str) -> List[Dict]:
        """Extrae contenido textual y visual de archivos PDF"""
        elementos = []
        
        try:
            documento = fitz.open(ruta_pdf)
            for num_pagina in range(len(documento)):
                pagina = documento[num_pagina]
                
                # Extraer texto
                texto = pagina.get_text()
                if texto.strip():
                    elementos.append({
                        'tipo': 'texto',
                        'contenido': texto,
                        'pagina': num_pagina + 1,
                        'origen': os.path.basename(ruta_pdf)
                    })
                
                # Extraer imágenes
                for indice_img,img in enumerate(pagina.get_images()):
                    xref = img[0]
                    pix = fitz.Pixmap(documento, xref)
                    if pix.n - pix.alpha < 4:
                        datos_img = pix.tobytes("png")
                        elementos.append({
                            'tipo': 'imagen',
                            'contenido': datos_img,
                            'pagina': num_pagina + 1,
                            'origen': os.path.basename(ruta_pdf),
                            'indice_imagen': indice_img
                        })
                    pix = None
            
            documento.close()
            
        except Exception as e:
            print(f"Error procesando PDF: {e}")
            
        return elementos
    
    def procesar_carpeta(self, ruta_carpeta: str) -> List[Dict]:
        """Procesa todos los documentos en un directorio"""
        todos_elementos = []
        extensiones_validas = ['.pdf', '.txt', '.md', '.jpg', '.jpeg', '.png']
        
        for raiz, _, archivos in os.walk(ruta_carpeta):
            for archivo in archivos:
                ruta_completa = os.path.join(raiz, archivo)
                extension = os.path.splitext(archivo)[1].lower()
                
                if extension == '.pdf':
                    elementos = self.extraer_pdf(ruta_completa)
                    todos_elementos.extend(elementos)
                    
                elif extension in ['.txt', '.md']:
                    with open(ruta_completa, 'r', encoding='utf-8') as f:
                        contenido = f.read()
                    todos_elementos.append({
                        'tipo': 'texto',
                        'contenido': contenido,
                        'origen': archivo
                    })
                    
                elif extension in ['.jpg', '.jpeg', '.png']:
                    with open(ruta_completa, 'rb') as f:
                        datos = f.read()
                    todos_elementos.append({
                        'tipo': 'imagen',
                        'contenido': datos,
                        'origen': archivo
                    })
        
        return todos_elementos
    
    def generar_embeddings(self, documentos: List[Dict]) -> List[Dict]:
        """Convierte documentos a vectores numéricos"""
        documentos_vectorizados = []
        
        for i, doc in enumerate(documentos):
            try:
                if doc['tipo'] == 'texto':
                    embedding = self.modelo.encode(doc['contenido'])
                else:
                    imagen = Image.open(io.BytesIO(doc['contenido']))
                    embedding = self.modelo.encode(imagen)
                
                doc['embedding'] = embedding.tolist()
                documentos_vectorizados.append(doc)
                
                if (i + 1) % 10 == 0:
                    print(f"Procesados {i + 1}/{len(documentos)}")
                    
            except Exception as e:
                print(f"Error en elemento {i}: {e}")
                continue
        
        return documentos_vectorizados
    
    def guardar_indice(self, documentos: List[Dict], ruta_salida: str):
        """Almacena el índice en formato JSON"""
        with open(ruta_salida, 'w', encoding='utf-8') as f:
            json.dump(documentos, f, ensure_ascii=False, indent=2)

3.2 Motor de Búsqueda

# motor_busqueda.py
import numpy as np
from PIL import Image
import io
import json
from typing import List, Dict, Union

class MotorBusqueda:
    def __init__(self, modelo, ruta_documentos: str = None):
        self.modelo = modelo
        self.documentos = []
        self.embeddings = np.array([])
        
        if ruta_documentos:
            self.cargar_desde_archivo(ruta_documentos)
    
    def cargar_desde_archivo(self, ruta: str):
        """Carga documentos previamente indexados"""
        with open(ruta, 'r', encoding='utf-8') as f:
            datos = json.load(f)
        
        self.documentos = datos
        self.embeddings = np.array([doc['embedding'] for doc in datos])
    
    def agregar_documentos(self, documentos: List[Dict]):
        """Añade nuevos documentos al índice"""
        for doc in documentos:
            if 'embedding' not in doc:
                if doc['tipo'] == 'texto':
                    emb = self.modelo.encode(doc['contenido'])
                else:
                    img = Image.open(io.BytesIO(doc['contenido']))
                    emb = self.modelo.encode(img)
                doc['embedding'] = emb.tolist()
            
            self.documentos.append(doc)
            self.embeddings = np.array([d['embedding'] for d in self.documentos])
    
    def buscar(self, consulta: Union[str, Image.Image], k: int = 5) -> List[Dict]:
        """Encuentra los documentos más similares"""
        if isinstance(consulta, str):
            vector_consulta = self.modelo.encode(consulta)
        else:
            vector_consulta = self.modelo.encode(consulta)
        
        if len(self.embeddings) == 0:
            return []
        
        # Normalización para similitud coseno
        consulta_norm = vector_consulta / np.linalg.norm(vector_consulta)
        docs_norm = self.embeddings / np.linalg.norm(self.embeddings, axis=1, keepdims=True)
        
        # Calcular similitudes
        similitudes = np.dot(docs_norm, consulta_norm)
        
        # Ordenar y obtener top-k
        indices_top = np.argsort(similitudes)[::-1][:k]
        
        resultados = []
        for idx in indices_top:
            resultado = self.documentos[idx].copy()
            resultado['similitud'] = float(similitudes[idx])
            resultados.append(resultado)
        
        return resultados
    
    def busqueda_hibrida(self, consulta_texto: str = None, 
                         consulta_imagen: Image.Image = None, 
                         k: int = 5) -> List[Dict]:
        """Combina búsquedas de texto e imagen"""
        if consulta_texto and consulta_imagen:
            v_texto = self.modelo.encode(consulta_texto)
            v_imagen = self.modelo.encode(consulta_imagen)
            vector_combinado = (v_texto + v_imagen) / 2
        elif consulta_texto:
            vector_combinado = self.modelo.encode(consulta_texto)
        elif consulta_imagen:
            vector_combinado = self.modelo.encode(consulta_imagen)
        else:
            return []
        
        return self._buscar_con_vector(vector_combinado, k)
    
    def _buscar_con_vector(self, vector: np.ndarray, k: int) -> List[Dict]:
        """Búsqueda usando vector precalculado"""
        if len(self.embeddings) == 0:
            return []
        
        vect_norm = vector / np.linalg.norm(vector)
        docs_norm = self.embeddings / np.linalg.norm(self.embeddings, axis=1, keepdims=True)
        
        similitudes = np.dot(docs_norm, vect_norm)
        indices = np.argsort(similitudes)[::-1][:k]
        
        return [
            {**self.documentos[idx], 'similitud': float(similitudes[idx])}
            for idx in indices
        ]

3.3 Interfaz Grádica

# aplicacion.py
import gradio as gr
from sentence_transformers import SentenceTransformer
from procesador_docs import ProcesadorDocumentos
from motor_busqueda import MotorBusqueda
import os
import json

# Inicialización
print("Inicializando sistema...")
modelo = SentenceTransformer('Alibaba-NLP/gte-multimodel-qwen2-vl-2b')
buscador = MotorBusqueda(modelo)
documentos-globales = []

def indexar_archivos(archivos):
    """Procesa y indexa documentos subidos"""
    if not archivos:
        return "Selecciona archivos primero"
    
    procesador = ProcesadorDocumentos(modelo)
    elementos_indexar = []
    
    for archivo in archivos:
        if archivo.endswith('.pdf'):
            elem = procesador.extraer_pdf(archivo)
        elif archivo.endswith(('.txt', '.md')):
            with open(archivo, 'r', encoding='utf-8') as f:
                contenido = f.read()
            elem = [{'tipo': 'texto', 'contenido': contenido, 
                    'origen': os.path.basename(archivo)}]
        elif archivo.endswith(('.jpg', '.jpeg', '.png')):
            with open(archivo, 'rb') as f:
                datos = f.read()
            elem = [{'tipo': 'imagen', 'contenido': datos,
                    'origen': os.path.basename(archivo)}]
        else:
            continue
        
        elementos_indexar.extend(elem)
    
    # Vectorización
    vectorizados = procesador.generar_embeddings(elementos_indexar)
    buscador.agregar_documentos(vectorizados)
    
    global documentos-globales
    documentos-globales.extend(vectorizados)
    
    if vectorizados:
        procesador.guardar_indice(vectorizados, 'indice_documentos.json')
    
    return f"Indexados {len(vectorizados)} elementos"

def realizar_busqueda(texto, imagen, top_k):
    """Ejecuta búsqueda multimodal"""
    if not buscador.documentos:
        return "Indexa documentos primero", []
    
    resultados = buscador.busqueda_hibrida(
        consulta_texto=texto,
        consulta_imagen=imagen,
        k=top_k
    )
    
    if not resultados:
        return "Sin resultados", []
    
    presentacion = "Resultados:\n\n"
    vista_detallada = []
    
    for i, res in enumerate(resultados):
        tipo = res['tipo']
        origen = res.get('origen', 'Desconocido')
        pagina = res.get('pagina', '')
        sim = res['similitud']
        
        presentacion += f"{i+1}. [{tipo}] {origen}"
        if pagina:
            presentacion += f" - Página {pagina}"
        presentacion += f" (Similitud: {sim:.3f})\n"
        
        if tipo == 'texto':
            fragmento = res['contenido'][:200] + "..." if len(res['contenido']) > 200 else res['contenido']
            presentacion += f"   {fragmento}\n"
        
        presentacion += "\n"
        
        vista_detallada.append({
            'tipo': tipo,
            'origen': origen,
            'similitud': sim,
            'contenido': res['contenido'][:500] + "..." if len(res.get('contenido', '')) > 500 else res.get('contenido', '')
        })
    
    return presentacion, vista_detallada

# Construcción de interfaz
with gr.Blocks(title="Asistente de Búsqueda Documental") as interfaz:
    gr.Markdown("# Asistente de Búsqueda Multimodal")
    gr.Markdown("Sistema de recuperación basado en GME con Qwen2-VL-2B")
    
    with gr.Tab("Gestión de Documentos"):
        with gr.Row():
            with gr.Column():
                entrada_archivos = gr.File(
                    label="Subir documentos",
                    file_count="multiple",
                    file_types=[".pdf", ".txt", ".md", ".jpg", ".jpeg", ".png"]
                )
                boton_procesar = gr.Button("Indexar", variant="primary")
            
            with gr.Column():
                salida_estado = gr.Textbox(label="Estado", interactive=False)
        
        boton_procesar.click(
            indexar_archivos,
            inputs=[entrada_archivos],
            outputs=[salida_estado]
        )
    
    with gr.Tab("Búsqueda"):
        with gr.Row():
            with gr.Column():
                consulta_texto = gr.Textbox(
                    label="Consulta textual",
                    placeholder="Describe lo que buscas...",
                    lines=3
                )
                consulta_imagen = gr.Image(
                    label="Imagen de referencia (opcional)",
                    type="pil"
                )
                control_top = gr.Slider(1, 20, value=5, step=1, label="Resultados")
                boton_buscar = gr.Button("Buscar", variant="primary")
            
            with gr.Column():
                salida_resultados = gr.Textbox(lines=15, label="Resultados")
                salida_detalle = gr.JSON(label="Detalles")
        
        boton_buscar.click(
            realizar_busqueda,
            inputs=[consulta_texto, consulta_imagen, control_top],
            outputs=[salida_resultados, salida_detalle]
        )
    
    with gr.Tab("Acerca de"):
        gr.Markdown("## Especificaciones")
        gr.Markdown(f"- Modelo: GME-Qwen2-VL-2B")
        gr.Markdown(f"- Documentos indexados: {len(documentos-globales)}")
        gr.Markdown("### Capacidades")
        gr.Markdown("""
        - Búsqueda texto→texto
        - Búsqueda imagen→imagen
        - Búsqueda cruzada texto↔imagen
        -检索 híbrida multimodal
        """)

if __name__ == "__main__":
    if os.path.exists("indice_documentos.json"):
        print("Cargando índice existente...")
        buscador.cargar_desde_archivo("indice_documentos.json")
        with open("indice_documentos.json", 'r') as f:
            documentos-globales = json.load(f)
    
    interfaz.launch(server_name="0.0.0.0", server_port=7860)

4. Casos de Uso Prácticos

4.1 Investigación Académica

Un investigador tiene 50 artigos sobre visión por computadora. Con este sistema puede:

• Buscar "detección de objetos en tiempo real" y encontrar artículos relacionados aunque usen terminología diferente
• Subir un diagrama de arquitectura de red y localizar artículos con enfoques similares
• Combinar descripción textual con imágenes de referencia

4.2 Documentación Técnica

Equipos de ingeniería de software manejan cientos de documentos con capturas de pantalla, diagramas de flujo y especificaciones. El sistema permite:

• Localizar documentación relacionada por similitud semántica
• Encontrar diagrams similares en diferentes proyectos
• Mantener consistencia en documentación existente

5. Optimizaciones de Rendimiento

5.1 Procesamiento por Lotes

def procesar_lotes(elementos, tamano_lote=32):
    """Procesa documentos en grupos para mayor velocidad"""
    resultados = []
    
    for i in range(0, len(elementos), tamano_lote):
        lote = elementos[i:i+tamano_lote]
        
        for elem in lote:
            try:
                if elem['tipo'] == 'texto':
                    emb = modelo.encode(elem['contenido'])
                else:
                    img = Image.open(io.BytesIO(elem['contenido']))
                    emb = modelo.encode(img)
                
                elem['embedding'] = emb.tolist()
                resultados.append(elem)
            except Exception as e:
                continue
        
        print(f"Progreso: {min(i+tamano_lote, len(elementos))}/{len(elementos)}")
    
    return resultados

5.2 Caché de Resultados

import hashlib
import pickle
from functools import lru_cache

class MotorBusquedaCache(MotorBusqueda):
    def __init__(self, modelo, directorio_cache="cache"):
        super().__init__(modelo)
        self.directorio_cache = directorio_cache
        os.makedirs(directorio_cache, exist_ok=True)
    
    def hash_consulta(self, consulta):
        """Genera identificador único para la consulta"""
        if isinstance(consulta, str):
            datos = consulta.encode()
        else:
            datos = consulta.tobytes()
        return hashlib.md5(datos).hexdigest()
    
    @lru_cache(maxsize=100)
    def buscar_cache(self, hash_consulta, k):
        """Almacena resultados de consultas repetidas"""
        ruta_cache = os.path.join(self.directorio_cache, f"{hash_consulta}_{k}.pkl")
        
        if os.path.exists(ruta_cache):
            with open(ruta_cache, 'rb') as f:
                return pickle.load(f)
        
        # Ejecutar búsqueda real requiere modificar lógica padre
        # Esta es una implementación conceptual
        return None

6. Consideraciones Finales

Este sistema de búsqueda multimodal ofrece mejoras significativas sobre métodos tradicionales:

Comprensión profunda: No busca palabras clave, sino conceptos y relaciones semánticas.

Flexibilidad total: Cualquier tipo de consulta (texto, imagen o ambas) produce resultados relevantes.

Escalabilidad: La arquitectura modular permite扩展 con bases de datos vectoriales para manejar millones de documentos.

Para implementaciones en producción, considera usar FAISS o Milvus para indexing eficiente, y arquitecturas de microservicios para escalar horizontalmente.