atareao / embeddings.py

#!/usr/bin/env python3 """ 🚀 Embeddings para todo — Demo completa ======================================== Episodio 817: Un solo script que recorre todos los casos de uso. Búsqueda semántica, clasificación, deduplicación, recomendación y ChromaDB. Requisitos: - Ollama corriendo en localhost:11434 - Modelo nomic-embed-text descargado: ollama pull nomic-embed-text - Python 3.10+ Uso: python3 setup_completo.py # Ejecuta todo python3 setup_completo.py --fast # Solo pasos principales, sin ChromaDB python3 setup_completo.py --list # Lista los pasos disponibles python3 setup_completo.py --step 3 # Ejecuta solo el paso 3 python3 setup_completo.py --step 2,4,6 # Ejecuta pasos específicos python3 setup_completo.py --interactive # Paso a paso con confirmación """ import os, sys, json, math, glob, time, shutil, subprocess, textwrap, argparse # ─── Configuración ──────────────────────────────────────────────────────── MODEL = "bge-m3" OLLAMA_URL = "http://localhost:11434" DIR_DOCS = "./docs_prueba" # Documentos de prueba para búsqueda semántica DIR_CHROMA = "./chroma_store" # Donde ChromaDB guarda los datos # Colores para la terminal VERDE = "\033[92m" AZUL = "\033[94m" AMAR = "\033[93m" ROJO = "\033[91m" FIN = "\033[0m" NEGR = "\033[1m" def ok(msg): print(f" {VERDE}✅{FIN} {msg}") def info(msg): print(f" {AZUL}ℹ️{FIN} {msg}") def warn(msg): print(f" {AMAR}⚠️{FIN} {msg}") def error(msg): print(f" {ROJO}❌{FIN} {msg}") def titulo(t): print(f"\n{NEGR}{'=' * 60}{FIN}\n{NEGR} {t}{FIN}\n{'=' * 60}") # ─── Helpers ────────────────────────────────────────────────────────────── def check_ollama(): """Verifica que Ollama está corriendo y el modelo está disponible.""" try: r = subprocess.run( ["curl", "-s", f"{OLLAMA_URL}/api/tags"], capture_output=True, text=True, timeout=5, ) if r.returncode != 0: error("Ollama no responde. ¿Está corriendo?") info("Ejecuta: ollama serve") return False modelos = json.loads(r.stdout) disponible = any(MODEL in m["name"] for m in modelos.get("models", [])) if not disponible: warn(f"Modelo {MODEL} no encontrado. Descargando...") subprocess.run(["ollama", "pull", MODEL], check=True) ok(f"Modelo {MODEL} descargado") return True except Exception as e: error(f"Error conectando con Ollama: {e}") return False def emb(texto): """Genera un embedding usando Ollama.""" payload = json.dumps({"model": MODEL, "prompt": texto}) r = subprocess.run( ["curl", "-s", f"{OLLAMA_URL}/api/embeddings", "-d", payload], capture_output=True, text=True, ) return json.loads(r.stdout)["embedding"] def cos_sim(a, b): """Similitud coseno entre dos vectores.""" dot = sum(x * y for x, y in zip(a, b)) na = math.sqrt(sum(x * x for x in a)) nb = math.sqrt(sum(y * y for y in b)) return dot / (na * nb) if na and nb else 0 # ─── 1. Crear documentos de prueba ──────────────────────────────────────── def crear_documentos(): titulo("1️⃣ CREAR DOCUMENTOS DE PRUEBA") if os.path.exists(DIR_DOCS): shutil.rmtree(DIR_DOCS) os.makedirs(DIR_DOCS) documentos = { "facturas.txt": """Factura hosting enero 2026: 12.99€ Factura dominio atareao.es 2026: 18.50€ Pago servidor VPS febrero: 24.99€ Total gastos infraestructura Q1: 156.47€ Próximo pago hosting: 1 de marzo""", "apuntes-linux.txt": """Cómo montar un servidor web con Nginx en Ubuntu 24.04 Configurar firewall con nftables: tabla inet filter con policy drop Instalar Docker: sudo apt install docker.io docker-compose Los logs del sistema están en /var/log/syslog systemctl enable --now para activar servicios al arranque""", "recetas.txt": """Paella valenciana: arroz bomba, pollo, conejo, judía verde, garrofón Tortilla de patatas: 4 huevos, 2 patatas grandes, cebolla, aceite de oliva Pan casero: 500g harina de fuerza, 300ml agua, 10g sal, 5g levadura La clave de un buen arroz es el sofrito y el caldo caliente""", "compras.txt": """Lista de la compra: leche, huevos, pan, arroz, tomates, lechuga Comprar regalo cumpleaños: libro de Rust Recoger paquete en Correos: número de seguimiento 123456789 Repostar coche: gasolina 95, 40€""", "ideas-proyectos.txt": """App web para gestionar gastos compartidos con amigos Script que monitoriza el precio de la luz y avisa cuando baja Bot de Telegram que resuma noticias de IA cada mañana Generador de menú semanal con recetas de un archivo de cocina""", } for nombre, contenido in documentos.items(): with open(os.path.join(DIR_DOCS, nombre), "w") as f: f.write(contenido.strip()) ok(f"Documento creado: {nombre}") info("5 documentos de prueba listos en ./docs_prueba/") return list(documentos.keys()) # ─── 2. Búsqueda semántica ──────────────────────────────────────────────── def busqueda_semantica(): titulo("2️⃣ BÚSQUEDA SEMÁNTICA — buscar por significado") # Indexar documentos info("Generando embeddings de los documentos de prueba...") archivos = sorted(glob.glob(f"{DIR_DOCS}/*.txt")) index = [] for path in archivos: with open(path) as f: texto = f.read() nombre = os.path.basename(path) vec = emb(texto) index.append((nombre, vec, texto)) print(f" {nombre}: {len(vec)} dimensiones") ok(f"{len(index)} documentos indexados") # Consultas de prueba consultas = [ "¿Cuánto dinero gastamos en servidores?", "¿Qué hay para cenar esta noche?", "Alguna idea para un proyecto con IA", ] for consulta in consultas: print(f'\n {NEGR}🔍 Consulta:{FIN} "{consulta}"') vec_q = emb(consulta) resultados = sorted( [ (nombre, cos_sim(vec_q, vec), texto[:100]) for nombre, vec, texto in index ], key=lambda x: -x[1], ) for nombre, sim, preview in resultados[:3]: barra = "█" * int(sim * 20) + "░" * (20 - int(sim * 20)) print(f" {barra} {sim:.4f} {nombre}") mejor = resultados[0] print(f" → Mejor resultado: {NEGR}{mejor[0]}{FIN} ({mejor[1]:.1%})") if mejor[1] > 0.7: ok("Coincidencia semántica alta") elif mejor[1] > 0.5: info("Coincidencia semántica media") else: warn("Coincidencia baja — prueba con otra consulta") # ─── 3. Clasificación ───────────────────────────────────────────────────── def clasificacion(): titulo("3️⃣ CLASIFICACIÓN — sin entrenar modelos") categorias = { "urgente": [ "se ha caído el servidor de producción", "necesito esto para mañana sin falta", "error crítico en la base de datos", ], "normal": [ "te envío el documento actualizado", "confirma si has recibido el email", ], "spam": [ "gana dinero rápido desde casa", "has ganado un premio exclusivo", ], "cocina": [ "receta de paella valenciana", "cómo hacer pan casero", ], } # Pre-calcular embedding de cada categoría (promedio de ejemplos) info("Calculando firmas de categorías...") firmas = {} for cat, ejemplos in categorias.items(): vecs = [emb(e) for e in ejemplos] firmas[cat] = [sum(vals) / len(vals) for vals in zip(*vecs)] ok(f"Categoría '{cat}' lista ({len(ejemplos)} ejemplos)") # Textos a clasificar textos_prueba = [ "el servidor web no responde desde las 3 de la mañana", "te adjunto la factura de este mes para tu revisión", "FELICIDADES HAS GANADO UN VIAJE AL CARIBE", "cómo hacer una tortilla de patatas con cebolla caramelizada", ] for texto in textos_prueba: vec = emb(texto) resultados = sorted( [(cat, cos_sim(vec, firma)) for cat, firma in firmas.items()], key=lambda x: -x[1], ) mejor = resultados[0] barra = "█" * int(mejor[1] * 20) + "░" * (20 - int(mejor[1] * 20)) print(f'\n 📝 "{texto[:60]}..."') print(f" → {barra} {NEGR}{mejor[0]}{FIN} ({mejor[1]:.1%})") # ─── 4. Deduplicación ───────────────────────────────────────────────────── def deduplicacion(): titulo("4️⃣ DEDUPLICACIÓN — encontrar contenido repetido") items = [ {"id": "doc1", "texto": "Cómo instalar Ollama en Ubuntu 24.04 paso a paso"}, {"id": "doc2", "texto": "Guía completa para instalar Ollama en Ubuntu 24.04"}, {"id": "doc3", "texto": "Receta de paella valenciana tradicional con marisco"}, {"id": "doc4", "texto": "Cómo configurar nftables como firewall en Linux"}, {"id": "doc5", "texto": "Instalación de Ollama en Ubuntu 24.04 - tutorial"}, ] info("Generando embeddings...") embebidos = [(item, emb(item["texto"])) for item in items] print(f"\n {'Item':<10} {'Similitud con':<20} {'Score':<8}") print(f" {'-' * 40}") for i, (item_a, emb_a) in enumerate(embebidos): for j, (item_b, emb_b) in enumerate(embebidos[i + 1 :], i + 1): sim = cos_sim(emb_a, emb_b) status = ( "🔴 DUPLICADO" if sim > 0.90 else "🟡 PARECIDO" if sim > 0.75 else "🟢 DISTINTO" ) print(f" {item_a['id']:<10} {item_b['id']:<20} {sim:.4f} {status}") # Detectar duplicados duplicados = [ (item_a, item_b, sim) for (item_a, emb_a), (item_b, emb_b) in [ ((items[i], embebidos[i][1]), (items[j], embebidos[j][1])) for i in range(len(items)) for j in range(i + 1, len(items)) ] if (sim := cos_sim(emb_a, emb_b)) > 0.85 ] if duplicados: print( f"\n {NEGR}🔴 Se detectaron {len(duplicados)} grupos de duplicados:{FIN}" ) for a, b, sim in duplicados: print(f' • "{a["texto"][:50]}..."') print(f' "{b["texto"][:50]}..."') print(f" → Similitud: {sim:.1%}") else: info("No se detectaron duplicados significativos") # ─── 5. Recomendación ───────────────────────────────────────────────────── def recomendacion(): titulo("5️⃣ RECOMENDACIÓN — más como esto") articulos = [ "Linux es el mejor sistema operativo para servidores", "Cómo instalar Docker en Ubuntu 24.04", "Introducción a Rust: seguridad de memoria sin recolector de basura", "PostgreSQL con pgvector para búsqueda semántica", "Ollama: modelos de lenguaje locales en tu servidor", "Configurar nftables como firewall en Linux", "Rust vs Go: comparativa para backend en 2026", "Embeddings: coordenadas en un mapa de significados", ] info("Generando embeddings de {len(articulos)} artículos...") index = [(tit, emb(tit)) for tit in articulos] # Elegir un artículo de referencia ref = "Ollama: modelos de lenguaje locales en tu servidor" print(f'\n 📖 Artículo de referencia: "{ref}"') print(f"\n {NEGR}📚 Recomendaciones:{FIN}") vec_ref = emb(ref) recomendados = sorted( [(tit, cos_sim(vec_ref, vec)) for tit, vec in index if tit != ref], key=lambda x: -x[1], ) for i, (tit, sim) in enumerate(recomendados[:4], 1): barra = "█" * int(sim * 20) + "░" * (20 - int(sim * 20)) print(f" {i}. {barra} {sim:.4f} {tit}") # ─── 6. ChromaDB ────────────────────────────────────────────────────────── def chroma_demo(): titulo("6️⃣ CHROMADB — base de datos vectorial") try: import chromadb from chromadb.utils import embedding_functions except ImportError: warn("chromadb no instalado. Ejecuta: pip install chromadb") return # Limpiar y crear if os.path.exists(DIR_CHROMA): shutil.rmtree(DIR_CHROMA) info("Creando base de datos vectorial con ChromaDB...") client = chromadb.PersistentClient(path=DIR_CHROMA) collection = client.get_or_create_collection( name="demo_817", embedding_function=embedding_functions.OllamaEmbeddingFunction( url=f"{OLLAMA_URL}/api/embeddings", model_name=MODEL ), ) # Documentos de prueba docs = [ "Linux es el mejor sistema operativo para servidores por su estabilidad", "Para instalar Docker en Ubuntu ejecuta sudo apt install docker.io", "La paella valenciana lleva arroz, pollo, conejo, judía y garrofón", "Ollama permite ejecutar modelos de lenguaje localmente en tu hardware", "Rust es un lenguaje de programación de sistemas seguro y rápido", "PostgreSQL con pgvector permite búsqueda semántica sobre documentos", "Los embeddings convierten texto en vectores numéricos", "El jamón ibérico de bellota se diferencia por la dieta del cerdo", ] metadatos = [ {"tema": "linux"}, {"tema": "linux"}, {"tema": "cocina"}, {"tema": "ia"}, {"tema": "programacion"}, {"tema": "ia"}, {"tema": "ia"}, {"tema": "cocina"}, ] ids = [f"doc_{i}" for i in range(len(docs))] collection.add(documents=docs, metadatas=metadatos, ids=ids) ok(f"{len(docs)} documentos indexados en ChromaDB") # Búsqueda consultas = [ "¿Qué framework usar para backend?", "Dime algo de cocina", ] for consulta in consultas: print(f'\n {NEGR}🔍 Consulta:{FIN} "{consulta}"') results = collection.query(query_texts=[consulta], n_results=3) for doc, meta, dist in zip( results["documents"][0], results["metadatas"][0], results["distances"][0] ): conf = 1 - dist barra = "█" * int(conf * 20) + "░" * (20 - int(conf * 20)) print(f" {barra} {conf:.1%} [{meta['tema']}] {doc}") # Filtro por metadatos print(f" 🔽 Con filtro [tema=linux]:") filtrados = collection.query( query_texts=[consulta], n_results=2, where={"tema": "linux"} ) for doc in filtrados["documents"][0]: print(f" 🐧 {doc}") # Persistencia if os.path.exists(DIR_CHROMA): ok(f"Datos persistentes en {DIR_CHROMA}/") info("Cierra y abre Python — los datos siguen ahí") # Limpiar shutil.rmtree(DIR_CHROMA) info("ChromaDB limpiada (demo temporal)") # ─── MAIN ────────────────────────────────────────────────────────────────── PASOS = { 1: ("Crear documentos de prueba", crear_documentos, True), 2: ("Búsqueda semántica", busqueda_semantica, True), 3: ("Clasificación", clasificacion, True), 4: ("Deduplicación", deduplicacion, True), 5: ("Recomendación", recomendacion, True), 6: ("ChromaDB", chroma_demo, False), } def listar_pasos(): print(f"\n{'=' * 60}") print(f" PASOS DISPONIBLES") print(f"{'=' * 60}") for n in sorted(PASOS): nombre, _, esencial = PASOS[n] req = " (requiere chromadb)" if not esencial else "" print(f" {n}. {nombre}{req}") print(f"\n --fast omite el paso 6 (ChromaDB)") print(f" --interactive pide confirmación entre pasos\n") def ejecutar_paso(n, limpiar_al_final=False): if n not in PASOS: warn(f"Paso {n} no válido. Usa --list para ver los pasos.") return nombre, fn, _ = PASOS[n] info(f"Ejecutando paso {n}: {nombre}...") fn() if limpiar_al_final and os.path.exists(DIR_DOCS): shutil.rmtree(DIR_DOCS) def main(): parser = argparse.ArgumentParser( description="Episodio 817 — Embeddings para todo. Demo completa.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent("""\ Ejemplos: python3 setup_completo.py # todo python3 setup_completo.py --fast # sin ChromaDB python3 setup_completo.py --list # listar pasos python3 setup_completo.py --step 3 # solo paso 3 python3 setup_completo.py --step 1,3,5 # pasos 1,3,5 python3 setup_completo.py --interactive # paso a paso """), ) parser.add_argument("--fast", action="store_true", help="Omite ChromaDB (paso 6)") parser.add_argument( "--list", "-l", action="store_true", help="Lista los pasos disponibles" ) parser.add_argument( "--step", "-s", type=str, help="Ejecuta paso(s) específicos: 3, 1,3,5, 1-4", ) parser.add_argument( "--interactive", "-i", action="store_true", help="Ejecuta paso a paso con confirmación", ) args = parser.parse_args() if args.list: listar_pasos() return print(f""" {NEGR}🚀 EPISODIO 817 — Embeddings para todo{FIN} {NEGR} Demo completa: búsqueda, clasificación, deduplicación, recomendación, ChromaDB{FIN} {NEGR}Requisitos:{FIN} • Ollama en {OLLAMA_URL} → {VERDE}ollama serve{FIN} • Modelo: {MODEL} → {VERDE}ollama pull {MODEL}{FIN} • Python 3.10+ • pip install chromadb (opcional, para el bloque 6) """) if not check_ollama(): sys.exit(1) # Resolver qué pasos ejecutar if args.step: pasos = set() for parte in args.step.split(","): parte = parte.strip() if "-" in parte: a, b = parte.split("-", 1) pasos.update(range(int(a), int(b) + 1)) else: pasos.add(int(parte)) pasos = sorted(pasos) info(f"Ejecutando pasos: {', '.join(str(p) for p in pasos)}") for p in pasos: ejecutar_paso(p) if os.path.exists(DIR_DOCS): shutil.rmtree(DIR_DOCS) elif args.interactive: info( "Modo interactivo: presiona Enter para avanzar, 's' para saltar, 'q' para salir" ) for n in sorted(PASOS): nombre, fn, esencial = PASOS[n] if args.fast and n == 6: info(f"Paso {n} omitido (--fast)") continue r = ( input(f"\n{NEGR}¿Ejecutar paso {n}: {nombre}?{FIN} [Enter/s/q] ") .strip() .lower() ) if r == "q": info("Interrumpido por el usuario") break if r == "s": info(f"Paso {n} saltado") continue fn() if os.path.exists(DIR_DOCS): shutil.rmtree(DIR_DOCS) else: crear_documentos() busqueda_semantica() clasificacion() deduplicacion() recomendacion() if not args.fast: chroma_demo() else: info("Modo rápido: omitiendo ChromaDB") info("Ejecuta sin --fast para incluir ChromaDB") if os.path.exists(DIR_DOCS): shutil.rmtree(DIR_DOCS) print(f"\n{NEGR}{'=' * 60}{FIN}") print(f"\n{VERDE}🎉 Demo completada{FIN}") print(f" • 6 casos de uso ejecutados") print(f" • 1 modelo de embeddings ({MODEL})") print(f" • 0 modelos entrenados, 0 GPUs, 0€") print( f"\n {AZUL}📖 Escaleta completa en: episodio-817-escaleta-embeddings.md{FIN}" ) print(f" {AZUL}🔗 https://github.com/aejimmi/fail2ban-rs{FIN}") if __name__ == "__main__": main()