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.
464 lines
18 KiB
Python
Executable file
464 lines
18 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Dos JARVIS hablando: uno examina, el otro ejecuta, el primero corrige.
|
|
|
|
python3 ~/COFRE/CODERS/JARVIS/config/autoescuela.py # una tanda
|
|
python3 ~/COFRE/CODERS/JARVIS/config/autoescuela.py --horas 6 # toda la noche
|
|
python3 ~/COFRE/CODERS/JARVIS/config/autoescuela.py --aplicar # guardar mejoras
|
|
|
|
La idea. Los bucles que teniamos probaban trozos: mejora_nocturna comprobaba
|
|
solo si la frase llega a la accion; probar_acciones, solo si la accion corre.
|
|
Esto los junta en una conversacion de punta a punta y por AUDIO:
|
|
|
|
Examinador -> dice una tarea en voz alta (Piper), con ruido y sala
|
|
Jarvis -> la oye (Whisper), decide que accion es, la ejecuta,
|
|
y contesta en voz alta
|
|
Examinador -> comprueba tres cosas:
|
|
1. ¿oyo la tarea correcta? (reconocimiento)
|
|
2. ¿hizo la accion que tocaba? (emparejamiento)
|
|
3. ¿el resultado tiene sentido? (ejecucion y respuesta)
|
|
y pone nota
|
|
|
|
Lo que se ejecuta de verdad: solo lo de solo lectura. Lo que abre cosas, apaga
|
|
o necesita root se comprueba hasta el emparejamiento, sin lanzarlo —una
|
|
autoescuela no deberia apagarte la pantalla ni pedir tu contraseña de noche.
|
|
|
|
Lo que aprende: los fallos se separan por tipo. Un fallo de reconocimiento
|
|
propone un alias (con el mismo filtro seguro de mejora_nocturna: sistematico,
|
|
sin ambiguedad, conserva palabra clave). Un fallo de ejecucion o de respuesta
|
|
se ANOTA para arreglarlo a mano —eso no se toca solo, que es codigo—.
|
|
|
|
Escenarios: casi todos son de un turno (una tarea, una respuesta), sacados del
|
|
catalogo. Unos pocos son de varios turnos para probar lo que tiene estado:
|
|
buscar -> "abre el segundo" -> "busca dentro X". Y hay charla que NO debe
|
|
disparar ninguna accion, como control negativo.
|
|
"""
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from collections import defaultdict
|
|
|
|
CASA = os.path.expanduser("~")
|
|
JARVIS = os.path.join(CASA, "COFRE/CODERS/JARVIS")
|
|
CONFIG = os.path.join(CASA, ".var/app/io.github.qwersyk.Newelle/config")
|
|
APP = "io.github.qwersyk.Newelle//master"
|
|
|
|
PIPER = os.path.join(JARVIS, "voz/piper/piper")
|
|
MODELOS_VOZ = os.path.join(JARVIS, "voz/modelos")
|
|
WHISPER = os.path.join(CONFIG, "models/whisper/whisper.cpp/build/bin/whisper-cli")
|
|
MODELOS_STT = os.path.join(CONFIG, "models/whisper/whisper.cpp/models")
|
|
|
|
CATALOGO = os.path.expanduser(
|
|
"~/.local/share/flatpak/app/io.github.qwersyk.Newelle/x86_64/master/"
|
|
"active/files/share/newelle/acciones/acciones.json")
|
|
CATALOGO_FUENTE = os.path.join(JARVIS, "newelle/data/acciones/acciones.json")
|
|
PROPIO = os.path.join(CONFIG, "jarvis-acciones.json")
|
|
MOTOR = os.path.join(JARVIS, "newelle/src/utility/acciones.py")
|
|
SELECCION = os.path.join(JARVIS, "newelle/src/utility/seleccion.py")
|
|
|
|
SALA = os.path.join(JARVIS, "muestras/autoescuela")
|
|
DIALOGO = os.path.join(SALA, "dialogo.log")
|
|
INFORME = os.path.join(SALA, "informe.json")
|
|
WAVS = os.path.join(CONFIG, "autoescuela_wavs")
|
|
|
|
VOCES = ["es_ES-sharvard-medium", "es_ES-davefx-medium", "es_ES-carlfm-x_low"]
|
|
VELOCIDADES = [0.9, 1.0, 1.12]
|
|
RUIDOS = ["", "aecho=0.8:0.7:22:0.12", "volume=0.72,highpass=f=120",
|
|
"highpass=f=150,lowpass=f=6500"]
|
|
|
|
# Lo que tiene efecto o pide permiso: se comprueba el emparejamiento pero no se
|
|
# ejecuta. Igual que en probar_acciones.
|
|
EFECTOS = ["firefox", "xdg-open", "loginctl", "xset", "pactl set", "pkill",
|
|
"brightnessctl", "systemctl start", "systemctl stop", "systemctl restart",
|
|
"import -window", "gnome-screenshot", "sudo", "rm ", "kill", "dpms"]
|
|
|
|
RELLENOS = ["notas", "el informe", "documentos", "reactor"]
|
|
|
|
|
|
def carga(ruta, nombre):
|
|
spec = importlib.util.spec_from_file_location(nombre, ruta)
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
return m
|
|
|
|
|
|
def log(msg, a_dialogo=False):
|
|
linea = f"[{time.strftime('%H:%M:%S')}] {msg}"
|
|
print(linea, flush=True)
|
|
fichero = DIALOGO if a_dialogo else DIALOGO
|
|
try:
|
|
with open(fichero, "a") as f:
|
|
f.write((msg if a_dialogo else linea) + "\n")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def dice(texto, voz, vel, ruido, destino):
|
|
"""El examinador dice una tarea: la sintetiza a un wav de 16 kHz."""
|
|
crudo = tempfile.mktemp(suffix=".wav")
|
|
try:
|
|
r = subprocess.run(
|
|
[PIPER, "--model", os.path.join(MODELOS_VOZ, voz + ".onnx"),
|
|
"--length_scale", str(1 / vel), "--output_file", crudo],
|
|
input=texto, capture_output=True, text=True, timeout=60)
|
|
if r.returncode != 0 or not os.path.exists(crudo):
|
|
return False
|
|
filtros = "aresample=16000" + ("," + ruido if ruido else "")
|
|
subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-i", crudo,
|
|
"-ac", "1", "-af", filtros, "-y", destino], check=True, timeout=60)
|
|
return True
|
|
except (OSError, subprocess.SubprocessError):
|
|
return False
|
|
finally:
|
|
if os.path.exists(crudo):
|
|
os.remove(crudo)
|
|
|
|
|
|
def oye(wavs, hilos=8, por_tanda=40):
|
|
"""Jarvis oye: transcribe los wav. {fichero: texto}."""
|
|
salida = {}
|
|
ficheros = sorted(wavs)
|
|
for i in range(0, len(ficheros), por_tanda):
|
|
tanda = ficheros[i:i + por_tanda]
|
|
lista = " ".join(f'"{WAVS}/{n}"' for n in tanda)
|
|
orden = (f'for f in {lista}; do echo "### $(basename "$f")"; '
|
|
f'{WHISPER} -m {MODELOS_STT}/ggml-base.bin -l es -nt -t {hilos} '
|
|
f'-f "$f" 2>/dev/null | tr "\\n" " "; echo; done')
|
|
try:
|
|
r = subprocess.run(["flatpak", "run", "--command=bash", APP, "-c", orden],
|
|
capture_output=True, text=True, errors="replace", timeout=1200)
|
|
except subprocess.SubprocessError:
|
|
continue
|
|
actual = None
|
|
for linea in r.stdout.splitlines():
|
|
if linea.startswith("### "):
|
|
actual = linea[4:].strip()
|
|
elif actual:
|
|
salida[actual] = linea.strip()
|
|
actual = None
|
|
return salida
|
|
|
|
|
|
def tiene_efecto(comando):
|
|
return any(m in comando for m in EFECTOS)
|
|
|
|
|
|
def ejecuta_lectura(comando, argumento=""):
|
|
"""Ejecuta un comando de solo lectura y devuelve su salida."""
|
|
c = comando.replace("{argumento}", shlex.quote(argumento))
|
|
c = c.replace("{argumento_url}", shlex.quote(argumento))
|
|
try:
|
|
r = subprocess.run(["bash", "-lc", c], capture_output=True, text=True,
|
|
errors="replace", timeout=15)
|
|
return (r.stdout or r.stderr or "").strip(), r.returncode
|
|
except subprocess.SubprocessError:
|
|
return "", 1
|
|
|
|
|
|
class Ejecutor:
|
|
"""La tuberia de Jarvis: oye una frase, decide y hace. Replica el enrutado
|
|
del panel (seleccion antes que catalogo) para poder probarlo sin GUI."""
|
|
|
|
def __init__(self, motor, seleccion):
|
|
self.motor = motor
|
|
self.sel = seleccion
|
|
self.cat = motor.Catalogo([p for p in (PROPIO, CATALOGO) if os.path.exists(p)]
|
|
or [CATALOGO_FUENTE])
|
|
self.lista = []
|
|
self.fichero = None
|
|
|
|
def recarga(self):
|
|
self.cat = self.motor.Catalogo(
|
|
[p for p in (PROPIO, CATALOGO) if os.path.exists(p)] or [CATALOGO_FUENTE])
|
|
|
|
def procesa(self, oido):
|
|
"""Devuelve (tipo, id_o_dato, salida). tipo:
|
|
'seleccion' | 'dentro' | 'accion' | 'modelo'."""
|
|
norm = self.motor.normaliza(oido)
|
|
|
|
s = self.sel.orden_de_seleccion(norm)
|
|
if s is not None and self.lista:
|
|
que, pos = s
|
|
idx = len(self.lista) - 1 if pos == -1 else pos - 1
|
|
if 0 <= idx < len(self.lista):
|
|
self.fichero = self.lista[idx]
|
|
return "seleccion", f"{que}:{pos}", self.fichero
|
|
return "seleccion", "fuera_de_rango", ""
|
|
|
|
termino = self.sel.orden_buscar_dentro(norm)
|
|
if termino and self.fichero:
|
|
r, _ = ejecuta_lectura(f"grep -in {shlex.quote(termino)} "
|
|
f"{shlex.quote(self.fichero)} | head", "")
|
|
return "dentro", termino, r
|
|
|
|
accion, arg = self.cat.busca(oido)
|
|
if accion is None:
|
|
return "modelo", "", ""
|
|
|
|
# las de lista alimentan self.lista
|
|
if accion.id in ("buscar_fichero", "buscar_carpeta", "grep_recursivo"):
|
|
salida, _ = ejecuta_lectura(accion.comando, arg)
|
|
rutas = [l.split(":", 1)[0] if accion.id == "grep_recursivo" else l
|
|
for l in salida.splitlines() if l.strip()]
|
|
self.lista = rutas
|
|
if rutas:
|
|
self.fichero = rutas[0]
|
|
return "accion", accion.id, salida
|
|
|
|
if accion.confirmar or tiene_efecto(accion.comando):
|
|
return "accion", accion.id, "(no se ejecuta; solo emparejamiento)"
|
|
|
|
salida, _ = ejecuta_lectura(accion.comando, arg)
|
|
return "accion", accion.id, salida
|
|
|
|
|
|
def escenarios(acciones):
|
|
"""Construye las tareas de examen. Cada una: (texto, comprobacion)."""
|
|
tareas = []
|
|
|
|
# un turno por accion: la tarea es su frase, se espera su id
|
|
for a in acciones:
|
|
if not a["frases"]:
|
|
continue
|
|
texto = a["frases"][0]
|
|
if a.get("captura"):
|
|
texto += " " + random.choice(RELLENOS)
|
|
tareas.append({
|
|
"tipo": "accion", "texto": texto, "espera_id": a["id"],
|
|
"turnos": [texto],
|
|
})
|
|
|
|
# multi-turno: probar el estado (lista -> elegir -> buscar dentro)
|
|
tareas.append({
|
|
"tipo": "multi", "texto": "buscar y elegir",
|
|
"turnos": ["busca el fichero reactor", "abre el primero", "busca dentro def"],
|
|
"espera": ["accion:grep_recursivo|accion:buscar_fichero", "seleccion", "dentro"],
|
|
})
|
|
|
|
# control negativo: charla que NO debe disparar accion
|
|
for frase in ["cuentame un chiste sobre gatos", "que opinas de todo esto",
|
|
"estoy cansado hoy la verdad", "no se muy bien que hacer ahora"]:
|
|
tareas.append({"tipo": "charla", "texto": frase, "turnos": [frase]})
|
|
|
|
return tareas
|
|
|
|
|
|
def una_tanda(ejecutor, motor, ronda, muestras):
|
|
acciones = ejecutor.cat.acciones
|
|
base_acciones = json.load(open(CATALOGO if os.path.exists(CATALOGO)
|
|
else CATALOGO_FUENTE))["acciones"]
|
|
tareas = escenarios(base_acciones)
|
|
|
|
# sintetizar todos los turnos
|
|
shutil.rmtree(WAVS, ignore_errors=True)
|
|
os.makedirs(WAVS, exist_ok=True)
|
|
plan = [] # (fichero, tarea_idx, turno_idx, texto)
|
|
n = 0
|
|
for ti, tarea in enumerate(tareas):
|
|
for tj, texto in enumerate(tarea["turnos"]):
|
|
n += 1
|
|
nombre = f"{n:05d}.wav"
|
|
if dice(texto, random.choice(VOCES), random.choice(VELOCIDADES),
|
|
random.choice(RUIDOS), os.path.join(WAVS, nombre)):
|
|
plan.append((nombre, ti, tj, texto))
|
|
|
|
log(f"ronda {ronda}: examinador dice {len(plan)} tareas, jarvis las oye...")
|
|
oidos = oye([p[0] for p in plan])
|
|
|
|
# correr los escenarios en orden (el estado importa en los multi-turno)
|
|
resultados = {"bien": 0, "reconoce_mal": 0, "accion_mal": 0,
|
|
"ejecuta_mal": 0, "charla_colada": 0}
|
|
candidatos = defaultdict(lambda: defaultdict(int))
|
|
por_tarea = defaultdict(list)
|
|
for nombre, ti, tj, texto in plan:
|
|
por_tarea[ti].append((tj, nombre, texto))
|
|
|
|
for ti in sorted(por_tarea):
|
|
tarea = tareas[ti]
|
|
ejecutor.lista = [] # cada escenario empieza limpio
|
|
ejecutor.fichero = None
|
|
for tj, nombre, texto in sorted(por_tarea[ti]):
|
|
oido = oidos.get(nombre, "").strip()
|
|
if not oido:
|
|
continue
|
|
tipo, dato, salida = ejecutor.procesa(oido)
|
|
|
|
# dejar la conversacion en el log, legible
|
|
log(f" examinador: «{texto}»", a_dialogo=True)
|
|
log(f" jarvis oye: «{oido}» -> {tipo}:{dato}", a_dialogo=True)
|
|
|
|
veredicto = verifica(tarea, tj, texto, oido, tipo, dato, salida,
|
|
motor, candidatos)
|
|
log(f" veredicto: {veredicto}\n", a_dialogo=True)
|
|
resultados[veredicto] = resultados.get(veredicto, 0) + 1
|
|
|
|
total = sum(resultados.values())
|
|
bien = resultados["bien"]
|
|
log(f"ronda {ronda}: {bien}/{total} bien · " +
|
|
" · ".join(f"{k} {v}" for k, v in resultados.items() if k != "bien" and v))
|
|
return resultados, candidatos
|
|
|
|
|
|
def verifica(tarea, tj, texto, oido, tipo, dato, salida, motor, candidatos):
|
|
"""Pone nota a un turno y, si es fallo de reconocimiento, anota el candidato."""
|
|
if tarea["tipo"] == "charla":
|
|
return "bien" if tipo == "modelo" else "charla_colada"
|
|
|
|
if tarea["tipo"] == "multi":
|
|
esperado = tarea["espera"][tj]
|
|
actual = f"{tipo}:{dato}"
|
|
if any(actual.startswith(e.split("|")[0].split(":")[0] + ":") and
|
|
(e == actual or e.split(":")[0] == tipo)
|
|
for e in esperado.split("|")):
|
|
# para multi basta con que el TIPO sea el correcto
|
|
if tipo in [e.split(":")[0] for e in esperado.split("|")]:
|
|
return "bien"
|
|
return "accion_mal"
|
|
|
|
# tarea de un turno: se espera un id concreto
|
|
espera_id = tarea["espera_id"]
|
|
if tipo != "accion":
|
|
# no reconocio la orden: candidato a alias si conserva sentido
|
|
candidatos[espera_id][motor.normaliza(oido)] += 1
|
|
return "reconoce_mal"
|
|
if dato != espera_id:
|
|
return "accion_mal"
|
|
# emparejo bien; ¿la ejecucion dio algo con sentido?
|
|
if salida and ("command not found" in salida or salida == "TIMEOUT"):
|
|
return "ejecuta_mal"
|
|
return "bien"
|
|
|
|
|
|
# --- aprender: mismos filtros seguros que mejora_nocturna --------------------
|
|
VACIAS = {"que", "de", "la", "el", "me", "se", "es", "esta", "estan", "hay",
|
|
"tengo", "un", "una", "por", "con", "como", "cuanto", "cuanta"}
|
|
|
|
|
|
def contenido(frase):
|
|
return {p for p in frase.split() if len(p) > 3 and p not in VACIAS}
|
|
|
|
|
|
def alias_seguros(candidatos, acciones, motor, cat):
|
|
porfrase = defaultdict(set)
|
|
for ident, frases in candidatos.items():
|
|
for f in frases:
|
|
porfrase[f].add(ident)
|
|
porid = {a["id"]: a for a in acciones}
|
|
propuestos = defaultdict(list)
|
|
for ident, frases in candidatos.items():
|
|
a = porid.get(ident)
|
|
if a is None or a.get("captura"):
|
|
continue
|
|
orig = motor.normaliza(a["frases"][0] if a["frases"] else "")
|
|
claves = contenido(orig)
|
|
for f, veces in frases.items():
|
|
if veces < 2 or len(porfrase[f]) > 1 or len(f.split()) < 2:
|
|
continue
|
|
if cat.busca(f)[0] is not None or motor.parecido(f, orig) < 0.6:
|
|
continue
|
|
pal = set(f.split())
|
|
if claves and not any(
|
|
c in pal or any(motor.parecido(c, w) > 0.8 for w in pal)
|
|
for c in claves):
|
|
continue
|
|
propuestos[ident].append((f, veces))
|
|
return propuestos
|
|
|
|
|
|
def guarda_alias(propuestos, acciones):
|
|
try:
|
|
propio = json.load(open(PROPIO))
|
|
except (OSError, json.JSONDecodeError):
|
|
propio = {"version": 1, "acciones": []}
|
|
porid = {a["id"]: a for a in propio.get("acciones", [])}
|
|
orig = {a["id"]: a for a in acciones}
|
|
n = 0
|
|
for ident, lista in propuestos.items():
|
|
e = porid.get(ident)
|
|
if e is None:
|
|
e = dict(orig[ident]); porid[ident] = e
|
|
propio.setdefault("acciones", []).append(e)
|
|
for f, _ in lista:
|
|
if f not in e["frases"]:
|
|
e["frases"].append(f); n += 1
|
|
json.dump(propio, open(PROPIO, "w"), indent=2, ensure_ascii=False)
|
|
return n
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--horas", type=float, default=0)
|
|
ap.add_argument("--aplicar", action="store_true", help="guardar los alias que encuentre")
|
|
ap.add_argument("--muestras", type=int, default=1)
|
|
args = ap.parse_args()
|
|
|
|
for ruta, q in ((PIPER, "piper"), (WHISPER, "whisper")):
|
|
if not os.path.exists(ruta):
|
|
print(f"falta {q}"); return 1
|
|
|
|
os.makedirs(SALA, exist_ok=True)
|
|
motor = carga(MOTOR, "acciones")
|
|
sel = carga(SELECCION, "seleccion")
|
|
ejecutor = Ejecutor(motor, sel)
|
|
|
|
limite = time.time() + args.horas * 3600 if args.horas else None
|
|
log("=" * 56)
|
|
log("arranca la autoescuela (examinador + jarvis)")
|
|
|
|
historial = []
|
|
ronda = 0
|
|
total_alias = 0
|
|
while True:
|
|
ronda += 1
|
|
ini = time.time()
|
|
try:
|
|
resultados, candidatos = una_tanda(ejecutor, motor, ronda, args.muestras)
|
|
except Exception as e:
|
|
log(f"ronda {ronda}: error, se salta — {type(e).__name__}: {e}")
|
|
time.sleep(5); continue
|
|
|
|
# aprender de los fallos de reconocimiento
|
|
base = json.load(open(CATALOGO if os.path.exists(CATALOGO)
|
|
else CATALOGO_FUENTE))["acciones"]
|
|
propuestos = alias_seguros(candidatos, base, motor, ejecutor.cat)
|
|
nuevos = sum(len(v) for v in propuestos.values())
|
|
if nuevos and args.aplicar:
|
|
total_alias += guarda_alias(propuestos, base)
|
|
ejecutor.recarga()
|
|
log(f"ronda {ronda}: +{nuevos} alias aplicados")
|
|
elif nuevos:
|
|
log(f"ronda {ronda}: {nuevos} alias propuestos (usa --aplicar para guardarlos)")
|
|
|
|
total = sum(resultados.values())
|
|
historial.append({
|
|
"ronda": ronda,
|
|
"acierto": round(100 * resultados["bien"] / max(1, total), 1),
|
|
"fallos": {k: v for k, v in resultados.items() if k != "bien" and v},
|
|
"minutos": round((time.time() - ini) / 60, 1),
|
|
})
|
|
json.dump({"actualizado": time.strftime("%Y-%m-%d %H:%M"),
|
|
"rondas": ronda, "alias_aplicados": total_alias,
|
|
"historial": historial[-30:]},
|
|
open(INFORME, "w"), indent=2, ensure_ascii=False)
|
|
|
|
if not args.horas:
|
|
break
|
|
if time.time() > limite:
|
|
log("se acabo el tiempo. Fin."); break
|
|
|
|
shutil.rmtree(WAVS, ignore_errors=True)
|
|
log(f"terminado tras {ronda} rondas · ultimo acierto "
|
|
f"{historial[-1]['acierto'] if historial else '?'} %")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|