1use rightkit_onnx::ort;
2use std::path::{Path, PathBuf};
3use std::sync::{Mutex, OnceLock};
4use std::time::{Duration, Instant};
5
6#[cfg(target_os = "macos")]
7use crate::coreml::{self, CoreMlEngine};
8#[cfg(target_os = "macos")]
9use crate::coreml_units::CoreMlComputeUnit;
10use crate::coreml_units::CoreMlUnits;
11use crate::e40::{self, E40Engine, E40SessionProfile};
12use crate::parity_cache;
13use rightkit_onnx::ort::session::Session;
14use rightkit_onnx::{
15 configure_runtime_with_options, init_environment, CandidateSource, EnvironmentOptions,
16 GlobalPool, RuntimeBindingOptions, RuntimeCandidate, SessionOptions,
17};
18use thiserror::Error;
19use tokenizers::{EncodeInput, Tokenizer};
20
21pub const EMBEDDINGGEMMA_MODEL_ID: &str = "embeddinggemma-300m-q4";
22pub const EMBEDDINGGEMMA2_MODEL_ID: &str = "embeddinggemma2-q4";
26pub const EMBEDDINGGEMMA2_E40_MODEL_ID: &str = "embeddinggemma2-e40-dml";
31pub const EMBEDDINGGEMMA2_COREML_MODEL_ID: &str = "embeddinggemma2-coreml";
37const GEMMA2_MODALITY_SENTINELS: [&str; 3] = ["<|image|>", "<|video|>", "<|audio|>"];
40pub const EMBEDDINGGEMMA_NATIVE_DIM: usize = 768;
41pub const DEFAULT_TRUNCATE_DIM: usize = 256;
42pub const MAX_SEQUENCE_TOKENS: usize = 2048;
43pub const DEFAULT_MAX_BATCH: usize = 16;
45pub const DEFAULT_INGEST_BATCH: usize = e40::DEFAULT_INGEST_BATCH;
47const MAX_PADDED_TOKENS_PER_CALL: usize = 16_384;
51const ACCELERATOR_BUCKETS: [usize; 6] = [64, 128, 256, 512, 1024, 2048];
54const DOCUMENT_PROMPT_PREFIX: &str = "title: none | text: ";
55const QUERY_PROMPT_PREFIX: &str = "task: search result | query: ";
56const PIPELINE_SCHEMA: &str = "rightkit-embed/pipeline/1";
59const POOLING_DESCRIPTION: &str = "model sentence_embedding [B,D] or masked mean over [B,T,D]";
60const NORMALIZATION_DESCRIPTION: &str = "l2 native, matryoshka prefix truncate, l2";
61const PROBE_PROMPT: &str = "title: none | text: rightkit embed provider probe";
62pub const ACCELERATOR_PARITY_MIN_COSINE: f32 = 0.995;
67const PARITY_QUERY: &str = "how are invoices matched to the ledger";
69const PARITY_DOCUMENT: &str = "The nightly billing job reads every invoice issued during the day, \
71looks up the matching ledger account, and compares the invoiced total with the posted amount. \
72Differences below one cent are rounded away. Larger differences are written to an exceptions \
73table together with the customer, the invoice number and the date, and the finance team reviews \
74that table each morning before the payment run starts.";
75const PARITY_LONG_SECTIONS: usize = 20;
77
78#[derive(Debug, Error)]
79pub enum EmbedError {
80 #[error("model directory is missing: {0}")]
81 MissingModelDirectory(PathBuf),
82 #[error("model artifact is missing: {0}")]
83 MissingArtifact(PathBuf),
84 #[error("invalid model configuration: {0}")]
85 InvalidConfig(String),
86 #[error("tokenizer: {0}")]
87 Tokenizer(String),
88 #[error("ONNX Runtime: {0}")]
89 Runtime(String),
90 #[error("model output: {0}")]
91 Output(String),
92 #[error("ingest batch {requested} is not supported: {reason}")]
95 UnsupportedIngestBatch {
96 requested: usize,
97 reason: &'static str,
98 },
99 #[error("{provider} failed the vector parity check against cpu: {detail}")]
106 AcceleratorParity {
107 provider: &'static str,
108 lowest_cosine: Option<f32>,
109 floor: f32,
110 detail: String,
111 },
112}
113
114#[derive(Debug, Clone, Copy, PartialEq)]
117pub struct AcceleratorParity {
118 pub lowest_cosine: f32,
120 pub at_tokens: usize,
122 pub pairs: usize,
124}
125
126#[derive(Debug, Clone, Copy, PartialEq, Eq)]
128pub struct SessionCreate {
129 pub provider: &'static str,
131 pub graph_tokens: usize,
133 pub duration: Duration,
135}
136
137#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
140#[non_exhaustive]
141pub enum ParityCacheOutcome {
142 #[default]
144 NotUsed,
145 Hit { passed: bool },
148 Miss { reason: &'static str, stored: bool },
151}
152
153#[derive(Debug, Clone, Default, PartialEq, Eq)]
157#[non_exhaustive]
158pub struct LoadTimings {
159 pub tokenizer_load: Duration,
161 pub token_table_read: Duration,
163 pub probe: Option<Duration>,
165 pub parity_check_total: Option<Duration>,
168 pub parity_cpu_reference: Option<Duration>,
171 pub parity_accelerator: Option<Duration>,
174 pub parity_cache: ParityCacheOutcome,
176 pub session_creates: Vec<SessionCreate>,
178 pub prewarm: Option<Duration>,
182 pub prewarm_per_length: Vec<(usize, Duration)>,
186 pub total: Duration,
188}
189
190#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
215pub enum ExecutionProvider {
216 #[default]
217 Cpu,
218 Auto,
219 DirectMl,
220 CoreMl,
221}
222
223impl From<rightkit_onnx::ExecutionProvider> for ExecutionProvider {
224 fn from(value: rightkit_onnx::ExecutionProvider) -> Self {
225 match value {
226 rightkit_onnx::ExecutionProvider::Cpu => Self::Cpu,
227 rightkit_onnx::ExecutionProvider::DirectMl => Self::DirectMl,
228 rightkit_onnx::ExecutionProvider::CoreMl => Self::CoreMl,
229 }
230 }
231}
232
233impl ExecutionProvider {
234 fn requested(self) -> (rightkit_onnx::ExecutionProvider, bool) {
235 match self {
236 Self::Cpu => (rightkit_onnx::ExecutionProvider::Cpu, false),
237 Self::Auto => (
238 rightkit_onnx::ExecutionProvider::platform_accelerator(),
239 true,
240 ),
241 Self::DirectMl => (rightkit_onnx::ExecutionProvider::DirectMl, false),
242 Self::CoreMl => (rightkit_onnx::ExecutionProvider::CoreMl, false),
243 }
244 }
245}
246
247pub(crate) fn provider_name(provider: rightkit_onnx::ExecutionProvider) -> &'static str {
248 match provider {
249 rightkit_onnx::ExecutionProvider::Cpu => "cpu",
250 rightkit_onnx::ExecutionProvider::DirectMl => "directml",
251 rightkit_onnx::ExecutionProvider::CoreMl => "coreml",
252 }
253}
254
255#[derive(Debug, Clone)]
256pub struct EmbedderConfig {
257 pub model_dir: PathBuf,
258 pub runtime_path: PathBuf,
259 pub provider: ExecutionProvider,
260 pub truncate_dim: usize,
261 pub max_tokens: usize,
262 pub max_batch: usize,
263 pub ingest_batch: usize,
266 pub intra_threads: usize,
267 pub fallback_model_dir: Option<PathBuf>,
270 pub coreml_cache_dir: Option<PathBuf>,
272 pub parity_cache_dir: Option<PathBuf>,
274 pub prewarm: Option<bool>,
277 pub coreml_units: Option<CoreMlUnits>,
280 pub(crate) e40_session_profile: Option<E40SessionProfile>,
282 pub(crate) runtime_binding: RuntimeBindingOptions,
284}
285
286impl EmbedderConfig {
287 pub fn new(model_dir: impl Into<PathBuf>, runtime_path: impl Into<PathBuf>) -> Self {
288 Self {
289 model_dir: model_dir.into(),
290 runtime_path: runtime_path.into(),
291 provider: ExecutionProvider::Cpu,
292 truncate_dim: DEFAULT_TRUNCATE_DIM,
293 max_tokens: MAX_SEQUENCE_TOKENS,
294 max_batch: DEFAULT_MAX_BATCH,
295 ingest_batch: DEFAULT_INGEST_BATCH,
296 intra_threads: 1,
297 fallback_model_dir: None,
298 coreml_cache_dir: None,
299 parity_cache_dir: None,
300 prewarm: None,
301 coreml_units: None,
302 e40_session_profile: None,
303 runtime_binding: RuntimeBindingOptions::default(),
304 }
305 }
306
307 pub fn with_prewarm(mut self, prewarm: bool) -> Self {
319 self.prewarm = Some(prewarm);
320 self
321 }
322
323 pub fn with_coreml_units(mut self, units: impl Into<CoreMlUnits>) -> Self {
331 self.coreml_units = Some(units.into());
332 self
333 }
334
335 #[cfg(feature = "bench-internals")]
338 pub fn with_e40_session_profile(mut self, profile: E40SessionProfile) -> Self {
339 self.e40_session_profile = Some(profile);
340 self
341 }
342
343 pub fn with_runtime_binding(mut self, options: RuntimeBindingOptions) -> Self {
349 self.runtime_binding = options;
350 self
351 }
352
353 pub fn with_fallback_model_dir(mut self, dir: impl Into<PathBuf>) -> Self {
363 self.fallback_model_dir = Some(dir.into());
364 self
365 }
366
367 pub fn with_coreml_cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
372 self.coreml_cache_dir = Some(dir.into());
373 self
374 }
375
376 pub fn with_parity_cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
393 self.parity_cache_dir = Some(dir.into());
394 self
395 }
396
397 pub fn with_provider(mut self, provider: impl Into<ExecutionProvider>) -> Self {
398 self.provider = provider.into();
399 self
400 }
401
402 pub fn with_truncate_dim(mut self, dim: usize) -> Self {
403 self.truncate_dim = dim;
404 self
405 }
406
407 pub fn with_max_tokens(mut self, max_tokens: usize) -> Self {
408 self.max_tokens = max_tokens;
409 self
410 }
411
412 pub fn with_max_batch(mut self, max_batch: usize) -> Self {
414 self.max_batch = max_batch;
415 self
416 }
417
418 pub fn with_ingest_batch(mut self, ingest_batch: usize) -> Self {
424 self.ingest_batch = ingest_batch;
425 self
426 }
427
428 pub fn with_intra_threads(mut self, threads: usize) -> Self {
430 self.intra_threads = threads;
431 self
432 }
433}
434
435enum Backend {
438 Onnx(Mutex<Session>),
439 E40(E40Engine),
440 #[cfg(target_os = "macos")]
441 CoreMl(CoreMlEngine),
442}
443
444pub struct Embedder {
445 tokenizer: Tokenizer,
446 backend: Backend,
447 model_dir: PathBuf,
448 native_dim: usize,
449 truncate_dim: usize,
450 max_tokens: usize,
451 max_batch: usize,
452 model_id: String,
453 gemma2: bool,
454 pad_id: i64,
455 takes_token_type_ids: bool,
456 provider_used: &'static str,
457 fallback_reason: Option<String>,
458 parity: Option<AcceleratorParity>,
459 load_timings: Option<LoadTimings>,
460 pipeline_id: OnceLock<Result<String, String>>,
461}
462
463struct BuiltEmbedSession {
464 session: Session,
465 provider: rightkit_onnx::ExecutionProvider,
466 fallback: Option<String>,
467}
468
469fn build_session(
470 config: &EmbedderConfig,
471 provider: rightkit_onnx::ExecutionProvider,
472 cpu_fallback: bool,
473) -> Result<BuiltEmbedSession, EmbedError> {
474 let built = SessionOptions::new(provider)
475 .with_threads(Some(config.intra_threads))
476 .with_cpu_fallback(cpu_fallback)
477 .build_from_file_reported(&config.model_dir.join("model_q4.onnx"))
478 .map_err(|e| EmbedError::Runtime(e.to_string()))?;
479 Ok(BuiltEmbedSession {
480 session: built.session,
481 provider: built.provider,
482 fallback: built.fallback,
483 })
484}
485
486fn takes_token_type_ids(session: &Session) -> bool {
487 session
488 .inputs()
489 .iter()
490 .any(|input| input.name() == "token_type_ids")
491}
492
493fn load_tokenizer(model_dir: &Path) -> Result<(Tokenizer, i64), EmbedError> {
496 let mut tokenizer = Tokenizer::from_file(model_dir.join("tokenizer.json"))
497 .map_err(|e| EmbedError::Tokenizer(e.to_string()))?;
498 let pad_id = tokenizer
499 .get_padding()
500 .map(|padding| i64::from(padding.pad_id))
501 .or_else(|| tokenizer.token_to_id("<pad>").map(i64::from))
502 .unwrap_or(0);
503 tokenizer.with_padding(None);
504 Ok((tokenizer, pad_id))
505}
506
507impl Embedder {
508 pub fn load(config: EmbedderConfig) -> Result<Self, EmbedError> {
509 let started = Instant::now();
510 validate_config(&config)?;
511 #[cfg(target_os = "macos")]
513 {
514 if coreml::is_bundle(&config.model_dir) {
515 let (tokenizer, pad_id) = load_tokenizer(&config.model_dir)?;
516 let tokenizer_load = started.elapsed();
517 let mut embedder = Self::load_coreml(config, tokenizer, pad_id)?;
518 if let Some(timings) = embedder.load_timings.as_mut() {
520 timings.tokenizer_load = tokenizer_load;
521 timings.total = started.elapsed();
522 }
523 return Ok(embedder);
524 }
525 }
526 let gemma2 = bundle_is_gemma2(&config.model_dir)?;
527 let tokenizer_started = Instant::now();
528 let (tokenizer, pad_id) = load_tokenizer(&config.model_dir)?;
529 let tokenizer_load = tokenizer_started.elapsed();
530 let selection = configure_runtime_with_options(
531 [RuntimeCandidate::new(
532 config.runtime_path.clone(),
533 CandidateSource::Explicit,
534 )],
535 &config.runtime_binding,
536 )
537 .map(|binding| binding.selection)
538 .map_err(|e| EmbedError::Runtime(e.to_string()))?;
539 init_environment(
540 &selection,
541 &EnvironmentOptions {
542 global_pool: Some(GlobalPool::from_budget()),
543 },
544 )
545 .map_err(EmbedError::Runtime)?;
546 if e40::is_bundle(&config.model_dir) {
547 let mut embedder = Self::load_e40(config, tokenizer, pad_id, tokenizer_load)?;
548 let sessions = match &embedder.backend {
550 Backend::E40(engine) => Some(engine.finish_session_log()),
551 _ => None,
552 };
553 if let (Some(timings), Some(sessions)) = (embedder.load_timings.as_mut(), sessions) {
554 timings.session_creates = sessions;
555 timings.total = started.elapsed();
556 }
557 return Ok(embedder);
558 }
559 let (requested, auto) = config.provider.requested();
560 let built = build_session(&config, requested, auto)?;
561 let accelerator = built.provider;
562 let mut embedder = Self::from_session(&config, tokenizer, pad_id, gemma2, built);
563 if auto && accelerator != rightkit_onnx::ExecutionProvider::Cpu {
564 if let Err(reason) = embedder.probe() {
565 let cpu = build_session(&config, rightkit_onnx::ExecutionProvider::Cpu, false)?;
566 embedder.takes_token_type_ids = takes_token_type_ids(&cpu.session);
567 embedder.backend = Backend::Onnx(Mutex::new(cpu.session));
568 embedder.provider_used = provider_name(cpu.provider);
569 embedder.fallback_reason = Some(format!(
570 "{} -> cpu: probe inference failed: {reason}",
571 provider_name(accelerator)
572 ));
573 }
574 }
575 if embedder.provider_used != "cpu" {
576 return embedder.admit_accelerator(&config, auto);
577 }
578 Ok(embedder)
579 }
580
581 fn from_session(
582 config: &EmbedderConfig,
583 tokenizer: Tokenizer,
584 pad_id: i64,
585 gemma2: bool,
586 built: BuiltEmbedSession,
587 ) -> Self {
588 Self {
589 tokenizer,
590 takes_token_type_ids: takes_token_type_ids(&built.session),
591 backend: Backend::Onnx(Mutex::new(built.session)),
592 model_dir: config.model_dir.clone(),
593 native_dim: EMBEDDINGGEMMA_NATIVE_DIM,
594 truncate_dim: config.truncate_dim,
595 max_tokens: config.max_tokens,
596 max_batch: config.max_batch,
597 model_id: if gemma2 {
598 EMBEDDINGGEMMA2_MODEL_ID
599 } else {
600 EMBEDDINGGEMMA_MODEL_ID
601 }
602 .to_owned(),
603 gemma2,
604 pad_id,
605 provider_used: provider_name(built.provider),
606 fallback_reason: built.fallback,
607 parity: None,
608 load_timings: None,
609 pipeline_id: OnceLock::new(),
610 }
611 }
612
613 fn from_e40(
614 config: &EmbedderConfig,
615 tokenizer: Tokenizer,
616 pad_id: i64,
617 engine: E40Engine,
618 ) -> Self {
619 Self {
620 tokenizer,
621 provider_used: provider_name(engine.provider()),
622 backend: Backend::E40(engine),
623 model_dir: config.model_dir.clone(),
624 native_dim: EMBEDDINGGEMMA_NATIVE_DIM,
625 truncate_dim: config.truncate_dim,
626 max_tokens: config.max_tokens.min(e40::MAX_TOKENS),
628 max_batch: config.max_batch,
629 model_id: EMBEDDINGGEMMA2_E40_MODEL_ID.to_owned(),
630 gemma2: true,
631 pad_id,
632 takes_token_type_ids: false,
633 fallback_reason: None,
634 parity: None,
635 load_timings: None,
636 pipeline_id: OnceLock::new(),
637 }
638 }
639
640 fn load_e40(
646 config: EmbedderConfig,
647 tokenizer: Tokenizer,
648 pad_id: i64,
649 tokenizer_load: Duration,
650 ) -> Result<Self, EmbedError> {
651 use rightkit_onnx::ExecutionProvider as Ep;
652 let (requested, auto) = match config.provider {
653 ExecutionProvider::Cpu => (Ep::Cpu, false),
654 ExecutionProvider::DirectMl => (Ep::DirectMl, false),
655 ExecutionProvider::CoreMl => {
656 return Err(EmbedError::InvalidConfig(
657 "an E40 bundle runs on DirectML or CPU, not CoreML".into(),
658 ))
659 }
660 ExecutionProvider::Auto => match Ep::platform_accelerator() {
661 Ep::DirectMl => (Ep::DirectMl, true),
662 _ => (Ep::Cpu, false),
663 },
664 };
665 let profile = config
666 .e40_session_profile
667 .unwrap_or_else(|| E40SessionProfile::shipped(config.intra_threads));
668 let engine = E40Engine::open(&config.model_dir, requested, profile, config.ingest_batch)?;
669 let timings = LoadTimings {
670 tokenizer_load,
671 token_table_read: engine.token_table_read(),
672 ..LoadTimings::default()
673 };
674 let mut embedder = Self::from_e40(&config, tokenizer, pad_id, engine);
675 embedder.load_timings = Some(timings);
676 if requested == Ep::Cpu {
677 return Ok(embedder);
678 }
679 Self::admit_or_fall_back(embedder, &config, auto)
680 }
681
682 #[cfg(target_os = "macos")]
683 fn from_coreml(
684 config: &EmbedderConfig,
685 tokenizer: Tokenizer,
686 pad_id: i64,
687 engine: CoreMlEngine,
688 ) -> Self {
689 Self {
690 tokenizer,
691 provider_used: engine.provider_name(),
692 backend: Backend::CoreMl(engine),
693 model_dir: config.model_dir.clone(),
694 native_dim: EMBEDDINGGEMMA_NATIVE_DIM,
695 truncate_dim: config.truncate_dim,
696 max_tokens: config.max_tokens.min(e40::MAX_TOKENS),
698 max_batch: config.max_batch,
699 model_id: EMBEDDINGGEMMA2_COREML_MODEL_ID.to_owned(),
700 gemma2: true,
701 pad_id,
702 takes_token_type_ids: false,
703 fallback_reason: None,
704 parity: None,
705 load_timings: Some(LoadTimings::default()),
706 pipeline_id: OnceLock::new(),
707 }
708 }
709
710 #[cfg(target_os = "macos")]
722 fn load_coreml(
723 config: EmbedderConfig,
724 tokenizer: Tokenizer,
725 pad_id: i64,
726 ) -> Result<Self, EmbedError> {
727 let accelerated = || config.coreml_units.clone().unwrap_or_default();
728 let (units, auto) = match config.provider {
729 ExecutionProvider::Cpu => (CoreMlUnits::uniform(CoreMlComputeUnit::CpuOnly), false),
730 ExecutionProvider::Auto => (accelerated(), true),
731 ExecutionProvider::CoreMl => (accelerated(), false),
732 ExecutionProvider::DirectMl => {
733 return Err(EmbedError::InvalidConfig(
734 "a Core ML bundle runs on Core ML or CPU, not DirectML".into(),
735 ))
736 }
737 };
738 let cpu_only = units.is_cpu_only();
739 let engine = match CoreMlEngine::open(
740 &config.model_dir,
741 units,
742 config.coreml_cache_dir.as_deref(),
743 ) {
744 Ok(engine) => engine,
745 Err(error) if auto => {
746 return match &config.fallback_model_dir {
747 Some(dir) => {
748 Self::load_fallback(&config, dir, format!("coreml -> cpu: {error}"))
749 }
750 None => Err(error),
751 }
752 }
753 Err(error) => return Err(error),
754 };
755 let embedder = Self::from_coreml(&config, tokenizer, pad_id, engine);
756 if cpu_only {
758 return Ok(embedder);
759 }
760 Self::admit_or_fall_back(embedder, &config, auto)
761 }
762
763 fn admit_or_fall_back(
767 mut embedder: Self,
768 config: &EmbedderConfig,
769 auto: bool,
770 ) -> Result<Self, EmbedError> {
771 let config = config.clone();
772 let accelerator = embedder.provider_used;
773 if auto {
774 let probe_started = Instant::now();
775 let probed = embedder.probe();
776 if let Some(timings) = embedder.load_timings.as_mut() {
777 timings.probe = Some(probe_started.elapsed());
778 }
779 if let Err(reason) = probed {
780 return embedder.e40_not_admitted(
781 &config,
782 format!("{accelerator} -> cpu: probe inference failed: {reason}"),
783 );
784 }
785 }
786 match embedder.admit_accelerator(&config, auto) {
787 Ok(admitted) if admitted.provider_used == "cpu" => {
789 match (&config.fallback_model_dir, admitted.fallback_reason.clone()) {
790 (Some(dir), Some(reason)) => Self::load_fallback(&config, dir, reason),
791 _ => Ok(admitted),
792 }
793 }
794 Ok(admitted) => Ok(admitted),
795 Err(error) if auto => match &config.fallback_model_dir {
796 Some(dir) => {
797 Self::load_fallback(&config, dir, format!("{accelerator} -> cpu: {error}"))
798 }
799 None => Err(error),
800 },
801 Err(error) => Err(error),
802 }
803 }
804
805 fn e40_not_admitted(
808 mut self,
809 config: &EmbedderConfig,
810 reason: String,
811 ) -> Result<Self, EmbedError> {
812 if let Some(dir) = &config.fallback_model_dir {
813 return Self::load_fallback(config, dir, reason);
814 }
815 let mut cpu = self.cpu_twin(config)?;
816 cpu.fallback_reason = Some(reason);
817 cpu.load_timings = self.load_timings.take();
818 Ok(cpu)
819 }
820
821 fn load_fallback(
824 config: &EmbedderConfig,
825 dir: &Path,
826 reason: String,
827 ) -> Result<Self, EmbedError> {
828 if e40::is_bundle(dir) || coreml_bundle_files(dir)?.is_some() {
829 return Err(EmbedError::InvalidConfig(
830 "the fallback model directory must hold a Q4 bundle, not an E40 or Core ML bundle"
831 .into(),
832 ));
833 }
834 let mut fallback = config.clone();
835 fallback.model_dir = dir.to_path_buf();
836 fallback.provider = ExecutionProvider::Cpu;
837 fallback.fallback_model_dir = None;
838 let mut embedder = Self::load(fallback)?;
839 embedder.fallback_reason = Some(format!(
840 "{reason}; serving {} on cpu from the fallback model directory",
841 embedder.model_id
842 ));
843 Ok(embedder)
844 }
845
846 fn cpu_twin(&self, config: &EmbedderConfig) -> Result<Self, EmbedError> {
848 let (tokenizer, pad_id) = load_tokenizer(&config.model_dir)?;
849 match &self.backend {
850 Backend::Onnx(_) => {
851 let built = build_session(config, rightkit_onnx::ExecutionProvider::Cpu, false)?;
852 Ok(Self::from_session(
853 config,
854 tokenizer,
855 pad_id,
856 self.gemma2,
857 built,
858 ))
859 }
860 Backend::E40(engine) => Ok(Self::from_e40(
861 config,
862 tokenizer,
863 pad_id,
864 engine.with_provider(rightkit_onnx::ExecutionProvider::Cpu),
865 )),
866 #[cfg(target_os = "macos")]
868 Backend::CoreMl(engine) => Ok(Self::from_coreml(
869 config,
870 tokenizer,
871 pad_id,
872 engine.with_units(CoreMlUnits::uniform(CoreMlComputeUnit::CpuOnly)),
873 )),
874 }
875 }
876
877 fn is_fixed_graph(&self) -> bool {
879 match self.backend {
880 Backend::Onnx(_) => false,
881 Backend::E40(_) => true,
882 #[cfg(target_os = "macos")]
883 Backend::CoreMl(_) => true,
884 }
885 }
886
887 fn release_graphs(&self) {
890 match &self.backend {
891 Backend::Onnx(_) => {}
892 Backend::E40(engine) => engine.release(),
893 #[cfg(target_os = "macos")]
894 Backend::CoreMl(engine) => engine.release(),
895 }
896 }
897
898 fn prewarms(&self, config: &EmbedderConfig) -> bool {
900 let (fixed_graph_accelerator, default_on) = match &self.backend {
902 Backend::E40(_) => (true, false),
903 #[cfg(target_os = "macos")]
904 Backend::CoreMl(_) => (true, true),
905 Backend::Onnx(_) => (false, false),
906 };
907 config.prewarm.unwrap_or(default_on)
908 && self.provider_used != "cpu"
909 && fixed_graph_accelerator
910 }
911
912 fn prewarm_graphs(&mut self) -> Result<(), EmbedError> {
916 let (elapsed, per_length) = match &self.backend {
917 Backend::E40(engine) => (engine.prewarm(self.pad_id)?, Vec::new()),
918 #[cfg(target_os = "macos")]
919 Backend::CoreMl(engine) => engine.prewarm(self.pad_id)?,
920 Backend::Onnx(_) => return Ok(()),
921 };
922 if let Some(timings) = self.load_timings.as_mut() {
923 timings.prewarm = Some(elapsed);
924 timings.prewarm_per_length = per_length;
925 }
926 Ok(())
927 }
928
929 fn admit_accelerator(
935 mut self,
936 config: &EmbedderConfig,
937 auto: bool,
938 ) -> Result<Self, EmbedError> {
939 let provider = self.provider_used;
940 let check_started = Instant::now();
941 let cache = self.parity_cache_target(config);
942 let mut miss_reason: Option<&'static str> = None;
943 if let Some(target) = &cache {
944 match parity_cache::read(
945 &target.dir,
946 &config.model_dir,
947 &target.key,
948 parity_cache::now_unix(),
949 ) {
950 parity_cache::Lookup::Hit(verdict) => {
951 self.record_parity_timings(
952 check_started,
953 None,
954 None,
955 ParityCacheOutcome::Hit {
956 passed: verdict.passed,
957 },
958 );
959 return self.apply_cached_verdict(&verdict, config, auto);
960 }
961 parity_cache::Lookup::Miss(reason) => miss_reason = Some(reason),
962 }
963 }
964 let outcome = |stored: bool| match miss_reason {
966 Some(reason) => ParityCacheOutcome::Miss { reason, stored },
967 None => ParityCacheOutcome::NotUsed,
968 };
969 let mut cpu = self.cpu_twin(config)?;
970 let prompts = cpu.parity_prompts();
971 let tokens: Vec<usize> = cpu
972 .encode_prompts(&prompts)?
973 .iter()
974 .map(|row| row.ids.len())
975 .collect();
976 let reference = cpu.parity_vectors(&prompts)?;
977 let cpu_reference = check_started.elapsed();
978 let accelerator_started = Instant::now();
979 let candidate = self.parity_vectors(&prompts);
980 let accelerator = accelerator_started.elapsed();
981 let (lowest_cosine, detail) = match candidate {
982 Ok(candidate) => match compare_parity(
983 &reference,
984 &candidate,
985 self.native_dim,
986 ACCELERATOR_PARITY_MIN_COSINE,
987 ) {
988 Ok(lowest) => {
989 let parity = AcceleratorParity {
990 lowest_cosine: lowest.cosine,
991 at_tokens: tokens[lowest.index % tokens.len()],
992 pairs: reference.len(),
993 };
994 self.parity = Some(parity);
995 let prewarm = self.prewarms(config);
998 if !prewarm {
999 self.release_graphs();
1000 }
1001 let stored = cache.as_ref().is_some_and(|target| {
1002 target.store(
1003 &config.model_dir,
1004 &parity_cache::ParityVerdict {
1005 passed: true,
1006 lowest_cosine: Some(parity.lowest_cosine),
1007 at_tokens: parity.at_tokens,
1008 pairs: parity.pairs,
1009 detail: String::new(),
1010 timestamp_unix: parity_cache::now_unix(),
1011 },
1012 )
1013 });
1014 self.record_parity_timings(
1015 check_started,
1016 Some(cpu_reference),
1017 Some(accelerator),
1018 outcome(stored),
1019 );
1020 if prewarm {
1021 self.prewarm_graphs()?;
1022 }
1023 return Ok(self);
1024 }
1025 Err(miss) => {
1026 let (lowest_cosine, detail) =
1027 parity_miss_detail(&miss, &tokens, ACCELERATOR_PARITY_MIN_COSINE);
1028 let stored = cache.as_ref().is_some_and(|target| {
1031 target.store(
1032 &config.model_dir,
1033 &parity_cache::ParityVerdict {
1034 passed: false,
1035 lowest_cosine,
1036 at_tokens: 0,
1037 pairs: reference.len(),
1038 detail: detail.clone(),
1039 timestamp_unix: parity_cache::now_unix(),
1040 },
1041 )
1042 });
1043 self.record_parity_timings(
1044 check_started,
1045 Some(cpu_reference),
1046 Some(accelerator),
1047 outcome(stored),
1048 );
1049 (lowest_cosine, detail)
1050 }
1051 },
1052 Err(error) if auto => {
1053 self.record_parity_timings(
1054 check_started,
1055 Some(cpu_reference),
1056 Some(accelerator),
1057 outcome(false),
1058 );
1059 (None, format!("inference error: {error}"))
1060 }
1061 Err(error) => return Err(error),
1062 };
1063 if auto {
1064 cpu.fallback_reason =
1065 Some(format!("{provider} -> cpu: vector parity failed: {detail}"));
1066 cpu.load_timings = self.load_timings.take();
1067 return Ok(cpu);
1068 }
1069 Err(EmbedError::AcceleratorParity {
1070 provider,
1071 lowest_cosine,
1072 floor: ACCELERATOR_PARITY_MIN_COSINE,
1073 detail,
1074 })
1075 }
1076
1077 fn parity_cache_target(&self, config: &EmbedderConfig) -> Option<ParityCacheTarget> {
1081 let dir = config.parity_cache_dir.clone()?;
1082 if !matches!(self.backend, Backend::E40(_)) {
1083 return None;
1084 }
1085 let artifacts = e40::artifacts(&config.model_dir).ok()?;
1086 let artifacts: Vec<&str> = artifacts.iter().map(String::as_str).collect();
1087 let key = parity_cache::ParityKey::build(
1088 &config.model_dir,
1089 &artifacts,
1090 &config.runtime_path,
1091 ort::info(),
1092 self.provider_used,
1093 ACCELERATOR_PARITY_MIN_COSINE,
1094 );
1095 Some(ParityCacheTarget { dir, key })
1096 }
1097
1098 fn record_parity_timings(
1100 &mut self,
1101 started: Instant,
1102 cpu_reference: Option<Duration>,
1103 accelerator: Option<Duration>,
1104 cache: ParityCacheOutcome,
1105 ) {
1106 if let Some(timings) = self.load_timings.as_mut() {
1107 timings.parity_check_total = Some(started.elapsed());
1108 timings.parity_cpu_reference = cpu_reference;
1109 timings.parity_accelerator = accelerator;
1110 timings.parity_cache = cache;
1111 }
1112 }
1113
1114 fn apply_cached_verdict(
1117 mut self,
1118 verdict: &parity_cache::ParityVerdict,
1119 config: &EmbedderConfig,
1120 auto: bool,
1121 ) -> Result<Self, EmbedError> {
1122 let provider = self.provider_used;
1123 if verdict.passed {
1124 self.parity = verdict
1125 .lowest_cosine
1126 .map(|lowest_cosine| AcceleratorParity {
1127 lowest_cosine,
1128 at_tokens: verdict.at_tokens,
1129 pairs: verdict.pairs,
1130 });
1131 if self.prewarms(config) {
1132 self.prewarm_graphs()?;
1133 }
1134 return Ok(self);
1135 }
1136 let detail = format!("{} (cached verdict)", verdict.detail);
1137 if auto {
1138 let mut cpu = self.cpu_twin(config)?;
1139 cpu.fallback_reason =
1140 Some(format!("{provider} -> cpu: vector parity failed: {detail}"));
1141 cpu.load_timings = self.load_timings.take();
1142 return Ok(cpu);
1143 }
1144 Err(EmbedError::AcceleratorParity {
1145 provider,
1146 lowest_cosine: verdict.lowest_cosine,
1147 floor: ACCELERATOR_PARITY_MIN_COSINE,
1148 detail,
1149 })
1150 }
1151
1152 fn parity_prompts(&self) -> Vec<String> {
1160 let mut prompts = vec![
1161 self.query_prompt(PARITY_QUERY),
1162 self.document_prompt(PARITY_DOCUMENT),
1163 self.document_prompt(&parity_long_document()),
1164 ];
1165 if self.is_fixed_graph() {
1166 prompts.insert(1, self.document_prompt(&parity_sections(1)));
1167 prompts.insert(3, self.document_prompt(&parity_sections(4)));
1168 }
1169 prompts
1170 }
1171
1172 fn parity_vectors(&self, prompts: &[String]) -> Result<Vec<Vec<f32>>, EmbedError> {
1176 let mut vectors = Vec::with_capacity(prompts.len() * 2);
1177 for prompt in prompts {
1178 vectors.extend(self.embed_prompts_native(vec![prompt.clone()])?);
1179 }
1180 vectors.extend(self.embed_prompts_native(prompts.to_vec())?);
1181 Ok(vectors)
1182 }
1183
1184 fn probe(&self) -> Result<(), String> {
1187 let rows = self
1188 .encode_prompts(&[PROBE_PROMPT.to_owned()])
1189 .map_err(|e| e.to_string())?;
1190 let chunk = [&rows[0]];
1191 let native = self
1192 .run_chunk(&chunk, true)
1193 .map_err(|e| e.to_string())?
1194 .into_iter()
1195 .next()
1196 .ok_or_else(|| "probe returned no row".to_owned())?;
1197 probe_output_ok(&native, self.native_dim)
1198 }
1199
1200 pub fn model_id(&self) -> &str {
1201 &self.model_id
1202 }
1203
1204 pub fn native_dim(&self) -> usize {
1205 self.native_dim
1206 }
1207
1208 pub fn dimension(&self) -> usize {
1209 self.truncate_dim
1210 }
1211
1212 pub fn provider_used(&self) -> &str {
1214 self.provider_used
1215 }
1216
1217 pub fn fallback_reason(&self) -> Option<&str> {
1221 self.fallback_reason.as_deref()
1222 }
1223
1224 pub fn accelerator_parity(&self) -> Option<AcceleratorParity> {
1231 self.parity
1232 }
1233
1234 pub fn load_timings(&self) -> Option<&LoadTimings> {
1241 self.load_timings.as_ref()
1242 }
1243
1244 pub fn coreml_units(&self) -> Option<&CoreMlUnits> {
1248 match &self.backend {
1249 #[cfg(target_os = "macos")]
1250 Backend::CoreMl(engine) => Some(engine.units()),
1251 _ => None,
1252 }
1253 }
1254
1255 #[cfg(feature = "bench-internals")]
1258 pub fn bench_query_tokens(&self, text: &str) -> Result<BenchTokens, EmbedError> {
1259 let prompt = self.query_prompt(text);
1260 let ids = self
1261 .encode_prompts(std::slice::from_ref(&prompt))?
1262 .remove(0)
1263 .ids;
1264 Ok(BenchTokens {
1265 graph_tokens: match &self.backend {
1266 Backend::E40(engine) => engine.query_bucket(ids.len()),
1267 _ => e40::bucket_for(ids.len()),
1268 },
1269 pad_id: self.pad_id,
1270 prompt,
1271 ids,
1272 })
1273 }
1274
1275 #[cfg(feature = "bench-internals")]
1279 pub fn bench_embed_query_split(&self, text: &str) -> Result<BenchEmbedSplit, EmbedError> {
1280 let Backend::E40(engine) = &self.backend else {
1281 return Err(EmbedError::InvalidConfig(
1282 "bench_embed_query_split needs an E40 bundle".into(),
1283 ));
1284 };
1285 let started = Instant::now();
1286 let prompt = self.query_prompt(text);
1287 let row = self
1288 .encode_prompts(std::slice::from_ref(&prompt))?
1289 .remove(0);
1290 let tokenize = started.elapsed();
1291 let (values, engine_split) = engine.run_split(&row.ids, self.pad_id)?;
1292 let post_started = Instant::now();
1293 let shape = [1, self.native_dim as i64];
1294 let mut pooled = pool_batch(&shape, &values, &[row.mask.as_slice()], self.native_dim)?;
1295 let _ = truncate_and_normalize(&pooled.remove(0), self.truncate_dim);
1296 Ok(BenchEmbedSplit {
1297 tokenize,
1298 engine: engine_split,
1299 post: post_started.elapsed(),
1300 total: started.elapsed(),
1301 })
1302 }
1303
1304 pub fn pipeline_id(&self) -> String {
1324 match self.pipeline_id.get_or_init(|| {
1325 let mut files = Vec::new();
1326 let e40_names = match &self.backend {
1327 Backend::E40(_) => e40::artifacts(&self.model_dir).map_err(|e| e.to_string())?,
1328 _ => Vec::new(),
1329 };
1330 let e40_refs: Vec<&str> = e40_names.iter().map(String::as_str).collect();
1331 let names: &[&str] = match self.backend {
1332 Backend::E40(_) => &e40_refs,
1333 #[cfg(target_os = "macos")]
1334 Backend::CoreMl(_) => &coreml::ARTIFACTS,
1335 Backend::Onnx(_) => &[
1336 "model_q4.onnx",
1337 "model_q4.onnx_data",
1338 "tokenizer.json",
1339 "config.json",
1340 ],
1341 };
1342 for name in names {
1343 files.push((
1344 e40::pipeline_name(name),
1345 hash_file(&self.model_dir.join(name))?,
1346 ));
1347 }
1348 Ok(pipeline_fingerprint(&PipelineParts {
1349 model_id: &self.model_id,
1350 native_dim: self.native_dim,
1351 truncate_dim: self.truncate_dim,
1352 max_tokens: self.max_tokens,
1353 scrubs_sentinels: self.gemma2,
1354 files: &files,
1355 }))
1356 }) {
1357 Ok(id) => id.clone(),
1358 Err(error) => format!("rkpipe1-unavailable-{}", sanitize(error)),
1360 }
1361 }
1362
1363 pub fn embed(&self, text: &str) -> Result<Vec<f32>, EmbedError> {
1364 self.embed_prompts(vec![self.document_prompt(text)])?
1365 .pop()
1366 .ok_or_else(|| EmbedError::Output("no embedding produced".into()))
1367 }
1368
1369 pub fn embed_query(&self, text: &str) -> Result<Vec<f32>, EmbedError> {
1370 self.embed_prompts(vec![self.query_prompt(text)])?
1371 .pop()
1372 .ok_or_else(|| EmbedError::Output("no embedding produced".into()))
1373 }
1374
1375 pub fn embed_batch<'a, I>(&self, texts: I) -> Result<Vec<Vec<f32>>, EmbedError>
1379 where
1380 I: IntoIterator<Item = &'a str>,
1381 {
1382 let prompts: Vec<String> = texts.into_iter().map(|t| self.document_prompt(t)).collect();
1383 Ok(self
1384 .embed_prompts_native_for(prompts, true)?
1385 .into_iter()
1386 .map(|native| truncate_and_normalize(&native, self.truncate_dim))
1387 .collect())
1388 }
1389
1390 pub fn embed_query_batch<'a, I>(&self, texts: I) -> Result<Vec<Vec<f32>>, EmbedError>
1392 where
1393 I: IntoIterator<Item = &'a str>,
1394 {
1395 self.embed_prompts(texts.into_iter().map(|t| self.query_prompt(t)).collect())
1396 }
1397
1398 fn document_prompt(&self, text: &str) -> String {
1399 format!("{DOCUMENT_PROMPT_PREFIX}{}", self.model_text(text))
1400 }
1401
1402 fn query_prompt(&self, text: &str) -> String {
1403 format!("{QUERY_PROMPT_PREFIX}{}", self.model_text(text))
1404 }
1405
1406 fn model_text<'a>(&self, text: &'a str) -> std::borrow::Cow<'a, str> {
1409 if !self.gemma2 || !GEMMA2_MODALITY_SENTINELS.iter().any(|s| text.contains(s)) {
1410 return std::borrow::Cow::Borrowed(text);
1411 }
1412 let mut owned = text.to_owned();
1413 for sentinel in GEMMA2_MODALITY_SENTINELS {
1414 owned = owned.replace(sentinel, &sentinel.replace('|', "\u{2502}"));
1415 }
1416 std::borrow::Cow::Owned(owned)
1417 }
1418
1419 fn encode_prompts(&self, prompts: &[String]) -> Result<Vec<TokenRow>, EmbedError> {
1420 let inputs: Vec<EncodeInput> = prompts
1421 .iter()
1422 .map(|prompt| EncodeInput::Single(prompt.as_str().into()))
1423 .collect();
1424 let encodings = self
1425 .tokenizer
1426 .encode_batch(inputs, true)
1427 .map_err(|e| EmbedError::Tokenizer(e.to_string()))?;
1428 encodings
1429 .iter()
1430 .map(|encoding| {
1431 let ids: Vec<i64> = encoding
1432 .get_ids()
1433 .iter()
1434 .take(self.max_tokens)
1435 .map(|id| i64::from(*id))
1436 .collect();
1437 if ids.is_empty() {
1438 return Err(EmbedError::Tokenizer("encoding returned no tokens".into()));
1439 }
1440 let mask: Vec<i64> = encoding
1441 .get_attention_mask()
1442 .iter()
1443 .take(ids.len())
1444 .map(|value| i64::from(*value))
1445 .collect();
1446 Ok(TokenRow { ids, mask })
1447 })
1448 .collect()
1449 }
1450
1451 fn embed_prompts(&self, prompts: Vec<String>) -> Result<Vec<Vec<f32>>, EmbedError> {
1452 Ok(self
1453 .embed_prompts_native(prompts)?
1454 .into_iter()
1455 .map(|native| truncate_and_normalize(&native, self.truncate_dim))
1456 .collect())
1457 }
1458
1459 fn embed_prompts_native(&self, prompts: Vec<String>) -> Result<Vec<Vec<f32>>, EmbedError> {
1462 self.embed_prompts_native_for(prompts, false)
1463 }
1464
1465 fn embed_prompts_native_for(
1470 &self,
1471 prompts: Vec<String>,
1472 ingest: bool,
1473 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1474 if prompts.is_empty() {
1475 return Ok(Vec::new());
1476 }
1477 let rows = self.encode_prompts(&prompts)?;
1478 if let (true, Backend::E40(engine)) = (ingest, &self.backend) {
1479 let ids: Vec<&[i64]> = rows.iter().map(|row| row.ids.as_slice()).collect();
1480 if let Some(vectors) = engine.run_ingest(&ids, self.pad_id)? {
1481 let shape = [1, self.native_dim as i64];
1482 return rows
1483 .iter()
1484 .zip(vectors)
1485 .map(|(row, values)| {
1486 pool_batch(&shape, &values, &[row.mask.as_slice()], self.native_dim)
1487 .map(|mut pooled| pooled.remove(0))
1488 })
1489 .collect();
1490 }
1491 }
1492 let lengths: Vec<usize> = rows.iter().map(|row| row.ids.len()).collect();
1493 let order = length_order(&lengths);
1494 let sorted_lengths: Vec<usize> = order.iter().map(|&i| lengths[i]).collect();
1495 let accelerated = self.provider_used != "cpu";
1496 let mut out: Vec<Option<Vec<f32>>> = vec![None; rows.len()];
1497 for range in plan_chunks(
1498 &sorted_lengths,
1499 self.max_batch.max(1),
1500 MAX_PADDED_TOKENS_PER_CALL,
1501 ) {
1502 let chunk: Vec<&TokenRow> = order[range.clone()].iter().map(|&i| &rows[i]).collect();
1503 let natives = self.run_chunk(&chunk, accelerated)?;
1504 for (&index, native) in order[range].iter().zip(natives) {
1505 out[index] = Some(native);
1506 }
1507 }
1508 out.into_iter()
1509 .map(|row| row.ok_or_else(|| EmbedError::Output("missing embedding row".into())))
1510 .collect()
1511 }
1512
1513 fn run_chunk(
1516 &self,
1517 chunk: &[&TokenRow],
1518 accelerated: bool,
1519 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1520 let session = match &self.backend {
1521 Backend::Onnx(session) => session,
1522 Backend::E40(engine) => {
1526 return self.run_fixed_rows(chunk, |ids| engine.run(ids, self.pad_id));
1527 }
1528 #[cfg(target_os = "macos")]
1531 Backend::CoreMl(engine) => {
1532 return self.run_fixed_rows(chunk, |ids| engine.run(ids, self.pad_id));
1533 }
1534 };
1535 let longest = chunk.iter().map(|row| row.ids.len()).max().unwrap_or(1);
1536 let (batch_rows, padded_len) = padded_shape(chunk.len(), longest, accelerated);
1537 let padded = pad_batch(chunk, self.pad_id, padded_len, batch_rows);
1538 let (rows, len) = (padded.rows, padded.len);
1539 let tensor = |data: Vec<i64>| {
1540 ort::value::Tensor::from_array(([rows, len], data))
1541 .map_err(|e| EmbedError::Runtime(e.to_string()))
1542 };
1543 let mut inputs = ort::inputs![
1544 "input_ids" => tensor(padded.ids)?,
1545 "attention_mask" => tensor(padded.mask)?
1546 ];
1547 if self.takes_token_type_ids {
1548 inputs.push(("token_type_ids".into(), tensor(vec![0; rows * len])?.into()));
1549 }
1550 let mut session = session
1551 .lock()
1552 .map_err(|_| EmbedError::Runtime("model mutex poisoned".into()))?;
1553 let outputs = session
1554 .run(inputs)
1555 .map_err(|e| EmbedError::Runtime(e.to_string()))?;
1556 let output = outputs
1557 .get("sentence_embedding")
1558 .ok_or_else(|| EmbedError::Output("missing sentence_embedding output".into()))?;
1559 let (shape, values) = output
1560 .try_extract_tensor::<f32>()
1561 .map_err(|e| EmbedError::Output(e.to_string()))?;
1562 let masks: Vec<&[i64]> = chunk.iter().map(|row| row.mask.as_slice()).collect();
1563 pool_batch(shape.as_ref(), values, &masks, self.native_dim)
1564 }
1565}
1566
1567impl Embedder {
1568 fn run_fixed_rows(
1571 &self,
1572 chunk: &[&TokenRow],
1573 run: impl Fn(&[i64]) -> Result<Vec<f32>, EmbedError>,
1574 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1575 let mut natives = Vec::with_capacity(chunk.len());
1576 for row in chunk {
1577 let values = run(&row.ids)?;
1578 let shape = [1, self.native_dim as i64];
1579 let mut pooled = pool_batch(&shape, &values, &[row.mask.as_slice()], self.native_dim)?;
1580 natives.push(pooled.remove(0));
1581 }
1582 Ok(natives)
1583 }
1584}
1585
1586struct ParityCacheTarget {
1588 dir: PathBuf,
1589 key: parity_cache::ParityKey,
1590}
1591
1592impl ParityCacheTarget {
1593 fn store(&self, model_dir: &Path, verdict: &parity_cache::ParityVerdict) -> bool {
1596 parity_cache::write(&self.dir, model_dir, &self.key, verdict).is_ok()
1597 }
1598}
1599
1600#[cfg(feature = "bench-internals")]
1602#[derive(Debug, Clone, PartialEq, Eq)]
1603pub struct BenchTokens {
1604 pub prompt: String,
1606 pub ids: Vec<i64>,
1608 pub graph_tokens: Option<usize>,
1610 pub pad_id: i64,
1612}
1613
1614#[cfg(feature = "bench-internals")]
1616#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1617pub struct BenchEmbedSplit {
1618 pub tokenize: Duration,
1620 pub engine: e40::RunSplit,
1622 pub post: Duration,
1624 pub total: Duration,
1626}
1627
1628struct TokenRow {
1630 ids: Vec<i64>,
1631 mask: Vec<i64>,
1632}
1633
1634#[derive(Debug, PartialEq, Eq)]
1636struct PaddedBatch {
1637 rows: usize,
1638 len: usize,
1639 ids: Vec<i64>,
1640 mask: Vec<i64>,
1641}
1642
1643fn length_order(lengths: &[usize]) -> Vec<usize> {
1646 let mut order: Vec<usize> = (0..lengths.len()).collect();
1647 order.sort_by_key(|&i| lengths[i]);
1648 order
1649}
1650
1651fn plan_chunks(
1655 sorted_lengths: &[usize],
1656 max_rows: usize,
1657 token_budget: usize,
1658) -> Vec<std::ops::Range<usize>> {
1659 let mut ranges = Vec::new();
1660 let mut start = 0;
1661 while start < sorted_lengths.len() {
1662 let mut end = start + 1;
1663 while end < sorted_lengths.len()
1664 && end - start < max_rows
1665 && (end - start + 1) * sorted_lengths[end] <= token_budget
1666 {
1667 end += 1;
1668 }
1669 ranges.push(start..end);
1670 start = end;
1671 }
1672 ranges
1673}
1674
1675fn padded_shape(rows: usize, longest: usize, accelerated: bool) -> (usize, usize) {
1679 if !accelerated {
1680 return (rows, longest);
1681 }
1682 let len = ACCELERATOR_BUCKETS
1683 .iter()
1684 .copied()
1685 .find(|bucket| *bucket >= longest)
1686 .unwrap_or(longest);
1687 (rows.next_power_of_two(), len)
1688}
1689
1690fn pad_batch(chunk: &[&TokenRow], pad_id: i64, len: usize, rows: usize) -> PaddedBatch {
1693 let mut ids = vec![pad_id; rows * len];
1694 let mut mask = vec![0i64; rows * len];
1695 for row in 0..rows {
1696 let source = chunk[row.min(chunk.len() - 1)];
1697 let used = source.ids.len().min(len);
1698 ids[row * len..row * len + used].copy_from_slice(&source.ids[..used]);
1699 mask[row * len..row * len + used].copy_from_slice(&source.mask[..used]);
1700 }
1701 PaddedBatch {
1702 rows,
1703 len,
1704 ids,
1705 mask,
1706 }
1707}
1708
1709fn pool_batch(
1714 shape: &[i64],
1715 values: &[f32],
1716 masks: &[&[i64]],
1717 native_dim: usize,
1718) -> Result<Vec<Vec<f32>>, EmbedError> {
1719 let real = masks.len();
1720 if shape.len() == 2 && shape[1] == native_dim as i64 {
1721 let batch = shape[0] as usize;
1722 if batch >= real && values.len() == batch * native_dim {
1723 return Ok((0..real)
1724 .map(|row| normalize(values[row * native_dim..(row + 1) * native_dim].to_vec()))
1725 .collect());
1726 }
1727 }
1728 if shape.len() == 3 && shape[2] == native_dim as i64 && shape[0] > 0 && shape[1] > 0 {
1729 let (batch, tokens) = (shape[0] as usize, shape[1] as usize);
1730 if batch >= real && values.len() == batch * tokens * native_dim {
1731 let mut pooled_rows = Vec::with_capacity(real);
1732 for (row, mask) in masks.iter().enumerate() {
1733 let mut pooled = vec![0.0f32; native_dim];
1734 let mut count = 0.0f32;
1735 for token in 0..tokens {
1736 if mask.get(token).copied().unwrap_or(0) == 0 {
1737 continue;
1738 }
1739 count += 1.0;
1740 let base = (row * tokens + token) * native_dim;
1741 for (dim, value) in pooled.iter_mut().enumerate() {
1742 *value += values[base + dim];
1743 }
1744 }
1745 if count == 0.0 {
1746 return Err(EmbedError::Output(
1747 "attention mask contains no tokens".into(),
1748 ));
1749 }
1750 for value in &mut pooled {
1751 *value /= count;
1752 }
1753 pooled_rows.push(normalize(pooled));
1754 }
1755 return Ok(pooled_rows);
1756 }
1757 }
1758 Err(EmbedError::Output(format!(
1759 "expected [B,{native_dim}] or [B,tokens,{native_dim}] with B >= {real}, got shape {shape:?}"
1760 )))
1761}
1762
1763fn probe_output_ok(native: &[f32], native_dim: usize) -> Result<(), String> {
1764 if native.len() != native_dim {
1765 return Err(format!(
1766 "expected {native_dim} values, got {}",
1767 native.len()
1768 ));
1769 }
1770 if native.iter().any(|value| !value.is_finite()) {
1771 return Err("non-finite values".into());
1772 }
1773 if native.iter().map(|value| value * value).sum::<f32>() <= f32::EPSILON {
1774 return Err("all-zero output".into());
1775 }
1776 Ok(())
1777}
1778
1779fn parity_long_document() -> String {
1782 parity_sections(PARITY_LONG_SECTIONS)
1783}
1784
1785fn parity_sections(count: usize) -> String {
1787 (1..=count)
1788 .map(|n| {
1789 format!(
1790 "Section {n}: the nightly billing job reconciles every invoice against the \
1791 ledger, flags mismatched totals for account {}, and sends the exceptions to \
1792 the finance team for review before payment run {n}.",
1793 1000 + n * 7
1794 )
1795 })
1796 .collect::<Vec<_>>()
1797 .join(" ")
1798}
1799
1800#[derive(Debug, Clone, Copy, PartialEq)]
1802struct ParityLowest {
1803 index: usize,
1804 cosine: f32,
1805}
1806
1807#[derive(Debug, Clone, PartialEq)]
1809enum ParityMiss {
1810 Invalid { index: usize, reason: String },
1813 Below(ParityLowest),
1815}
1816
1817fn cosine(a: &[f32], b: &[f32]) -> f32 {
1818 let (mut dot, mut na, mut nb) = (0.0f64, 0.0f64, 0.0f64);
1819 for (x, y) in a.iter().zip(b) {
1820 let (x, y) = (f64::from(*x), f64::from(*y));
1821 dot += x * y;
1822 na += x * x;
1823 nb += y * y;
1824 }
1825 (dot / (na.sqrt() * nb.sqrt())) as f32
1826}
1827
1828fn compare_parity(
1833 reference: &[Vec<f32>],
1834 candidate: &[Vec<f32>],
1835 native_dim: usize,
1836 floor: f32,
1837) -> Result<ParityLowest, ParityMiss> {
1838 if reference.is_empty() || reference.len() != candidate.len() {
1839 return Err(ParityMiss::Invalid {
1840 index: 0,
1841 reason: format!(
1842 "expected {} vectors to compare, got {}",
1843 reference.len(),
1844 candidate.len()
1845 ),
1846 });
1847 }
1848 let mut lowest = ParityLowest {
1849 index: 0,
1850 cosine: f32::INFINITY,
1851 };
1852 for (index, (r, c)) in reference.iter().zip(candidate).enumerate() {
1853 probe_output_ok(r, native_dim).map_err(|reason| ParityMiss::Invalid {
1854 index,
1855 reason: format!("cpu reference: {reason}"),
1856 })?;
1857 probe_output_ok(c, native_dim).map_err(|reason| ParityMiss::Invalid {
1858 index,
1859 reason: format!("accelerator output: {reason}"),
1860 })?;
1861 let value = cosine(r, c);
1862 if value < lowest.cosine {
1863 lowest = ParityLowest {
1864 index,
1865 cosine: value,
1866 };
1867 }
1868 }
1869 if lowest.cosine >= floor {
1870 Ok(lowest)
1871 } else {
1872 Err(ParityMiss::Below(lowest))
1873 }
1874}
1875
1876fn parity_miss_detail(miss: &ParityMiss, tokens: &[usize], floor: f32) -> (Option<f32>, String) {
1880 let at = |index: usize| {
1881 let batched = if index >= tokens.len() {
1882 " (batched)"
1883 } else {
1884 ""
1885 };
1886 format!("~{} tokens{batched}", tokens[index % tokens.len()])
1887 };
1888 match miss {
1889 ParityMiss::Below(lowest) => (
1890 Some(lowest.cosine),
1891 format!(
1892 "cosine {:.3} at {}, floor {floor}",
1893 lowest.cosine,
1894 at(lowest.index)
1895 ),
1896 ),
1897 ParityMiss::Invalid { index, reason } => (None, format!("{reason} at {}", at(*index))),
1898 }
1899}
1900
1901struct PipelineParts<'a> {
1902 model_id: &'a str,
1903 native_dim: usize,
1904 truncate_dim: usize,
1905 max_tokens: usize,
1906 scrubs_sentinels: bool,
1907 files: &'a [(&'a str, String)],
1908}
1909
1910fn pipeline_fingerprint(parts: &PipelineParts<'_>) -> String {
1912 let mut text = String::new();
1913 let mut line = |key: &str, value: &str| {
1914 text.push_str(key);
1915 text.push('=');
1916 text.push_str(&value.len().to_string());
1917 text.push(':');
1918 text.push_str(value);
1919 text.push('\n');
1920 };
1921 line("schema", PIPELINE_SCHEMA);
1922 line("model_id", parts.model_id);
1923 line("native_dim", &parts.native_dim.to_string());
1924 line("truncate_dim", &parts.truncate_dim.to_string());
1925 line("max_tokens", &parts.max_tokens.to_string());
1926 line("document_prompt", DOCUMENT_PROMPT_PREFIX);
1927 line("query_prompt", QUERY_PROMPT_PREFIX);
1928 line("pooling", POOLING_DESCRIPTION);
1929 line("normalization", NORMALIZATION_DESCRIPTION);
1930 line("scrubs_sentinels", &parts.scrubs_sentinels.to_string());
1931 for (name, hash) in parts.files {
1932 line(&format!("file:{name}"), hash);
1933 }
1934 let digest = rightkit_fs::digest::sha256_bytes(text.as_bytes());
1936 format!("rkpipe1-{}", &digest[..32])
1937}
1938
1939fn sanitize(text: &str) -> String {
1940 text.chars()
1941 .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
1942 .take(48)
1943 .collect()
1944}
1945
1946pub(crate) fn hash_file(path: &Path) -> Result<String, String> {
1949 let mut file = match std::fs::File::open(path) {
1950 Ok(file) => file,
1951 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok("absent".into()),
1952 Err(error) => return Err(format!("{}: {error}", path.display())),
1953 };
1954 rightkit_fs::digest::sha256_reader(&mut file).map_err(|e| format!("{}: {e}", path.display()))
1955}
1956
1957fn bundle_is_gemma2(model_dir: &std::path::Path) -> Result<bool, EmbedError> {
1961 let path = model_dir.join("config.json");
1962 let Ok(bytes) = std::fs::read(&path) else {
1963 return Ok(false);
1964 };
1965 let config: serde_json::Value = serde_json::from_slice(&bytes)
1966 .map_err(|e| EmbedError::InvalidConfig(format!("{}: {e}", path.display())))?;
1967 Ok(config["model_type"] == "embedding_gemma2")
1968}
1969
1970#[cfg(target_os = "macos")]
1973fn coreml_bundle_files(dir: &Path) -> Result<Option<&'static [&'static str]>, EmbedError> {
1974 let files: &'static [&'static str] = &coreml::ARTIFACTS;
1975 Ok(coreml::is_bundle(dir).then_some(files))
1976}
1977
1978#[cfg(not(target_os = "macos"))]
1979fn coreml_bundle_files(dir: &Path) -> Result<Option<&'static [&'static str]>, EmbedError> {
1980 if dir.join("EmbeddingGemma2Text.mlpackage").is_dir() {
1981 return Err(EmbedError::InvalidConfig(
1982 "a Core ML bundle (EmbeddingGemma2Text.mlpackage) runs on macOS only".into(),
1983 ));
1984 }
1985 Ok(None)
1986}
1987
1988fn validate_config(config: &EmbedderConfig) -> Result<(), EmbedError> {
1989 e40::check_ingest_batch(config.ingest_batch)?;
1990 if !config.model_dir.is_dir() {
1991 return Err(EmbedError::MissingModelDirectory(config.model_dir.clone()));
1992 }
1993 if config.truncate_dim == 0 || config.truncate_dim > EMBEDDINGGEMMA_NATIVE_DIM {
1994 return Err(EmbedError::InvalidConfig(format!(
1995 "truncate dimension must be 1..={EMBEDDINGGEMMA_NATIVE_DIM}"
1996 )));
1997 }
1998 if config.max_tokens == 0 || config.max_tokens > MAX_SEQUENCE_TOKENS {
1999 return Err(EmbedError::InvalidConfig(format!(
2000 "max tokens must be 1..={MAX_SEQUENCE_TOKENS}"
2001 )));
2002 }
2003 if config.max_batch == 0 {
2004 return Err(EmbedError::InvalidConfig("max batch must be >= 1".into()));
2005 }
2006 if config.intra_threads == 0 {
2007 return Err(EmbedError::InvalidConfig(
2008 "intra threads must be >= 1".into(),
2009 ));
2010 }
2011 let coreml_files = coreml_bundle_files(&config.model_dir)?;
2012 let coreml_bundle = coreml_files.is_some();
2013 let e40_files;
2014 let e40_refs: Vec<&str>;
2015 let required: &[&str] = if let Some(files) = coreml_files {
2016 files
2017 } else if e40::is_bundle(&config.model_dir) {
2018 e40_files = e40::artifacts(&config.model_dir)?;
2019 e40_refs = e40_files.iter().map(String::as_str).collect();
2020 &e40_refs
2021 } else {
2022 &["model_q4.onnx", "model_q4.onnx_data", "tokenizer.json"]
2023 };
2024 for name in required {
2025 let path = config.model_dir.join(name);
2026 if !path.is_file() {
2027 return Err(EmbedError::MissingArtifact(path));
2028 }
2029 }
2030 if !coreml_bundle && !config.runtime_path.is_file() {
2032 return Err(EmbedError::MissingArtifact(config.runtime_path.clone()));
2033 }
2034 Ok(())
2035}
2036
2037fn normalize(mut values: Vec<f32>) -> Vec<f32> {
2038 let norm = values.iter().map(|value| value * value).sum::<f32>().sqrt();
2039 if norm > 0.0 {
2040 for value in &mut values {
2041 *value /= norm;
2042 }
2043 }
2044 values
2045}
2046
2047pub fn bundle_files(dir: &Path) -> Result<Option<Vec<String>>, EmbedError> {
2053 let files: Vec<String> = if let Some(files) = coreml_bundle_files(dir)? {
2054 files.iter().map(|file| (*file).to_owned()).collect()
2055 } else if e40::is_bundle(dir) {
2056 e40::artifacts(dir)?
2057 } else if dir.join("model_q4.onnx").is_file() {
2058 ["model_q4.onnx", "model_q4.onnx_data", "tokenizer.json"]
2059 .iter()
2060 .map(|file| (*file).to_owned())
2061 .collect()
2062 } else {
2063 return Ok(None);
2064 };
2065 for file in &files {
2066 let path = dir.join(file);
2067 if !path.is_file() {
2068 return Err(EmbedError::MissingArtifact(path));
2069 }
2070 }
2071 Ok(Some(files))
2072}
2073
2074pub fn installed_model_dir(executable: &Path, subdir: &str) -> Option<PathBuf> {
2085 let mut parts = Path::new(subdir).components();
2086 let plain = matches!(
2087 (parts.next(), parts.next()),
2088 (Some(std::path::Component::Normal(_)), None)
2089 );
2090 if !plain {
2091 return None;
2092 }
2093 rightkit_onnx::installed_resource_dir(executable).map(|dir| dir.join(subdir))
2094}
2095
2096pub fn truncate_and_normalize(native: &[f32], dimension: usize) -> Vec<f32> {
2097 let mut values = native[..native.len().min(dimension)].to_vec();
2098 values.resize(dimension, 0.0);
2099 normalize(values)
2100}
2101
2102#[cfg(test)]
2103mod layout_tests {
2104 use super::*;
2105
2106 #[test]
2107 fn installed_model_dir_refuses_relative_and_non_plain_names() {
2108 assert_eq!(
2109 installed_model_dir(Path::new("app/bin/app"), "models"),
2110 None
2111 );
2112 let exe = if cfg!(windows) {
2113 Path::new(r"C:\app\app.exe")
2114 } else {
2115 Path::new("/opt/app/app")
2116 };
2117 assert_eq!(installed_model_dir(exe, ""), None);
2118 assert_eq!(installed_model_dir(exe, "../models"), None);
2119 assert_eq!(installed_model_dir(exe, "/models"), None);
2120 assert_eq!(installed_model_dir(exe, "a/b"), None);
2121 assert_eq!(installed_model_dir(exe, "./models"), None);
2122 let dir = installed_model_dir(exe, "embed").expect("absolute executable");
2123 assert!(dir.ends_with("embed"));
2124 }
2125
2126 #[test]
2127 fn bundle_files_names_the_q4_layout_and_none_for_an_empty_directory() {
2128 let dir = tempfile::tempdir().unwrap();
2129 assert_eq!(bundle_files(dir.path()).unwrap(), None);
2130 for name in ["model_q4.onnx", "model_q4.onnx_data", "tokenizer.json"] {
2131 std::fs::write(dir.path().join(name), b"x").unwrap();
2132 }
2133 assert_eq!(
2134 bundle_files(dir.path()).unwrap(),
2135 Some(vec![
2136 "model_q4.onnx".to_owned(),
2137 "model_q4.onnx_data".to_owned(),
2138 "tokenizer.json".to_owned(),
2139 ])
2140 );
2141 }
2142
2143 #[test]
2144 fn bundle_files_refuses_a_partial_e40_bundle() {
2145 let dir = tempfile::tempdir().unwrap();
2146 std::fs::write(dir.path().join(e40::TABLE_FILE), b"x").unwrap();
2147 std::fs::write(dir.path().join("model_32.onnx"), b"x").unwrap();
2148 assert!(matches!(
2149 bundle_files(dir.path()),
2150 Err(EmbedError::MissingArtifact(_))
2151 ));
2152 }
2153}
2154
2155#[cfg(test)]
2156mod tests {
2157 use super::*;
2158
2159 fn row(ids: &[i64]) -> TokenRow {
2160 TokenRow {
2161 ids: ids.to_vec(),
2162 mask: vec![1; ids.len()],
2163 }
2164 }
2165
2166 fn state(ids: &[i64], dim: usize) -> Vec<f32> {
2169 ids.iter()
2170 .enumerate()
2171 .flat_map(|(pos, id)| {
2172 (0..dim)
2173 .map(move |d| ((*id as f32) * 0.37 + pos as f32 * 0.11 + d as f32 * 0.05).sin())
2174 })
2175 .collect()
2176 }
2177
2178 #[test]
2179 fn pad_batch_pads_right_with_zero_mask_and_repeats_last_row() {
2180 let a = row(&[5, 6, 7]);
2181 let b = row(&[9]);
2182 let padded = pad_batch(&[&a, &b], 0, 4, 3);
2183 assert_eq!((padded.rows, padded.len), (3, 4));
2184 assert_eq!(padded.ids, vec![5, 6, 7, 0, 9, 0, 0, 0, 9, 0, 0, 0]);
2185 assert_eq!(padded.mask, vec![1, 1, 1, 0, 1, 0, 0, 0, 1, 0, 0, 0]);
2186 }
2187
2188 #[test]
2189 fn pad_batch_uses_the_pad_id_for_masked_positions() {
2190 let a = row(&[5]);
2191 let padded = pad_batch(&[&a], 3, 3, 1);
2192 assert_eq!(padded.ids, vec![5, 3, 3]);
2193 assert_eq!(padded.mask, vec![1, 0, 0]);
2194 }
2195
2196 #[test]
2197 fn padded_mean_pool_matches_unpadded_rows() {
2198 let dim = 8;
2199 let rows_ids: [&[i64]; 3] = [&[3, 4, 5, 6, 7], &[10, 11], &[20, 21, 22]];
2200 let rows: Vec<TokenRow> = rows_ids.iter().map(|ids| row(ids)).collect();
2201 let refs: Vec<&TokenRow> = rows.iter().collect();
2202 let padded = pad_batch(&refs, 0, 5, 3);
2203 let mut values = vec![9999.0f32; padded.rows * padded.len * dim];
2206 for (r, ids) in rows_ids.iter().enumerate() {
2207 let s = state(ids, dim);
2208 let base = r * padded.len * dim;
2209 values[base..base + s.len()].copy_from_slice(&s);
2210 }
2211 let masks: Vec<&[i64]> = rows.iter().map(|r| r.mask.as_slice()).collect();
2212 let batched =
2213 pool_batch(&[3, padded.len as i64, dim as i64], &values, &masks, dim).unwrap();
2214 for (r, ids) in rows_ids.iter().enumerate() {
2215 let single = pool_batch(
2216 &[1, ids.len() as i64, dim as i64],
2217 &state(ids, dim),
2218 &[rows[r].mask.as_slice()],
2219 dim,
2220 )
2221 .unwrap();
2222 let diff = batched[r]
2223 .iter()
2224 .zip(&single[0])
2225 .map(|(a, b)| (a - b).abs())
2226 .fold(0.0f32, f32::max);
2227 assert!(diff <= 1e-6, "row {r} differs by {diff}");
2228 let norm: f32 = batched[r].iter().map(|v| v * v).sum::<f32>().sqrt();
2229 assert!((norm - 1.0).abs() <= 1e-5);
2230 }
2231 }
2232
2233 #[test]
2234 fn pooled_output_rows_are_normalized_and_extra_rows_dropped() {
2235 let dim = 4;
2236 let values = vec![3.0, 4.0, 0.0, 0.0, 0.0, 0.0, 2.0, 0.0, 1.0, 1.0, 1.0, 1.0];
2237 let m = [1i64];
2238 let out = pool_batch(&[3, 4], &values, &[&m, &m], dim).unwrap();
2239 assert_eq!(out.len(), 2);
2240 assert_eq!(out[0], vec![0.6, 0.8, 0.0, 0.0]);
2241 assert_eq!(out[1], vec![0.0, 0.0, 1.0, 0.0]);
2242 }
2243
2244 #[test]
2245 fn pool_rejects_bad_shapes_and_empty_masks() {
2246 let m = [0i64, 0];
2247 assert!(pool_batch(&[1, 2, 2], &[1.0; 4], &[&m], 2).is_err());
2248 assert!(pool_batch(&[1, 3], &[1.0; 3], &[&[1]], 2).is_err());
2249 assert!(pool_batch(&[1, 2], &[1.0; 2], &[&[1], &[1]], 2).is_err());
2250 }
2251
2252 #[test]
2253 fn chunks_respect_row_cap_token_budget_and_cover_everything() {
2254 let lengths = [1, 2, 3, 4, 5, 6, 7];
2255 assert_eq!(plan_chunks(&lengths, 3, 1000), vec![0..3, 3..6, 6..7]);
2256 assert_eq!(plan_chunks(&[1, 3, 4, 4], 8, 10), vec![0..2, 2..4]);
2258 assert_eq!(plan_chunks(&[50, 60], 8, 10), vec![0..1, 1..2]);
2260 assert!(plan_chunks(&[], 4, 10).is_empty());
2261 }
2262
2263 #[test]
2264 fn length_order_is_stable_ascending() {
2265 assert_eq!(length_order(&[5, 2, 5, 1, 2]), vec![3, 1, 4, 0, 2]);
2266 }
2267
2268 #[test]
2269 fn accelerators_use_bucketed_shapes_cpu_uses_exact() {
2270 assert_eq!(padded_shape(3, 70, false), (3, 70));
2271 assert_eq!(padded_shape(3, 70, true), (4, 128));
2272 assert_eq!(padded_shape(1, 64, true), (1, 64));
2273 assert_eq!(padded_shape(16, 2048, true), (16, 2048));
2274 }
2275
2276 #[test]
2277 fn probe_rejects_zero_nonfinite_and_wrong_size() {
2278 assert!(probe_output_ok(&[0.0; 4], 4).is_err());
2279 assert!(probe_output_ok(&[f32::NAN, 1.0, 0.0, 0.0], 4).is_err());
2280 assert!(probe_output_ok(&[1.0; 3], 4).is_err());
2281 assert!(probe_output_ok(&[0.5, 0.5, 0.5, 0.5], 4).is_ok());
2282 }
2283
2284 #[test]
2285 fn parity_floor_is_pinned() {
2286 assert_eq!(ACCELERATOR_PARITY_MIN_COSINE, 0.995);
2287 }
2288
2289 #[test]
2290 fn identical_vectors_pass_parity_and_report_the_lowest_pair() {
2291 let reference = vec![vec![1.0, 0.0, 0.0], vec![0.0, 1.0, 0.0]];
2292 let lowest = compare_parity(&reference, &reference.clone(), 3, 0.995).unwrap();
2293 assert!((lowest.cosine - 1.0).abs() <= 1e-6);
2294 let candidate = vec![vec![1.0, 0.0, 0.0], vec![0.0, 0.6, 0.8]];
2296 let at_floor = cosine(&reference[1], &candidate[1]);
2297 let lowest = compare_parity(&reference, &candidate, 3, at_floor).unwrap();
2298 assert_eq!((lowest.index, lowest.cosine), (1, at_floor));
2299 }
2300
2301 #[test]
2302 fn a_low_cosine_pair_fails_and_reports_value_and_index() {
2303 let wrong = vec![(1.0f32 - 0.58 * 0.58).sqrt(), 0.58];
2305 let reference = vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 0.0]];
2306 let candidate = vec![vec![1.0, 0.0], wrong.clone(), vec![1.0, 0.0]];
2307 match compare_parity(&reference, &candidate, 2, 0.995) {
2308 Err(ParityMiss::Below(lowest)) => {
2309 assert_eq!(lowest.index, 1);
2310 assert!((lowest.cosine - 0.58).abs() <= 1e-6, "{lowest:?}");
2311 }
2312 other => panic!("expected a Below miss, got {other:?}"),
2313 }
2314 let single = compare_parity(&[vec![0.0, 1.0]], &[wrong], 2, 0.995);
2315 match single {
2316 Err(ParityMiss::Below(lowest)) => assert!((lowest.cosine - 0.58).abs() <= 1e-6),
2317 other => panic!("expected a Below miss, got {other:?}"),
2318 }
2319 }
2320
2321 #[test]
2322 fn nonfinite_zero_or_wrongly_sized_vectors_fail_parity() {
2323 let good = vec![vec![1.0, 0.0]];
2324 for bad in [
2325 vec![f32::NAN, 1.0],
2326 vec![f32::INFINITY, 0.0],
2327 vec![0.0, 0.0],
2328 vec![1.0, 0.0, 0.0],
2329 ] {
2330 assert!(
2331 matches!(
2332 compare_parity(&good, std::slice::from_ref(&bad), 2, 0.995),
2333 Err(ParityMiss::Invalid { index: 0, .. })
2334 ),
2335 "candidate {bad:?}"
2336 );
2337 assert!(
2338 matches!(
2339 compare_parity(std::slice::from_ref(&bad), &good, 2, 0.995),
2340 Err(ParityMiss::Invalid { index: 0, .. })
2341 ),
2342 "reference {bad:?}"
2343 );
2344 }
2345 }
2346
2347 #[test]
2348 fn mismatched_or_empty_parity_inputs_fail() {
2349 let one = vec![vec![1.0, 0.0]];
2350 let two = vec![vec![1.0, 0.0], vec![1.0, 0.0]];
2351 assert!(matches!(
2352 compare_parity(&one, &two, 2, 0.995),
2353 Err(ParityMiss::Invalid { .. })
2354 ));
2355 assert!(matches!(
2356 compare_parity(&two, &one, 2, 0.995),
2357 Err(ParityMiss::Invalid { .. })
2358 ));
2359 assert!(matches!(
2360 compare_parity(&[], &[], 2, 0.995),
2361 Err(ParityMiss::Invalid { .. })
2362 ));
2363 }
2364
2365 #[test]
2366 fn parity_miss_detail_names_cosine_length_and_floor() {
2367 let tokens = [20, 100, 600];
2368 let miss = ParityMiss::Below(ParityLowest {
2369 index: 0,
2370 cosine: 0.5791,
2371 });
2372 assert_eq!(
2373 parity_miss_detail(&miss, &tokens, 0.995),
2374 (
2375 Some(0.5791),
2376 "cosine 0.579 at ~20 tokens, floor 0.995".to_owned()
2377 )
2378 );
2379 let batched = ParityMiss::Below(ParityLowest {
2380 index: 5,
2381 cosine: 0.71,
2382 });
2383 assert_eq!(
2384 parity_miss_detail(&batched, &tokens, 0.995).1,
2385 "cosine 0.710 at ~600 tokens (batched), floor 0.995"
2386 );
2387 let invalid = ParityMiss::Invalid {
2388 index: 1,
2389 reason: "accelerator output: non-finite values".into(),
2390 };
2391 assert_eq!(
2392 parity_miss_detail(&invalid, &tokens, 0.995),
2393 (
2394 None,
2395 "accelerator output: non-finite values at ~100 tokens".to_owned()
2396 )
2397 );
2398 }
2399
2400 #[test]
2401 fn parity_long_document_is_deterministic_and_long() {
2402 let text = parity_long_document();
2403 assert_eq!(text, parity_long_document());
2404 assert!(text.split_whitespace().count() >= 400, "{}", text.len());
2405 assert!(PARITY_DOCUMENT.split_whitespace().count() >= 50);
2406 }
2407
2408 #[test]
2409 fn parity_error_names_provider_cosine_and_floor() {
2410 let error = EmbedError::AcceleratorParity {
2411 provider: "directml",
2412 lowest_cosine: Some(0.579),
2413 floor: ACCELERATOR_PARITY_MIN_COSINE,
2414 detail: "cosine 0.579 at ~20 tokens, floor 0.995".into(),
2415 };
2416 assert_eq!(
2417 error.to_string(),
2418 "directml failed the vector parity check against cpu: cosine 0.579 at ~20 tokens, floor 0.995"
2419 );
2420 }
2421
2422 fn parts<'a>(files: &'a [(&'a str, String)], truncate: usize) -> PipelineParts<'a> {
2423 PipelineParts {
2424 model_id: EMBEDDINGGEMMA_MODEL_ID,
2425 native_dim: 768,
2426 truncate_dim: truncate,
2427 max_tokens: 2048,
2428 scrubs_sentinels: false,
2429 files,
2430 }
2431 }
2432
2433 #[test]
2434 fn pipeline_fingerprint_is_stable_and_sensitive() {
2435 let files = [("model_q4.onnx", "aa".to_owned())];
2436 let base = pipeline_fingerprint(&parts(&files, 256));
2437 assert_eq!(base, pipeline_fingerprint(&parts(&files, 256)));
2438 assert!(base.starts_with("rkpipe1-") && base.len() == "rkpipe1-".len() + 32);
2439 assert_ne!(base, pipeline_fingerprint(&parts(&files, 128)));
2440 let changed = [("model_q4.onnx", "ab".to_owned())];
2441 assert_ne!(base, pipeline_fingerprint(&parts(&changed, 256)));
2442 let mut other = parts(&files, 256);
2443 other.scrubs_sentinels = true;
2444 assert_ne!(base, pipeline_fingerprint(&other));
2445 other = parts(&files, 256);
2446 other.max_tokens = 512;
2447 assert_ne!(base, pipeline_fingerprint(&other));
2448 }
2449
2450 #[test]
2451 fn hash_file_reports_absent_and_content() {
2452 let dir = tempfile::tempdir().unwrap();
2453 assert_eq!(hash_file(&dir.path().join("none")).unwrap(), "absent");
2454 let path = dir.path().join("f");
2455 std::fs::write(&path, b"abc").unwrap();
2456 assert_eq!(
2457 hash_file(&path).unwrap(),
2458 "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
2459 );
2460 }
2461
2462 #[test]
2463 fn config_defaults_and_builders() {
2464 let c = EmbedderConfig::new("m", "r");
2465 assert!(!c.runtime_binding.allow_known_bad_runtime);
2466 let qa = c.clone().with_runtime_binding(RuntimeBindingOptions {
2467 allow_known_bad_runtime: true,
2468 ..Default::default()
2469 });
2470 assert!(qa.runtime_binding.allow_known_bad_runtime);
2471 assert!(
2472 !EmbedderConfig::new("m", "r")
2473 .runtime_binding
2474 .allow_known_bad_runtime
2475 );
2476 assert_eq!((c.max_batch, c.intra_threads), (16, 1));
2477 assert_eq!(c.ingest_batch, 8);
2478 assert_eq!(c.clone().with_ingest_batch(4).ingest_batch, 4);
2479 assert!(matches!(
2480 validate_config(&c.clone().with_ingest_batch(16)),
2481 Err(EmbedError::UnsupportedIngestBatch { requested: 16, .. })
2482 ));
2483 assert_eq!(c.provider, ExecutionProvider::Cpu);
2484 let c = c
2485 .with_max_batch(4)
2486 .with_intra_threads(3)
2487 .with_provider(ExecutionProvider::Auto);
2488 assert_eq!((c.max_batch, c.intra_threads), (4, 3));
2489 assert_eq!(c.provider, ExecutionProvider::Auto);
2490 assert!(!c.runtime_binding.allow_known_bad_runtime);
2491 let legacy =
2492 EmbedderConfig::new("m", "r").with_provider(rightkit_onnx::ExecutionProvider::DirectMl);
2493 assert_eq!(legacy.provider, ExecutionProvider::DirectMl);
2494 assert!(EmbedderConfig::new("m", "r").fallback_model_dir.is_none());
2495 let with_fallback = EmbedderConfig::new("m", "r").with_fallback_model_dir("q4");
2496 assert_eq!(with_fallback.fallback_model_dir, Some(PathBuf::from("q4")));
2497 assert!(EmbedderConfig::new("m", "r").coreml_cache_dir.is_none());
2498 let with_cache = EmbedderConfig::new("m", "r").with_coreml_cache_dir("cache");
2499 assert_eq!(with_cache.coreml_cache_dir, Some(PathBuf::from("cache")));
2500 assert!(EmbedderConfig::new("m", "r").coreml_units.is_none());
2501 let forced =
2502 EmbedderConfig::new("m", "r").with_coreml_units(crate::CoreMlComputeUnit::CpuAndGpu);
2503 assert_eq!(
2504 forced.coreml_units,
2505 Some(CoreMlUnits::uniform(crate::CoreMlComputeUnit::CpuAndGpu))
2506 );
2507 let custom = EmbedderConfig::new("m", "r")
2508 .with_coreml_units("256=cpu_only".parse::<CoreMlUnits>().unwrap());
2509 assert_eq!(
2510 custom.coreml_units.unwrap().unit_for(256),
2511 Some(crate::CoreMlComputeUnit::CpuOnly)
2512 );
2513 }
2514
2515 #[test]
2516 fn model_ids_are_distinct_so_pipeline_ids_differ_per_model() {
2517 let ids = [
2518 EMBEDDINGGEMMA_MODEL_ID,
2519 EMBEDDINGGEMMA2_MODEL_ID,
2520 EMBEDDINGGEMMA2_E40_MODEL_ID,
2521 EMBEDDINGGEMMA2_COREML_MODEL_ID,
2522 ];
2523 for (i, a) in ids.iter().enumerate() {
2524 for b in &ids[i + 1..] {
2525 assert_ne!(a, b);
2526 }
2527 }
2528 let files = [("model_32.onnx", "aa".to_owned())];
2529 let parts = |model_id| PipelineParts {
2530 model_id,
2531 native_dim: 768,
2532 truncate_dim: 256,
2533 max_tokens: 512,
2534 scrubs_sentinels: true,
2535 files: &files,
2536 };
2537 assert_ne!(
2538 pipeline_fingerprint(&parts(EMBEDDINGGEMMA2_E40_MODEL_ID)),
2539 pipeline_fingerprint(&parts(EMBEDDINGGEMMA2_MODEL_ID))
2540 );
2541 }
2542
2543 #[test]
2544 fn parity_sections_are_prefixes_of_the_long_document() {
2545 assert!(parity_long_document().starts_with(&parity_sections(4)));
2546 assert!(parity_sections(4).starts_with(&parity_sections(1)));
2547 assert!(
2548 parity_sections(1).split_whitespace().count()
2549 < parity_sections(4).split_whitespace().count()
2550 );
2551 }
2552}