Modelos pequeños en Ollama: Llama 3.2, Qwen2.5, Phi-3 y Gemma 2

# Datos y preparación que el video da por hechos: ejecuta esto antes del resto del código.
# Requiere Ollama con bge-m3 y los cinco modelos: ollama pull llama3.2:1b llama3.2:3b qwen2.5:3b phi3:mini gemma2:2b
import io, json, re, textwrap, urllib.request, zipfile
import chromadb, ollama

BASE = "https://pypro-assets.sfo3.digitaloceanspaces.com/cursos/rag-con-ollama/datos"
NO_ESTA = "No está en la documentación."
MODELO = "llama3.2:3b"
MODELOS = ['llama3.2:1b', 'llama3.2:3b', 'qwen2.5:3b', 'phi3:mini', 'gemma2:2b']
OPC = {"temperature": 0, "seed": 42, "num_predict": 400}


def cargar_corpus(url=f"{BASE}/pyshop-kb.zip"):
    """Descarga el zip y devuelve una lista de documentos (dict con doc, titulo, tipo, area, texto)."""
    datos = urllib.request.urlopen(url).read()
    docs = []
    with zipfile.ZipFile(io.BytesIO(datos)) as z:
        for nombre in sorted(n for n in z.namelist() if n.endswith(".md")):
            campos, texto = quitar_front_matter(z.read(nombre).decode("utf-8"))
            docs.append({"doc": nombre.split("/")[-1], "titulo": campos["titulo"],
                         "tipo": campos["tipo"], "area": campos["area"], "texto": texto})
    return docs


def cargar_preguntas(url=f"{BASE}/preguntas.jsonl"):
    lineas = urllib.request.urlopen(url).read().decode("utf-8").splitlines()
    return [json.loads(l) for l in lineas if l.strip()]


def quitar_front_matter(crudo):
    """Separa el bloque `---` inicial: devuelve (metadatos, texto)."""
    meta, texto = re.match(r"---\n(.*?)\n---\n(.*)", crudo, re.S).groups()
    campos = dict(linea.split(": ", 1) for linea in meta.splitlines())
    return campos, texto.strip()


def secciones_de(doc):
    """Parte el texto por encabezados `##`: lista de (seccion, texto). La parte antes del
    primer `##` (título e introducción) se llama «Introducción»."""
    partes = re.split(r"^(?=## )", doc["texto"], flags=re.M)
    salida = []
    for parte in partes:
        if not parte.strip():
            continue
        m = re.match(r"## (.+)", parte)
        salida.append((m.group(1).strip() if m else "Introducción", parte.strip()))
    return salida


def chunk_fijo(texto, tam=800, solape=100):
    """Ventana fija de `tam` caracteres que avanza `tam - solape`."""
    paso = tam - solape
    return [texto[i:i + tam] for i in range(0, max(len(texto) - solape, 1), paso)]


def _chunk(doc, texto, seccion, n):
    return {"id": f"{doc['doc']}#{n}", "texto": texto, "doc": doc["doc"],
            "titulo": doc["titulo"], "tipo": doc["tipo"], "area": doc["area"],
            "seccion": seccion}


def chunk_markdown(doc, tam=800, solape=100):
    """Chunking por defecto del curso: una sección `##` por chunk y, si pasa de `tam`
    caracteres, ventana fija `tam`/`solape` dentro de la sección."""
    chunks = []
    for seccion, texto in secciones_de(doc):
        trozos = [texto] if len(texto) <= tam else chunk_fijo(texto, tam, solape)
        for trozo in trozos:
            chunks.append(_chunk(doc, trozo, seccion, len(chunks)))
    return chunks


def chunks_markdown(docs, tam=800, solape=100):
    return [c for doc in docs for c in chunk_markdown(doc, tam, solape)]


def prompt_rag(pregunta, chunks):
    """Prompt de RAG: solo el contexto, citas [doc#sección] y salida explícita si no está.
    (Las reglas numeradas con un ejemplo de cita hacen que los modelos de 3B citen mucho más
    que una instrucción suelta.)"""
    contexto = "\n\n".join(f"Fuente: [{c['doc']}#{c['seccion']}]\n{c['texto']}" for c in chunks)
    return (
        "Eres el asistente interno de PyShop. Responde la pregunta usando solo el contexto de abajo.\n"
        "Reglas:\n"
        "1. Al final de cada oración pon entre corchetes la fuente que la respalda, copiada tal cual, "
        "por ejemplo: El envío es gratis desde 999 pesos [envios-nacionales.md#Costos de envío].\n"
        f"2. Si el contexto no contiene la respuesta, responde solo: «{NO_ESTA}»\n\n"
        f"Contexto:\n{contexto}\n\n"
        f"Pregunta: {pregunta}"
    )


# Colección del video de ChromaDB: chunking por defecto + bge-m3 (160 chunks en la carpeta pyshop_chroma)
docs = cargar_corpus()
chunks = chunks_markdown(docs)
P = {p["id"]: p for p in cargar_preguntas()}
cliente = chromadb.PersistentClient(path="pyshop_chroma")
if "pyshop_kb" in [c.name for c in cliente.list_collections()]:
    cliente.delete_collection("pyshop_kb")
col = cliente.create_collection("pyshop_kb", metadata={"hnsw:space": "cosine"})
textos = [c["texto"] for c in chunks]
E = []
for i in range(0, len(textos), 64):
    E += ollama.embed(model="bge-m3", input=textos[i:i + 64]).embeddings
META = ("doc", "titulo", "tipo", "area", "seccion")
col.add(ids=[c["id"] for c in chunks], documents=textos, embeddings=E,
        metadatas=[{k: c[k] for k in META} for c in chunks])


def buscar(pregunta, k=4):
    """Top-k chunks (bge-m3): lista de dict con id, texto, distancia y metadatos."""
    q = ollama.embed(model="bge-m3", input=pregunta).embeddings
    r = col.query(query_embeddings=q, n_results=k)
    return [{"id": i, "texto": t, "distancia": d, **m}
            for i, t, d, m in zip(r["ids"][0], r["documents"][0],
                                  r["distances"][0], r["metadatas"][0])]


def chat(prompt, modelo=MODELO):
    """Una llamada a Ollama con temperatura 0, semilla 42 y tope de 400 tokens."""
    mensajes = prompt if isinstance(prompt, list) else [{"role": "user", "content": prompt}]
    return ollama.chat(model=modelo, messages=mensajes, options=OPC).message.content


def ver(texto):
    print(textwrap.fill(texto, 33, break_on_hyphens=False))


# Del video de prompts (755): extraer y verificar citas [doc#sección]
PATRON = r"\[([^\[\]]+\.md#[^\[\]]+)\]"


def citas(r):
    return re.findall(PATRON, r)


def verificar(r, ch):
    ok = {c["doc"] + "#" + c["seccion"] for c in ch}
    return [x for x in citas(r) if x not in ok]

for m in MODELOS:
    d = ollama.show(m).details
    print(f"{m:11}", d.family,
          d.parameter_size,
          d.quantization_level)

for m in MODELOS:
    chat("Hola", m)  # carga el modelo
    for x in ollama.ps().models:
        if x.model == m:
            gb = x.size_vram / 1e9
            print(f"{m:11}", round(gb, 2))

IDS = ["q01", "q03", "q05", "q08", "q26",
       "q33", "q46", "q55", "q58", "q60"]
REC = {q: buscar(P[q]["pregunta"])
       for q in IDS}

def responder(q, m):
    p = P[q]["pregunta"]
    return chat(prompt_rag(p, REC[q]), m)

def calificar(r, ch):
    if verificar(r, ch):
        return "inventada"
    if NO_ESTA in r:
        return "no está"
    return "citada" if citas(r) else "sin cita"
r = responder("q01", "llama3.2:1b")
ver(calificar(r, REC["q01"]) + ": " + r)

L = {"citada": "C", "no está": "N",
     "sin cita": "S", "inventada": "I"}
for m in MODELOS:
    f = [L[calificar(responder(q, m), REC[q])]
         for q in IDS]
    print(f"{m:11}", *f[:7], "|", *f[7:])

for m in MODELOS:
    r = responder("q60", m)
    print(f"{m:11}", r[:40])