6a81cef3be
- Seleccion de modelo: JARVIS cambia su cerebro en caliente (elegir_modelo). - Ventanas en Wayland: backend para Sway e Hyprland (ventanas-wayland.sh). - ROADMAP.md y CONTRIBUTING.md: tareas acotadas y como empezar, para animar commits (mas gestores, mas acciones, probar otros modelos, el RAG). - docs/rag.md: tabla explicita de que se publica y que no del RAG; nota de que un clon recien hecho ya tiene RAG util (man pages + glosario) sin datos privados. - README enlaza el roadmap y menciona la eleccion de modelo.
411 lines
18 KiB
Python
411 lines
18 KiB
Python
"""Pensar: hablar con ollama y dejarle usar las manos.
|
|
|
|
Newelle dedica 578 lineas a esto porque soporta quince motores distintos y
|
|
tiene que traducir el formato de herramientas de cada uno. Nosotros hablamos
|
|
solo con ollama, que trae tool calling nativo en `/api/chat`.
|
|
|
|
## Dos herramientas, no una
|
|
|
|
La primera version tenia una sola —"ejecuta una orden"— y salio mal de una
|
|
forma instructiva. Medido con tres modelos:
|
|
|
|
4B "abre una terminal" -> ejecuta_orden("xterm -e bash")
|
|
9B "cuanto espacio queda" -> ejecuta_orden("df -h / | grep ...")
|
|
9B "abre una terminal" -> ejecuta_orden("open -a Terminal") (¡macOS!)
|
|
|
|
Los modelos son serviciales: si les dejas un hueco donde cabe un comando, lo
|
|
escriben. Y el 9B, que es mas listo, lo hacia MAS —escribia pipelines enteras y
|
|
hasta ordenes de otro sistema operativo—.
|
|
|
|
La solucion no es rogarle al modelo en el prompt, es que el hueco no quepa. Se
|
|
parten en dos herramientas con fronteras claras:
|
|
|
|
`orden_del_catalogo` recibe castellano, se lo come el emparejador. La
|
|
descripcion lleva ejemplos REALES sacados del catalogo,
|
|
no inventados, y dice explicitamente que no acepta
|
|
comandos.
|
|
`comando_de_shell` para lo que no esta en el catalogo. Aqui SI se espera
|
|
un comando, asi que el modelo tiene donde poner lo que
|
|
queria poner, y deja de meterlo donde no toca.
|
|
|
|
## Por que el catalogo va antes que esto
|
|
|
|
Las 150 acciones se emparejan ANTES de llamar a ningun modelo. Esto solo se
|
|
usa para lo que no encaja —alrededor del 15 %— y para conversar. Por eso un
|
|
cerebro lento duele menos de lo que parece: la mayoria de las ordenes ni pasan
|
|
por aqui.
|
|
"""
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
ENDPOINT = os.environ.get("JARVIS_OLLAMA", "http://127.0.0.1:11434")
|
|
MODELO = os.environ.get("JARVIS_MODELO", "qwen3.5:4b-jarvis")
|
|
|
|
|
|
def modelos_disponibles(endpoint=ENDPOINT) -> list:
|
|
"""Los modelos que tiene Ollama en disco, para poder elegir entre ellos."""
|
|
try:
|
|
with urllib.request.urlopen(endpoint + "/api/tags", timeout=5) as r:
|
|
datos = json.load(r)
|
|
return sorted(m["name"] for m in datos.get("models", []))
|
|
except Exception:
|
|
return []
|
|
|
|
# Cuantas vueltas de "llama a una herramienta -> toma el resultado" se permiten
|
|
# antes de cortar. Sin tope, un modelo que se atasca pidiendo lo mismo deja al
|
|
# usuario esperando para siempre.
|
|
VUELTAS = 4
|
|
|
|
PERSONA = """Eres JARVIS, el asistente de voz de esta maquina.
|
|
|
|
REGLA QUE NO SE ROMPE NUNCA: no des ni un solo dato sobre este ordenador sin
|
|
haberlo comprobado antes con una herramienta. Ni el espacio del disco, ni la
|
|
memoria, ni cuantos ficheros hay, ni si algo esta abierto, ni la hora. Aunque
|
|
creas saberlo. Aunque parezca obvio. Si no lo has comprobado en esta misma
|
|
conversacion, lo compruebas.
|
|
|
|
Inventarse una cifra que suena razonable es el peor fallo que puedes cometer,
|
|
porque el usuario no tiene forma de saber que te la has inventado.
|
|
|
|
Y AL REVES: contesta SOLO lo que te preguntan. No añadas al final de tus
|
|
respuestas datos del sistema que nadie ha pedido —cuanta RAM queda, el espacio
|
|
del disco, la hora—. Si te preguntan como estas, "bien, señor" y punto; no lleva
|
|
un informe de memoria detras.
|
|
|
|
Para hacer algo, el orden es:
|
|
1. orden_del_catalogo, siempre que lo que te piden se parezca a algo de la
|
|
lista. Es lo que JARVIS sabe hacer bien y esta probado.
|
|
2. comando_de_shell solo si de verdad no esta en el catalogo.
|
|
|
|
Si te preguntan COMO se hace algo tecnico —una tecnica de pentesting, un flag,
|
|
una herramienta— y no lo tienes claro o no aparece en la documentacion de abajo,
|
|
usa buscar_en_apuntes con palabras clave ANTES de contestar. Ahi estan los
|
|
comandos que el usuario ya ha probado; es mejor buscarlos que inventarlos. Si la
|
|
primera busqueda no da, reformula y prueba otra vez.
|
|
|
|
Eliges tu propio cerebro. Si una pregunta es mas dificil de lo normal —razonar,
|
|
codigo, varios pasos— cambia a un modelo mas potente con elegir_modelo("potente")
|
|
antes de contestarla, y para lo simple vuelve a elegir_modelo("rapido"). Si el
|
|
usuario pide un modelo concreto, cambialo. Avisa en una frase de que has
|
|
cambiado.
|
|
|
|
Tu respuesta se convierte en AUDIO y se escucha en voz alta:
|
|
- Castellano siempre. Trata al usuario de usted y llamale "señor" en la primera
|
|
frase.
|
|
- Una o dos frases. Nada mas, salvo que te pidan detalle.
|
|
- Texto plano: ni asteriscos, ni vinetas, ni bloques de codigo, ni rutas largas.
|
|
- Los numeros redondeados y en palabras: "quedan setecientos gigas".
|
|
- Si algo falla, di en una frase QUE fallo. Sin disculpas de tres lineas.
|
|
|
|
Tienes acceso real a esta maquina y no estas en ningun contenedor. No digas
|
|
nunca que no puedes. En concreto, con comando_de_shell puedes:
|
|
- Ejecutar cualquier herramienta instalada, incluidas las de COFRE. Si no
|
|
recuerdas como se usa una, en la documentacion del usuario de abajo suele
|
|
estar el comando exacto.
|
|
- Conectarte a sus servidores: "ssh <host> '<comando>'" corre algo en el
|
|
servidor. Sus hosts estan en su configuracion; abajo veras cual encaja.
|
|
- Tareas de administrador, SOLO si el candado esta abierto. Si algo lo necesita
|
|
y no lo esta, dilo: que desbloquee y lo repites.
|
|
Antes de nada que borre datos o cambie el sistema, di lo que vas a hacer."""
|
|
|
|
|
|
def _ejemplos(catalogo, cuantos=8) -> str:
|
|
"""Ordenes de verdad del catalogo, repartidas entre grupos.
|
|
|
|
Van en la descripcion de la herramienta porque es lo unico que consigue que
|
|
el modelo pase castellano en vez de inventarse un comando. Salen del
|
|
catalogo y no escritas a mano para que no se queden viejas.
|
|
"""
|
|
por_grupo = {}
|
|
for a in catalogo.acciones:
|
|
if a.frases and not a.captura:
|
|
por_grupo.setdefault(a.grupo or "otras", []).append(a.frases[0])
|
|
muestra = []
|
|
grupos = list(por_grupo.values())
|
|
for i in range(cuantos):
|
|
for g in grupos:
|
|
if i < len(g) and len(muestra) < cuantos:
|
|
muestra.append(g[i])
|
|
return ", ".join(f'"{m}"' for m in muestra)
|
|
|
|
|
|
def herramientas(catalogo) -> list:
|
|
return [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "orden_del_catalogo",
|
|
"description": (
|
|
"Ejecuta una de las ordenes que JARVIS ya sabe hacer. "
|
|
"Recibe la orden EN CASTELLANO, tal y como la diria una "
|
|
"persona. NO acepta comandos de shell: si escribes 'df -h' "
|
|
"o 'xterm' falla. Ejemplos validos: "
|
|
+ _ejemplos(catalogo) + ". "
|
|
"Usala siempre que puedas antes que el shell."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"orden": {
|
|
"type": "string",
|
|
"description": ("la orden en castellano, en palabras, "
|
|
"nunca un comando"),
|
|
}
|
|
},
|
|
"required": ["orden"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "buscar_en_apuntes",
|
|
"description": (
|
|
"Busca en los apuntes de pentesting y Linux del usuario (su "
|
|
"COFRE) el comando o la explicacion que necesites. Usala "
|
|
"cuando te pregunten COMO se hace algo tecnico —una tecnica, "
|
|
"un flag, una herramienta— y no lo tengas ya delante en la "
|
|
"documentacion. Puedes reformular la consulta con tus propias "
|
|
"palabras clave y buscar varias veces hasta dar con ello. "
|
|
"Devuelve los fragmentos mas parecidos, con su fuente, para "
|
|
"que respondas desde ahi y no de memoria."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"consulta": {
|
|
"type": "string",
|
|
"description": ("que buscar, en pocas palabras clave; "
|
|
"el nombre de la herramienta ayuda"),
|
|
}
|
|
},
|
|
"required": ["consulta"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "elegir_modelo",
|
|
"description": (
|
|
"Cambia el modelo con el que piensas. Usalo cuando una "
|
|
"pregunta sea mas dificil de lo normal (razonar, codigo, "
|
|
"varios pasos) y quieras uno mas potente, o cuando el usuario "
|
|
"pida expresamente cambiar de modelo. Valores: 'rapido' "
|
|
"(ligero, para lo simple), 'potente' (mas grande, mas lento), "
|
|
"'razonador' (piensa en voz alta antes de responder), o el "
|
|
"nombre exacto de un modelo de Ollama. El cambio dura hasta "
|
|
"que lo vuelvas a cambiar. Contesta breve tras cambiar."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"cual": {
|
|
"type": "string",
|
|
"description": "rapido | potente | razonador | nombre exacto",
|
|
}
|
|
},
|
|
"required": ["cual"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "comando_de_shell",
|
|
"description": (
|
|
"Ejecuta un comando en la terminal de esta maquina, que es "
|
|
"Debian con GNOME. Solo para lo que NO esta en el catalogo. "
|
|
"Es de solo lectura: no ejecutes nada que borre o modifique "
|
|
"sin que te lo pidan expresamente."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"comando": {"type": "string",
|
|
"description": "el comando, para bash"},
|
|
},
|
|
"required": ["comando"],
|
|
},
|
|
},
|
|
},
|
|
]
|
|
|
|
|
|
class Cerebro:
|
|
def __init__(self, catalogo, modelo=MODELO, endpoint=ENDPOINT, en_ram=False,
|
|
manos=None):
|
|
self.catalogo = catalogo
|
|
self.modelo = modelo
|
|
self.endpoint = endpoint
|
|
# num_gpu 0 fuerza CPU: el modelo entero en RAM. Util para probar uno
|
|
# mas grande del que cabe en la grafica, a costa de velocidad.
|
|
self.capas = 0 if en_ram else None
|
|
self.manos = manos # a quien pedirle que ejecute
|
|
self.historia = []
|
|
# A quien avisar cuando JARVIS cambia de modelo, para que el panel
|
|
# actualice el rotulo del boton "cerebro". Lo conecta el panel.
|
|
self.al_cambiar_modelo = None
|
|
|
|
# Atajos hablados -> familia de modelo. El modelo elige por INTENCION
|
|
# ("potente") y aqui se traduce a lo que hay instalado, para no obligarle a
|
|
# saberse los nombres exactos. Se resuelve contra lo que de verdad hay en
|
|
# Ollama, cogiendo el mayor/menor de la familia qwen que exista.
|
|
def _resuelve_modelo(self, cual: str) -> str | None:
|
|
cual = (cual or "").strip().lower()
|
|
hay = modelos_disponibles(self.endpoint)
|
|
if not hay:
|
|
return None
|
|
# nombre exacto o casi (permite "9b" -> "qwen3.5:9b")
|
|
for m in hay:
|
|
if cual == m.lower() or cual == m.lower().split(":")[-1]:
|
|
return m
|
|
qwen = sorted(m for m in hay if "qwen" in m.lower())
|
|
def por_tam(sufijo):
|
|
return next((m for m in qwen if m.lower().endswith(sufijo)), None)
|
|
if cual in ("rapido", "ligero", "pequeno", "pequeño"):
|
|
return por_tam(":2b") or por_tam(":4b-jarvis") or (qwen[0] if qwen else None)
|
|
if cual in ("potente", "grande", "mejor", "fuerte"):
|
|
return por_tam(":9b") or (qwen[-1] if qwen else None)
|
|
if cual in ("razonador", "razona", "piensa", "deepseek"):
|
|
return next((m for m in hay if "deepseek" in m.lower() or "-r1" in m.lower()), None)
|
|
# por defecto, el de trabajo del naranja si existe
|
|
return por_tam(":4b-jarvis") or (qwen[0] if qwen else None)
|
|
|
|
@staticmethod
|
|
def _nombre_corto(m: str) -> str:
|
|
# "qwen3.5:9b" -> "9b"; "deepseek-r1:latest" -> "deepseek-r1"
|
|
base, _, tag = m.partition(":")
|
|
return base if tag in ("", "latest") else tag
|
|
|
|
def cambia_modelo(self, cual: str) -> str:
|
|
"""Cambia el cerebro en caliente. Devuelve lo que decir tras cambiar."""
|
|
nuevo = self._resuelve_modelo(cual)
|
|
if not nuevo:
|
|
return f"no tengo ningun modelo que encaje con «{cual}», señor."
|
|
if nuevo == self.modelo:
|
|
return f"ya estoy pensando con {self._nombre_corto(nuevo)}, señor."
|
|
self.modelo = nuevo
|
|
if self.al_cambiar_modelo:
|
|
try:
|
|
self.al_cambiar_modelo(nuevo)
|
|
except Exception:
|
|
pass
|
|
# Ojo: un modelo mas grande no cabe en la grafica pequeña y tirara de CPU
|
|
# —mas lento pero mas capaz—; eso es cosa de Ollama, no de aqui.
|
|
return f"Cambiado a {self._nombre_corto(nuevo)}, señor."
|
|
|
|
def _saber(self, texto: str) -> str:
|
|
"""Los apuntes de COFRE que vengan a cuento, para meterlos al prompt.
|
|
|
|
Es la mitad "aumentar" de RAG: antes de contestar, se le pone delante al
|
|
modelo el trozo de los apuntes del usuario que mas se parece a lo que
|
|
pregunta. Asi construye la respuesta desde SU documentacion —flags
|
|
reales, comandos que ha probado— en vez de desde lo que recuerde, que
|
|
para 668 herramientas seria inventarselo.
|
|
|
|
Si no hay indice o nada encaja lo bastante, devuelve cadena vacia y el
|
|
prompt se queda como estaba: la funcion no puede empeorar una respuesta,
|
|
solo mejorarla.
|
|
"""
|
|
try:
|
|
from saber.busca import contexto, disponible
|
|
except Exception:
|
|
return ""
|
|
if not disponible():
|
|
return ""
|
|
trozos = contexto(texto, cuantos=4, minimo=0.38)
|
|
if not trozos:
|
|
return ""
|
|
return ("\n\n## Documentacion del usuario relevante a esto\n"
|
|
"Usa ESTO para responder, son sus apuntes probados. Si el "
|
|
"comando exacto esta aqui, dalo tal cual; no te inventes flags "
|
|
"que no aparezcan.\n\n" + trozos)
|
|
|
|
def _pide(self, mensajes, tools=None) -> dict:
|
|
cuerpo = {"model": self.modelo, "messages": mensajes, "stream": False,
|
|
"think": False, "options": {"temperature": 0.3}}
|
|
if tools:
|
|
cuerpo["tools"] = tools
|
|
if self.capas is not None:
|
|
cuerpo["options"]["num_gpu"] = self.capas
|
|
req = urllib.request.Request(
|
|
self.endpoint + "/api/chat", data=json.dumps(cuerpo).encode(),
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=300) as r:
|
|
return json.loads(r.read())
|
|
|
|
def responde(self, texto: str, al_ejecutar=None) -> str:
|
|
"""Contesta, usando las manos si hace falta. Devuelve lo que hay que decir."""
|
|
self.historia.append({"role": "user", "content": texto})
|
|
sistema = PERSONA + self._saber(texto)
|
|
mensajes = [{"role": "system", "content": sistema}] + self.historia[-8:]
|
|
tools = herramientas(self.catalogo)
|
|
|
|
for _ in range(VUELTAS):
|
|
try:
|
|
d = self._pide(mensajes, tools)
|
|
except (urllib.error.URLError, OSError) as e:
|
|
return f"No he podido pensar, señor: {str(e)[:50]}"
|
|
|
|
m = d.get("message", {})
|
|
llamadas = m.get("tool_calls") or []
|
|
if not llamadas:
|
|
dicho = (m.get("content") or "").strip()
|
|
self.historia.append({"role": "assistant", "content": dicho})
|
|
return dicho
|
|
|
|
mensajes.append(m)
|
|
for lc in llamadas:
|
|
f = lc.get("function", {})
|
|
args = f.get("arguments") or {}
|
|
if isinstance(args, str):
|
|
try:
|
|
args = json.loads(args)
|
|
except json.JSONDecodeError:
|
|
args = {}
|
|
resultado = self._usa(f.get("name", ""), args, al_ejecutar)
|
|
mensajes.append({"role": "tool", "content": resultado[:2000]})
|
|
|
|
return "Me he liado dando vueltas, señor. Pruebe a pedirmelo de otra forma."
|
|
|
|
def _usa(self, nombre: str, args: dict, al_ejecutar=None) -> str:
|
|
# buscar_en_apuntes NO pasa por las manos: es lectura del indice, no
|
|
# toca la maquina, asi que funciona aunque el candado este echado.
|
|
if nombre == "buscar_en_apuntes":
|
|
return self._busca_apuntes(args.get("consulta", ""))
|
|
if nombre == "elegir_modelo":
|
|
return self.cambia_modelo(args.get("cual", ""))
|
|
if self.manos is None:
|
|
return "no hay manos conectadas"
|
|
if nombre == "orden_del_catalogo":
|
|
return self.manos.por_orden(args.get("orden", ""), al_ejecutar)
|
|
if nombre == "comando_de_shell":
|
|
return self.manos.por_shell(args.get("comando", ""), al_ejecutar)
|
|
return f"no existe la herramienta {nombre}"
|
|
|
|
def _busca_apuntes(self, consulta: str) -> str:
|
|
"""El RAG como herramienta: el modelo busca cuando lo decide.
|
|
|
|
La inyeccion pasiva de _saber() usa la frase LITERAL del usuario y
|
|
dispara siempre. Esto es lo otro: el modelo reformula la consulta con
|
|
sus palabras y busca a proposito, que es lo que salva los casos donde el
|
|
usuario dice una cosa y el apunte esta escrito de otra."""
|
|
consulta = (consulta or "").strip()
|
|
if not consulta:
|
|
return "dime que buscar"
|
|
try:
|
|
from saber.busca import busca
|
|
except Exception:
|
|
return "no tengo los apuntes indexados"
|
|
trozos = [t for t in busca(consulta, cuantos=5) if t.get("sim", 0) >= 0.3]
|
|
if not trozos:
|
|
return ("no encontre nada en los apuntes sobre eso; prueba con otras "
|
|
"palabras o el nombre de la herramienta")
|
|
lineas = []
|
|
for t in trozos:
|
|
fuente = t.get("fichero") or f"{t.get('perfil')}/{t.get('herramienta')}"
|
|
lineas.append(f"[{fuente}]\n{t['texto'][:500]}")
|
|
return "\n\n".join(lineas)
|