JARVIS/nucleo/saber/enriquece/sintetiza_catalogo.py
sito 7c5381eea7 saber: lo instalado en la maquina y los docs de repos clonados entran al indice
instalado.py lee los .desktop y los paquetes para saber que hay en el equipo y
para que sirve cada cosa. indexa.py lo engancha, y con el los docs markdown de
las herramientas clonadas en CODERS. Las dos fuentes son personales: se quedan
en el indice de la maquina y no entran en el publico.

busca_cli.py expone el mismo motor que usa el cerebro (saber.busca) para
consultar el indice desde la terminal, con --n y --json.

reentrena_noche.sh encadena las fases del reentrenamiento largo y las deja
corriendo horas; cada fase es reanudable por su cuenta.

eval del indice vivo: hit@1 43/57 (75%), hit@5 55/57 (96%), sim 0.909.
2026-08-18 12:14:57 +02:00

135 lines
5.5 KiB
Python

#!/usr/bin/env python3
"""Genera preguntas-gancho para el pack de conocimiento y el catalogo de awesome.
./sintetiza_catalogo.py procesa todo (reanudable, ~1-2 h)
./sintetiza_catalogo.py --muestra 5 ensena 5 y para
Misma idea multi-representacion que sintetiza.py, pero sobre el contenido que
VIENE con el repo (conocimiento/ + awesome lists) en vez de sobre la cosecha de
COFRE. Por que importa: el indice PUBLICO pre-construido trae los fragmentos en
crudo pero SIN preguntas-gancho, asi que un clon recien hecho no tiene el salto de
recall que dan (48->82 % hit@1 medido en la metodologia). Esto lo genera para que
si venga horneado.
Respeta la marca `publico` de cada fragmento: los ganchos heredan esa marca, y
indexa.py mete en el indice publico solo los publicos (los de awesome-hacking, sin
licencia, se quedan en el personal).
Reanudable: apunta en un .hecho por que fragmento iba. El contenido publico se
procesa PRIMERO, para que si se corta, lo que se sube ya este.
"""
import argparse
import json
import os
import sys
AQUI = os.path.dirname(os.path.abspath(__file__))
RAIZ = os.path.dirname(os.path.dirname(AQUI))
SALIDA = os.path.join(RAIZ, "datos", "sintesis-catalogo.jsonl")
MARCA = SALIDA + ".hecho"
# Reutiliza el modelo, la llamada a ollama y el reanudado de sintetiza.py.
sys.path.insert(0, RAIZ) # nucleo/ en el path para saber.*
from saber import awesome # noqa: E402
from saber import indexa # noqa: E402 (_conocimiento)
from saber.enriquece import sintetiza # noqa: E402
# Instruccion mas general que la de sintetiza.py (aquello era solo pentest).
sintetiza.INSTRUCCION = """Eres un generador de preguntas para un buscador de
herramientas y software. Te doy un fragmento con una lista de programas o tecnicas
de una categoria, cada uno con una breve descripcion. Devuelve de UNA a TRES
preguntas, en castellano, que una persona haria por voz buscando algo de esa
categoria y que este fragmento ayuda a responder.
Reglas:
- Preguntas naturales: "que programa hay para...", "con que puedo montar un...",
"cual es una alternativa libre a...", "que herramienta sirve para...".
- Piensa en la NECESIDAD (que quiere hacer), no en nombres concretos.
- NO respondas, NO copies nombres ni comandos: solo las preguntas.
- Si el fragmento es ruido (un titulo suelto, sin descripciones), lista vacia.
Responde SOLO con JSON: {"preguntas": ["...", "..."]}"""
def _fragmentos():
"""Conocimiento (publico) primero, luego catalogo publico, y al final el
personal (awesome-hacking). Cada uno con su marca `publico`."""
frags = []
for e in indexa._conocimiento(): # el pack del repo: todo publico
e = dict(e)
e["publico"] = True
frags.append(e)
for e in awesome.entradas(): # catalogo: ya trae `publico`
frags.append(e)
for e in indexa._instalado(): # apps de esta maquina: personal
e = dict(e)
e["publico"] = False
frags.append(e)
for e in indexa._docs_extra(): # docs de strix, etc.: personal
e = dict(e)
e["publico"] = False
frags.append(e)
frags.sort(key=lambda d: (0 if d.get("publico") else 1,
0 if d.get("tipo") == "metodologia" else 1))
return frags
def main():
p = argparse.ArgumentParser()
p.add_argument("--muestra", type=int, default=0)
a = p.parse_args()
frags = _fragmentos()
modelo = sintetiza._modelo()
npub = sum(1 for f in frags if f.get("publico"))
print(f" {len(frags)} fragmentos ({npub} publicos) · modelo {modelo}", flush=True)
if a.muestra:
for fr in frags[:a.muestra]:
print(f"\n ── {fr.get('fichero')} ──")
for q in sintetiza._pide(fr["texto"], modelo):
print(f" P: {q}")
return 0
desde = 0
if os.path.exists(MARCA) and os.path.exists(SALIDA):
try:
desde = int(open(MARCA).read().strip()) + 1
except (ValueError, OSError):
desde = 0
modo = "a" if desde else "w"
hechas = sum(1 for _ in open(SALIDA)) if desde and os.path.exists(SALIDA) else 0
if desde:
print(f" reanudando desde {desde} ({hechas} preguntas)", flush=True)
total = len(frags)
with open(SALIDA, modo, encoding="utf-8", buffering=1) as out:
for i in range(desde, total):
if i % 25 == 0:
modelo = sintetiza._modelo()
fr = frags[i]
for q in sintetiza._pide(fr["texto"], modelo):
out.write(json.dumps({
"tipo": "pregunta",
"herramienta": fr.get("herramienta"),
"perfil": fr.get("perfil"),
"fichero": fr.get("fichero"),
"tema": fr.get("tema", ""),
"fuente": fr.get("fuente"),
"licencia": fr.get("licencia"),
"publico": bool(fr.get("publico")),
"texto": fr["texto"],
"indexar": q,
}, ensure_ascii=False) + "\n")
hechas += 1
with open(MARCA, "w") as m:
m.write(str(i))
if (i + 1) % 25 == 0:
print(f" PROGRESO catalogo hecho={i+1} total={total} "
f"preguntas={hechas}", flush=True)
print(f"\n hecho: {hechas} preguntas en {SALIDA}", flush=True)
return 0
if __name__ == "__main__":
sys.exit(main())