From f8f98e83481e7376a235fb305a095552433f79ab Mon Sep 17 00:00:00 2001 From: elvis Date: Sun, 6 Sep 2026 19:21:16 -0300 Subject: Local voice assistant on top of Canary, llama.cpp and qwentts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pipeline en Rust de hilos y canales que une los tres motores: el micrófono alimenta un segmentador con VAD, las intervenciones cerradas van al reconocedor, la transcripción al modelo y cada frase que este cierra sale hacia el sintetizador sin esperar al resto de la respuesta. Dos reglas sostienen el diseño: ninguna etapa bloquea a la anterior —quien va sobrado descarta trabajo en lugar de acumular retraso— y todo lo que viaja por los canales lleva el turno al que pertenece, así que interrumpir es subir el contador y levantar dos banderas de cancelación. Seis crates: core (configuración, eventos, HTTP, telemetría, herramientas), audio (cpal, VAD, anillo de reproducción), asr, llm, tts y app (supervisor de procesos y orquestador). Los motores van como submódulos fijados a un commit, con los cambios locales en vendor/patches. Midiendo el pipeline aparecieron tres cuellos de botella de configuración que valieron más que cualquier cambio de código, todos documentados en docs/RENDIMIENTO.md: - tts-server decodificaba el audio en bloques de 24 s, de modo que el modo «streaming» llegaba de una pieza: 4948 ms -> 585 ms hasta el primer audio. - La plantilla de chat del modelo abre y no lo cierra nunca, sin variable que lo apague: 8630 ms -> 413 ms hasta el primer token, con una copia de la plantilla que deja el bloque cerrado de entrada. - Cualquier indicación de estilo junto a la guía de herramientas hace que este modelo de 2B deje de llamarlas y se invente el dato (8/8 aciertos con la guía sola, 0/8 con la persona de asistente de voz). El turno alterna ahora entre dos instrucciones de sistema. La ejecución de órdenes del sistema queda implementada y apagada, tras cuatro barreras: lista blanca sobre el ejecutable, rutas rechazadas, sin shell que interprete metacaracteres y plazo máximo. 68 pruebas unitarias sin modelos, más seis de integración que se saltan solas si no hay servidores y se turnan la GPU: en paralelo, los dos servidores no caben en 4 GB y miden contención en vez de latencia. Claude-Session: https://claude.ai/code/session_01FNxz5cSdQSscJH9H7b8uGU --- crates/asist-llm/src/chat.rs | 275 +++++++++++++++++++++++++++++++++ crates/asist-llm/src/client.rs | 342 +++++++++++++++++++++++++++++++++++++++++ crates/asist-llm/src/lib.rs | 12 ++ 3 files changed, 629 insertions(+) create mode 100644 crates/asist-llm/src/chat.rs create mode 100644 crates/asist-llm/src/client.rs create mode 100644 crates/asist-llm/src/lib.rs (limited to 'crates/asist-llm/src') diff --git a/crates/asist-llm/src/chat.rs b/crates/asist-llm/src/chat.rs new file mode 100644 index 0000000..01a9a74 --- /dev/null +++ b/crates/asist-llm/src/chat.rs @@ -0,0 +1,275 @@ +//! Historial de conversación en el formato que espera la API de chat. + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use asist_core::tools::{ToolCall, ToolOutcome}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Role { + System, + User, + Assistant, + Tool, +} + +impl Role { + fn as_str(self) -> &'static str { + match self { + Role::System => "system", + Role::User => "user", + Role::Assistant => "assistant", + Role::Tool => "tool", + } + } +} + +#[derive(Debug, Clone)] +pub struct Message { + pub role: Role, + pub content: String, + /// Herramientas que el asistente pidió en este turno. + pub tool_calls: Vec, + /// Para los mensajes de rol `tool`: a qué llamada responden. + pub tool_call_id: Option, +} + +impl Message { + pub fn system(content: impl Into) -> Self { + Self::plain(Role::System, content) + } + + pub fn user(content: impl Into) -> Self { + Self::plain(Role::User, content) + } + + pub fn assistant(content: impl Into) -> Self { + Self::plain(Role::Assistant, content) + } + + fn plain(role: Role, content: impl Into) -> Self { + Self { + role, + content: content.into(), + tool_calls: Vec::new(), + tool_call_id: None, + } + } + + /// Turno del asistente que en vez de hablar pidió herramientas. + pub fn tool_request(content: String, tool_calls: Vec) -> Self { + Self { + role: Role::Assistant, + content, + tool_calls, + tool_call_id: None, + } + } + + /// Resultado devuelto al modelo. + pub fn tool_result(outcome: &ToolOutcome) -> Self { + Self { + role: Role::Tool, + content: outcome.output.clone(), + tool_calls: Vec::new(), + tool_call_id: Some(outcome.id.clone()), + } + } + + pub fn to_json(&self) -> Value { + let mut object = json!({ "role": self.role.as_str(), "content": self.content }); + if !self.tool_calls.is_empty() { + object["tool_calls"] = Value::Array( + self.tool_calls + .iter() + .map(|call| { + json!({ + "id": call.id, + "type": "function", + "function": { "name": call.name, "arguments": call.arguments } + }) + }) + .collect(), + ); + } + if let Some(id) = &self.tool_call_id { + object["tool_call_id"] = json!(id); + } + object + } +} + +/// Historial con la instrucción de sistema fija y una ventana deslizante de +/// turnos, para que la conversación no crezca sin fin. +#[derive(Debug, Clone)] +pub struct Conversation { + system: Message, + turns: Vec, + max_turns: usize, +} + +impl Conversation { + pub fn new(system_prompt: impl Into, max_turns: usize) -> Self { + Self { + system: Message::system(system_prompt), + turns: Vec::new(), + max_turns, + } + } + + /// Cambia la instrucción de sistema sin tocar el historial. + /// + /// El turno alterna entre la guía de herramientas y la de estilo, y el + /// historial tiene que sobrevivir al cambio: si se reiniciara, el modelo + /// perdería los resultados que acaba de pedir. + pub fn set_system(&mut self, prompt: &str) { + self.system = Message::system(prompt); + } + + pub fn push(&mut self, message: Message) { + self.turns.push(message); + self.trim(); + } + + pub fn extend(&mut self, messages: impl IntoIterator) { + self.turns.extend(messages); + self.trim(); + } + + /// Recorta a `max_turns` intervenciones de usuario, sin dejar nunca un + /// mensaje de rol `tool` huérfano al principio: la API lo rechaza si no + /// va precedido de la llamada que lo originó. + fn trim(&mut self) { + if self.max_turns == 0 { + self.turns.clear(); + return; + } + let user_positions: Vec = self + .turns + .iter() + .enumerate() + .filter(|(_, m)| m.role == Role::User) + .map(|(i, _)| i) + .collect(); + if user_positions.len() <= self.max_turns { + return; + } + let cut = user_positions[user_positions.len() - self.max_turns]; + self.turns.drain(..cut); + } + + pub fn messages(&self) -> Vec<&Message> { + std::iter::once(&self.system) + .chain(self.turns.iter()) + .collect() + } + + pub fn to_json(&self) -> Value { + Value::Array(self.messages().into_iter().map(Message::to_json).collect()) + } + + pub fn len(&self) -> usize { + self.turns.len() + } + + pub fn is_empty(&self) -> bool { + self.turns.is_empty() + } + + pub fn clear(&mut self) { + self.turns.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn la_instruccion_de_sistema_va_siempre_delante() { + let mut chat = Conversation::new("sé breve", 4); + chat.push(Message::user("hola")); + let json = chat.to_json(); + assert_eq!(json[0]["role"], "system"); + assert_eq!(json[0]["content"], "sé breve"); + assert_eq!(json[1]["role"], "user"); + } + + #[test] + fn cambiar_la_instruccion_de_sistema_conserva_el_historial() { + let mut chat = Conversation::new("primera", 4); + chat.push(Message::user("hola")); + chat.set_system("segunda"); + let json = chat.to_json(); + assert_eq!(json[0]["content"], "segunda"); + assert_eq!( + json[1]["content"], "hola", + "el historial no puede perderse al cambiar" + ); + } + + #[test] + fn el_historial_se_recorta_por_turnos_de_usuario() { + let mut chat = Conversation::new("s", 2); + for i in 0..5 { + chat.push(Message::user(format!("p{i}"))); + chat.push(Message::assistant(format!("r{i}"))); + } + let messages = chat.messages(); + let users: Vec<&str> = messages + .iter() + .filter(|m| m.role == Role::User) + .map(|m| m.content.as_str()) + .collect(); + assert_eq!(users, vec!["p3", "p4"]); + } + + #[test] + fn el_recorte_no_deja_un_resultado_de_herramienta_huerfano() { + // La API rechaza un mensaje `tool` que no venga detrás de la llamada + // que lo pidió, así que el corte tiene que caer en un turno de usuario. + let mut chat = Conversation::new("s", 1); + chat.push(Message::user("p0")); + chat.push(Message::tool_request( + String::new(), + vec![ToolCall { + id: "a".into(), + name: "t".into(), + arguments: "{}".into(), + }], + )); + chat.push(Message { + role: Role::Tool, + content: "ok".into(), + tool_calls: vec![], + tool_call_id: Some("a".into()), + }); + chat.push(Message::user("p1")); + + let roles: Vec = chat.messages().iter().map(|m| m.role).collect(); + assert_eq!(roles, vec![Role::System, Role::User]); + } + + #[test] + fn con_cero_turnos_no_hay_memoria() { + let mut chat = Conversation::new("s", 0); + chat.push(Message::user("hola")); + assert_eq!(chat.messages().len(), 1); + } + + #[test] + fn una_llamada_a_herramienta_se_serializa_como_espera_la_api() { + let message = Message::tool_request( + String::new(), + vec![ToolCall { + id: "call_1".into(), + name: "hora_actual".into(), + arguments: "{}".into(), + }], + ); + let json = message.to_json(); + assert_eq!(json["tool_calls"][0]["type"], "function"); + assert_eq!(json["tool_calls"][0]["function"]["name"], "hora_actual"); + } +} diff --git a/crates/asist-llm/src/client.rs b/crates/asist-llm/src/client.rs new file mode 100644 index 0000000..1dca2af --- /dev/null +++ b/crates/asist-llm/src/client.rs @@ -0,0 +1,342 @@ +//! Transporte hacia llama-server. + +use std::time::{Duration, Instant}; + +use serde_json::{json, Value}; + +use asist_core::config::LlmConfig; +use asist_core::error::{Error, Result}; +use asist_core::http::{Cancel, HttpClient}; +use asist_core::tools::{ToolCall, ToolRegistry}; + +use crate::chat::Conversation; + +/// Un trozo de respuesta según sale del modelo. +#[derive(Debug, Clone)] +pub enum Delta { + /// Texto para el usuario. + Text(String), + /// El modelo ha decidido llamar a herramientas; no habrá más texto. + ToolCalls(Vec), +} + +/// Cómo terminó una respuesta en streaming. +#[derive(Debug, Clone, Default)] +pub struct StreamOutcome { + pub text: String, + pub tool_calls: Vec, + /// Tiempo hasta el primer fragmento con contenido. + pub ttft: Option, + pub stopped_early: bool, +} + +pub struct LlmClient { + http: HttpClient, + config: LlmConfig, +} + +impl LlmClient { + pub fn new(authority: String, config: &LlmConfig) -> Self { + Self { + http: HttpClient::new(authority) + .with_read_timeout(Duration::from_secs(config.request_timeout_secs)), + config: config.clone(), + } + } + + pub fn healthy(&self) -> bool { + self.http.healthy("/health") + } + + /// Espera a que el servidor termine de cargar el modelo. + pub fn wait_ready(&self, timeout: Duration) -> Result<()> { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if self.healthy() { + return Ok(()); + } + std::thread::sleep(Duration::from_millis(500)); + } + Err(Error::Llm(format!( + "{} no respondió a /health en {} s", + self.http.authority, + timeout.as_secs() + ))) + } + + /// Comprueba que la plantilla de chat no arranca un bloque de + /// razonamiento sin cerrar. + /// + /// Merece la pena avisar: con la plantilla original de Qwen3.5 el modelo + /// se pasa entre 7 y 9 segundos «pensando» antes de la primera palabra + /// audible, y desde fuera parece que el asistente se ha colgado. + pub fn warn_if_thinking_template(&self) { + let Ok(body) = self.http.get("/props") else { + return; + }; + let Ok(props) = serde_json::from_slice::(&body) else { + return; + }; + let Some(template) = props.get("chat_template").and_then(Value::as_str) else { + return; + }; + + let opens = template.matches("").count(); + let closes = template.matches("").count(); + if opens > closes { + tracing::warn!( + target: "llm", + "la plantilla de chat abre sin cerrarlo: el modelo razonará \ + varios segundos antes de contestar. Arranca llama-server con \ + --chat-template-file config/qwen35-no-think.jinja" + ); + } + } + + fn request_body(&self, chat: &Conversation, tools: Option<&ToolRegistry>) -> Value { + let mut body = json!({ + "messages": chat.to_json(), + "stream": true, + "temperature": self.config.temperature, + "top_p": self.config.top_p, + "top_k": self.config.top_k, + "min_p": self.config.min_p, + "repeat_penalty": self.config.repeat_penalty, + "max_tokens": self.config.max_tokens, + }); + if !self.config.model.is_empty() { + body["model"] = json!(self.config.model); + } + if let Some(tools) = tools.filter(|t| !t.is_empty()) { + body["tools"] = tools.schema(); + body["tool_choice"] = json!("auto"); + } + body + } + + /// Envía la conversación y entrega la respuesta a trozos. + /// + /// `on_delta` devuelve `false` para cortar —lo que hace el orquestador + /// cuando el usuario interrumpe—; `cancel` hace lo mismo desde otro hilo. + pub fn stream( + &self, + chat: &Conversation, + tools: Option<&ToolRegistry>, + cancel: &Cancel, + mut on_delta: impl FnMut(&Delta) -> bool, + ) -> Result { + let body = self.request_body(chat, tools); + let started = Instant::now(); + let mut outcome = StreamOutcome::default(); + let mut assembler = ToolCallAssembler::default(); + let mut finished = false; + + let result = self + .http + .post_json_lines("/v1/chat/completions", &body, cancel, |line| { + let Some(payload) = line.strip_prefix("data: ") else { + return true; + }; + if payload.trim() == "[DONE]" { + finished = true; + return false; + } + let Ok(event) = serde_json::from_str::(payload) else { + tracing::debug!(target: "llm", %payload, "fragmento SSE ilegible"); + return true; + }; + let Some(choice) = event.get("choices").and_then(|c| c.get(0)) else { + return true; + }; + let delta = choice.get("delta").unwrap_or(&Value::Null); + + if let Some(calls) = delta.get("tool_calls").and_then(Value::as_array) { + assembler.absorb(calls); + } + if let Some(text) = delta.get("content").and_then(Value::as_str) { + if !text.is_empty() { + if outcome.ttft.is_none() { + outcome.ttft = Some(started.elapsed()); + } + outcome.text.push_str(text); + if !on_delta(&Delta::Text(text.to_string())) { + outcome.stopped_early = true; + return false; + } + } + } + // Un `finish_reason` cierra la respuesta aunque el servidor no + // llegue a mandar el [DONE] (pasa al cortar la conexión). + if choice.get("finish_reason").is_some_and(|r| !r.is_null()) { + finished = true; + return false; + } + true + }); + + match result { + Ok(()) => {} + // Cortar a propósito no es un fallo: `on_delta` devolvió false o + // se activó la cancelación. + Err(Error::Cancelled) if finished || outcome.stopped_early || cancel.is_cancelled() => { + } + Err(err) => return Err(err), + } + + outcome.tool_calls = assembler.finish(); + if !outcome.tool_calls.is_empty() + && !on_delta(&Delta::ToolCalls(outcome.tool_calls.clone())) + { + outcome.stopped_early = true; + } + Ok(outcome) + } +} + +/// Reensambla las llamadas a herramientas, que llegan repartidas en fragmentos. +/// +/// El streaming manda el nombre en un fragmento y los argumentos en varios +/// más, identificados sólo por su `index`; hasta que no termina el flujo no +/// hay una llamada completa que ejecutar. +#[derive(Default)] +struct ToolCallAssembler { + partial: Vec, +} + +#[derive(Default, Clone)] +struct PartialCall { + id: String, + name: String, + arguments: String, +} + +impl ToolCallAssembler { + fn absorb(&mut self, calls: &[Value]) { + for call in calls { + let index = call.get("index").and_then(Value::as_u64).unwrap_or(0) as usize; + if self.partial.len() <= index { + self.partial.resize(index + 1, PartialCall::default()); + } + let slot = &mut self.partial[index]; + if let Some(id) = call.get("id").and_then(Value::as_str) { + if !id.is_empty() { + slot.id = id.to_string(); + } + } + let Some(function) = call.get("function") else { + continue; + }; + if let Some(name) = function.get("name").and_then(Value::as_str) { + if !name.is_empty() { + slot.name = name.to_string(); + } + } + if let Some(args) = function.get("arguments").and_then(Value::as_str) { + slot.arguments.push_str(args); + } + } + } + + fn finish(self) -> Vec { + self.partial + .into_iter() + .enumerate() + .filter(|(_, call)| !call.name.is_empty()) + .map(|(i, call)| ToolCall { + id: if call.id.is_empty() { + format!("call_{i}") + } else { + call.id + }, + name: call.name, + arguments: if call.arguments.is_empty() { + "{}".into() + } else { + call.arguments + }, + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn absorb(assembler: &mut ToolCallAssembler, raw: &str) { + let value: Value = serde_json::from_str(raw).unwrap(); + assembler.absorb(value.as_array().unwrap()); + } + + #[test] + fn los_argumentos_troceados_se_reensamblan() { + let mut assembler = ToolCallAssembler::default(); + absorb( + &mut assembler, + r#"[{"index":0,"id":"c1","function":{"name":"eco","arguments":"{\"a\":"}}]"#, + ); + absorb( + &mut assembler, + r#"[{"index":0,"function":{"arguments":"1}"}}]"#, + ); + let calls = assembler.finish(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].id, "c1"); + assert_eq!(calls[0].name, "eco"); + assert_eq!(calls[0].arguments, r#"{"a":1}"#); + } + + #[test] + fn se_reensamblan_varias_llamadas_en_paralelo() { + let mut assembler = ToolCallAssembler::default(); + absorb( + &mut assembler, + r#"[{"index":0,"id":"a","function":{"name":"uno","arguments":"{}"}}]"#, + ); + absorb( + &mut assembler, + r#"[{"index":1,"id":"b","function":{"name":"dos","arguments":"{}"}}]"#, + ); + let calls = assembler.finish(); + assert_eq!(calls.len(), 2); + assert_eq!(calls[0].name, "uno"); + assert_eq!(calls[1].name, "dos"); + } + + #[test] + fn sin_argumentos_se_asume_el_objeto_vacio() { + let mut assembler = ToolCallAssembler::default(); + absorb( + &mut assembler, + r#"[{"index":0,"id":"c","function":{"name":"hora_actual"}}]"#, + ); + assert_eq!(assembler.finish()[0].arguments, "{}"); + } + + #[test] + fn un_hueco_sin_nombre_no_produce_una_llamada() { + // El servidor puede numerar el índice 1 sin haber mandado nunca el 0. + let mut assembler = ToolCallAssembler::default(); + absorb( + &mut assembler, + r#"[{"index":1,"id":"b","function":{"name":"dos","arguments":"{}"}}]"#, + ); + let calls = assembler.finish(); + assert_eq!( + calls.len(), + 1, + "el hueco no debe convertirse en una llamada vacía" + ); + assert_eq!(calls[0].name, "dos"); + } + + #[test] + fn sin_id_se_genera_uno_estable() { + let mut assembler = ToolCallAssembler::default(); + absorb( + &mut assembler, + r#"[{"index":0,"function":{"name":"x","arguments":"{}"}}]"#, + ); + assert_eq!(assembler.finish()[0].id, "call_0"); + } +} diff --git a/crates/asist-llm/src/lib.rs b/crates/asist-llm/src/lib.rs new file mode 100644 index 0000000..4c860df --- /dev/null +++ b/crates/asist-llm/src/lib.rs @@ -0,0 +1,12 @@ +//! Cliente de llama-server. +//! +//! Habla la API de chat de OpenAI en modo streaming, porque en un asistente de +//! voz la respuesta no se espera: se trocea en frases y se va sintetizando +//! mientras el modelo sigue escribiendo. Esperar a la respuesta entera sumaría +//! el tiempo del modelo al del sintetizador en lugar de solaparlos. + +pub mod chat; +pub mod client; + +pub use chat::{Message, Role}; +pub use client::{Delta, LlmClient, StreamOutcome}; -- cgit v1.2.3