1use crate::catalog::CatalogModel;
10use crate::engine::chat_template::{merge_kwargs, render_chat, TemplateVars};
11use crate::engine::llm_core::{
12 chat_messages, completion_json, effective_context, finish_for, plan_budget, should_add_bos,
13 StopHold,
14};
15use crate::engine::{Engine, EngineCapabilities};
16use crate::host::{ChatModel, LoadedModel};
17use crate::types::*;
18use anyhow::{anyhow, bail, Context, Result};
19use llama_cpp_2::context::params::LlamaContextParams;
20use llama_cpp_2::llama_backend::LlamaBackend;
21use llama_cpp_2::llama_batch::LlamaBatch;
22use llama_cpp_2::model::params::LlamaModelParams;
23use llama_cpp_2::model::{AddBos, LlamaModel};
24use llama_cpp_2::sampling::LlamaSampler;
25use parking_lot::Mutex;
26use std::collections::BTreeMap;
27use std::num::NonZeroU32;
28use std::path::{Path, PathBuf};
29use std::sync::Arc;
30use std::time::Instant;
31use tracing::{debug, info, warn};
32
33const TRACE_TARGET: &str = "studio_worker::engine::llama";
36
37pub struct LlamaEngine {
38 backend: Arc<LlamaBackend>,
39 models_root: PathBuf,
40}
41
42static GLOBAL_BACKEND: std::sync::OnceLock<Arc<LlamaBackend>> = std::sync::OnceLock::new();
46
47static BACKEND_INIT_LOCK: Mutex<()> = Mutex::new(());
54
55fn global_backend() -> Result<Arc<LlamaBackend>> {
56 if let Some(b) = GLOBAL_BACKEND.get() {
57 return Ok(b.clone());
58 }
59 let _guard = BACKEND_INIT_LOCK.lock();
60 if let Some(b) = GLOBAL_BACKEND.get() {
63 return Ok(b.clone());
64 }
65 let backend = LlamaBackend::init().map_err(|e| match e {
66 llama_cpp_2::LlamaCppError::BackendAlreadyInitialized => anyhow!(
70 "llama backend was initialised outside global_backend(); no shared handle available"
71 ),
72 other => anyhow!(other),
73 })?;
74 let arc = Arc::new(backend);
75 let _ = GLOBAL_BACKEND.set(arc.clone());
76 Ok(arc)
77}
78
79impl LlamaEngine {
80 pub fn new(models_root: PathBuf) -> Result<Self> {
81 let backend = global_backend().context("initialising llama backend")?;
82 Ok(Self {
83 backend,
84 models_root,
85 })
86 }
87
88 fn llm_dir(&self) -> PathBuf {
89 self.models_root.join("llm")
90 }
91
92 fn list_models(&self) -> Vec<(String, PathBuf)> {
93 let dir = self.llm_dir();
94 let Ok(read) = std::fs::read_dir(&dir) else {
95 return Vec::new();
96 };
97 let mut out = Vec::new();
98 for entry in read.flatten() {
99 let p = entry.path();
100 if p.extension().and_then(|s| s.to_str()) == Some("gguf") {
101 if let Some(stem) = p.file_stem().and_then(|s| s.to_str()) {
102 out.push((stem.to_string(), p));
103 }
104 }
105 }
106 out
107 }
108
109 fn resolve_path(&self, model: &str) -> Option<PathBuf> {
110 self.list_models()
111 .into_iter()
112 .find(|(stem, _)| stem == model)
113 .map(|(_, p)| p)
114 }
115
116 fn load_transient(&self, model: &str, path: &Path) -> Result<LlamaModel> {
120 info!(
121 target: TRACE_TARGET,
122 op = "load",
123 model,
124 path = %path.display(),
125 "loading model for one job"
126 );
127 let started = Instant::now();
128 let loaded = load_model(&self.backend, model, path).inspect_err(|e| {
129 warn!(
130 target: TRACE_TARGET,
131 op = "load",
132 model,
133 path = %path.display(),
134 elapsed_ms = started.elapsed().as_millis() as u64,
135 error = %e,
136 "failed to load model"
137 );
138 })?;
139 info!(
140 target: TRACE_TARGET,
141 op = "load",
142 model,
143 elapsed_ms = started.elapsed().as_millis() as u64,
144 "model loaded"
145 );
146 Ok(loaded)
147 }
148}
149
150fn append_piece<E: std::fmt::Display>(
161 out: &mut String,
162 step: usize,
163 piece: std::result::Result<String, E>,
164 decode_failures: &mut u32,
165) {
166 match piece {
167 Ok(s) => out.push_str(&s),
168 Err(e) => {
169 *decode_failures += 1;
170 warn!(
171 target: TRACE_TARGET,
172 op = "generate",
173 step,
174 error = %e,
175 "llama token piece decode failed; dropping it from the completion"
176 );
177 }
178 }
179}
180
181fn gpu_layers() -> u32 {
183 if cfg!(feature = "cuda") {
184 999
185 } else {
186 0
187 }
188}
189
190const PROMPT_BATCH: usize = 2048;
193
194fn load_model(backend: &LlamaBackend, id: &str, path: &Path) -> Result<LlamaModel> {
195 let params = LlamaModelParams::default().with_n_gpu_layers(gpu_layers());
196 LlamaModel::load_from_file(backend, path, ¶ms)
197 .with_context(|| format!("loading model {id} from {}", path.display()))
198}
199
200fn template_vars(model: &LlamaModel) -> TemplateVars {
202 let text = |token| {
203 let mut decoder = encoding_rs::UTF_8.new_decoder();
204 model
205 .token_to_piece(token, &mut decoder, true, None)
206 .unwrap_or_default()
207 };
208 TemplateVars {
209 bos_token: text(model.token_bos()),
210 eos_token: text(model.token_eos()),
211 }
212}
213
214fn render_request(
216 model: &LlamaModel,
217 defaults: &ModelCliDefaults,
218 params: &LlmParams,
219) -> Result<String> {
220 let template = model
221 .meta_val_str("tokenizer.chat_template")
222 .map_err(|_| anyhow!("model has no chat template (tokenizer.chat_template)"))?;
223 let kwargs = merge_kwargs(
224 defaults.chat_template_kwargs.as_ref(),
225 params.chat_template_kwargs.as_ref(),
226 );
227 Ok(render_chat(
228 &template,
229 &chat_messages(params),
230 &kwargs,
231 &template_vars(model),
232 )?)
233}
234
235fn model_adds_bos(model: &LlamaModel) -> bool {
236 model
237 .meta_val_str("tokenizer.ggml.add_bos_token")
238 .map(|v| v == "true")
239 .unwrap_or(false)
240}
241
242pub fn complete(
245 model: &LlamaModel,
246 backend: &LlamaBackend,
247 id: &str,
248 defaults: &ModelCliDefaults,
249 params: LlmParams,
250 cancelled: &dyn Fn() -> bool,
251 on_piece: &mut dyn FnMut(&str),
252) -> Result<serde_json::Value> {
253 let started = Instant::now();
254 let prompt = render_request(model, defaults, ¶ms)?;
255 let bos = template_vars(model).bos_token;
256 let add_bos = if should_add_bos(model_adds_bos(model), &prompt, &bos) {
257 AddBos::Always
258 } else {
259 AddBos::Never
260 };
261 let tokens = model
262 .str_to_token(&prompt, add_bos)
263 .map_err(|e| anyhow!("tokenize prompt: {e:?}"))?;
264 let n_ctx = effective_context(defaults.context_size, model.n_ctx_train());
265 let budget = plan_budget(tokens.len(), params.max_tokens, n_ctx)?;
266 debug!(
267 target: TRACE_TARGET,
268 op = "generate",
269 model = id,
270 prompt_tokens = tokens.len(),
271 budget,
272 n_ctx,
273 "starting generation"
274 );
275 let ctx_params = LlamaContextParams::default()
276 .with_n_ctx(NonZeroU32::new(n_ctx))
277 .with_n_batch(PROMPT_BATCH as u32);
278 let mut ctx = model
279 .new_context(backend, ctx_params)
280 .context("creating llama context")?;
281
282 let mut batch = LlamaBatch::new(PROMPT_BATCH, 1);
283 let mut pos: i32 = 0;
284 let last = tokens.len() - 1;
285 for chunk in tokens.chunks(PROMPT_BATCH) {
286 if cancelled() {
287 bail!("cancelled: the model is unloading or the client left");
288 }
289 batch.clear();
290 for token in chunk {
291 batch
292 .add(*token, pos, &[0], pos as usize == last)
293 .map_err(|e| anyhow!("batch add: {e:?}"))?;
294 pos += 1;
295 }
296 ctx.decode(&mut batch).context("decoding prompt")?;
297 }
298
299 let mut sampler = if params.temperature <= 0.0 {
300 LlamaSampler::greedy()
301 } else {
302 let mut chain = Vec::new();
303 if let Some(p) = params.top_p {
304 chain.push(LlamaSampler::top_p(p, 1));
305 }
306 chain.push(LlamaSampler::temp(params.temperature));
307 chain.push(LlamaSampler::dist(1234));
308 LlamaSampler::chain_simple(chain)
309 };
310 let mut hold = StopHold::new(¶ms.stop.clone().unwrap_or_default());
311 let mut decoder = encoding_rs::UTF_8.new_decoder();
314 let mut piece = String::new();
315 let (mut generated, mut hit_end, mut decode_failures) = (0u32, false, 0u32);
316 while generated < budget {
317 if cancelled() {
318 bail!("cancelled: the model is unloading or the client left");
319 }
320 let token = sampler.sample(&ctx, batch.n_tokens() - 1);
321 sampler.accept(token);
322 if model.is_eog_token(token) {
323 hit_end = true;
324 break;
325 }
326 piece.clear();
327 append_piece(
328 &mut piece,
329 generated as usize,
330 model.token_to_piece(token, &mut decoder, false, None),
331 &mut decode_failures,
332 );
333 generated += 1;
334 let safe = hold.push(&piece);
335 if !safe.is_empty() {
336 on_piece(&safe);
337 }
338 if hold.stopped().is_some() {
339 hit_end = true;
340 break;
341 }
342 batch.clear();
343 batch
344 .add(token, pos, &[0], true)
345 .map_err(|e| anyhow!("batch add (token): {e:?}"))?;
346 pos += 1;
347 ctx.decode(&mut batch).context("decoding token")?;
348 }
349 let rest = hold.finish();
350 if !rest.is_empty() {
351 on_piece(&rest);
352 }
353 let out = hold.text().to_string();
354 let elapsed_ms = started.elapsed().as_millis() as u64;
355 let finish = finish_for(generated, budget, hit_end);
356 info!(
357 target: TRACE_TARGET,
358 op = "generate",
359 model = id,
360 prompt_tokens = tokens.len(),
361 completion_tokens = generated,
362 finish = finish.as_str(),
363 decode_failures,
364 elapsed_ms,
365 "generation complete"
366 );
367 Ok(completion_json(
368 id,
369 &out,
370 tokens.len(),
371 generated,
372 finish,
373 elapsed_ms,
374 ))
375}
376
377pub struct LoadedLlm {
379 id: String,
380 model: LlamaModel,
381 backend: Arc<LlamaBackend>,
382 defaults: ModelCliDefaults,
383}
384
385impl LoadedModel for LoadedLlm {
386 fn as_any(&self) -> &dyn std::any::Any {
387 self
388 }
389
390 fn as_chat(&self) -> Option<&dyn ChatModel> {
391 Some(self)
392 }
393}
394
395impl ChatModel for LoadedLlm {
396 fn chat(
397 &self,
398 params: LlmParams,
399 cancelled: &dyn Fn() -> bool,
400 on_piece: &mut dyn FnMut(&str),
401 ) -> Result<serde_json::Value> {
402 complete(
403 &self.model,
404 &self.backend,
405 &self.id,
406 &self.defaults,
407 params,
408 cancelled,
409 on_piece,
410 )
411 }
412
413 fn tokenize(&self, text: &str, add_special: bool) -> Result<Vec<i32>> {
414 let add_bos = if add_special {
415 AddBos::Always
416 } else {
417 AddBos::Never
418 };
419 Ok(self
420 .model
421 .str_to_token(text, add_bos)
422 .map_err(|e| anyhow!("tokenize: {e:?}"))?
423 .into_iter()
424 .map(|t| t.0)
425 .collect())
426 }
427}
428
429#[cfg_attr(coverage_nightly, coverage(off))]
431pub fn load_resident(models_root: &Path, model: &CatalogModel) -> Result<LoadedLlm> {
432 let backend = global_backend().context("initialising llama backend")?;
433 let files = ensure_llm_files(models_root, &model.source)?;
434 let path = pick_gguf(&files).ok_or_else(|| {
435 anyhow!(
436 "llama modelSource for `{}` contained no .gguf file",
437 model.id
438 )
439 })?;
440 let started = Instant::now();
441 let loaded = load_model(&backend, &model.id, &path)?;
442 info!(
443 target: TRACE_TARGET,
444 op = "load",
445 model = %model.id,
446 gpu_layers = gpu_layers(),
447 elapsed_ms = started.elapsed().as_millis() as u64,
448 "resident model loaded"
449 );
450 Ok(LoadedLlm {
451 id: model.id.clone(),
452 model: loaded,
453 backend,
454 defaults: model.source.cli_defaults.clone(),
455 })
456}
457
458#[cfg_attr(coverage_nightly, coverage(off))]
460fn ensure_llm_files(
461 models_root: &Path,
462 source: &ModelSource,
463) -> Result<Vec<(ModelFileRole, PathBuf)>> {
464 let dir = models_root.join("llm");
465 source
466 .files
467 .iter()
468 .map(|file| {
469 Ok((
470 file.role,
471 crate::engine::download::ensure_file_reusing(&dir, models_root, file)?,
472 ))
473 })
474 .collect()
475}
476
477const LLAMA_MODEL_WILDCARD: &str = "llama-cpp:*";
484
485fn is_gguf(path: &Path) -> bool {
486 path.extension()
487 .and_then(|s| s.to_str())
488 .map(|e| e.eq_ignore_ascii_case("gguf"))
489 .unwrap_or(false)
490}
491
492fn pick_gguf(files: &[(ModelFileRole, PathBuf)]) -> Option<PathBuf> {
496 files
497 .iter()
498 .find(|(role, path)| matches!(role, ModelFileRole::Model) && is_gguf(path))
499 .or_else(|| files.iter().find(|(_, path)| is_gguf(path)))
500 .map(|(_, path)| path.clone())
501}
502
503fn as_llm(task: Task, model: &str) -> Result<LlmParams> {
506 match task {
507 Task::Llm(p) => Ok(p),
508 other => {
509 warn!(
510 target: TRACE_TARGET,
511 op = "dispatch",
512 kind = other.kind().as_str(),
513 model,
514 "unsupported task kind"
515 );
516 Err(crate::engine::UnsupportedTask::new("llama", other.kind()).into())
517 }
518 }
519}
520
521impl LlamaEngine {
522 fn run_llm(
526 &self,
527 model: &str,
528 path: &Path,
529 defaults: &ModelCliDefaults,
530 llm: LlmParams,
531 ) -> Result<TaskResult> {
532 let loaded = self.load_transient(model, path)?;
533 let json = complete(
534 &loaded,
535 &self.backend,
536 model,
537 defaults,
538 llm,
539 &|| false,
540 &mut |_| {},
541 )
542 .inspect_err(|e| {
543 warn!(
544 target: TRACE_TARGET,
545 op = "dispatch",
546 kind = "llm",
547 model,
548 error = %e,
549 "generation failed"
550 );
551 })?;
552 Ok(TaskResult::Llm { json })
553 }
554}
555
556impl Engine for LlamaEngine {
557 fn name(&self) -> &'static str {
558 "llama"
559 }
560
561 fn capabilities(&self) -> EngineCapabilities {
562 let mut models: Vec<String> = self.list_models().into_iter().map(|(s, _)| s).collect();
567 models.push(LLAMA_MODEL_WILDCARD.to_string());
568 let mut map: BTreeMap<TaskKind, Vec<String>> = BTreeMap::new();
569 map.insert(TaskKind::Llm, models);
570 EngineCapabilities {
571 supported_models_per_kind: map,
572 }
573 }
574
575 fn dispatch(&self, model: &str, task: Task) -> Result<TaskResult> {
576 let llm = as_llm(task, model)?;
577 let path = self.resolve_path(model).ok_or_else(|| {
578 warn!(
579 target: TRACE_TARGET,
580 op = "dispatch",
581 model,
582 models_root = %self.llm_dir().display(),
583 "model not found"
584 );
585 anyhow!(
586 "model `{model}` not found in {} and the offer carried no \
587 modelSource to download it from",
588 self.llm_dir().display()
589 )
590 })?;
591 self.run_llm(model, &path, &ModelCliDefaults::default(), llm)
592 }
593
594 fn dispatch_with_source(
595 &self,
596 model: &str,
597 task: Task,
598 source: &ModelSource,
599 ) -> Result<TaskResult> {
600 let llm = as_llm(task, model)?;
601 let path = if source.files.is_empty() {
604 self.resolve_path(model).ok_or_else(|| {
605 anyhow!(
606 "model `{model}` not found in {} and the offer's modelSource \
607 listed no files to download",
608 self.llm_dir().display()
609 )
610 })?
611 } else {
612 let resolved = ensure_llm_files(&self.models_root, source)?;
613 pick_gguf(&resolved)
614 .ok_or_else(|| anyhow!("llama modelSource for `{model}` contained no .gguf file"))?
615 };
616 self.run_llm(model, &path, &source.cli_defaults, llm)
617 }
618}
619
620#[cfg(test)]
621mod tests {
622 use super::*;
623
624 #[test]
635 fn append_piece_concatenates_ok_pieces() {
636 let mut out = String::new();
637 let mut failures = 0u32;
638 append_piece(&mut out, 0, Ok::<_, &str>("hel".to_string()), &mut failures);
639 append_piece(&mut out, 1, Ok::<_, &str>("lo".to_string()), &mut failures);
640 assert_eq!(out, "hello");
641 assert_eq!(failures, 0);
642 }
643
644 #[test]
645 fn append_piece_drops_and_counts_failed_pieces() {
646 let mut out = String::new();
650 let mut failures = 0u32;
651 append_piece(
652 &mut out,
653 0,
654 Ok::<_, &str>("kept".to_string()),
655 &mut failures,
656 );
657 append_piece(&mut out, 1, Err("invalid utf-8"), &mut failures);
658 append_piece(
659 &mut out,
660 2,
661 Ok::<_, &str>(" tail".to_string()),
662 &mut failures,
663 );
664 assert_eq!(out, "kept tail");
665 assert_eq!(failures, 1);
666 }
667
668 #[test]
669 fn append_piece_warns_on_each_decode_failure() {
670 let logs = crate::test_support::capture(|| {
671 let mut out = String::new();
672 let mut failures = 0u32;
673 append_piece(&mut out, 7, Err("decode boom"), &mut failures);
674 });
675 assert!(
676 logs.contains("studio_worker::engine::llama"),
677 "expected llama target, got: {logs}"
678 );
679 assert!(logs.contains("WARN"), "expected WARN level, got: {logs}");
680 assert!(
681 logs.contains("decode boom"),
682 "expected the underlying error, got: {logs}"
683 );
684 assert!(
685 logs.contains("step=7"),
686 "expected the step index, got: {logs}"
687 );
688 }
689
690 #[test]
694 fn global_backend_never_fails_under_contention() {
695 let handles: Vec<_> = (0..32)
696 .map(|_| std::thread::spawn(|| global_backend().map(|_| ())))
697 .collect();
698 for h in handles {
699 h.join()
700 .expect("thread panicked")
701 .expect("global_backend must never fail under contention");
702 }
703 }
704
705 #[test]
706 fn capabilities_advertise_wildcard_even_with_no_local_models() {
707 let tmp = tempfile::tempdir().unwrap();
711 let engine = LlamaEngine::new(tmp.path().to_path_buf()).expect("init backend");
712 let caps = engine.capabilities();
713 let models = &caps.supported_models_per_kind[&TaskKind::Llm];
714 assert_eq!(models, &vec![LLAMA_MODEL_WILDCARD.to_string()]);
715 assert!(caps.supports(TaskKind::Llm, LLAMA_MODEL_WILDCARD));
716 }
717
718 #[test]
719 fn capabilities_picks_up_gguf_files_and_keeps_wildcard() {
720 let tmp = tempfile::tempdir().unwrap();
721 let llm_dir = tmp.path().join("llm");
722 std::fs::create_dir_all(&llm_dir).unwrap();
723 std::fs::write(llm_dir.join("smollm-135m-q8.gguf"), b"not-real").unwrap();
725 std::fs::write(llm_dir.join("ignored.txt"), b"x").unwrap();
726 let engine = LlamaEngine::new(tmp.path().to_path_buf()).expect("init backend");
727 let caps = engine.capabilities();
728 let models = &caps.supported_models_per_kind[&TaskKind::Llm];
729 assert_eq!(
730 models,
731 &vec![
732 "smollm-135m-q8".to_string(),
733 LLAMA_MODEL_WILDCARD.to_string()
734 ]
735 );
736 }
737
738 #[test]
739 fn is_gguf_matches_extension_case_insensitively() {
740 assert!(is_gguf(Path::new("/m/model.gguf")));
741 assert!(is_gguf(Path::new("/m/model.GGUF")));
742 assert!(!is_gguf(Path::new("/m/model.safetensors")));
743 assert!(!is_gguf(Path::new("/m/model")));
744 }
745
746 #[test]
747 fn pick_gguf_prefers_model_role_then_first_gguf() {
748 let files = vec![
750 (ModelFileRole::TextEncoder, PathBuf::from("/m/clip.gguf")),
751 (ModelFileRole::Model, PathBuf::from("/m/weights.gguf")),
752 ];
753 assert_eq!(pick_gguf(&files), Some(PathBuf::from("/m/weights.gguf")));
754 let files = vec![
756 (ModelFileRole::Vae, PathBuf::from("/m/vae.safetensors")),
757 (ModelFileRole::TextEncoder, PathBuf::from("/m/first.gguf")),
758 (ModelFileRole::Lora, PathBuf::from("/m/second.gguf")),
759 ];
760 assert_eq!(pick_gguf(&files), Some(PathBuf::from("/m/first.gguf")));
761 let files = vec![(ModelFileRole::Vae, PathBuf::from("/m/vae.safetensors"))];
763 assert_eq!(pick_gguf(&files), None);
764 }
765
766 #[test]
767 fn as_llm_extracts_llm_params_and_rejects_other_kinds() {
768 let llm = Task::Llm(LlmParams {
769 messages: vec![ChatMessage {
770 role: "user".into(),
771 content: "hi".into(),
772 }],
773 max_tokens: 8,
774 temperature: 0.1,
775 ..Default::default()
776 });
777 assert!(as_llm(llm, "m").is_ok());
778 let image = Task::Image(ImageParams {
779 prompt: "x".into(),
780 ..Default::default()
781 });
782 let err = as_llm(image, "m").unwrap_err().to_string();
783 assert!(err.contains("cannot serve image"), "got: {err}");
784 }
785
786 #[test]
787 fn dispatch_returns_error_when_model_missing() {
788 let tmp = tempfile::tempdir().unwrap();
789 let engine = LlamaEngine::new(tmp.path().to_path_buf()).expect("init backend");
790 let task = Task::Llm(LlmParams {
791 messages: vec![ChatMessage {
792 role: "user".into(),
793 content: "hi".into(),
794 }],
795 max_tokens: 1,
796 temperature: 0.0,
797 ..Default::default()
798 });
799 let err = engine.dispatch("no-such-model", task).unwrap_err();
800 assert!(err.to_string().contains("not found"));
801 }
802
803 #[test]
804 fn dispatch_rejects_non_llm_tasks() {
805 let tmp = tempfile::tempdir().unwrap();
806 let engine = LlamaEngine::new(tmp.path().to_path_buf()).expect("init backend");
807 let task = Task::Image(ImageParams {
808 prompt: "x".into(),
809 width: 64,
810 height: 64,
811 steps: 1,
812 seed: None,
813 ext: "webp".into(),
814 ..Default::default()
815 });
816 let err = engine.dispatch("anything", task).unwrap_err();
817 assert!(err.to_string().contains("cannot serve image"));
818 }
819}