8e4bc8ad94
Nucleo propio: oye con whisper.cpp, piensa con un modelo de Ollama, habla con Piper, y hace RAG sobre los apuntes del usuario. 100% local, sin cuentas ni claves. Escrito bajo una restriccion dura, 4 GB de VRAM: el cerebro y whisper comparten tarjeta y solo caben porque estan dimensionados para ello. El RAG usa embeddings estaticos con busqueda hibrida; la voz clonada se sirve de una cache de frases. Incluye instalador (install.sh), requisitos, y documentacion del stack, del manejo de root y de las acciones. Los apuntes indexados y el diario NO se incluyen: son privados y el .gitignore los bloquea.
113 lines
4.1 KiB
Python
Executable file
113 lines
4.1 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Construye el indice candidato uniendo lo viejo, lo cosechado y lo sintetizado.
|
|
|
|
./reindexa.py escribe datos/saber.jsonl.nuevo (NO toca el vivo)
|
|
|
|
## Que une
|
|
|
|
1. El indice vivo (datos/saber.jsonl): comandos de Linux, glosario, fichas,
|
|
servidores... todo lo que cosecha no produce. Se conserva entero.
|
|
2. cosecha.jsonl: la metodologia de pentesting que faltaba.
|
|
3. sintesis.jsonl: preguntas en castellano que apuntan a esos fragmentos.
|
|
|
|
## Indexacion multi-representacion
|
|
|
|
Cada entrada se EMBEBE por su campo `indexar` si lo tiene (las preguntas
|
|
sintetizadas), y si no por `texto` (todo lo demas). Asi una entrada de pregunta
|
|
se compara con la consulta del usuario pregunta-contra-pregunta, pero lo que se
|
|
le muestra al cerebro sigue siendo el fragmento literal de `texto`. busca.py no
|
|
cambia: solo lee `texto` y `v`.
|
|
|
|
## No destruye nada
|
|
|
|
Escribe a `saber.jsonl.nuevo`. Quien decide si sustituye al vivo es evalua.py +
|
|
correr_noche.sh, y solo si mide que es mejor, guardando antes una copia.
|
|
|
|
## Deduplicado
|
|
|
|
Por los primeros 120 caracteres del texto normalizado (para las entradas
|
|
normales) o de la pregunta (para las sintetizadas). OSCP_APUNTES duplica apuntes
|
|
de primer nivel; esto los colapsa.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import unicodedata
|
|
|
|
AQUI = os.path.dirname(os.path.abspath(__file__))
|
|
RAIZ = os.path.dirname(os.path.dirname(AQUI))
|
|
DATOS = os.path.join(RAIZ, "datos")
|
|
# La base son las entradas que cosecha NO produce (comandos Linux, glosario,
|
|
# fichas). Por defecto el indice vivo, pero se puede fijar con SABER_BASE al
|
|
# indice ORIGINAL para una reconstruccion limpia que no apile sintesis sobre
|
|
# sintesis y no infle el indice a cada pasada.
|
|
VIVO = os.environ.get("SABER_BASE") or os.path.join(DATOS, "saber.jsonl")
|
|
COSECHA = os.path.join(DATOS, "cosecha.jsonl")
|
|
SINTESIS = os.path.join(DATOS, "sintesis.jsonl")
|
|
NUEVO = os.path.join(DATOS, "saber.jsonl.nuevo")
|
|
|
|
MODELO = "minishlab/potion-multilingual-128M"
|
|
|
|
|
|
def _norm(t):
|
|
t = unicodedata.normalize("NFD", (t or "").lower())
|
|
t = "".join(c for c in t if unicodedata.category(c) != "Mn")
|
|
return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9 ]", " ", t)).strip()
|
|
|
|
|
|
def _lee(ruta, quita_v=False):
|
|
if not os.path.exists(ruta):
|
|
return
|
|
with open(ruta, encoding="utf-8") as f:
|
|
for l in f:
|
|
try:
|
|
d = json.loads(l)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if quita_v:
|
|
d.pop("v", None)
|
|
yield d
|
|
|
|
|
|
def main():
|
|
entradas, vistos = [], set()
|
|
fuentes = [("vivo", VIVO, True), ("cosecha", COSECHA, False),
|
|
("sintesis", SINTESIS, False)]
|
|
cuenta = {}
|
|
for nombre, ruta, quita in fuentes:
|
|
n = 0
|
|
for d in _lee(ruta, quita_v=quita):
|
|
clave_txt = d.get("indexar") or d.get("texto", "")
|
|
firma = _norm(clave_txt)[:120]
|
|
if not firma or firma in vistos:
|
|
continue
|
|
vistos.add(firma)
|
|
entradas.append(d)
|
|
n += 1
|
|
cuenta[nombre] = n
|
|
print(f" {nombre:9s} {n:6d} entradas nuevas", flush=True)
|
|
|
|
print(f"\n total tras deduplicar: {len(entradas)}", flush=True)
|
|
|
|
print(" cargando el modelo de embeddings...", flush=True)
|
|
from model2vec import StaticModel
|
|
modelo = StaticModel.from_pretrained(MODELO)
|
|
|
|
# se embebe `indexar` si existe (la pregunta), si no `texto`
|
|
a_embeber = [(e.get("indexar") or e["texto"]) for e in entradas]
|
|
print(f" calculando {len(a_embeber)} vectores...", flush=True)
|
|
vectores = modelo.encode(a_embeber, show_progress_bar=False)
|
|
|
|
with open(NUEVO, "w", encoding="utf-8") as f:
|
|
for e, v in zip(entradas, vectores):
|
|
e["v"] = [round(float(x), 5) for x in v]
|
|
f.write(json.dumps(e, ensure_ascii=False) + "\n")
|
|
print(f" escrito: {NUEVO} ({os.path.getsize(NUEVO)//1024} KB)", flush=True)
|
|
print(f" RESUMEN vivo={cuenta['vivo']} cosecha={cuenta['cosecha']} "
|
|
f"sintesis={cuenta['sintesis']} total={len(entradas)}", flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|