diff options
Diffstat (limited to 'crates/asist-app/src/main.rs')
| -rw-r--r-- | crates/asist-app/src/main.rs | 474 |
1 files changed, 474 insertions, 0 deletions
diff --git a/crates/asist-app/src/main.rs b/crates/asist-app/src/main.rs new file mode 100644 index 0000000..2788d97 --- /dev/null +++ b/crates/asist-app/src/main.rs @@ -0,0 +1,474 @@ +//! Asistente de voz local: escucha, piensa y responde hablando. +//! +//! Une tres motores que ya existen —Canary para oír, llama.cpp para pensar y +//! qwentts para hablar— en un pipeline de hilos y canales donde ninguna etapa +//! espera a la siguiente. + +mod pipeline; +mod render; +mod session; +mod supervisor; + +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::{bail, Context, Result}; +use crossbeam_channel::unbounded; + +use asist_asr::Recognizer; +use asist_audio::{Capture, CaptureBlock, Playback}; +use asist_core::config::Config; +use asist_core::event::Event; +use asist_core::tools::ToolRegistry; +use asist_llm::LlmClient; +use asist_tts::TtsClient; + +use pipeline::Pipeline; +use render::Renderer; +use session::Session; +use supervisor::Supervisor; + +fn main() -> Result<()> { + let args = Args::parse()?; + if args.help { + println!("{USAGE}"); + return Ok(()); + } + init_logging(&args); + + let mut config = Config::load(&args.config) + .with_context(|| format!("no se pudo cargar {}", args.config.display()))?; + args.apply(&mut config); + + match args.command { + Command::Run => run(config, &args), + Command::Devices => devices(), + Command::Check => check(&config), + } +} + +fn run(config: Config, args: &Args) -> Result<()> { + let session = Session::new(); + + // Los clientes se crean antes de arrancar nada: así se detecta lo que ya + // esté escuchando y no se levanta un servidor por duplicado. + let llm = LlmClient::new(config.llm_authority(), &config.llm); + let tts = TtsClient::new(config.tts_authority(), &config.tts); + let (llm_up, tts_up) = (llm.healthy(), tts.healthy()); + + let mut supervisor = Supervisor::new(&args.log_dir)?; + supervisor.start(&config, llm_up, tts_up)?; + + let timeout = Duration::from_secs(config.supervisor.startup_timeout_secs); + eprintln!("Esperando a los motores (hasta {} s)…", timeout.as_secs()); + llm.wait_ready(timeout).map_err(|e| { + anyhow::anyhow!( + "{e}\nRevisa {}", + supervisor.log_path("llama-server").display() + ) + })?; + tts.wait_ready(timeout).map_err(|e| { + anyhow::anyhow!( + "{e}\nRevisa {}", + supervisor.log_path("tts-server").display() + ) + })?; + llm.warn_if_thinking_template(); + + if let Some(reference) = &config.tts.reference { + tts.register_voice(reference) + .context("no se pudo registrar la voz clonada")?; + } + // Se paga aquí la construcción de los grafos del sintetizador, unos 3,5 s + // que si no se los comería el primer turno de verdad. + if let Err(err) = tts.warmup() { + tracing::warn!(target: "tts", %err, "falló el precalentado; el primer turno irá lento"); + } + + eprintln!("Cargando el modelo de voz a texto…"); + let recognizer = Recognizer::load(&config.asr)?; + + let playback = Playback::open(&config.audio)?; + let (capture_tx, capture_rx) = unbounded::<CaptureBlock>(); + let capture = Capture::open(&config.audio, capture_tx.clone())?; + + let tools = ToolRegistry::from_config(&config.tools); + if tools.is_empty() { + eprintln!("Herramientas: ninguna"); + } else { + eprintln!("Herramientas: {}", tools.names().join(", ")); + } + if config.tools.shell { + eprintln!( + "Ejecución de órdenes ACTIVA — permitidas: {}{}", + config.tools.shell_allowlist.join(", "), + if config.tools.shell_dry_run { + " (simulación)" + } else { + "" + } + ); + } + + let pipeline = Pipeline::spawn( + &config, + session.clone(), + recognizer, + llm, + tts, + tools, + playback.handle(), + capture_rx, + capture_tx, + capture.format, + )?; + + eprintln!( + "\nListo. Micrófono: {} · Altavoz: {} ({} Hz)\nHabla cuando quieras. Enter para salir.\n", + capture.device_name, playback.device_name, playback.sample_rate + ); + + // Enter cierra, pero sólo si hay alguien delante para pulsarlo. Con la + // entrada redirigida —bajo systemd, en un contenedor, con `< /dev/null`— + // `read_line` devuelve EOF al instante y el asistente se cerraría nada + // más arrancar. En ese caso se espera a una señal. + if stdin_is_tty() { + let session = session.clone(); + std::thread::spawn(move || { + let mut line = String::new(); + let _ = std::io::stdin().read_line(&mut line); + session.request_stop(); + }); + } else { + eprintln!("(entrada no interactiva: para cerrar, manda SIGINT o SIGTERM)"); + install_signal_handler(session.clone()); + } + + let mut renderer = Renderer::new(config.asr.partials); + let metrics = std::sync::Arc::clone(&pipeline.metrics); + let mut next_health_check = std::time::Instant::now() + Duration::from_secs(5); + loop { + match pipeline.events.recv_timeout(Duration::from_millis(150)) { + Ok(event) => { + let shutdown = matches!(event, Event::Shutdown); + renderer.apply(&event); + if shutdown { + break; + } + } + Err(crossbeam_channel::RecvTimeoutError::Timeout) => {} + Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break, + } + if session.is_stopping() { + break; + } + // Si un servidor se muere a mitad de sesión, todos los turnos + // siguientes fallarían con un error de transporte sin explicar por + // qué. Mejor decirlo una vez, con el registro a mano, y cerrar. + if std::time::Instant::now() >= next_health_check { + next_health_check = std::time::Instant::now() + Duration::from_secs(5); + if let Some((name, code)) = supervisor.crashed() { + renderer.finish(); + eprintln!( + "\n{name} ha terminado inesperadamente (código {}). Revisa {}", + code.map(|c| c.to_string()) + .unwrap_or_else(|| "desconocido".into()), + supervisor.log_path(name).display() + ); + break; + } + } + } + + renderer.finish(); + capture.stop(); + pipeline.shutdown(); + playback.stop(); + supervisor.shutdown(); + + if metrics.turns() > 0 { + eprintln!("\n{}", metrics.summary()); + } + Ok(()) +} + +/// Lista los dispositivos de audio, para poder nombrarlos en la configuración. +fn devices() -> Result<()> { + use asist_audio::describe; + use cpal_reexport::traits::HostTrait; + let host = cpal_reexport::default_host(); + + let default_in = host.default_input_device().map(|d| describe(&d)); + println!("Entradas:"); + for device in host.input_devices()? { + let name = describe(&device); + let mark = if Some(&name) == default_in.as_ref() { + " (por defecto)" + } else { + "" + }; + println!(" {name}{mark}"); + } + + let default_out = host.default_output_device().map(|d| describe(&d)); + println!("\nSalidas:"); + for device in host.output_devices()? { + let name = describe(&device); + let mark = if Some(&name) == default_out.as_ref() { + " (por defecto)" + } else { + "" + }; + println!(" {name}{mark}"); + } + Ok(()) +} + +/// Comprueba que está todo en su sitio, sin abrir el micrófono. +fn check(config: &Config) -> Result<()> { + let mut problems = Vec::new(); + let ok = |label: &str, detail: String| println!(" ok {label:<22} {detail}"); + + let paths: [(&str, &std::path::Path); 6] = [ + ("modelo asr", &config.asr.model_dir), + ("binario llama", &config.supervisor.llama.binary), + ("modelo llm", &config.supervisor.llama.model), + ("plantilla chat", &config.supervisor.llama.chat_template), + ("binario tts", &config.supervisor.tts.binary), + ("modelo tts", &config.supervisor.tts.model), + ]; + println!("Ficheros:"); + for (label, path) in paths { + if path.exists() { + ok(label, path.display().to_string()); + } else { + println!(" FALTA {label:<22} {}", path.display()); + problems.push(format!("falta {label}: {}", path.display())); + } + } + + if let Some(reference) = &config.tts.reference { + println!("Voz de referencia «{}»:", reference.name); + for (label, path) in [ + ("hablante (.spk)", &reference.speaker), + ("códigos (.rvq)", &reference.codes), + ("transcripción", &reference.transcript), + ] { + if path.exists() { + ok(label, path.display().to_string()); + } else { + println!(" FALTA {label:<22} {}", path.display()); + problems.push(format!("falta {label}")); + } + } + } + + println!("Servidores:"); + let llm = LlmClient::new(config.llm_authority(), &config.llm); + let tts = TtsClient::new(config.tts_authority(), &config.tts); + for (label, up, authority) in [ + ("llama-server", llm.healthy(), config.llm_authority()), + ("tts-server", tts.healthy(), config.tts_authority()), + ] { + println!( + " {:<5} {label:<22} {authority}", + if up { "ok" } else { "-" } + ); + } + if llm.healthy() { + llm.warn_if_thinking_template(); + } + if tts.healthy() { + match tts.voices() { + Ok(voices) if voices.contains(&config.tts.voice) => { + println!(" ok voz «{}» registrada", config.tts.voice); + } + Ok(voices) => println!( + " - voz «{}» aún no registrada (hay: {})", + config.tts.voice, + voices.join(", ") + ), + Err(err) => println!(" - no se pudieron listar las voces: {err}"), + } + } + + if problems.is_empty() { + println!("\nTodo listo."); + Ok(()) + } else { + bail!( + "{} problema(s):\n - {}", + problems.len(), + problems.join("\n - ") + ) + } +} + +// cpal se usa aquí sólo para listar dispositivos; llega a través de asist-audio. +use asist_audio::cpal as cpal_reexport; + +const USAGE: &str = "\ +asistente — asistente de voz local + +USO: + asistente [ORDEN] [OPCIONES] + +ÓRDENES: + run Escucha y responde (por defecto) + check Comprueba ficheros, modelos y servidores, y sale + devices Lista los dispositivos de audio + +OPCIONES: + -c, --config <RUTA> Fichero de configuración [config/asistente.toml] + --log-dir <RUTA> Carpeta de los registros de los servidores [logs/] + --voice <NOMBRE> Voz del sintetizador + --no-manage No lanzar los servidores; suponerlos ya arriba + --no-partials Sin transcripción provisional (ahorra CPU) + --barge-in Permitir cortar al asistente hablando encima + --shell Activar la ejecución de órdenes del sistema + -v, --verbose Registro de depuración (equivale a RUST_LOG=debug) + -h, --help Esta ayuda + +VARIABLES: + RUST_LOG Filtro de registro, p. ej. «info,tts=debug,latencia=info» +"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Command { + Run, + Check, + Devices, +} + +struct Args { + command: Command, + config: PathBuf, + log_dir: PathBuf, + voice: Option<String>, + no_manage: bool, + no_partials: bool, + barge_in: bool, + shell: bool, + verbose: bool, + help: bool, +} + +impl Args { + fn parse() -> Result<Self> { + let mut args = Self { + command: Command::Run, + config: PathBuf::from("config/asistente.toml"), + log_dir: PathBuf::from("logs"), + voice: None, + no_manage: false, + no_partials: false, + barge_in: false, + shell: false, + verbose: false, + help: false, + }; + let mut raw = std::env::args().skip(1); + while let Some(arg) = raw.next() { + match arg.as_str() { + "run" => args.command = Command::Run, + "check" => args.command = Command::Check, + "devices" => args.command = Command::Devices, + "-c" | "--config" => { + args.config = raw.next().context("--config necesita una ruta")?.into() + } + "--log-dir" => { + args.log_dir = raw.next().context("--log-dir necesita una ruta")?.into() + } + "--voice" => args.voice = Some(raw.next().context("--voice necesita un nombre")?), + "--no-manage" => args.no_manage = true, + "--no-partials" => args.no_partials = true, + "--barge-in" => args.barge_in = true, + "--shell" => args.shell = true, + "-v" | "--verbose" => args.verbose = true, + "-h" | "--help" => args.help = true, + other => bail!("opción desconocida: {other}\n\n{USAGE}"), + } + } + Ok(args) + } + + /// La línea de órdenes manda sobre el fichero. + fn apply(&self, config: &mut Config) { + if let Some(voice) = &self.voice { + config.tts.voice = voice.clone(); + } + if self.no_manage { + config.supervisor.manage = false; + } + if self.no_partials { + config.asr.partials = false; + } + if self.barge_in { + config.vad.barge_in = true; + } + if self.shell { + config.tools.shell = true; + } + } +} + +/// ¿Hay una persona al otro lado de la entrada estándar? +fn stdin_is_tty() -> bool { + #[cfg(unix)] + { + unsafe extern "C" { + fn isatty(fd: i32) -> i32; + } + unsafe { isatty(0) == 1 } + } + #[cfg(not(unix))] + true +} + +/// Cierre ordenado ante SIGINT/SIGTERM. +/// +/// Importa más de lo que parece: matar el proceso en seco deja a los +/// servidores hijos con la memoria de la GPU tomada. +#[cfg(unix)] +fn install_signal_handler(session: Session) { + use std::sync::OnceLock; + static SESSION: OnceLock<Session> = OnceLock::new(); + let _ = SESSION.set(session); + + extern "C" fn on_signal(_: i32) { + if let Some(session) = SESSION.get() { + session.request_stop(); + } + } + unsafe extern "C" { + fn signal(sig: i32, handler: extern "C" fn(i32)) -> usize; + } + unsafe { + signal(2, on_signal); // SIGINT + signal(15, on_signal); // SIGTERM + } +} + +#[cfg(not(unix))] +fn install_signal_handler(_session: Session) {} + +fn init_logging(args: &Args) { + use tracing_subscriber::{fmt, EnvFilter}; + // ONNX Runtime informa a nivel INFO de cada transformación del grafo: + // varios cientos de líneas que tapan la transcripción. Se silencia salvo + // que se pida expresamente por RUST_LOG. + let default = if args.verbose { + "debug,ort=warn" + } else { + "info,ort=warn" + }; + let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(default)); + fmt() + .with_env_filter(filter) + // Al terminal de texto van la transcripción y la respuesta; el registro + // va a stderr para que se puedan separar con una redirección. + .with_writer(std::io::stderr) + .with_target(true) + .without_time() + .init(); +} |