RAG con SQLite y Ollama, base de conocimiento desde cero

Vistas: 1
0:00 / 0:00
RAG con SQLite y Ollama, base de conocimiento desde cero

Llevo quince años escribiendo notas, artículos y tutoriales. Quince años acumulando conocimiento en forma de archivos markdown, repartidos por mi vault de Obsidian como si fueran las fichas de una biblioteca de Babel. El problema es que cuando alguien me pregunta oye, ¿tú no hablaste de cómo configurar certificados SSL en Traefik? me quedaba en blanco. Sabía que lo había escrito. Sabía que estaba ahí, en algún sitio. Pero rastrear cinco mil archivos a mano para encontrar una aguja en un pajar digital… no, gracias. Y de eso va exactamente el episodio de hoy, el primero de dos capítulos sobre RAG —Retrieval-Augmented Generation— donde vamos a construir una base de conocimiento 100% local, con SQLite y Ollama, para que puedas preguntarle a tu propio vault de notas en lenguaje natural. Sin ChatGPT, sin nubes, sin enviar tus datos a nadie. Vamos al turrón.

El problema: quince años de notas sin buscar

Imagina la escena. Llevas desde 2010 escribiendo. Artículos para el blog, tutoriales, notas rápidas, apuntes de conferencias, fragmentos de código que te funcionaron, comandos que no quieres olvidar. Con los años, eso se convierte en cinco mil, ocho mil, quizá diez mil archivos markdown. Y en mi caso, todo eso está en mi vault de notas, que no es más que un directorio con subdirectorios, archivos .md y mucho frontmatter YAML. El problema no es tener la información. El problema es encontrarla.

Puedes usar grep, por supuesto. O ripgrep si eres elegante. Pero grep busca cadenas de texto, no conceptos. Si buscas cómo configurar systemd timers y escribiste timers en systemd, grep te lo encuentra. Pero si escribiste programar tareas con systemd o automatizar servicios con temporizadores, grep se queda mudo. Necesitas búsqueda semántica, no solo textual. Y ahí entra el RAG.

Pero ojo, que no estamos hablando de esos sistemas RAG que ves en las demos de OpenAI, donde mandas un PDF a la nube y te responden como si fueras el presidente de la junta directiva. No. Estamos hablando de RAG local, de tu máquina, tus datos, tus modelos. Sin que nadie más meta las narices.

La solución: SQLite + FTS5 + Ollama, 100% local

Después de darle muchas vueltas —y de probar varias alternativas que te cuento más abajo—, he decidido simplificarlo al máximo. Nada de montar un PostgreSQL, nada de servidores, nada de APIs externas. El stack es tan sencillo que casi parece mentira:

  • SQLite con FTS5 para búsqueda textual — que ya viene incluido en Python, no necesitas instalar nada raro
  • Ollama con el modelo bge-m3 para generar embeddings
  • Python para orquestarlo todo
  • Y unos cuantos miles de documentos markdown como fuente de conocimiento

El flujo es este: escaneamos los ficheros *.md, extraemos el frontmatter YAML, troceamos el contenido en fragmentos manejables, generamos embeddings de cada fragmento con Ollama, y lo guardamos todo en SQLite. Luego montamos un índice FTS5 para poder buscar por texto. Y todo, absolutamente todo, corre en local.

¿El resultado? Una base de datos de conocimiento que, en el próximo episodio (el 821), vamos a consultar con búsqueda híbrida: lo mejor del texto exacto con FTS5 y lo mejor de la semántica con vectores. Pero eso será en el 821. Hoy toca sudar la gota gorda con el pipeline.

¿Por qué este stack y no otro?

Seguro que te estás preguntando por qué no usar PostgreSQL con pgvector, o Chroma, o Qdrant. Y te diré que lo he probado casi todo. PostgreSQL es una maravilla, pero para esto es como poner un reactor nuclear en una bicicleta. Tienes que montar el servidor, configurar la extensión, gestionar conexiones, hacer backups… para acabar indexando unos pocos miles de documentos. Chroma y Qdrant son estupendos, pero son servicios adicionales que mantener.

SQLite, en cambio, es un solo fichero. Lo copias, lo mueves, lo versionas con git. No necesita servidor, no necesita demonio, no necesita configuración. Arranca en milisegundos. Y con FTS5 tienes búsqueda textual de primera clase sin instalar nada extra.

Para 5000 documentos —que no es big data ni de lejos—, SQLite va sobrado. El límite teórico de SQLite son 140 TB. Créeme, no vas a llegar.

Preparación del entorno: Ollama, uv y dependencias

Lo primero de todo: asegurarnos de que tenemos las herramientas. Si no tienes Ollama instalado, ya sabes, a la web de ollama.com o al gestor de paquetes de tu distro. En Linux es tan sencillo como

curl -fsSL https://ollama.com/install.sh | sh

Una vez instalado, comprobamos que está vivo:

curl http://localhost:11434/api/tags

Si te devuelve un JSON con un array de modelos, perfecto. Si te da error, arranca Ollama con ollama serve o con systemd.

Ahora, el modelo de embeddings. Usamos bge-m3 de BAAI (Beijing Academy of Artificial Intelligence). Es un modelo multilenguaje de 568 millones de parámetros, genera vectores de 1024 dimensiones, y lo bueno es que ya viene normalizado —la norma L2 es 1, lo que simplifica mucho la búsqueda por similitud de coseno. Ocupa un poco más de un gigabyte en disco.

ollama pull bge-m3

Mientras se descarga —que tarda un rato, son 1.1 GB—, vamos instalando las dependencias Python. Yo he preparado un fichero requirements.txt:

ollama>=0.4.0
numpy>=2.0.0
pyyaml>=6.0
tiktoken>=0.7.0

Fíjate en algo importante: no necesitamos LangChain. Ni frameworks raros. Python stdlib más cuatro librerías. Eso es todo. ¿Por qué cuatro y no veinte? Porque el chunking lo vamos a hacer nosotros con regex y lógica simple, no necesitamos la navaja suiza de LangChain para un problema que se resuelve con treinta líneas de código.

Ah, un detalle importante. Comprueba la versión de SQLite de tu Python:

python3 -c "import sqlite3; print(sqlite3.sqlite_version)"

Necesitas la 3.9.0 o superior para FTS5. La mía es la 3.53.3, así que voy sobrado. Si tu Python lleva una versión antigua de SQLite —pasa en algunos sistemas antiguos—, puedes instalar pysqlite3-binario o actualizar Python.

Chunking: cómo trocear los documentos

Y ahora llega lo divertido. O lo tedioso, según se mire: el chunking. ¿Por qué hay que trocear los documentos? Por dos razones. Primero, porque los modelos de embeddings tienen un límite de contexto. El bge-m3 aguanta hasta 8192 tokens, pero si mandas un documento entero de diez mil palabras, te vas a quedar corto. Y segundo, porque cuando busques, quieres fragmentos precisos, no un ladrillazo de tres páginas.

La estrategia que uso es híbrida: respetamos las cabeceras markdown como fronteras naturales, y si una sección es demasiado larga, la subdividimos por tamaño con solapamiento. Si es demasiado corta, la fusionamos con la siguiente.

Vamos a escribir la función de chunking. Abre tu editor favorito —o Neovim, que sabemos que molas más— y crea un fichero rag_chunker.py.

Lo primero, detectar las cabeceras markdown con una regex:

import re

def chunk_markdown(texto, max_tokens=512, overlap=128):
    patron = r'(^|\n)(#{1,3})\s+(.+?)(?=\n|$)'
    splits = list(re.finditer(patron, texto))

Si el documento no tiene cabeceras —que pasa, hay markdowns planos—, hacemos un chunking simple por tamaño con solapamiento. Pero si las tiene, dividimos por cada cabecera y procesamos cada sección.

La función completa queda más o menos así:

def chunk_markdown(texto, max_tokens=512, overlap=128):
    patron = r'(^|\n)(#{1,3})\s+(.+?)(?=\n|$)'
    splits = list(re.finditer(patron, texto))

    if not splits:
        return chunk_por_tamano(texto, max_tokens, overlap)

    secciones = []
    for i, match in enumerate(splits):
        inicio = match.start()
        fin = splits[i+1].start() if i+1 < len(splits) else len(texto)
        secciones.append({
            'cabecera': match.group(0).strip(),
            'nivel': len(match.group(2)),
            'texto': texto[inicio:fin].strip()
        })

    chunks = []
    buffer = []
    buffer_tokens = 0

    for sec in secciones:
        tok_sec = contar_tokens(sec['texto'])

        if tok_sec > max_tokens:
            if buffer:
                chunks.append('\n\n'.join(buffer))
                buffer = []
                buffer_tokens = 0
            sub = chunk_por_tamano(sec['texto'], max_tokens, overlap,
                                   prefix=sec['cabecera'] + '\n')
            chunks.extend(sub)
        elif buffer_tokens + tok_sec <= max_tokens:
            buffer.append(sec['texto'])
            buffer_tokens += tok_sec
        else:
            chunks.append('\n\n'.join(buffer))
            buffer = [sec['texto']]
            buffer_tokens = tok_sec

    if buffer:
        chunks.append('\n\n'.join(buffer))

    return chunks

Fíjate en los detalles. Si una sección tiene más de 512 tokens, la partimos con solapamiento de 128. Eso es un 25% de solapamiento, suficiente para no perder contexto entre un chunk y el siguiente. Si viene una sección pequeña, la acumulamos en un buffer y la fusionamos con la siguiente. Así evitamos chunks de tres palabras que no aportan nada.

La función auxiliar chunk_por_tamano es simple pero efectiva:

def chunk_por_tamano(texto, max_tokens, overlap, prefix=""):
    tokens = texto.split()
    chunks = []
    for i in range(0, len(tokens), max_tokens - overlap):
        trozo = tokens[i:i + max_tokens]
        chunk_text = prefix + ' '.join(trozo) if prefix and i == 0 else ' '.join(trozo)
        chunks.append(chunk_text)
        prefix = ""
        if i + max_tokens >= len(tokens):
            break
    return chunks

Ya, ya sé que texto.split() no es lo mismo que contar tokens de verdad. No es preciso. Pero para un 80% de los casos funciona perfectamente. Si quieres precisión, puedes usar tiktoken, que es la librería de OpenAI para contar tokens:

import tiktoken

def contar_tokens(texto):
    enc = tiktoken.get_encoding("cl100k_base")
    return len(enc.encode(texto))

O si prefieres una aproximación rápida sin dependencias, vale len(texto) // 4. En español, un token son aproximadamente cuatro caracteres. No es exacto, pero para el chunking nos sirve. Ponytail mode, que le llaman 😅

Frontmatter YAML: limpiar antes de trocear

Todos mis artículos tienen frontmatter YAML. Ese bloque entre --- al principio del fichero que guarda el título, la fecha, las etiquetas. Algo así:

---
title: "Empaquetar y desempaquetar con tar"
created: "2020-06-03T16:00:00+00:00"
tags:
    - ubuntu
---

Eso no lo queremos en los chunks. El frontmatter son metadatos, no contenido. Así que antes de chunkear, lo extraemos y lo guardamos aparte. Luego el contenido limpio va a los chunks, y los metadatos a la tabla de documentos.

Tenemos dos opciones. La opción A: instalar python-frontmatter, que es una librería que hace exactamente esto y además maneja edge cases como archivos con BOM o YAML mal formado.

import frontmatter
post = frontmatter.load("archivo.md")
print(post.content)      # texto sin frontmatter
print(post.metadata)     # dict con title, tags, etc

La opción B: hacerlo a mano con regex y yaml.safe_load. Así no necesitas instalar nada más que pyyaml, que total ya lo tienes del requirements.txt.

import re, yaml

def extraer_frontmatter(texto):
    patron = r'^---\s*\n(.*?)\n---\s*\n'
    match = re.match(patron, texto, re.DOTALL)
    if match:
        try:
            metadata = yaml.safe_load(match.group(1)) or {}
        except yaml.YAMLError:
            metadata = {}
        contenido = texto[match.end():]
        return metadata, contenido
    return {}, texto

¿Cuál elijo? Para el proyecto, la manual. Es más didáctica, no oculta la magia. Y además, si algo falla, sabes exactamente dónde mirar.

Esta función la vamos a usar en el pipeline principal. Cada archivo que procesamos, primero extraemos frontmatter, luego chunkeamos el contenido limpio. Los metadatos van a la tabla documentos, los chunks a la tabla chunks.

Estructura de la base de datos SQLite

Tenemos los chunks, tenemos los metadatos. Ahora toca meterlo en una base de datos que aguante el tirón. Y para eso, SQLite con FTS5.

El esquema lo he dividido en tres tablas:

Primera: documentos. Un registro por cada fichero markdown. Guardamos la ruta, el hash MD5 para detectar cambios, y la fecha de última indexación.

CREATE TABLE IF NOT EXISTS documentos (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    path TEXT UNIQUE NOT NULL,
    md5_hash TEXT NOT NULL,
    last_indexed TEXT DEFAULT (datetime('now'))
);

Segunda: chunks. Los fragmentos de texto. Cada chunk pertenece a un documento, tiene un índice dentro de ese documento, el contenido, y metadatos como el título y las etiquetas.

CREATE TABLE IF NOT EXISTS chunks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    doc_id INTEGER NOT NULL,
    chunk_index INTEGER NOT NULL,
    content TEXT NOT NULL,
    tokens INTEGER,
    doc_path TEXT NOT NULL,
    title TEXT,
    tags TEXT,
    created_at TEXT DEFAULT (datetime('now')),
    FOREIGN KEY (doc_id) REFERENCES documentos(id) ON DELETE CASCADE
);

Fíjate en la clave foránea con ON DELETE CASCADE. Si borramos un documento, se borran todos sus chunks. Eso nos ahorra tener que andar limpiando manualmente.

Tercera: la tabla virtual FTS5. Aquí viene la magia.

CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(
    content,
    tokenize='porter unicode61 remove_diacritics 1',
    content='chunks',
    content_rowid='id'
);

Un par de detalles importantes. El tokenize lo configuro con porter unicode61 remove_diacritics 1. Vamos por partes. El unicode61 separa por caracteres Unicode —ideal para español, con sus tildes y eñes— y remove_diacritics quita los acentos para que buscar configuración te encuentre también configuracion. El porter es un stemmer de inglés que… bueno, para español no hace gran cosa, pero no estorba.

¿Y por qué no uso un tokenizador específico para español? Porque habría que escribirlo en C y compilarlo como extensión dinámica de SQLite. Para el 99% de los casos, unicode61 con remove_diacritics hace un trabajo más que aceptable. Las búsquedas en español funcionan, las tildes no son un problema, y la precisión es alta. Si necesitas stemming en español de verdad —que reduzca corriendo, corría y correré a la raíz corr-—, puedes añadirlo después en la capa de aplicación, o directamente esperar a la búsqueda semántica del episodio 821, que es inmune a estas variaciones morfológicas.

Y el parámetro content='chunks' es clave. Esto hace que FTS5 sea una tabla de contenido externo: no duplica el texto, lo lee de la tabla chunks. Esto ahorra aproximadamente la mitad del espacio en disco. La contrapartida es que hay que sincronizar manualmente.

¿Cómo se sincroniza? Con triggers. Después de insertar, actualizar o borrar chunks, los triggers mantienen el índice FTS5 al día:

CREATE TRIGGER chunks_ai AFTER INSERT ON chunks
BEGIN
    INSERT INTO chunks_fts(rowid, content) VALUES (new.id, new.content);
END;

CREATE TRIGGER chunks_ad AFTER DELETE ON chunks
BEGIN
    INSERT INTO chunks_fts(chunks_fts, rowid, content) VALUES('delete', old.id, old.content);
END;

CREATE TRIGGER chunks_au AFTER UPDATE ON chunks
BEGIN
    INSERT INTO chunks_fts(chunks_fts, rowid, content) VALUES('delete', old.id, old.content);
    INSERT INTO chunks_fts(rowid, content) VALUES (new.id, new.content);
END;

Y al final del pipeline, un rebuild general por si acaso:

def rebuild_fts5(conn):
    conn.executescript("INSERT INTO chunks_fts(chunks_fts) VALUES('rebuild');")

Sin este paso, el índice FTS5 se queda desactualizado. Es un error clásico: insertas datos, haces consultas FTS5 y no encuentras nada. Te vuelves loco, miras las tablas, los datos están ahí… pero FTS5 no los ve. Porque no has hecho el rebuild. Ya sabes, error de desincronización, que lo llaman los modernos. Yo lo llamo olvidarme y perder veinte minutos 🤦

Cuarta tabla: embeddings. Los vectores numéricos de cada chunk.

CREATE TABLE IF NOT EXISTS embeddings (
    chunk_id INTEGER PRIMARY KEY,
    vector BLOB NOT NULL,
    model TEXT DEFAULT 'bge-m3',
    dimensions INTEGER DEFAULT 1024,
    FOREIGN KEY (chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
);

Los embeddings se guardan como BLOB binarios. Cada vector de 1024 floats ocupa exactamente 4096 bytes (1024 × 4 bytes por float). Para empaquetar y desempaquetar:

import struct

def pack_vector(vec):
    return struct.pack(f'{len(vec)}f', *vec)

def unpack_vector(data):
    n = len(data) // 4
    return list(struct.unpack(f'{n}f', data))

Si te parece que guardar vectores en SQLite es una chapuza…

¿Por qué bge-m3 y no otro modelo?

Buena pregunta. bge-m3 de BAAI es, a día de hoy, uno de los mejores modelos de embeddings multilenguaje que puedes ejecutar en local. Con 568 millones de parámetros, genera vectores de 1024 dimensiones y soporta más de 100 idiomas, incluido el español. Esto último es importante porque muchos modelos de embeddings populares están entrenados principalmente con texto en inglés y dan resultados mediocres con otros idiomas.

Otra ventaja clave: los vectores que genera bge-m3 vienen normalizados (la norma L2 es exactamente 1). Esto significa que la similitud por coseno se reduce a un producto punto, que es mucho más rápido de calcular:

def cosine_similarity(a, b):
    return float(np.dot(a, b))  # solo si están normalizados

Alternativas que consideré: nomic-embed-text (137M params, 768 dimensiones) es más pequeño y rápido, pero pierde calidad en español. all-MiniLM-L6-v2 (23M params) es rapidísimo pero solo inglés. mxbai-embed-large (334M params) va bien, pero bge-m3 le gana en benchmarks multilenguaje.

Para el contexto del episodio, con ~5000 documentos y ~30-40K chunks, bge-m3 es el punto óptimo entre calidad y velocidad. En CPU, generar todos los embeddings lleva entre 2 y 4 horas. En una GPU básica, entre 20 y 40 minutos. La paciencia tiene recompensa.

Pipeline incremental: solo lo que cambia

Vale, tenemos el esquema, tenemos el chunking, tenemos los embeddings. Pero hay un detalle: no vamos a reprocesar los cinco mil documentos cada vez. Eso sería una locura. Necesitamos un pipeline incremental: que detecte qué archivos son nuevos, cuáles han cambiado y cuáles se han borrado, y solo reprocese los afectados.

Para detectar cambios, uso MD5 hash. Calculamos el hash de cada archivo y lo comparamos con el que tenemos guardado en la base de datos. Si coincide, no se toca. Si no coincide, se reprocesa.

import hashlib

def compute_md5(path):
    h = hashlib.md5()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()

La función get_doc_status recorre todos los markdowns, calcula su MD5 y los compara con los registros de la base de datos. Devuelve tres listas: nuevos, modificados y eliminados.

def get_doc_status(db, files):
    cur = db.execute("SELECT path, md5_hash FROM documentos")
    conocidos = {row[0]: row[1] for row in cur.fetchall()}

    actual = {}
    for f in files:
        md5 = compute_md5(f["path"])
        actual[f["path"]] = md5

    nuevos = [r for r in actual if r not in conocidos]
    modificados = [r for r in actual if conocidos.get(r) != actual[r]]
    eliminados = list(set(conocidos.keys()) - set(actual.keys()))

    return {"nuevos": nuevos, "modificados": modificados, "eliminados": eliminados}

Para recorrer los archivos, una función simple que excluye .git, node_modules y otras carpetas que no nos interesan:

import os

def walk_markdowns(root_dir):
    exclude = {'.git', 'node_modules', '__pycache__', 'venv', '.venv', '.opencode'}
    for root, dirs, files in os.walk(root_dir):
        dirs[:] = [d for d in dirs if d not in exclude]
        for f in files:
            if f.endswith('.md'):
                yield os.path.join(root, f)

Luego, para cada archivo nuevo o modificado, llamamos a reindex_file, que elimina los chunks viejos, actualiza los metadatos y regenera los nuevos chunks y embeddings.

def reindex_file(db, file_path):
    md5 = compute_md5(file_path)
    with open(file_path, encoding="utf-8", errors="replace") as f:
        texto = f.read()

    metadata, contenido = strip_frontmatter(texto)
    if not contenido.strip():
        return 0

    chunks = chunk_markdown(contenido, max_tokens=512, overlap=128)
    if not chunks:
        return 0

    # Verificar si el documento ya existe
    cur = db.execute("SELECT id FROM documentos WHERE path = ?", (file_path,))
    row = cur.fetchone()

    if row:
        doc_id = row[0]
        limpiar_chunks_por_doc_id(db, doc_id)
        db.execute("UPDATE documentos SET md5_hash = ?, last_indexed = datetime('now') WHERE id = ?",
                   (md5, doc_id))
    else:
        cur = db.execute("INSERT INTO documentos (path, md5_hash) VALUES (?, ?)",
                         (file_path, md5))
        doc_id = cur.lastrowid

    # Insertar chunks
    for idx, chunk in enumerate(chunks):
        db.execute("""INSERT INTO chunks
                       (doc_id, chunk_index, content, tokens, doc_path, title, tags)
                       VALUES (?, ?, ?, ?, ?, ?, ?)""",
                   (doc_id, idx, chunk["content"], chunk["tokens"],
                    relpath, chunk.get("title") or titulo, tags_json))

    db.commit()

    # Generar embeddings en batch
    textos = [c["content"] for c in chunks]
    embeddings = get_embeddings_batch(textos, batch_size=16)

    # Guardar embeddings
    for chunk_id, vec in zip(chunk_ids, embeddings):
        blob = pack_vector(vec)
        db.execute("INSERT INTO embeddings (chunk_id, vector, model, dimensions) VALUES (?, ?, ?, ?)",
                   (chunk_id, blob, MODELO_EMBEDDINGS, len(vec)))

    db.commit()
    return len(chunks)

Y al final de todo, un rebuild del índice FTS5 para que esté sincronizado. Así, cada vez que ejecutes el pipeline, solo se procesan los cambios. La primera ejecución tardará un rato —con CPU, unas dos o tres horas para cinco mil documentos—, pero las siguientes serán cuestión de segundos.

Demo en vivo: consultas y resultados

Vale, hemos montado todo el tinglado. Ahora vamos a verificar que funciona. Para eso, un script sencillo que haga una consulta FTS5 y nos devuelva los chunks más relevantes.

import sqlite3
import sys

db = sqlite3.connect("rag_conocimiento.db")

query = sys.argv[1] if len(sys.argv) > 1 else "systemd timers"

cur = db.execute("""
    SELECT
        c.id, c.doc_path, c.title,
        snippet(chunks_fts, 0, '<<<', '>>>', '...', 50) AS fragmento,
        rank, c.tokens
    FROM chunks_fts
    JOIN chunks c ON chunks_fts.rowid = c.id
    WHERE chunks_fts MATCH ?
    ORDER BY rank
    LIMIT 10
""", (query,))

for row in cur:
    print(f"\n📄 {row[2]}")
    print(f"   {row[1]}")
    print(f"   {row[3]}")
    print(f"   Score: {row[4]:.2f}")

Y lo ejecutamos:

python rag_consulta.py "cómo configurar systemd timers"

La función snippet() es maravillosa. Te devuelve un fragmento del texto con los términos de búsqueda resaltados —yo le he puesto <<< y >>>— y con contexto alrededor. Es lo que hace que las búsquedas sean útiles de verdad.

La columna rank es el score BM25 que calcula FTS5 automáticamente. A menor valor, mejor coincidencia. Los resultados vienen ordenados por relevancia. Y como hemos puesto LIMIT 10, nos quedamos con los diez mejores.

¿Y qué pasa si la consulta tiene varias palabras? FTS5 soporta operadores:

  • "frase exacta" entre comillas
  • termino AND otro — las dos palabras
  • termino OR otro — cualquiera de las dos
  • term* — búsqueda por prefijo

Así que puedes hacer cosas como:

python rag_consulta.py "docker AND compose AND volumes"

Y te saca todos los chunks donde aparecen esos tres términos. No está mal para una base de datos sin servidor, ¿eh? 🔥

Casos de uso reales

Para que te hagas una idea de lo que puedes hacer con esto, te pongo algunos ejemplos de mi día a día.

Buscar artículos que escribiste sobre un tema concreto. Hace unos meses necesitaba recordar cómo configuraba los certificados SSL en Traefik con Let’s Encrypt. Una consulta rápida: python rag_consulta.py "traefik AND letsencrypt AND certificados". En menos de un segundo, los tres artículos que había escrito sobre el tema, con el fragmento exacto donde explicaba la configuración.

Encontrar comandos que sabes que existen pero no recuerdas. Vale, ¿cómo era eso de montar un volumen tmpfs en Docker? python rag_consulta.py "docker tmpfs mount". Y ahí estaba, en una nota de 2021, el comando exacto con la sintaxis que necesitaba.

Recuperar fragmentos de código de tutoriales antiguos. A veces escribo un tutorial y meses después necesito el bloque de código concreto, no todo el contexto. Con FTS5 y la función snippet() tienes el fragmento exacto con el código alrededor.

Saber si ya has escrito sobre algo antes de repetirte. Esto es clave para mantener el blog. Si voy a escribir sobre un tema, primero consulto la base de conocimiento. Si ya lo traté, enlazo al artículo anterior en lugar de repetirme. Si no, adelante con el nuevo.

Y todo esto, insisto, sin enviar una sola petición a internet. Todo corre en tu máquina, con tus datos, bajo tu control.

Errores comunes y cómo solucionarlos

Antes de soltarte a tu aire, un pequeño checklist de cosas que probablemente se te van a olvidar. Porque a mí me ha pasado con todas y cada una de ellas.

Uno: Comprueba que Ollama está corriendo antes de lanzar el pipeline. Un curl http://localhost:11434/api/tags y te aseguras. Si no, el script se va a quejar y no vas a entender por qué.

Dos: No te olvides del commit después de las inserciones. SQLite no hace autocommit por defecto. Si no llamas a db.commit(), todo lo que insertaste se pierde. Y no, no te va a avisar.

Tres: El rebuild de FTS5. Te he visto venir. Metes datos, haces la consulta, no sale nada, y empiezas a debuggear. Haz el rebuild. Siempre.

Cuatro: Algunos archivos pueden tener el frontmatter YAML mal formado. Captura la excepción con un try/except y salta esos archivos. Un fichero con errores no debería parar todo el pipeline.

try:
    metadata = yaml.safe_load(match.group(1)) or {}
except yaml.YAMLError:
    print(f"  ⚠️ Frontmatter inválido, saltando")
    return {}, texto

Cinco: Excluye archivos binarios. Parece obvio, pero si tu walk_markdowns no filtra por extensión *.md, te vas a encontrar con imágenes, PDFs, o el archivo .bash_history que tienes perdido por ahí. Solo *.md.

Seis: La primera ejecución va a tardar. Prepárate un café, sal a pasear, o escucha otro episodio del podcast —el 809, por ejemplo— mientras los embeddings se generan. Con CPU, calcula unas dos o tres horas para cinco mil documentos. Si tienes GPU, en veinte minutos está.

Siete: Si cambias de modelo de embeddings, verifica que los vectores sigan normalizados. bge-m3 los devuelve con norma L2 = 1, pero otros modelos puede que no. Si los vectores no están normalizados, la similitud por coseno no funciona como esperas.

Ventajas de hacerlo tú mismo frente a usar LangChain

Buena pregunta. Porque no lo necesitas. LangChain es un framework fantástico si estás haciendo cosas complejas con cadenas de agentes, herramientas y memoria conversacional. Pero para un pipeline de indexación de documentos, es como usar un lanzamisiles para matar una mosca.

Con seis scripts Python y ~1300 líneas de código tienes todo lo necesario. Sin dependencias pesadas, sin capas de abstracción que ocultan lo que pasa, sin tener que aprender una API nueva cada tres meses.

Mira, esta es la tabla comparativa:

AspectoEste proyectoLangChain
Dependencias4 librerías15+ (langchain, langchain-text-splitters, etc.)
FlexibilidadControl totalAlta pero compleja
Curva de aprendizajeBajaMedia-alta
Chunking markdown30 líneas de regexMarkdownHeaderTextSplitter
TokenizacióntiktokenRecursiveCharacterTextSplitter
DebuggearAbres el script y vesCapas sobre capas de abstracción

Además, hacerlo tú mismo te da control total. Sabes exactamente cómo se trocean los documentos, cómo se generan los embeddings, cómo se almacenan. Cuando algo falla —y va a fallar—, sabes dónde mirar. No tienes que leer la documentación de LangChain para entender por qué un chunk se ha partido por un sitio que no esperabas.

Eso sí, si ya tienes un proyecto con LangChain y quieres añadirle esta funcionalidad, no te voy a decir que lo tires. Pero para empezar desde cero, hacerlo a pelo es más didáctico, más ligero y más divertido.

Lo que viene en el episodio 821

Y hasta aquí el episodio de hoy. Hemos construido la base de conocimiento: pipeline de chunking, extracción de frontmatter, esquema SQLite con FTS5, generación de embeddings con Ollama y bge-m3, y detección incremental de cambios. No está mal para una tarde de domingo, ¿eh?

El key takeaway de hoy es este: con Python, SQLite y Ollama tienes todo lo necesario para construir una base de conocimiento RAG 100% local. Sin dependencias externas, sin nubes, sin que tus notas acaben siendo el training data de otro.

En el próximo episodio, el 821, subimos de nivel. Vamos a implementar búsqueda semántica de verdad con cosine similarity, combinaremos los resultados de FTS5 con los de los embeddings para tener búsqueda híbrida, y como guinda del pastel, haremos un plugin para Neovim para consultar la base de conocimiento sin salir del editor. Sí, has oído bien. Un plugin para Neovim.

Todo el código está disponible en los scripts del episodio. Si te ha gustado, ya sabes, comparte, deja un comentario, o simplemente pruébalo y cuéntame qué tal. Y si tienes alguna duda, ya sabes donde encontrarme.

Nos escuchamos en el episodio 821. ¡Hasta entonces! 🐧


Más información

Deja una respuesta