1use std::path::{Path, PathBuf};
24
25use ort::session::builder::GraphOptimizationLevel;
26use ort::session::Session;
27use ort::value::Tensor;
28use tokenizers::Tokenizer;
29
30use crate::semantic_index::{format_embedding_init_error, pre_validate_onnx_runtime};
31use crate::slog_info;
32
33const MINILM_REPO: &str = "Qdrant/all-MiniLM-L6-v2-onnx";
36const MINILM_MODEL_FILE: &str = "model.onnx";
37const MINILM_TOKENIZER_FILE: &str = "tokenizer.json";
38const MINILM_MAX_LENGTH: usize = 512;
43const MAX_BATCH_ATTENTION_UNITS: usize = 4_000_000;
55
56const MAX_ORT_INTRA_THREADS: usize = 8;
57
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59enum CgroupCpuQuota {
60 Limited(usize),
61 Unlimited,
62 Invalid,
63}
64
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66struct IntraThreadDerivation {
67 threads: usize,
68 source: &'static str,
69 available_parallelism: usize,
70 quota_threads: Option<usize>,
71}
72
73fn quota_threads(quota: u64, period: u64) -> Option<usize> {
74 if quota == 0 || period == 0 {
75 return None;
76 }
77 usize::try_from(quota.div_ceil(period))
78 .ok()
79 .map(|value| value.max(1))
80}
81
82fn parse_cgroup_v2_cpu_max(contents: &str) -> CgroupCpuQuota {
83 let mut fields = contents.split_whitespace();
84 let Some(quota) = fields.next() else {
85 return CgroupCpuQuota::Invalid;
86 };
87 let Some(period) = fields.next() else {
88 return CgroupCpuQuota::Invalid;
89 };
90 if fields.next().is_some() {
91 return CgroupCpuQuota::Invalid;
92 }
93 if quota == "max" {
94 return period
95 .parse::<u64>()
96 .ok()
97 .filter(|period| *period > 0)
98 .map_or(CgroupCpuQuota::Invalid, |_| CgroupCpuQuota::Unlimited);
99 }
100 match (quota.parse::<u64>().ok(), period.parse::<u64>().ok()) {
101 (Some(quota), Some(period)) => quota_threads(quota, period)
102 .map(CgroupCpuQuota::Limited)
103 .unwrap_or(CgroupCpuQuota::Invalid),
104 _ => CgroupCpuQuota::Invalid,
105 }
106}
107
108fn parse_cgroup_v1_cpu_quota(quota: &str, period: &str) -> CgroupCpuQuota {
109 let Ok(quota) = quota.trim().parse::<i64>() else {
110 return CgroupCpuQuota::Invalid;
111 };
112 let Ok(period) = period.trim().parse::<u64>() else {
113 return CgroupCpuQuota::Invalid;
114 };
115 if quota < 0 {
116 return if period > 0 {
117 CgroupCpuQuota::Unlimited
118 } else {
119 CgroupCpuQuota::Invalid
120 };
121 }
122 quota_threads(quota as u64, period)
123 .map(CgroupCpuQuota::Limited)
124 .unwrap_or(CgroupCpuQuota::Invalid)
125}
126
127fn derive_intra_threads(
128 available_parallelism: usize,
129 v2_cpu_max: Option<&str>,
130 v1_cpu_quota_us: Option<&str>,
131 v1_cpu_period_us: Option<&str>,
132) -> IntraThreadDerivation {
133 let available_parallelism = available_parallelism.max(1);
134 let parallelism_threads = available_parallelism.div_ceil(2).max(1);
135 let quota = match v2_cpu_max.map(parse_cgroup_v2_cpu_max) {
136 Some(CgroupCpuQuota::Invalid) | None => match (v1_cpu_quota_us, v1_cpu_period_us) {
137 (Some(quota), Some(period)) => parse_cgroup_v1_cpu_quota(quota, period),
138 _ => CgroupCpuQuota::Invalid,
139 },
140 Some(quota) => quota,
141 };
142 let quota_threads = match quota {
143 CgroupCpuQuota::Limited(threads) => Some(threads),
144 CgroupCpuQuota::Unlimited | CgroupCpuQuota::Invalid => None,
145 };
146 let threads = parallelism_threads
147 .min(quota_threads.unwrap_or(usize::MAX))
148 .min(MAX_ORT_INTRA_THREADS)
149 .max(1);
150 let source = if quota_threads.is_some_and(|quota| quota <= threads) {
151 "quota"
152 } else if parallelism_threads > MAX_ORT_INTRA_THREADS {
153 "cap"
154 } else {
155 "parallelism"
156 };
157 IntraThreadDerivation {
158 threads,
159 source,
160 available_parallelism,
161 quota_threads,
162 }
163}
164
165#[cfg(target_os = "linux")]
166fn read_first(paths: &[&str]) -> Option<String> {
167 paths
168 .iter()
169 .find_map(|path| std::fs::read_to_string(path).ok())
170}
171
172fn intra_thread_derivation() -> IntraThreadDerivation {
173 let available_parallelism = std::thread::available_parallelism()
174 .map(|parallelism| parallelism.get())
175 .unwrap_or(1);
176 #[cfg(target_os = "linux")]
177 {
178 let v2 = read_first(&["/sys/fs/cgroup/cpu.max"]);
179 let v1_quota = read_first(&[
180 "/sys/fs/cgroup/cpu/cpu.cfs_quota_us",
181 "/sys/fs/cgroup/cpu.cfs_quota_us",
182 ]);
183 let v1_period = read_first(&[
184 "/sys/fs/cgroup/cpu/cpu.cfs_period_us",
185 "/sys/fs/cgroup/cpu.cfs_period_us",
186 ]);
187 return derive_intra_threads(
188 available_parallelism,
189 v2.as_deref(),
190 v1_quota.as_deref(),
191 v1_period.as_deref(),
192 );
193 }
194 #[cfg(not(target_os = "linux"))]
195 derive_intra_threads(available_parallelism, None, None, None)
196}
197
198pub struct LocalEmbedder {
199 session: Session,
200 tokenizer: Tokenizer,
201 wants_token_type_ids: bool,
202}
203
204impl LocalEmbedder {
205 pub fn new(model: &str) -> Result<Self, String> {
208 match model {
209 "all-MiniLM-L6-v2" | "all-minilm-l6-v2" => {}
210 other => {
211 return Err(format!(
212 "unsupported local embedding model '{other}'. Supported: all-MiniLM-L6-v2"
213 ))
214 }
215 }
216
217 pre_validate_onnx_runtime()?;
220
221 let (model_path, tokenizer_path) = resolve_model_files()?;
222
223 let thread_derivation = intra_thread_derivation();
224 let threads = thread_derivation.threads;
225 let session = Session::builder()
226 .map_err(|e| format!("failed to create ONNX session builder: {e}"))?
227 .with_optimization_level(GraphOptimizationLevel::Level3)
228 .map_err(|e| format!("failed to set ONNX optimization level: {e}"))?
229 .with_intra_threads(threads)
230 .map_err(|e| format!("failed to set ONNX intra-op threads: {e}"))?
231 .commit_from_file(&model_path)
232 .map_err(format_embedding_init_error)?;
236
237 let mut tokenizer = Tokenizer::from_file(&tokenizer_path)
238 .map_err(|e| format!("failed to load tokenizer {}: {e}", tokenizer_path.display()))?;
239 tokenizer
242 .with_truncation(Some(tokenizers::TruncationParams {
243 max_length: MINILM_MAX_LENGTH,
244 ..Default::default()
245 }))
246 .map_err(|e| format!("failed to set tokenizer truncation: {e}"))?;
247
248 let wants_token_type_ids = session
249 .inputs()
250 .iter()
251 .any(|input| input.name() == "token_type_ids");
252
253 slog_info!(
254 "local embedder ready: model=all-MiniLM-L6-v2 intra_threads={} intra_threads_source={} available_parallelism={} cgroup_quota_threads={} token_type_ids={}",
255 threads,
256 thread_derivation.source,
257 thread_derivation.available_parallelism,
258 thread_derivation
259 .quota_threads
260 .map(|value| value.to_string())
261 .unwrap_or_else(|| "none".to_string()),
262 wants_token_type_ids
263 );
264
265 Ok(Self {
266 session,
267 tokenizer,
268 wants_token_type_ids,
269 })
270 }
271
272 pub fn embed(&mut self, texts: &[String]) -> Result<Vec<Vec<f32>>, String> {
283 if texts.is_empty() {
284 return Ok(Vec::new());
285 }
286
287 let text_refs: Vec<&str> = texts.iter().map(String::as_str).collect();
288 let encodings = self
289 .tokenizer
290 .encode_batch(text_refs, true)
291 .map_err(|e| format!("tokenize batch: {e}"))?;
292
293 let mut result = Vec::with_capacity(encodings.len());
297 let mut batch_start = 0usize;
298 let mut batch_max = 0usize;
299 for (i, enc) in encodings.iter().enumerate() {
300 let len = enc.get_ids().len().max(1);
301 let count = i - batch_start; let candidate_max = batch_max.max(len);
303 let cost = (count + 1)
304 .saturating_mul(candidate_max)
305 .saturating_mul(candidate_max);
306 if count > 0 && cost > MAX_BATCH_ATTENTION_UNITS {
307 let vecs = self.run_inference(&encodings[batch_start..i])?;
308 result.extend(vecs);
309 batch_start = i;
310 batch_max = len;
311 } else {
312 batch_max = candidate_max;
313 }
314 }
315 let vecs = self.run_inference(&encodings[batch_start..])?;
317 result.extend(vecs);
318 Ok(result)
319 }
320
321 fn run_inference(
326 &mut self,
327 encodings: &[tokenizers::Encoding],
328 ) -> Result<Vec<Vec<f32>>, String> {
329 if encodings.is_empty() {
330 return Ok(Vec::new());
331 }
332
333 let batch = encodings.len();
334 let max_len = encodings
335 .iter()
336 .map(|e| e.get_ids().len())
337 .max()
338 .unwrap_or(1)
339 .max(1);
340
341 let mut ids = vec![0i64; batch * max_len];
345 let mut mask = vec![0i64; batch * max_len];
346 for (row, enc) in encodings.iter().enumerate() {
347 let row_ids = enc.get_ids();
348 let row_mask = enc.get_attention_mask();
349 let base = row * max_len;
350 for col in 0..row_ids.len() {
351 ids[base + col] = row_ids[col] as i64;
352 mask[base + col] = row_mask[col] as i64;
353 }
354 }
355
356 let input_ids = ndarray::Array2::<i64>::from_shape_vec((batch, max_len), ids)
357 .map_err(|e| format!("build input_ids tensor: {e}"))?;
358 let attention_mask = ndarray::Array2::<i64>::from_shape_vec((batch, max_len), mask)
359 .map_err(|e| format!("build attention_mask tensor: {e}"))?;
360
361 let mut inputs = ort::inputs![
362 "input_ids" => Tensor::from_array(input_ids).map_err(|e| format!("input_ids: {e}"))?,
363 "attention_mask" => Tensor::from_array(attention_mask.clone())
364 .map_err(|e| format!("attention_mask: {e}"))?,
365 ];
366 if self.wants_token_type_ids {
367 let token_type_ids = ndarray::Array2::<i64>::zeros((batch, max_len));
368 inputs.push((
369 "token_type_ids".into(),
370 Tensor::from_array(token_type_ids)
371 .map_err(|e| format!("token_type_ids: {e}"))?
372 .into(),
373 ));
374 }
375
376 let outputs = self
377 .session
378 .run(inputs)
379 .map_err(|e| format!("ONNX inference failed: {e}"))?;
380 let output = outputs
381 .values()
382 .next()
383 .ok_or_else(|| "ONNX model produced no output".to_string())?;
384
385 let (shape, data): (Vec<i64>, Vec<f32>) = match output.try_extract_tensor::<f32>() {
387 Ok((s, d)) => (s.to_vec(), d.to_vec()),
388 Err(_) => {
389 let (s, d) = output
390 .try_extract_tensor::<half::f16>()
391 .map_err(|e| format!("extract output tensor: {e}"))?;
392 (s.to_vec(), d.iter().map(|h| h.to_f32()).collect())
393 }
394 };
395 if shape.len() != 3 {
396 return Err(format!(
397 "unexpected ONNX output rank {} (expected 3: [batch, seq, dim])",
398 shape.len()
399 ));
400 }
401 let seq = shape[1] as usize;
402 let dim = shape[2] as usize;
403
404 let mut result = Vec::with_capacity(batch);
405 for row in 0..batch {
406 let mut emb = vec![0.0f32; dim];
407 let mut valid = 0.0f32;
408 for col in 0..seq {
409 if mask_at(&attention_mask, row, col) == 1 {
410 valid += 1.0;
411 let base = (row * seq + col) * dim;
412 for (d, slot) in emb.iter_mut().enumerate() {
413 *slot += data[base + d];
414 }
415 }
416 }
417 let denom = if valid == 0.0 { 1.0 } else { valid };
418 for slot in &mut emb {
419 *slot /= denom;
420 }
421 let norm = emb.iter().map(|x| x * x).sum::<f32>().sqrt();
422 for slot in &mut emb {
423 *slot /= norm + 1e-12;
424 }
425 result.push(emb);
426 }
427 Ok(result)
428 }
429}
430
431#[inline]
432fn mask_at(mask: &ndarray::Array2<i64>, row: usize, col: usize) -> i64 {
433 mask[[row, col]]
434}
435
436fn resolve_model_files() -> Result<(PathBuf, PathBuf), String> {
439 let cache_dir = embedding_cache_dir()?;
440
441 if let Some(found) = scan_local_snapshot(&cache_dir) {
442 return Ok(found);
443 }
444
445 download_via_hf_hub(&cache_dir)
448}
449
450fn embedding_cache_dir() -> Result<PathBuf, String> {
454 embedding_cache_dir_from(
455 |name| crate::environment::non_empty_os_var(name),
456 std::env::home_dir().as_deref(),
457 )
458 .ok_or_else(|| "could not determine a home directory for the fastembed cache".to_string())
459}
460
461fn embedding_cache_dir_from(
462 lookup: impl Fn(&str) -> Option<std::ffi::OsString>,
463 fallback_home: Option<&Path>,
464) -> Option<PathBuf> {
465 let non_empty = |name| lookup(name).filter(|value| !value.is_empty());
466 if let Some(dir) = non_empty("FASTEMBED_CACHE_DIR") {
467 return Some(PathBuf::from(dir));
468 }
469 non_empty("HOME")
470 .or_else(|| non_empty("USERPROFILE"))
471 .map(PathBuf::from)
472 .or_else(|| fallback_home.map(PathBuf::from))
473 .map(|home| home.join(".cache").join("fastembed"))
474}
475
476fn scan_local_snapshot(cache_dir: &std::path::Path) -> Option<(PathBuf, PathBuf)> {
479 let repo_dir = cache_dir.join("models--Qdrant--all-MiniLM-L6-v2-onnx");
480 let snapshots = repo_dir.join("snapshots");
481 let mut candidates: Vec<PathBuf> = std::fs::read_dir(&snapshots)
482 .ok()?
483 .filter_map(|entry| entry.ok().map(|e| e.path()))
484 .filter(|p| p.is_dir())
485 .collect();
486 candidates.sort_by_key(|p| {
488 std::fs::metadata(p)
489 .and_then(|m| m.modified())
490 .unwrap_or(std::time::UNIX_EPOCH)
491 });
492 candidates.reverse();
493 for snap in candidates {
494 let model = snap.join(MINILM_MODEL_FILE);
495 let tokenizer = snap.join(MINILM_TOKENIZER_FILE);
496 if model.is_file() && tokenizer.is_file() {
497 return Some((model, tokenizer));
498 }
499 }
500 None
501}
502
503fn download_via_hf_hub(cache_dir: &std::path::Path) -> Result<(PathBuf, PathBuf), String> {
504 use hf_hub::api::sync::ApiBuilder;
505
506 slog_info!(
507 "downloading all-MiniLM-L6-v2 ({}) to {}",
508 MINILM_REPO,
509 cache_dir.display()
510 );
511 let api = ApiBuilder::new()
512 .with_progress(false)
513 .with_cache_dir(cache_dir.to_path_buf())
514 .build()
515 .map_err(|e| format!("failed to init hf-hub api: {e}"))?;
516 let repo = api.model(MINILM_REPO.to_string());
517 let model = repo
518 .get(MINILM_MODEL_FILE)
519 .map_err(|e| format!("failed to download {MINILM_MODEL_FILE}: {e}"))?;
520 let tokenizer = repo
521 .get(MINILM_TOKENIZER_FILE)
522 .map_err(|e| format!("failed to download {MINILM_TOKENIZER_FILE}: {e}"))?;
523 Ok((model, tokenizer))
524}
525
526#[cfg(test)]
527mod tests {
528 use super::{
529 derive_intra_threads, embedding_cache_dir_from, parse_cgroup_v1_cpu_quota,
530 parse_cgroup_v2_cpu_max, CgroupCpuQuota, MINILM_MAX_LENGTH,
531 };
532 use std::io::Write;
533 use tokenizers::Tokenizer;
534
535 #[test]
536 fn empty_fastembed_and_home_rungs_are_unset_with_an_injected_lookup() {
537 let empty = std::ffi::OsString::new();
538 let cache = embedding_cache_dir_from(
539 |name| match name {
540 "FASTEMBED_CACHE_DIR" | "HOME" => Some(empty.clone()),
541 "USERPROFILE" => Some(std::ffi::OsString::from("/profile")),
542 _ => None,
543 },
544 None,
545 );
546 assert_eq!(
547 cache,
548 Some(std::path::PathBuf::from("/profile/.cache/fastembed"))
549 );
550 assert_eq!(embedding_cache_dir_from(|_| None, None), None);
551 }
552
553 fn minilm_like_tokenizer_json() -> Vec<u8> {
554 serde_json::json!({
555 "version": "1.0",
556 "truncation": {
557 "direction": "Right",
558 "max_length": MINILM_MAX_LENGTH,
559 "strategy": "LongestFirst",
560 "stride": 0
561 },
562 "padding": null,
563 "added_tokens": [
564 {"id": 0, "content": "[PAD]", "single_word": false, "lstrip": false, "rstrip": false, "normalized": false, "special": true},
565 {"id": 1, "content": "[CLS]", "single_word": false, "lstrip": false, "rstrip": false, "normalized": false, "special": true},
566 {"id": 2, "content": "[SEP]", "single_word": false, "lstrip": false, "rstrip": false, "normalized": false, "special": true},
567 {"id": 3, "content": "[UNK]", "single_word": false, "lstrip": false, "rstrip": false, "normalized": false, "special": true}
568 ],
569 "normalizer": {
570 "type": "BertNormalizer",
571 "clean_text": true,
572 "handle_chinese_chars": true,
573 "strip_accents": null,
574 "lowercase": true
575 },
576 "pre_tokenizer": {"type": "BertPreTokenizer"},
577 "post_processor": {"type": "BertProcessing", "sep": ["[SEP]", 2], "cls": ["[CLS]", 1]},
578 "decoder": null,
579 "model": {
580 "type": "WordPiece",
581 "unk_token": "[UNK]",
582 "continuing_subword_prefix": "##",
583 "max_input_chars_per_word": 100,
584 "vocab": {
585 "[PAD]": 0,
586 "[CLS]": 1,
587 "[SEP]": 2,
588 "[UNK]": 3,
589 "hello": 4,
590 "world": 5,
591 "!": 6,
592 "cafe": 7,
593 "naive": 8,
594 "##ly": 9
595 }
596 }
597 })
598 .to_string()
599 .into_bytes()
600 }
601
602 fn assert_load_encode_parity(tokenizer: Tokenizer) {
603 let ascii = tokenizer.encode("Hello WORLD!", true).unwrap();
604 assert_eq!(ascii.get_ids(), &[1, 4, 5, 6, 2]);
605
606 let unicode = tokenizer.encode("Café naïvely", true).unwrap();
607 assert_eq!(unicode.get_ids(), &[1, 7, 8, 9, 2]);
608
609 let long_text = std::iter::repeat("hello")
610 .take(MINILM_MAX_LENGTH + 20)
611 .collect::<Vec<_>>()
612 .join(" ");
613 let long = tokenizer.encode(long_text.as_str(), true).unwrap();
614 let ids = long.get_ids();
615 assert_eq!(ids.len(), MINILM_MAX_LENGTH);
616 assert_eq!(ids.first(), Some(&1));
617 assert_eq!(ids.last(), Some(&2));
618 assert!(ids[1..MINILM_MAX_LENGTH - 1].iter().all(|id| *id == 4));
619 }
620
621 #[test]
622 fn cgroup_cpu_quota_parsing_and_thread_derivation_cover_v2_v1_and_absence() {
623 assert_eq!(
624 parse_cgroup_v2_cpu_max("max 100000\n"),
625 CgroupCpuQuota::Unlimited
626 );
627 assert_eq!(
628 parse_cgroup_v2_cpu_max("200000 100000\n"),
629 CgroupCpuQuota::Limited(2)
630 );
631 assert_eq!(
632 parse_cgroup_v1_cpu_quota("-1\n", "100000\n"),
633 CgroupCpuQuota::Unlimited
634 );
635 assert_eq!(
636 parse_cgroup_v1_cpu_quota("200000\n", "100000\n"),
637 CgroupCpuQuota::Limited(2)
638 );
639
640 let v2_limited = derive_intra_threads(64, Some("200000 100000"), None, None);
641 assert_eq!(v2_limited.threads, 2);
642 assert_eq!(v2_limited.source, "quota");
643
644 let v1_limited = derive_intra_threads(64, None, Some("200000"), Some("100000"));
645 assert_eq!(v1_limited.threads, 2);
646 assert_eq!(v1_limited.source, "quota");
647
648 let v2_unlimited = derive_intra_threads(64, Some("max 100000"), None, None);
649 assert_eq!(v2_unlimited.threads, 8);
650 assert_eq!(v2_unlimited.source, "cap");
651 assert_eq!(v2_unlimited.quota_threads, None);
652
653 let v1_unlimited = derive_intra_threads(64, None, Some("-1"), Some("100000"));
654 assert_eq!(v1_unlimited.threads, 8);
655 assert_eq!(v1_unlimited.source, "cap");
656 assert_eq!(v1_unlimited.quota_threads, None);
657
658 let absent = derive_intra_threads(64, None, None, None);
659 assert_eq!(absent.threads, 8);
660 assert_eq!(absent.source, "cap");
661 assert_eq!(absent.quota_threads, None);
662 }
663
664 #[test]
665 fn tokenizers_slim_features_load_and_encode_minilm_wordpiece() {
666 let json = minilm_like_tokenizer_json();
667
668 assert_load_encode_parity(Tokenizer::from_bytes(&json).unwrap());
669
670 let mut file = tempfile::NamedTempFile::new().unwrap();
671 file.write_all(&json).unwrap();
672 file.flush().unwrap();
673 assert_load_encode_parity(Tokenizer::from_file(file.path()).unwrap());
674 }
675}