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.
310 lines
13 KiB
Python
310 lines
13 KiB
Python
"""Acciones rapidas: caminos predefinidos que se saltan el modelo.
|
|
|
|
Por que existe esto. Preguntar "cuanto espacio queda" y esperar a que un modelo
|
|
de 4B razone, decida llamar a una herramienta, la ejecute y redacte la
|
|
respuesta cuesta segundos. Pero esa pregunta no necesita pensarse: es un `df`.
|
|
El catalogo empareja la frase con un comando y contesta con una plantilla, y
|
|
eso baja de segundos a decimas.
|
|
|
|
No es entrenar un modelo: es una tabla. Y esa es la gracia, porque una tabla es
|
|
predecible, se lee, se corrige y no alucina. Lo que no encaje en la tabla sigue
|
|
yendo al modelo como siempre.
|
|
|
|
Formato del catalogo (JSON):
|
|
|
|
{
|
|
"acciones": [
|
|
{
|
|
"id": "disco_libre",
|
|
"frases": ["cuanto espacio queda", "espacio en disco"],
|
|
"acuse": "Un momento, lo miro.",
|
|
"comando": "df -h / | tail -1",
|
|
"host": true,
|
|
"respuesta": "Quedan {campo4} libres de {campo2}.",
|
|
"confirmar": false
|
|
}
|
|
]
|
|
}
|
|
|
|
En `respuesta` se puede usar:
|
|
{salida} la salida completa, recortada
|
|
{primera} la primera linea
|
|
{ultima} la ultima linea no vacia
|
|
{campoN} la columna N de la ultima linea (1 en adelante)
|
|
{lineas} cuantas lineas ha devuelto
|
|
"""
|
|
import difflib
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import urllib.parse
|
|
import unicodedata
|
|
|
|
UMBRAL_PARECIDO = 0.86 # por debajo empieza a confundir ordenes
|
|
TIEMPO_LIMITE = 20 # ningun atajo deberia tardar mas
|
|
SALIDA_MAXIMA = 600 # lo que se guarda para enseñar y para la plantilla
|
|
|
|
|
|
def normaliza(texto: str) -> str:
|
|
"""Minusculas, sin tildes, sin puntuacion y con los espacios apretados.
|
|
|
|
Whisper devuelve "¿Cuánto espacio queda?" y el catalogo dice "cuanto
|
|
espacio queda". Sin normalizar no casaria ni una.
|
|
|
|
Lo de apretar los espacios no es cosmetico: al sustituir la puntuacion por
|
|
un espacio, "cuanto, espacio queda" quedaba con dos seguidos y dejaba de
|
|
coincidir con la frase del catalogo. El sintoma era un asistente que a
|
|
veces no reacciona a una orden que ha entendido bien, segun donde el
|
|
reconocedor decida poner una coma.
|
|
"""
|
|
texto = unicodedata.normalize("NFD", texto.lower())
|
|
texto = "".join(c for c in texto if unicodedata.category(c) != "Mn")
|
|
return " ".join(re.sub(r"[^\w\s]", " ", texto).split())
|
|
|
|
|
|
# El envoltorio de cortesia con el que habla cualquiera. Nadie dice "pon la
|
|
# pelicula" a secas: dice "necesito que pongas la pelicula" o "puedes ponerme".
|
|
# Con eso delante, ninguna frase del catalogo encaja como subcadena y la orden
|
|
# se pierde entera.
|
|
#
|
|
# Medido sobre las 462 frases que el panel anoto por no entender: de las 18 que
|
|
# eran ordenes de verdad, la mayoria fallaba solo por esto. Cinco de ellas eran
|
|
# el mismo intento de poner una pelicula, dicho de cinco maneras.
|
|
CORTESIA = re.compile(
|
|
r"^(por favor,?\s+)?"
|
|
r"(necesito que( me)?|quiero que( me)?|me gustaria que( me)?|"
|
|
r"puedes|podrias|me puedes|me podrias|te importa|hazme el favor de|"
|
|
r"a ver si|va(le|)?,? |anda,? |porfa,? |quiero|necesito)\s+",
|
|
re.I)
|
|
|
|
|
|
class Accion:
|
|
def __init__(self, datos: dict):
|
|
self.id = datos.get("id", "")
|
|
self.frases = [normaliza(f) for f in datos.get("frases", [])]
|
|
self.acuse = datos.get("acuse", "Voy.")
|
|
self.comando = datos.get("comando", "")
|
|
self.host = bool(datos.get("host", True))
|
|
self.respuesta = datos.get("respuesta", "{salida}")
|
|
self.confirmar = bool(datos.get("confirmar", False))
|
|
self.captura = bool(datos.get("captura", False))
|
|
self.grupo = datos.get("grupo", "")
|
|
|
|
def coincide(self, texto: str):
|
|
"""(longitud de la frase que encaja, lo que venia detras).
|
|
|
|
Se devuelve la longitud y no un si/no para poder quedarse con la
|
|
coincidencia mas especifica: "abre la carpeta de descargas" debe ganar
|
|
a "abre la carpeta" cuando las dos encajan.
|
|
|
|
El resto es el argumento: en "busca gatos en youtube" la frase es
|
|
"busca" ... y lo demas es lo que hay que buscar. Solo se recoge si la
|
|
accion lo pide, para que una frase suelta no arrastre basura.
|
|
"""
|
|
mejor, resto = 0, ""
|
|
for f in self.frases:
|
|
if not f or f not in texto:
|
|
continue
|
|
if len(f) > mejor:
|
|
mejor = len(f)
|
|
resto = texto.split(f, 1)[1].strip() if self.captura else ""
|
|
return mejor, resto
|
|
|
|
|
|
def parecido(a: str, b: str) -> float:
|
|
"""Cuanto se parecen dos cadenas, de 0 a 1, por letra y por sonido.
|
|
|
|
Para los casi-aciertos del reconocedor: "yo tuve" por "yo tube", "casis"
|
|
por "oasis". Se usa difflib y no una distancia de edicion propia porque
|
|
viene en la biblioteca estandar y aqui no hay que hilar tan fino.
|
|
|
|
Se queda con el MAYOR de los dos parecidos, el de la letra y el del sonido.
|
|
Muchos fallos del reconocedor no son de significado sino de ortografia
|
|
—"Abre van Kamp" por "abre bandcamp", "poops" por "pubs"— y por escrito
|
|
quedan lejos aunque suenen igual. Medido contra el banco: sube el acierto
|
|
de 79,3 a 84,7 % sin que ninguna de las 50 frases de charla dispare nada.
|
|
|
|
Que sea el maximo y no la media importa: la media hundiria los aciertos
|
|
normales, donde la letra ya casaba bien y el sonido no aporta.
|
|
"""
|
|
literal = difflib.SequenceMatcher(None, a, b).ratio()
|
|
if literal >= 0.98:
|
|
return literal # ya casa: no hace falta el sonido
|
|
from .fonetica import fonetica
|
|
return max(literal,
|
|
difflib.SequenceMatcher(None, fonetica(a), fonetica(b)).ratio())
|
|
|
|
|
|
class Catalogo:
|
|
def __init__(self, rutas):
|
|
self.acciones = []
|
|
self.errores = []
|
|
for ruta in rutas:
|
|
if ruta and os.path.exists(ruta):
|
|
self._carga(ruta)
|
|
|
|
def _carga(self, ruta):
|
|
try:
|
|
with open(ruta) as f:
|
|
datos = json.load(f)
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
self.errores.append(f"{os.path.basename(ruta)}: {e}")
|
|
return
|
|
|
|
vistos = {a.id for a in self.acciones}
|
|
for entrada in datos.get("acciones", []):
|
|
accion = Accion(entrada)
|
|
if not accion.id or not accion.frases:
|
|
self.errores.append(f"{os.path.basename(ruta)}: accion sin id o sin frases")
|
|
continue
|
|
# el primero que se carga manda: asi el catalogo del usuario puede
|
|
# redefinir uno del paquete sin tener que borrarlo
|
|
if accion.id not in vistos:
|
|
self.acciones.append(accion)
|
|
vistos.add(accion.id)
|
|
|
|
def busca(self, texto: str):
|
|
"""(accion mas especifica que encaje, argumento) o (None, "").
|
|
|
|
Si no encaja nada, se reintenta quitando la cortesia de delante. Se
|
|
hace como SEGUNDA pasada y no normalizando desde el principio a
|
|
proposito: asi lo que ya funcionaba sigue funcionando igual, y esto
|
|
solo añade aciertos donde antes no habia ninguno.
|
|
"""
|
|
encontrada, argumento = self._busca_directa(normaliza(texto))
|
|
if encontrada is not None:
|
|
return encontrada, argumento
|
|
sin_cortesia = CORTESIA.sub("", normaliza(texto)).strip()
|
|
if sin_cortesia and sin_cortesia != normaliza(texto):
|
|
return self._busca_directa(sin_cortesia)
|
|
return None, ""
|
|
|
|
def _busca_directa(self, texto: str):
|
|
mejor, puntos, argumento = None, 0, ""
|
|
for accion in self.acciones:
|
|
p, resto = accion.coincide(texto)
|
|
if p > puntos:
|
|
mejor, puntos, argumento = accion, p, resto
|
|
|
|
# Nada exacto: se prueba por parecido. El reconocedor se come letras
|
|
# ("yo tuve" por "yo tube") y sin esto la orden se pierde entera. El
|
|
# umbral es alto a proposito: mas abajo empieza a confundir ordenes
|
|
# distintas entre si, que es peor que no reaccionar.
|
|
if mejor is None:
|
|
self._resto_parecido = ""
|
|
mejor, puntos = self._por_parecido(texto)
|
|
if mejor is not None and mejor.captura:
|
|
argumento = self._resto_parecido
|
|
|
|
# una accion que captura y no recibe nada no sirve: "busca" a secas
|
|
if mejor is not None and mejor.captura and not argumento:
|
|
return None, ""
|
|
return mejor, argumento
|
|
|
|
def _por_parecido(self, texto: str, umbral: float = UMBRAL_PARECIDO):
|
|
"""La accion cuya frase mas se parezca, si pasa del umbral."""
|
|
palabras = texto.split()
|
|
mejor, mejor_p = None, umbral
|
|
for accion in self.acciones:
|
|
if accion.captura:
|
|
# Con argumento no vale comparar la frase entera: "busca en
|
|
# youtube conciertos de techno" no se parece a "busca en
|
|
# youtube". Se compara solo el arranque y lo demas es el
|
|
# argumento. Se exigen tres palabras para que un disparador
|
|
# corto no se coma media conversacion.
|
|
for frase in accion.frases:
|
|
trozos = frase.split()
|
|
# Dos palabras, no tres. Con tres, disparadores cortos como
|
|
# "receta de" no entraban NUNCA en esta fase, y "receta de
|
|
# documentos" mal oido se perdia entero. El riesgo de bajar
|
|
# a dos lo cubre que la comparacion va anclada al principio:
|
|
# solo encaja si el arranque suena a la frase.
|
|
if len(trozos) < 2 or len(palabras) <= len(trozos):
|
|
continue
|
|
arranque = " ".join(palabras[:len(trozos)])
|
|
p = parecido(arranque, frase)
|
|
if p > mejor_p:
|
|
mejor, mejor_p = accion, p
|
|
self._resto_parecido = " ".join(palabras[len(trozos):])
|
|
continue
|
|
|
|
# Frase ENTERA contra frase entera, no por ventana deslizante. La
|
|
# ventana encontraba un sub-trozo que puntuaba alto aunque la orden
|
|
# fuera otra: "que carne el tiempo" (mal oido de "que kernel tengo")
|
|
# colaba en la accion del tiempo con 0.90, y se ejecutaba la accion
|
|
# equivocada —lo peor que puede pasar en algo que lanza comandos—.
|
|
# Una orden se dice completa; si va incrustada en una frase larga,
|
|
# el emparejador exacto (por subcadena) ya la coge antes de llegar
|
|
# aqui. Comparar el todo con el todo deja fuera esos falsos
|
|
# positivos sin perder los casi-aciertos reales.
|
|
propio = " ".join(palabras)
|
|
for frase in accion.frases:
|
|
if not frase:
|
|
continue
|
|
p = parecido(propio, frase)
|
|
if p > mejor_p:
|
|
mejor, mejor_p = accion, p
|
|
return mejor, mejor_p
|
|
|
|
def __len__(self):
|
|
return len(self.acciones)
|
|
|
|
|
|
def ejecuta(accion: Accion, prefijo_host, argumento: str = "") -> tuple:
|
|
"""Corre el comando y devuelve (texto para decir, salida cruda).
|
|
|
|
prefijo_host es lo que hay que anteponer para salir del sandbox; se pasa
|
|
como argumento en vez de importarlo para poder probar esto sin flatpak.
|
|
"""
|
|
if not accion.comando:
|
|
return accion.acuse, ""
|
|
|
|
# El argumento viene de un microfono, o sea de fuera: va entrecomillado con
|
|
# shlex antes de tocar un shell. Sin esto, decir "borra punto y coma rm"
|
|
# seria una orden ejecutable.
|
|
comando = accion.comando.replace("{argumento}", shlex.quote(argumento))
|
|
comando = comando.replace("{argumento_url}",
|
|
shlex.quote(urllib.parse.quote_plus(argumento)))
|
|
|
|
orden = ["bash", "-lc", comando]
|
|
if accion.host:
|
|
orden = list(prefijo_host) + orden
|
|
|
|
try:
|
|
# errors="replace": un grep o un cat que toque un binario devuelve
|
|
# bytes que no son utf-8, y sin esto la orden entera reventaria con
|
|
# UnicodeDecodeError. Mejor un caracter raro que un fallo.
|
|
r = subprocess.run(orden, capture_output=True, text=True,
|
|
errors="replace", timeout=TIEMPO_LIMITE)
|
|
salida = (r.stdout or r.stderr or "").strip()
|
|
except subprocess.TimeoutExpired:
|
|
return "La orden ha tardado demasiado y la he cancelado.", ""
|
|
except OSError as e:
|
|
return f"No he podido ejecutarlo: {e}", ""
|
|
|
|
return formatea(accion.respuesta, salida, argumento), salida[:SALIDA_MAXIMA]
|
|
|
|
|
|
def formatea(plantilla: str, salida: str, argumento: str = "") -> str:
|
|
lineas = [l for l in salida.splitlines() if l.strip()]
|
|
ultima = lineas[-1] if lineas else ""
|
|
campos = ultima.split()
|
|
|
|
valores = {
|
|
"argumento": argumento,
|
|
"salida": salida[:SALIDA_MAXIMA] if salida else "sin resultado",
|
|
"primera": lineas[0] if lineas else "",
|
|
"ultima": ultima,
|
|
"lineas": len(lineas),
|
|
}
|
|
for i, campo in enumerate(campos, start=1):
|
|
valores[f"campo{i}"] = campo
|
|
|
|
try:
|
|
return plantilla.format(**valores)
|
|
except (KeyError, IndexError, ValueError):
|
|
# plantilla que pide un campo que no existe: mejor decir la salida
|
|
# cruda que soltar una excepcion en mitad de una conversacion
|
|
return valores["salida"]
|