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, init_environment, CandidateSource, EnvironmentOptions, GlobalPool,
16 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}
283
284impl EmbedderConfig {
285 pub fn new(model_dir: impl Into<PathBuf>, runtime_path: impl Into<PathBuf>) -> Self {
286 Self {
287 model_dir: model_dir.into(),
288 runtime_path: runtime_path.into(),
289 provider: ExecutionProvider::Cpu,
290 truncate_dim: DEFAULT_TRUNCATE_DIM,
291 max_tokens: MAX_SEQUENCE_TOKENS,
292 max_batch: DEFAULT_MAX_BATCH,
293 ingest_batch: DEFAULT_INGEST_BATCH,
294 intra_threads: 1,
295 fallback_model_dir: None,
296 coreml_cache_dir: None,
297 parity_cache_dir: None,
298 prewarm: None,
299 coreml_units: None,
300 e40_session_profile: None,
301 }
302 }
303
304 pub fn with_prewarm(mut self, prewarm: bool) -> Self {
316 self.prewarm = Some(prewarm);
317 self
318 }
319
320 pub fn with_coreml_units(mut self, units: impl Into<CoreMlUnits>) -> Self {
328 self.coreml_units = Some(units.into());
329 self
330 }
331
332 #[cfg(feature = "bench-internals")]
335 pub fn with_e40_session_profile(mut self, profile: E40SessionProfile) -> Self {
336 self.e40_session_profile = Some(profile);
337 self
338 }
339
340 pub fn with_fallback_model_dir(mut self, dir: impl Into<PathBuf>) -> Self {
350 self.fallback_model_dir = Some(dir.into());
351 self
352 }
353
354 pub fn with_coreml_cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
359 self.coreml_cache_dir = Some(dir.into());
360 self
361 }
362
363 pub fn with_parity_cache_dir(mut self, dir: impl Into<PathBuf>) -> Self {
380 self.parity_cache_dir = Some(dir.into());
381 self
382 }
383
384 pub fn with_provider(mut self, provider: impl Into<ExecutionProvider>) -> Self {
385 self.provider = provider.into();
386 self
387 }
388
389 pub fn with_truncate_dim(mut self, dim: usize) -> Self {
390 self.truncate_dim = dim;
391 self
392 }
393
394 pub fn with_max_tokens(mut self, max_tokens: usize) -> Self {
395 self.max_tokens = max_tokens;
396 self
397 }
398
399 pub fn with_max_batch(mut self, max_batch: usize) -> Self {
401 self.max_batch = max_batch;
402 self
403 }
404
405 pub fn with_ingest_batch(mut self, ingest_batch: usize) -> Self {
411 self.ingest_batch = ingest_batch;
412 self
413 }
414
415 pub fn with_intra_threads(mut self, threads: usize) -> Self {
417 self.intra_threads = threads;
418 self
419 }
420}
421
422enum Backend {
425 Onnx(Mutex<Session>),
426 E40(E40Engine),
427 #[cfg(target_os = "macos")]
428 CoreMl(CoreMlEngine),
429}
430
431pub struct Embedder {
432 tokenizer: Tokenizer,
433 backend: Backend,
434 model_dir: PathBuf,
435 native_dim: usize,
436 truncate_dim: usize,
437 max_tokens: usize,
438 max_batch: usize,
439 model_id: String,
440 gemma2: bool,
441 pad_id: i64,
442 takes_token_type_ids: bool,
443 provider_used: &'static str,
444 fallback_reason: Option<String>,
445 parity: Option<AcceleratorParity>,
446 load_timings: Option<LoadTimings>,
447 pipeline_id: OnceLock<Result<String, String>>,
448}
449
450struct BuiltEmbedSession {
451 session: Session,
452 provider: rightkit_onnx::ExecutionProvider,
453 fallback: Option<String>,
454}
455
456fn build_session(
457 config: &EmbedderConfig,
458 provider: rightkit_onnx::ExecutionProvider,
459 cpu_fallback: bool,
460) -> Result<BuiltEmbedSession, EmbedError> {
461 let built = SessionOptions::new(provider)
462 .with_threads(Some(config.intra_threads))
463 .with_cpu_fallback(cpu_fallback)
464 .build_from_file_reported(&config.model_dir.join("model_q4.onnx"))
465 .map_err(|e| EmbedError::Runtime(e.to_string()))?;
466 Ok(BuiltEmbedSession {
467 session: built.session,
468 provider: built.provider,
469 fallback: built.fallback,
470 })
471}
472
473fn takes_token_type_ids(session: &Session) -> bool {
474 session
475 .inputs()
476 .iter()
477 .any(|input| input.name() == "token_type_ids")
478}
479
480fn load_tokenizer(model_dir: &Path) -> Result<(Tokenizer, i64), EmbedError> {
483 let mut tokenizer = Tokenizer::from_file(model_dir.join("tokenizer.json"))
484 .map_err(|e| EmbedError::Tokenizer(e.to_string()))?;
485 let pad_id = tokenizer
486 .get_padding()
487 .map(|padding| i64::from(padding.pad_id))
488 .or_else(|| tokenizer.token_to_id("<pad>").map(i64::from))
489 .unwrap_or(0);
490 tokenizer.with_padding(None);
491 Ok((tokenizer, pad_id))
492}
493
494impl Embedder {
495 pub fn load(config: EmbedderConfig) -> Result<Self, EmbedError> {
496 let started = Instant::now();
497 validate_config(&config)?;
498 #[cfg(target_os = "macos")]
500 {
501 if coreml::is_bundle(&config.model_dir) {
502 let (tokenizer, pad_id) = load_tokenizer(&config.model_dir)?;
503 let tokenizer_load = started.elapsed();
504 let mut embedder = Self::load_coreml(config, tokenizer, pad_id)?;
505 if let Some(timings) = embedder.load_timings.as_mut() {
507 timings.tokenizer_load = tokenizer_load;
508 timings.total = started.elapsed();
509 }
510 return Ok(embedder);
511 }
512 }
513 let gemma2 = bundle_is_gemma2(&config.model_dir)?;
514 let tokenizer_started = Instant::now();
515 let (tokenizer, pad_id) = load_tokenizer(&config.model_dir)?;
516 let tokenizer_load = tokenizer_started.elapsed();
517 let selection = configure_runtime([RuntimeCandidate::new(
518 config.runtime_path.clone(),
519 CandidateSource::Explicit,
520 )])
521 .map_err(|e| EmbedError::Runtime(e.to_string()))?;
522 init_environment(
523 &selection,
524 &EnvironmentOptions {
525 global_pool: Some(GlobalPool::from_budget()),
526 },
527 )
528 .map_err(EmbedError::Runtime)?;
529 if e40::is_bundle(&config.model_dir) {
530 let mut embedder = Self::load_e40(config, tokenizer, pad_id, tokenizer_load)?;
531 let sessions = match &embedder.backend {
533 Backend::E40(engine) => Some(engine.finish_session_log()),
534 _ => None,
535 };
536 if let (Some(timings), Some(sessions)) = (embedder.load_timings.as_mut(), sessions) {
537 timings.session_creates = sessions;
538 timings.total = started.elapsed();
539 }
540 return Ok(embedder);
541 }
542 let (requested, auto) = config.provider.requested();
543 let built = build_session(&config, requested, auto)?;
544 let accelerator = built.provider;
545 let mut embedder = Self::from_session(&config, tokenizer, pad_id, gemma2, built);
546 if auto && accelerator != rightkit_onnx::ExecutionProvider::Cpu {
547 if let Err(reason) = embedder.probe() {
548 let cpu = build_session(&config, rightkit_onnx::ExecutionProvider::Cpu, false)?;
549 embedder.takes_token_type_ids = takes_token_type_ids(&cpu.session);
550 embedder.backend = Backend::Onnx(Mutex::new(cpu.session));
551 embedder.provider_used = provider_name(cpu.provider);
552 embedder.fallback_reason = Some(format!(
553 "{} -> cpu: probe inference failed: {reason}",
554 provider_name(accelerator)
555 ));
556 }
557 }
558 if embedder.provider_used != "cpu" {
559 return embedder.admit_accelerator(&config, auto);
560 }
561 Ok(embedder)
562 }
563
564 fn from_session(
565 config: &EmbedderConfig,
566 tokenizer: Tokenizer,
567 pad_id: i64,
568 gemma2: bool,
569 built: BuiltEmbedSession,
570 ) -> Self {
571 Self {
572 tokenizer,
573 takes_token_type_ids: takes_token_type_ids(&built.session),
574 backend: Backend::Onnx(Mutex::new(built.session)),
575 model_dir: config.model_dir.clone(),
576 native_dim: EMBEDDINGGEMMA_NATIVE_DIM,
577 truncate_dim: config.truncate_dim,
578 max_tokens: config.max_tokens,
579 max_batch: config.max_batch,
580 model_id: if gemma2 {
581 EMBEDDINGGEMMA2_MODEL_ID
582 } else {
583 EMBEDDINGGEMMA_MODEL_ID
584 }
585 .to_owned(),
586 gemma2,
587 pad_id,
588 provider_used: provider_name(built.provider),
589 fallback_reason: built.fallback,
590 parity: None,
591 load_timings: None,
592 pipeline_id: OnceLock::new(),
593 }
594 }
595
596 fn from_e40(
597 config: &EmbedderConfig,
598 tokenizer: Tokenizer,
599 pad_id: i64,
600 engine: E40Engine,
601 ) -> Self {
602 Self {
603 tokenizer,
604 provider_used: provider_name(engine.provider()),
605 backend: Backend::E40(engine),
606 model_dir: config.model_dir.clone(),
607 native_dim: EMBEDDINGGEMMA_NATIVE_DIM,
608 truncate_dim: config.truncate_dim,
609 max_tokens: config.max_tokens.min(e40::MAX_TOKENS),
611 max_batch: config.max_batch,
612 model_id: EMBEDDINGGEMMA2_E40_MODEL_ID.to_owned(),
613 gemma2: true,
614 pad_id,
615 takes_token_type_ids: false,
616 fallback_reason: None,
617 parity: None,
618 load_timings: None,
619 pipeline_id: OnceLock::new(),
620 }
621 }
622
623 fn load_e40(
629 config: EmbedderConfig,
630 tokenizer: Tokenizer,
631 pad_id: i64,
632 tokenizer_load: Duration,
633 ) -> Result<Self, EmbedError> {
634 use rightkit_onnx::ExecutionProvider as Ep;
635 let (requested, auto) = match config.provider {
636 ExecutionProvider::Cpu => (Ep::Cpu, false),
637 ExecutionProvider::DirectMl => (Ep::DirectMl, false),
638 ExecutionProvider::CoreMl => {
639 return Err(EmbedError::InvalidConfig(
640 "an E40 bundle runs on DirectML or CPU, not CoreML".into(),
641 ))
642 }
643 ExecutionProvider::Auto => match Ep::platform_accelerator() {
644 Ep::DirectMl => (Ep::DirectMl, true),
645 _ => (Ep::Cpu, false),
646 },
647 };
648 let profile = config
649 .e40_session_profile
650 .unwrap_or_else(|| E40SessionProfile::shipped(config.intra_threads));
651 let engine = E40Engine::open(&config.model_dir, requested, profile, config.ingest_batch)?;
652 let timings = LoadTimings {
653 tokenizer_load,
654 token_table_read: engine.token_table_read(),
655 ..LoadTimings::default()
656 };
657 let mut embedder = Self::from_e40(&config, tokenizer, pad_id, engine);
658 embedder.load_timings = Some(timings);
659 if requested == Ep::Cpu {
660 return Ok(embedder);
661 }
662 Self::admit_or_fall_back(embedder, &config, auto)
663 }
664
665 #[cfg(target_os = "macos")]
666 fn from_coreml(
667 config: &EmbedderConfig,
668 tokenizer: Tokenizer,
669 pad_id: i64,
670 engine: CoreMlEngine,
671 ) -> Self {
672 Self {
673 tokenizer,
674 provider_used: engine.provider_name(),
675 backend: Backend::CoreMl(engine),
676 model_dir: config.model_dir.clone(),
677 native_dim: EMBEDDINGGEMMA_NATIVE_DIM,
678 truncate_dim: config.truncate_dim,
679 max_tokens: config.max_tokens.min(e40::MAX_TOKENS),
681 max_batch: config.max_batch,
682 model_id: EMBEDDINGGEMMA2_COREML_MODEL_ID.to_owned(),
683 gemma2: true,
684 pad_id,
685 takes_token_type_ids: false,
686 fallback_reason: None,
687 parity: None,
688 load_timings: Some(LoadTimings::default()),
689 pipeline_id: OnceLock::new(),
690 }
691 }
692
693 #[cfg(target_os = "macos")]
705 fn load_coreml(
706 config: EmbedderConfig,
707 tokenizer: Tokenizer,
708 pad_id: i64,
709 ) -> Result<Self, EmbedError> {
710 let accelerated = || config.coreml_units.clone().unwrap_or_default();
711 let (units, auto) = match config.provider {
712 ExecutionProvider::Cpu => (CoreMlUnits::uniform(CoreMlComputeUnit::CpuOnly), false),
713 ExecutionProvider::Auto => (accelerated(), true),
714 ExecutionProvider::CoreMl => (accelerated(), false),
715 ExecutionProvider::DirectMl => {
716 return Err(EmbedError::InvalidConfig(
717 "a Core ML bundle runs on Core ML or CPU, not DirectML".into(),
718 ))
719 }
720 };
721 let cpu_only = units.is_cpu_only();
722 let engine = match CoreMlEngine::open(
723 &config.model_dir,
724 units,
725 config.coreml_cache_dir.as_deref(),
726 ) {
727 Ok(engine) => engine,
728 Err(error) if auto => {
729 return match &config.fallback_model_dir {
730 Some(dir) => {
731 Self::load_fallback(&config, dir, format!("coreml -> cpu: {error}"))
732 }
733 None => Err(error),
734 }
735 }
736 Err(error) => return Err(error),
737 };
738 let embedder = Self::from_coreml(&config, tokenizer, pad_id, engine);
739 if cpu_only {
741 return Ok(embedder);
742 }
743 Self::admit_or_fall_back(embedder, &config, auto)
744 }
745
746 fn admit_or_fall_back(
750 mut embedder: Self,
751 config: &EmbedderConfig,
752 auto: bool,
753 ) -> Result<Self, EmbedError> {
754 let config = config.clone();
755 let accelerator = embedder.provider_used;
756 if auto {
757 let probe_started = Instant::now();
758 let probed = embedder.probe();
759 if let Some(timings) = embedder.load_timings.as_mut() {
760 timings.probe = Some(probe_started.elapsed());
761 }
762 if let Err(reason) = probed {
763 return embedder.e40_not_admitted(
764 &config,
765 format!("{accelerator} -> cpu: probe inference failed: {reason}"),
766 );
767 }
768 }
769 match embedder.admit_accelerator(&config, auto) {
770 Ok(admitted) if admitted.provider_used == "cpu" => {
772 match (&config.fallback_model_dir, admitted.fallback_reason.clone()) {
773 (Some(dir), Some(reason)) => Self::load_fallback(&config, dir, reason),
774 _ => Ok(admitted),
775 }
776 }
777 Ok(admitted) => Ok(admitted),
778 Err(error) if auto => match &config.fallback_model_dir {
779 Some(dir) => {
780 Self::load_fallback(&config, dir, format!("{accelerator} -> cpu: {error}"))
781 }
782 None => Err(error),
783 },
784 Err(error) => Err(error),
785 }
786 }
787
788 fn e40_not_admitted(
791 mut self,
792 config: &EmbedderConfig,
793 reason: String,
794 ) -> Result<Self, EmbedError> {
795 if let Some(dir) = &config.fallback_model_dir {
796 return Self::load_fallback(config, dir, reason);
797 }
798 let mut cpu = self.cpu_twin(config)?;
799 cpu.fallback_reason = Some(reason);
800 cpu.load_timings = self.load_timings.take();
801 Ok(cpu)
802 }
803
804 fn load_fallback(
807 config: &EmbedderConfig,
808 dir: &Path,
809 reason: String,
810 ) -> Result<Self, EmbedError> {
811 if e40::is_bundle(dir) || coreml_bundle_files(dir)?.is_some() {
812 return Err(EmbedError::InvalidConfig(
813 "the fallback model directory must hold a Q4 bundle, not an E40 or Core ML bundle"
814 .into(),
815 ));
816 }
817 let mut fallback = config.clone();
818 fallback.model_dir = dir.to_path_buf();
819 fallback.provider = ExecutionProvider::Cpu;
820 fallback.fallback_model_dir = None;
821 let mut embedder = Self::load(fallback)?;
822 embedder.fallback_reason = Some(format!(
823 "{reason}; serving {} on cpu from the fallback model directory",
824 embedder.model_id
825 ));
826 Ok(embedder)
827 }
828
829 fn cpu_twin(&self, config: &EmbedderConfig) -> Result<Self, EmbedError> {
831 let (tokenizer, pad_id) = load_tokenizer(&config.model_dir)?;
832 match &self.backend {
833 Backend::Onnx(_) => {
834 let built = build_session(config, rightkit_onnx::ExecutionProvider::Cpu, false)?;
835 Ok(Self::from_session(
836 config,
837 tokenizer,
838 pad_id,
839 self.gemma2,
840 built,
841 ))
842 }
843 Backend::E40(engine) => Ok(Self::from_e40(
844 config,
845 tokenizer,
846 pad_id,
847 engine.with_provider(rightkit_onnx::ExecutionProvider::Cpu),
848 )),
849 #[cfg(target_os = "macos")]
851 Backend::CoreMl(engine) => Ok(Self::from_coreml(
852 config,
853 tokenizer,
854 pad_id,
855 engine.with_units(CoreMlUnits::uniform(CoreMlComputeUnit::CpuOnly)),
856 )),
857 }
858 }
859
860 fn is_fixed_graph(&self) -> bool {
862 match self.backend {
863 Backend::Onnx(_) => false,
864 Backend::E40(_) => true,
865 #[cfg(target_os = "macos")]
866 Backend::CoreMl(_) => true,
867 }
868 }
869
870 fn release_graphs(&self) {
873 match &self.backend {
874 Backend::Onnx(_) => {}
875 Backend::E40(engine) => engine.release(),
876 #[cfg(target_os = "macos")]
877 Backend::CoreMl(engine) => engine.release(),
878 }
879 }
880
881 fn prewarms(&self, config: &EmbedderConfig) -> bool {
883 let (fixed_graph_accelerator, default_on) = match &self.backend {
885 Backend::E40(_) => (true, false),
886 #[cfg(target_os = "macos")]
887 Backend::CoreMl(_) => (true, true),
888 Backend::Onnx(_) => (false, false),
889 };
890 config.prewarm.unwrap_or(default_on)
891 && self.provider_used != "cpu"
892 && fixed_graph_accelerator
893 }
894
895 fn prewarm_graphs(&mut self) -> Result<(), EmbedError> {
899 let (elapsed, per_length) = match &self.backend {
900 Backend::E40(engine) => (engine.prewarm(self.pad_id)?, Vec::new()),
901 #[cfg(target_os = "macos")]
902 Backend::CoreMl(engine) => engine.prewarm(self.pad_id)?,
903 Backend::Onnx(_) => return Ok(()),
904 };
905 if let Some(timings) = self.load_timings.as_mut() {
906 timings.prewarm = Some(elapsed);
907 timings.prewarm_per_length = per_length;
908 }
909 Ok(())
910 }
911
912 fn admit_accelerator(
918 mut self,
919 config: &EmbedderConfig,
920 auto: bool,
921 ) -> Result<Self, EmbedError> {
922 let provider = self.provider_used;
923 let check_started = Instant::now();
924 let cache = self.parity_cache_target(config);
925 let mut miss_reason: Option<&'static str> = None;
926 if let Some(target) = &cache {
927 match parity_cache::read(
928 &target.dir,
929 &config.model_dir,
930 &target.key,
931 parity_cache::now_unix(),
932 ) {
933 parity_cache::Lookup::Hit(verdict) => {
934 self.record_parity_timings(
935 check_started,
936 None,
937 None,
938 ParityCacheOutcome::Hit {
939 passed: verdict.passed,
940 },
941 );
942 return self.apply_cached_verdict(&verdict, config, auto);
943 }
944 parity_cache::Lookup::Miss(reason) => miss_reason = Some(reason),
945 }
946 }
947 let outcome = |stored: bool| match miss_reason {
949 Some(reason) => ParityCacheOutcome::Miss { reason, stored },
950 None => ParityCacheOutcome::NotUsed,
951 };
952 let mut cpu = self.cpu_twin(config)?;
953 let prompts = cpu.parity_prompts();
954 let tokens: Vec<usize> = cpu
955 .encode_prompts(&prompts)?
956 .iter()
957 .map(|row| row.ids.len())
958 .collect();
959 let reference = cpu.parity_vectors(&prompts)?;
960 let cpu_reference = check_started.elapsed();
961 let accelerator_started = Instant::now();
962 let candidate = self.parity_vectors(&prompts);
963 let accelerator = accelerator_started.elapsed();
964 let (lowest_cosine, detail) = match candidate {
965 Ok(candidate) => match compare_parity(
966 &reference,
967 &candidate,
968 self.native_dim,
969 ACCELERATOR_PARITY_MIN_COSINE,
970 ) {
971 Ok(lowest) => {
972 let parity = AcceleratorParity {
973 lowest_cosine: lowest.cosine,
974 at_tokens: tokens[lowest.index % tokens.len()],
975 pairs: reference.len(),
976 };
977 self.parity = Some(parity);
978 let prewarm = self.prewarms(config);
981 if !prewarm {
982 self.release_graphs();
983 }
984 let stored = cache.as_ref().is_some_and(|target| {
985 target.store(
986 &config.model_dir,
987 &parity_cache::ParityVerdict {
988 passed: true,
989 lowest_cosine: Some(parity.lowest_cosine),
990 at_tokens: parity.at_tokens,
991 pairs: parity.pairs,
992 detail: String::new(),
993 timestamp_unix: parity_cache::now_unix(),
994 },
995 )
996 });
997 self.record_parity_timings(
998 check_started,
999 Some(cpu_reference),
1000 Some(accelerator),
1001 outcome(stored),
1002 );
1003 if prewarm {
1004 self.prewarm_graphs()?;
1005 }
1006 return Ok(self);
1007 }
1008 Err(miss) => {
1009 let (lowest_cosine, detail) =
1010 parity_miss_detail(&miss, &tokens, ACCELERATOR_PARITY_MIN_COSINE);
1011 let stored = cache.as_ref().is_some_and(|target| {
1014 target.store(
1015 &config.model_dir,
1016 &parity_cache::ParityVerdict {
1017 passed: false,
1018 lowest_cosine,
1019 at_tokens: 0,
1020 pairs: reference.len(),
1021 detail: detail.clone(),
1022 timestamp_unix: parity_cache::now_unix(),
1023 },
1024 )
1025 });
1026 self.record_parity_timings(
1027 check_started,
1028 Some(cpu_reference),
1029 Some(accelerator),
1030 outcome(stored),
1031 );
1032 (lowest_cosine, detail)
1033 }
1034 },
1035 Err(error) if auto => {
1036 self.record_parity_timings(
1037 check_started,
1038 Some(cpu_reference),
1039 Some(accelerator),
1040 outcome(false),
1041 );
1042 (None, format!("inference error: {error}"))
1043 }
1044 Err(error) => return Err(error),
1045 };
1046 if auto {
1047 cpu.fallback_reason =
1048 Some(format!("{provider} -> cpu: vector parity failed: {detail}"));
1049 cpu.load_timings = self.load_timings.take();
1050 return Ok(cpu);
1051 }
1052 Err(EmbedError::AcceleratorParity {
1053 provider,
1054 lowest_cosine,
1055 floor: ACCELERATOR_PARITY_MIN_COSINE,
1056 detail,
1057 })
1058 }
1059
1060 fn parity_cache_target(&self, config: &EmbedderConfig) -> Option<ParityCacheTarget> {
1064 let dir = config.parity_cache_dir.clone()?;
1065 if !matches!(self.backend, Backend::E40(_)) {
1066 return None;
1067 }
1068 let artifacts = e40::artifacts(&config.model_dir).ok()?;
1069 let artifacts: Vec<&str> = artifacts.iter().map(String::as_str).collect();
1070 let key = parity_cache::ParityKey::build(
1071 &config.model_dir,
1072 &artifacts,
1073 &config.runtime_path,
1074 ort::info(),
1075 self.provider_used,
1076 ACCELERATOR_PARITY_MIN_COSINE,
1077 );
1078 Some(ParityCacheTarget { dir, key })
1079 }
1080
1081 fn record_parity_timings(
1083 &mut self,
1084 started: Instant,
1085 cpu_reference: Option<Duration>,
1086 accelerator: Option<Duration>,
1087 cache: ParityCacheOutcome,
1088 ) {
1089 if let Some(timings) = self.load_timings.as_mut() {
1090 timings.parity_check_total = Some(started.elapsed());
1091 timings.parity_cpu_reference = cpu_reference;
1092 timings.parity_accelerator = accelerator;
1093 timings.parity_cache = cache;
1094 }
1095 }
1096
1097 fn apply_cached_verdict(
1100 mut self,
1101 verdict: &parity_cache::ParityVerdict,
1102 config: &EmbedderConfig,
1103 auto: bool,
1104 ) -> Result<Self, EmbedError> {
1105 let provider = self.provider_used;
1106 if verdict.passed {
1107 self.parity = verdict
1108 .lowest_cosine
1109 .map(|lowest_cosine| AcceleratorParity {
1110 lowest_cosine,
1111 at_tokens: verdict.at_tokens,
1112 pairs: verdict.pairs,
1113 });
1114 if self.prewarms(config) {
1115 self.prewarm_graphs()?;
1116 }
1117 return Ok(self);
1118 }
1119 let detail = format!("{} (cached verdict)", verdict.detail);
1120 if auto {
1121 let mut cpu = self.cpu_twin(config)?;
1122 cpu.fallback_reason =
1123 Some(format!("{provider} -> cpu: vector parity failed: {detail}"));
1124 cpu.load_timings = self.load_timings.take();
1125 return Ok(cpu);
1126 }
1127 Err(EmbedError::AcceleratorParity {
1128 provider,
1129 lowest_cosine: verdict.lowest_cosine,
1130 floor: ACCELERATOR_PARITY_MIN_COSINE,
1131 detail,
1132 })
1133 }
1134
1135 fn parity_prompts(&self) -> Vec<String> {
1143 let mut prompts = vec![
1144 self.query_prompt(PARITY_QUERY),
1145 self.document_prompt(PARITY_DOCUMENT),
1146 self.document_prompt(&parity_long_document()),
1147 ];
1148 if self.is_fixed_graph() {
1149 prompts.insert(1, self.document_prompt(&parity_sections(1)));
1150 prompts.insert(3, self.document_prompt(&parity_sections(4)));
1151 }
1152 prompts
1153 }
1154
1155 fn parity_vectors(&self, prompts: &[String]) -> Result<Vec<Vec<f32>>, EmbedError> {
1159 let mut vectors = Vec::with_capacity(prompts.len() * 2);
1160 for prompt in prompts {
1161 vectors.extend(self.embed_prompts_native(vec![prompt.clone()])?);
1162 }
1163 vectors.extend(self.embed_prompts_native(prompts.to_vec())?);
1164 Ok(vectors)
1165 }
1166
1167 fn probe(&self) -> Result<(), String> {
1170 let rows = self
1171 .encode_prompts(&[PROBE_PROMPT.to_owned()])
1172 .map_err(|e| e.to_string())?;
1173 let chunk = [&rows[0]];
1174 let native = self
1175 .run_chunk(&chunk, true)
1176 .map_err(|e| e.to_string())?
1177 .into_iter()
1178 .next()
1179 .ok_or_else(|| "probe returned no row".to_owned())?;
1180 probe_output_ok(&native, self.native_dim)
1181 }
1182
1183 pub fn model_id(&self) -> &str {
1184 &self.model_id
1185 }
1186
1187 pub fn native_dim(&self) -> usize {
1188 self.native_dim
1189 }
1190
1191 pub fn dimension(&self) -> usize {
1192 self.truncate_dim
1193 }
1194
1195 pub fn provider_used(&self) -> &str {
1197 self.provider_used
1198 }
1199
1200 pub fn fallback_reason(&self) -> Option<&str> {
1204 self.fallback_reason.as_deref()
1205 }
1206
1207 pub fn accelerator_parity(&self) -> Option<AcceleratorParity> {
1214 self.parity
1215 }
1216
1217 pub fn load_timings(&self) -> Option<&LoadTimings> {
1224 self.load_timings.as_ref()
1225 }
1226
1227 pub fn coreml_units(&self) -> Option<&CoreMlUnits> {
1231 match &self.backend {
1232 #[cfg(target_os = "macos")]
1233 Backend::CoreMl(engine) => Some(engine.units()),
1234 _ => None,
1235 }
1236 }
1237
1238 #[cfg(feature = "bench-internals")]
1241 pub fn bench_query_tokens(&self, text: &str) -> Result<BenchTokens, EmbedError> {
1242 let prompt = self.query_prompt(text);
1243 let ids = self
1244 .encode_prompts(std::slice::from_ref(&prompt))?
1245 .remove(0)
1246 .ids;
1247 Ok(BenchTokens {
1248 graph_tokens: match &self.backend {
1249 Backend::E40(engine) => engine.query_bucket(ids.len()),
1250 _ => e40::bucket_for(ids.len()),
1251 },
1252 pad_id: self.pad_id,
1253 prompt,
1254 ids,
1255 })
1256 }
1257
1258 #[cfg(feature = "bench-internals")]
1262 pub fn bench_embed_query_split(&self, text: &str) -> Result<BenchEmbedSplit, EmbedError> {
1263 let Backend::E40(engine) = &self.backend else {
1264 return Err(EmbedError::InvalidConfig(
1265 "bench_embed_query_split needs an E40 bundle".into(),
1266 ));
1267 };
1268 let started = Instant::now();
1269 let prompt = self.query_prompt(text);
1270 let row = self
1271 .encode_prompts(std::slice::from_ref(&prompt))?
1272 .remove(0);
1273 let tokenize = started.elapsed();
1274 let (values, engine_split) = engine.run_split(&row.ids, self.pad_id)?;
1275 let post_started = Instant::now();
1276 let shape = [1, self.native_dim as i64];
1277 let mut pooled = pool_batch(&shape, &values, &[row.mask.as_slice()], self.native_dim)?;
1278 let _ = truncate_and_normalize(&pooled.remove(0), self.truncate_dim);
1279 Ok(BenchEmbedSplit {
1280 tokenize,
1281 engine: engine_split,
1282 post: post_started.elapsed(),
1283 total: started.elapsed(),
1284 })
1285 }
1286
1287 pub fn pipeline_id(&self) -> String {
1307 match self.pipeline_id.get_or_init(|| {
1308 let mut files = Vec::new();
1309 let e40_names = match &self.backend {
1310 Backend::E40(_) => e40::artifacts(&self.model_dir).map_err(|e| e.to_string())?,
1311 _ => Vec::new(),
1312 };
1313 let e40_refs: Vec<&str> = e40_names.iter().map(String::as_str).collect();
1314 let names: &[&str] = match self.backend {
1315 Backend::E40(_) => &e40_refs,
1316 #[cfg(target_os = "macos")]
1317 Backend::CoreMl(_) => &coreml::ARTIFACTS,
1318 Backend::Onnx(_) => &[
1319 "model_q4.onnx",
1320 "model_q4.onnx_data",
1321 "tokenizer.json",
1322 "config.json",
1323 ],
1324 };
1325 for name in names {
1326 files.push((*name, hash_file(&self.model_dir.join(name))?));
1327 }
1328 Ok(pipeline_fingerprint(&PipelineParts {
1329 model_id: &self.model_id,
1330 native_dim: self.native_dim,
1331 truncate_dim: self.truncate_dim,
1332 max_tokens: self.max_tokens,
1333 scrubs_sentinels: self.gemma2,
1334 files: &files,
1335 }))
1336 }) {
1337 Ok(id) => id.clone(),
1338 Err(error) => format!("rkpipe1-unavailable-{}", sanitize(error)),
1340 }
1341 }
1342
1343 pub fn embed(&self, text: &str) -> Result<Vec<f32>, EmbedError> {
1344 self.embed_prompts(vec![self.document_prompt(text)])?
1345 .pop()
1346 .ok_or_else(|| EmbedError::Output("no embedding produced".into()))
1347 }
1348
1349 pub fn embed_query(&self, text: &str) -> Result<Vec<f32>, EmbedError> {
1350 self.embed_prompts(vec![self.query_prompt(text)])?
1351 .pop()
1352 .ok_or_else(|| EmbedError::Output("no embedding produced".into()))
1353 }
1354
1355 pub fn embed_batch<'a, I>(&self, texts: I) -> Result<Vec<Vec<f32>>, EmbedError>
1359 where
1360 I: IntoIterator<Item = &'a str>,
1361 {
1362 let prompts: Vec<String> = texts.into_iter().map(|t| self.document_prompt(t)).collect();
1363 Ok(self
1364 .embed_prompts_native_for(prompts, true)?
1365 .into_iter()
1366 .map(|native| truncate_and_normalize(&native, self.truncate_dim))
1367 .collect())
1368 }
1369
1370 pub fn embed_query_batch<'a, I>(&self, texts: I) -> Result<Vec<Vec<f32>>, EmbedError>
1372 where
1373 I: IntoIterator<Item = &'a str>,
1374 {
1375 self.embed_prompts(texts.into_iter().map(|t| self.query_prompt(t)).collect())
1376 }
1377
1378 fn document_prompt(&self, text: &str) -> String {
1379 format!("{DOCUMENT_PROMPT_PREFIX}{}", self.model_text(text))
1380 }
1381
1382 fn query_prompt(&self, text: &str) -> String {
1383 format!("{QUERY_PROMPT_PREFIX}{}", self.model_text(text))
1384 }
1385
1386 fn model_text<'a>(&self, text: &'a str) -> std::borrow::Cow<'a, str> {
1389 if !self.gemma2 || !GEMMA2_MODALITY_SENTINELS.iter().any(|s| text.contains(s)) {
1390 return std::borrow::Cow::Borrowed(text);
1391 }
1392 let mut owned = text.to_owned();
1393 for sentinel in GEMMA2_MODALITY_SENTINELS {
1394 owned = owned.replace(sentinel, &sentinel.replace('|', "\u{2502}"));
1395 }
1396 std::borrow::Cow::Owned(owned)
1397 }
1398
1399 fn encode_prompts(&self, prompts: &[String]) -> Result<Vec<TokenRow>, EmbedError> {
1400 let inputs: Vec<EncodeInput> = prompts
1401 .iter()
1402 .map(|prompt| EncodeInput::Single(prompt.as_str().into()))
1403 .collect();
1404 let encodings = self
1405 .tokenizer
1406 .encode_batch(inputs, true)
1407 .map_err(|e| EmbedError::Tokenizer(e.to_string()))?;
1408 encodings
1409 .iter()
1410 .map(|encoding| {
1411 let ids: Vec<i64> = encoding
1412 .get_ids()
1413 .iter()
1414 .take(self.max_tokens)
1415 .map(|id| i64::from(*id))
1416 .collect();
1417 if ids.is_empty() {
1418 return Err(EmbedError::Tokenizer("encoding returned no tokens".into()));
1419 }
1420 let mask: Vec<i64> = encoding
1421 .get_attention_mask()
1422 .iter()
1423 .take(ids.len())
1424 .map(|value| i64::from(*value))
1425 .collect();
1426 Ok(TokenRow { ids, mask })
1427 })
1428 .collect()
1429 }
1430
1431 fn embed_prompts(&self, prompts: Vec<String>) -> Result<Vec<Vec<f32>>, EmbedError> {
1432 Ok(self
1433 .embed_prompts_native(prompts)?
1434 .into_iter()
1435 .map(|native| truncate_and_normalize(&native, self.truncate_dim))
1436 .collect())
1437 }
1438
1439 fn embed_prompts_native(&self, prompts: Vec<String>) -> Result<Vec<Vec<f32>>, EmbedError> {
1442 self.embed_prompts_native_for(prompts, false)
1443 }
1444
1445 fn embed_prompts_native_for(
1450 &self,
1451 prompts: Vec<String>,
1452 ingest: bool,
1453 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1454 if prompts.is_empty() {
1455 return Ok(Vec::new());
1456 }
1457 let rows = self.encode_prompts(&prompts)?;
1458 if let (true, Backend::E40(engine)) = (ingest, &self.backend) {
1459 let ids: Vec<&[i64]> = rows.iter().map(|row| row.ids.as_slice()).collect();
1460 if let Some(vectors) = engine.run_ingest(&ids, self.pad_id)? {
1461 let shape = [1, self.native_dim as i64];
1462 return rows
1463 .iter()
1464 .zip(vectors)
1465 .map(|(row, values)| {
1466 pool_batch(&shape, &values, &[row.mask.as_slice()], self.native_dim)
1467 .map(|mut pooled| pooled.remove(0))
1468 })
1469 .collect();
1470 }
1471 }
1472 let lengths: Vec<usize> = rows.iter().map(|row| row.ids.len()).collect();
1473 let order = length_order(&lengths);
1474 let sorted_lengths: Vec<usize> = order.iter().map(|&i| lengths[i]).collect();
1475 let accelerated = self.provider_used != "cpu";
1476 let mut out: Vec<Option<Vec<f32>>> = vec![None; rows.len()];
1477 for range in plan_chunks(
1478 &sorted_lengths,
1479 self.max_batch.max(1),
1480 MAX_PADDED_TOKENS_PER_CALL,
1481 ) {
1482 let chunk: Vec<&TokenRow> = order[range.clone()].iter().map(|&i| &rows[i]).collect();
1483 let natives = self.run_chunk(&chunk, accelerated)?;
1484 for (&index, native) in order[range].iter().zip(natives) {
1485 out[index] = Some(native);
1486 }
1487 }
1488 out.into_iter()
1489 .map(|row| row.ok_or_else(|| EmbedError::Output("missing embedding row".into())))
1490 .collect()
1491 }
1492
1493 fn run_chunk(
1496 &self,
1497 chunk: &[&TokenRow],
1498 accelerated: bool,
1499 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1500 let session = match &self.backend {
1501 Backend::Onnx(session) => session,
1502 Backend::E40(engine) => {
1506 return self.run_fixed_rows(chunk, |ids| engine.run(ids, self.pad_id));
1507 }
1508 #[cfg(target_os = "macos")]
1511 Backend::CoreMl(engine) => {
1512 return self.run_fixed_rows(chunk, |ids| engine.run(ids, self.pad_id));
1513 }
1514 };
1515 let longest = chunk.iter().map(|row| row.ids.len()).max().unwrap_or(1);
1516 let (batch_rows, padded_len) = padded_shape(chunk.len(), longest, accelerated);
1517 let padded = pad_batch(chunk, self.pad_id, padded_len, batch_rows);
1518 let (rows, len) = (padded.rows, padded.len);
1519 let tensor = |data: Vec<i64>| {
1520 ort::value::Tensor::from_array(([rows, len], data))
1521 .map_err(|e| EmbedError::Runtime(e.to_string()))
1522 };
1523 let mut inputs = ort::inputs![
1524 "input_ids" => tensor(padded.ids)?,
1525 "attention_mask" => tensor(padded.mask)?
1526 ];
1527 if self.takes_token_type_ids {
1528 inputs.push(("token_type_ids".into(), tensor(vec![0; rows * len])?.into()));
1529 }
1530 let mut session = session
1531 .lock()
1532 .map_err(|_| EmbedError::Runtime("model mutex poisoned".into()))?;
1533 let outputs = session
1534 .run(inputs)
1535 .map_err(|e| EmbedError::Runtime(e.to_string()))?;
1536 let output = outputs
1537 .get("sentence_embedding")
1538 .ok_or_else(|| EmbedError::Output("missing sentence_embedding output".into()))?;
1539 let (shape, values) = output
1540 .try_extract_tensor::<f32>()
1541 .map_err(|e| EmbedError::Output(e.to_string()))?;
1542 let masks: Vec<&[i64]> = chunk.iter().map(|row| row.mask.as_slice()).collect();
1543 pool_batch(shape.as_ref(), values, &masks, self.native_dim)
1544 }
1545}
1546
1547impl Embedder {
1548 fn run_fixed_rows(
1551 &self,
1552 chunk: &[&TokenRow],
1553 run: impl Fn(&[i64]) -> Result<Vec<f32>, EmbedError>,
1554 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1555 let mut natives = Vec::with_capacity(chunk.len());
1556 for row in chunk {
1557 let values = run(&row.ids)?;
1558 let shape = [1, self.native_dim as i64];
1559 let mut pooled = pool_batch(&shape, &values, &[row.mask.as_slice()], self.native_dim)?;
1560 natives.push(pooled.remove(0));
1561 }
1562 Ok(natives)
1563 }
1564}
1565
1566struct ParityCacheTarget {
1568 dir: PathBuf,
1569 key: parity_cache::ParityKey,
1570}
1571
1572impl ParityCacheTarget {
1573 fn store(&self, model_dir: &Path, verdict: &parity_cache::ParityVerdict) -> bool {
1576 parity_cache::write(&self.dir, model_dir, &self.key, verdict).is_ok()
1577 }
1578}
1579
1580#[cfg(feature = "bench-internals")]
1582#[derive(Debug, Clone, PartialEq, Eq)]
1583pub struct BenchTokens {
1584 pub prompt: String,
1586 pub ids: Vec<i64>,
1588 pub graph_tokens: Option<usize>,
1590 pub pad_id: i64,
1592}
1593
1594#[cfg(feature = "bench-internals")]
1596#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1597pub struct BenchEmbedSplit {
1598 pub tokenize: Duration,
1600 pub engine: e40::RunSplit,
1602 pub post: Duration,
1604 pub total: Duration,
1606}
1607
1608struct TokenRow {
1610 ids: Vec<i64>,
1611 mask: Vec<i64>,
1612}
1613
1614#[derive(Debug, PartialEq, Eq)]
1616struct PaddedBatch {
1617 rows: usize,
1618 len: usize,
1619 ids: Vec<i64>,
1620 mask: Vec<i64>,
1621}
1622
1623fn length_order(lengths: &[usize]) -> Vec<usize> {
1626 let mut order: Vec<usize> = (0..lengths.len()).collect();
1627 order.sort_by_key(|&i| lengths[i]);
1628 order
1629}
1630
1631fn plan_chunks(
1635 sorted_lengths: &[usize],
1636 max_rows: usize,
1637 token_budget: usize,
1638) -> Vec<std::ops::Range<usize>> {
1639 let mut ranges = Vec::new();
1640 let mut start = 0;
1641 while start < sorted_lengths.len() {
1642 let mut end = start + 1;
1643 while end < sorted_lengths.len()
1644 && end - start < max_rows
1645 && (end - start + 1) * sorted_lengths[end] <= token_budget
1646 {
1647 end += 1;
1648 }
1649 ranges.push(start..end);
1650 start = end;
1651 }
1652 ranges
1653}
1654
1655fn padded_shape(rows: usize, longest: usize, accelerated: bool) -> (usize, usize) {
1659 if !accelerated {
1660 return (rows, longest);
1661 }
1662 let len = ACCELERATOR_BUCKETS
1663 .iter()
1664 .copied()
1665 .find(|bucket| *bucket >= longest)
1666 .unwrap_or(longest);
1667 (rows.next_power_of_two(), len)
1668}
1669
1670fn pad_batch(chunk: &[&TokenRow], pad_id: i64, len: usize, rows: usize) -> PaddedBatch {
1673 let mut ids = vec![pad_id; rows * len];
1674 let mut mask = vec![0i64; rows * len];
1675 for row in 0..rows {
1676 let source = chunk[row.min(chunk.len() - 1)];
1677 let used = source.ids.len().min(len);
1678 ids[row * len..row * len + used].copy_from_slice(&source.ids[..used]);
1679 mask[row * len..row * len + used].copy_from_slice(&source.mask[..used]);
1680 }
1681 PaddedBatch {
1682 rows,
1683 len,
1684 ids,
1685 mask,
1686 }
1687}
1688
1689fn pool_batch(
1694 shape: &[i64],
1695 values: &[f32],
1696 masks: &[&[i64]],
1697 native_dim: usize,
1698) -> Result<Vec<Vec<f32>>, EmbedError> {
1699 let real = masks.len();
1700 if shape.len() == 2 && shape[1] == native_dim as i64 {
1701 let batch = shape[0] as usize;
1702 if batch >= real && values.len() == batch * native_dim {
1703 return Ok((0..real)
1704 .map(|row| normalize(values[row * native_dim..(row + 1) * native_dim].to_vec()))
1705 .collect());
1706 }
1707 }
1708 if shape.len() == 3 && shape[2] == native_dim as i64 && shape[0] > 0 && shape[1] > 0 {
1709 let (batch, tokens) = (shape[0] as usize, shape[1] as usize);
1710 if batch >= real && values.len() == batch * tokens * native_dim {
1711 let mut pooled_rows = Vec::with_capacity(real);
1712 for (row, mask) in masks.iter().enumerate() {
1713 let mut pooled = vec![0.0f32; native_dim];
1714 let mut count = 0.0f32;
1715 for token in 0..tokens {
1716 if mask.get(token).copied().unwrap_or(0) == 0 {
1717 continue;
1718 }
1719 count += 1.0;
1720 let base = (row * tokens + token) * native_dim;
1721 for (dim, value) in pooled.iter_mut().enumerate() {
1722 *value += values[base + dim];
1723 }
1724 }
1725 if count == 0.0 {
1726 return Err(EmbedError::Output(
1727 "attention mask contains no tokens".into(),
1728 ));
1729 }
1730 for value in &mut pooled {
1731 *value /= count;
1732 }
1733 pooled_rows.push(normalize(pooled));
1734 }
1735 return Ok(pooled_rows);
1736 }
1737 }
1738 Err(EmbedError::Output(format!(
1739 "expected [B,{native_dim}] or [B,tokens,{native_dim}] with B >= {real}, got shape {shape:?}"
1740 )))
1741}
1742
1743fn probe_output_ok(native: &[f32], native_dim: usize) -> Result<(), String> {
1744 if native.len() != native_dim {
1745 return Err(format!(
1746 "expected {native_dim} values, got {}",
1747 native.len()
1748 ));
1749 }
1750 if native.iter().any(|value| !value.is_finite()) {
1751 return Err("non-finite values".into());
1752 }
1753 if native.iter().map(|value| value * value).sum::<f32>() <= f32::EPSILON {
1754 return Err("all-zero output".into());
1755 }
1756 Ok(())
1757}
1758
1759fn parity_long_document() -> String {
1762 parity_sections(PARITY_LONG_SECTIONS)
1763}
1764
1765fn parity_sections(count: usize) -> String {
1767 (1..=count)
1768 .map(|n| {
1769 format!(
1770 "Section {n}: the nightly billing job reconciles every invoice against the \
1771 ledger, flags mismatched totals for account {}, and sends the exceptions to \
1772 the finance team for review before payment run {n}.",
1773 1000 + n * 7
1774 )
1775 })
1776 .collect::<Vec<_>>()
1777 .join(" ")
1778}
1779
1780#[derive(Debug, Clone, Copy, PartialEq)]
1782struct ParityLowest {
1783 index: usize,
1784 cosine: f32,
1785}
1786
1787#[derive(Debug, Clone, PartialEq)]
1789enum ParityMiss {
1790 Invalid { index: usize, reason: String },
1793 Below(ParityLowest),
1795}
1796
1797fn cosine(a: &[f32], b: &[f32]) -> f32 {
1798 let (mut dot, mut na, mut nb) = (0.0f64, 0.0f64, 0.0f64);
1799 for (x, y) in a.iter().zip(b) {
1800 let (x, y) = (f64::from(*x), f64::from(*y));
1801 dot += x * y;
1802 na += x * x;
1803 nb += y * y;
1804 }
1805 (dot / (na.sqrt() * nb.sqrt())) as f32
1806}
1807
1808fn compare_parity(
1813 reference: &[Vec<f32>],
1814 candidate: &[Vec<f32>],
1815 native_dim: usize,
1816 floor: f32,
1817) -> Result<ParityLowest, ParityMiss> {
1818 if reference.is_empty() || reference.len() != candidate.len() {
1819 return Err(ParityMiss::Invalid {
1820 index: 0,
1821 reason: format!(
1822 "expected {} vectors to compare, got {}",
1823 reference.len(),
1824 candidate.len()
1825 ),
1826 });
1827 }
1828 let mut lowest = ParityLowest {
1829 index: 0,
1830 cosine: f32::INFINITY,
1831 };
1832 for (index, (r, c)) in reference.iter().zip(candidate).enumerate() {
1833 probe_output_ok(r, native_dim).map_err(|reason| ParityMiss::Invalid {
1834 index,
1835 reason: format!("cpu reference: {reason}"),
1836 })?;
1837 probe_output_ok(c, native_dim).map_err(|reason| ParityMiss::Invalid {
1838 index,
1839 reason: format!("accelerator output: {reason}"),
1840 })?;
1841 let value = cosine(r, c);
1842 if value < lowest.cosine {
1843 lowest = ParityLowest {
1844 index,
1845 cosine: value,
1846 };
1847 }
1848 }
1849 if lowest.cosine >= floor {
1850 Ok(lowest)
1851 } else {
1852 Err(ParityMiss::Below(lowest))
1853 }
1854}
1855
1856fn parity_miss_detail(miss: &ParityMiss, tokens: &[usize], floor: f32) -> (Option<f32>, String) {
1860 let at = |index: usize| {
1861 let batched = if index >= tokens.len() {
1862 " (batched)"
1863 } else {
1864 ""
1865 };
1866 format!("~{} tokens{batched}", tokens[index % tokens.len()])
1867 };
1868 match miss {
1869 ParityMiss::Below(lowest) => (
1870 Some(lowest.cosine),
1871 format!(
1872 "cosine {:.3} at {}, floor {floor}",
1873 lowest.cosine,
1874 at(lowest.index)
1875 ),
1876 ),
1877 ParityMiss::Invalid { index, reason } => (None, format!("{reason} at {}", at(*index))),
1878 }
1879}
1880
1881struct PipelineParts<'a> {
1882 model_id: &'a str,
1883 native_dim: usize,
1884 truncate_dim: usize,
1885 max_tokens: usize,
1886 scrubs_sentinels: bool,
1887 files: &'a [(&'a str, String)],
1888}
1889
1890fn pipeline_fingerprint(parts: &PipelineParts<'_>) -> String {
1892 let mut text = String::new();
1893 let mut line = |key: &str, value: &str| {
1894 text.push_str(key);
1895 text.push('=');
1896 text.push_str(&value.len().to_string());
1897 text.push(':');
1898 text.push_str(value);
1899 text.push('\n');
1900 };
1901 line("schema", PIPELINE_SCHEMA);
1902 line("model_id", parts.model_id);
1903 line("native_dim", &parts.native_dim.to_string());
1904 line("truncate_dim", &parts.truncate_dim.to_string());
1905 line("max_tokens", &parts.max_tokens.to_string());
1906 line("document_prompt", DOCUMENT_PROMPT_PREFIX);
1907 line("query_prompt", QUERY_PROMPT_PREFIX);
1908 line("pooling", POOLING_DESCRIPTION);
1909 line("normalization", NORMALIZATION_DESCRIPTION);
1910 line("scrubs_sentinels", &parts.scrubs_sentinels.to_string());
1911 for (name, hash) in parts.files {
1912 line(&format!("file:{name}"), hash);
1913 }
1914 let digest = rightkit_fs::digest::sha256_bytes(text.as_bytes());
1916 format!("rkpipe1-{}", &digest[..32])
1917}
1918
1919fn sanitize(text: &str) -> String {
1920 text.chars()
1921 .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
1922 .take(48)
1923 .collect()
1924}
1925
1926pub(crate) fn hash_file(path: &Path) -> Result<String, String> {
1929 let mut file = match std::fs::File::open(path) {
1930 Ok(file) => file,
1931 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok("absent".into()),
1932 Err(error) => return Err(format!("{}: {error}", path.display())),
1933 };
1934 rightkit_fs::digest::sha256_reader(&mut file).map_err(|e| format!("{}: {e}", path.display()))
1935}
1936
1937fn bundle_is_gemma2(model_dir: &std::path::Path) -> Result<bool, EmbedError> {
1941 let path = model_dir.join("config.json");
1942 let Ok(bytes) = std::fs::read(&path) else {
1943 return Ok(false);
1944 };
1945 let config: serde_json::Value = serde_json::from_slice(&bytes)
1946 .map_err(|e| EmbedError::InvalidConfig(format!("{}: {e}", path.display())))?;
1947 Ok(config["model_type"] == "embedding_gemma2")
1948}
1949
1950#[cfg(target_os = "macos")]
1953fn coreml_bundle_files(dir: &Path) -> Result<Option<&'static [&'static str]>, EmbedError> {
1954 let files: &'static [&'static str] = &coreml::ARTIFACTS;
1955 Ok(coreml::is_bundle(dir).then_some(files))
1956}
1957
1958#[cfg(not(target_os = "macos"))]
1959fn coreml_bundle_files(dir: &Path) -> Result<Option<&'static [&'static str]>, EmbedError> {
1960 if dir.join("EmbeddingGemma2Text.mlpackage").is_dir() {
1961 return Err(EmbedError::InvalidConfig(
1962 "a Core ML bundle (EmbeddingGemma2Text.mlpackage) runs on macOS only".into(),
1963 ));
1964 }
1965 Ok(None)
1966}
1967
1968fn validate_config(config: &EmbedderConfig) -> Result<(), EmbedError> {
1969 e40::check_ingest_batch(config.ingest_batch)?;
1970 if !config.model_dir.is_dir() {
1971 return Err(EmbedError::MissingModelDirectory(config.model_dir.clone()));
1972 }
1973 if config.truncate_dim == 0 || config.truncate_dim > EMBEDDINGGEMMA_NATIVE_DIM {
1974 return Err(EmbedError::InvalidConfig(format!(
1975 "truncate dimension must be 1..={EMBEDDINGGEMMA_NATIVE_DIM}"
1976 )));
1977 }
1978 if config.max_tokens == 0 || config.max_tokens > MAX_SEQUENCE_TOKENS {
1979 return Err(EmbedError::InvalidConfig(format!(
1980 "max tokens must be 1..={MAX_SEQUENCE_TOKENS}"
1981 )));
1982 }
1983 if config.max_batch == 0 {
1984 return Err(EmbedError::InvalidConfig("max batch must be >= 1".into()));
1985 }
1986 if config.intra_threads == 0 {
1987 return Err(EmbedError::InvalidConfig(
1988 "intra threads must be >= 1".into(),
1989 ));
1990 }
1991 let coreml_files = coreml_bundle_files(&config.model_dir)?;
1992 let coreml_bundle = coreml_files.is_some();
1993 let e40_files;
1994 let e40_refs: Vec<&str>;
1995 let required: &[&str] = if let Some(files) = coreml_files {
1996 files
1997 } else if e40::is_bundle(&config.model_dir) {
1998 e40_files = e40::artifacts(&config.model_dir)?;
1999 e40_refs = e40_files.iter().map(String::as_str).collect();
2000 &e40_refs
2001 } else {
2002 &["model_q4.onnx", "model_q4.onnx_data", "tokenizer.json"]
2003 };
2004 for name in required {
2005 let path = config.model_dir.join(name);
2006 if !path.is_file() {
2007 return Err(EmbedError::MissingArtifact(path));
2008 }
2009 }
2010 if !coreml_bundle && !config.runtime_path.is_file() {
2012 return Err(EmbedError::MissingArtifact(config.runtime_path.clone()));
2013 }
2014 Ok(())
2015}
2016
2017fn normalize(mut values: Vec<f32>) -> Vec<f32> {
2018 let norm = values.iter().map(|value| value * value).sum::<f32>().sqrt();
2019 if norm > 0.0 {
2020 for value in &mut values {
2021 *value /= norm;
2022 }
2023 }
2024 values
2025}
2026
2027pub fn bundle_files(dir: &Path) -> Result<Option<Vec<String>>, EmbedError> {
2033 let files: Vec<String> = if let Some(files) = coreml_bundle_files(dir)? {
2034 files.iter().map(|file| (*file).to_owned()).collect()
2035 } else if e40::is_bundle(dir) {
2036 e40::artifacts(dir)?
2037 } else if dir.join("model_q4.onnx").is_file() {
2038 ["model_q4.onnx", "model_q4.onnx_data", "tokenizer.json"]
2039 .iter()
2040 .map(|file| (*file).to_owned())
2041 .collect()
2042 } else {
2043 return Ok(None);
2044 };
2045 for file in &files {
2046 let path = dir.join(file);
2047 if !path.is_file() {
2048 return Err(EmbedError::MissingArtifact(path));
2049 }
2050 }
2051 Ok(Some(files))
2052}
2053
2054pub fn installed_model_dir(executable: &Path, subdir: &str) -> Option<PathBuf> {
2065 let mut parts = Path::new(subdir).components();
2066 let plain = matches!(
2067 (parts.next(), parts.next()),
2068 (Some(std::path::Component::Normal(_)), None)
2069 );
2070 if !plain {
2071 return None;
2072 }
2073 rightkit_onnx::installed_resource_dir(executable).map(|dir| dir.join(subdir))
2074}
2075
2076pub fn truncate_and_normalize(native: &[f32], dimension: usize) -> Vec<f32> {
2077 let mut values = native[..native.len().min(dimension)].to_vec();
2078 values.resize(dimension, 0.0);
2079 normalize(values)
2080}
2081
2082#[cfg(test)]
2083mod layout_tests {
2084 use super::*;
2085
2086 #[test]
2087 fn installed_model_dir_refuses_relative_and_non_plain_names() {
2088 assert_eq!(
2089 installed_model_dir(Path::new("app/bin/app"), "models"),
2090 None
2091 );
2092 let exe = if cfg!(windows) {
2093 Path::new(r"C:\app\app.exe")
2094 } else {
2095 Path::new("/opt/app/app")
2096 };
2097 assert_eq!(installed_model_dir(exe, ""), None);
2098 assert_eq!(installed_model_dir(exe, "../models"), None);
2099 assert_eq!(installed_model_dir(exe, "/models"), None);
2100 assert_eq!(installed_model_dir(exe, "a/b"), None);
2101 assert_eq!(installed_model_dir(exe, "./models"), None);
2102 let dir = installed_model_dir(exe, "embed").expect("absolute executable");
2103 assert!(dir.ends_with("embed"));
2104 }
2105
2106 #[test]
2107 fn bundle_files_names_the_q4_layout_and_none_for_an_empty_directory() {
2108 let dir = tempfile::tempdir().unwrap();
2109 assert_eq!(bundle_files(dir.path()).unwrap(), None);
2110 for name in ["model_q4.onnx", "model_q4.onnx_data", "tokenizer.json"] {
2111 std::fs::write(dir.path().join(name), b"x").unwrap();
2112 }
2113 assert_eq!(
2114 bundle_files(dir.path()).unwrap(),
2115 Some(vec![
2116 "model_q4.onnx".to_owned(),
2117 "model_q4.onnx_data".to_owned(),
2118 "tokenizer.json".to_owned(),
2119 ])
2120 );
2121 }
2122
2123 #[test]
2124 fn bundle_files_refuses_a_partial_e40_bundle() {
2125 let dir = tempfile::tempdir().unwrap();
2126 std::fs::write(dir.path().join(e40::TABLE_FILE), b"x").unwrap();
2127 std::fs::write(dir.path().join("model_32.onnx"), b"x").unwrap();
2128 assert!(matches!(
2129 bundle_files(dir.path()),
2130 Err(EmbedError::MissingArtifact(_))
2131 ));
2132 }
2133}
2134
2135#[cfg(test)]
2136mod tests {
2137 use super::*;
2138
2139 fn row(ids: &[i64]) -> TokenRow {
2140 TokenRow {
2141 ids: ids.to_vec(),
2142 mask: vec![1; ids.len()],
2143 }
2144 }
2145
2146 fn state(ids: &[i64], dim: usize) -> Vec<f32> {
2149 ids.iter()
2150 .enumerate()
2151 .flat_map(|(pos, id)| {
2152 (0..dim)
2153 .map(move |d| ((*id as f32) * 0.37 + pos as f32 * 0.11 + d as f32 * 0.05).sin())
2154 })
2155 .collect()
2156 }
2157
2158 #[test]
2159 fn pad_batch_pads_right_with_zero_mask_and_repeats_last_row() {
2160 let a = row(&[5, 6, 7]);
2161 let b = row(&[9]);
2162 let padded = pad_batch(&[&a, &b], 0, 4, 3);
2163 assert_eq!((padded.rows, padded.len), (3, 4));
2164 assert_eq!(padded.ids, vec![5, 6, 7, 0, 9, 0, 0, 0, 9, 0, 0, 0]);
2165 assert_eq!(padded.mask, vec![1, 1, 1, 0, 1, 0, 0, 0, 1, 0, 0, 0]);
2166 }
2167
2168 #[test]
2169 fn pad_batch_uses_the_pad_id_for_masked_positions() {
2170 let a = row(&[5]);
2171 let padded = pad_batch(&[&a], 3, 3, 1);
2172 assert_eq!(padded.ids, vec![5, 3, 3]);
2173 assert_eq!(padded.mask, vec![1, 0, 0]);
2174 }
2175
2176 #[test]
2177 fn padded_mean_pool_matches_unpadded_rows() {
2178 let dim = 8;
2179 let rows_ids: [&[i64]; 3] = [&[3, 4, 5, 6, 7], &[10, 11], &[20, 21, 22]];
2180 let rows: Vec<TokenRow> = rows_ids.iter().map(|ids| row(ids)).collect();
2181 let refs: Vec<&TokenRow> = rows.iter().collect();
2182 let padded = pad_batch(&refs, 0, 5, 3);
2183 let mut values = vec![9999.0f32; padded.rows * padded.len * dim];
2186 for (r, ids) in rows_ids.iter().enumerate() {
2187 let s = state(ids, dim);
2188 let base = r * padded.len * dim;
2189 values[base..base + s.len()].copy_from_slice(&s);
2190 }
2191 let masks: Vec<&[i64]> = rows.iter().map(|r| r.mask.as_slice()).collect();
2192 let batched =
2193 pool_batch(&[3, padded.len as i64, dim as i64], &values, &masks, dim).unwrap();
2194 for (r, ids) in rows_ids.iter().enumerate() {
2195 let single = pool_batch(
2196 &[1, ids.len() as i64, dim as i64],
2197 &state(ids, dim),
2198 &[rows[r].mask.as_slice()],
2199 dim,
2200 )
2201 .unwrap();
2202 let diff = batched[r]
2203 .iter()
2204 .zip(&single[0])
2205 .map(|(a, b)| (a - b).abs())
2206 .fold(0.0f32, f32::max);
2207 assert!(diff <= 1e-6, "row {r} differs by {diff}");
2208 let norm: f32 = batched[r].iter().map(|v| v * v).sum::<f32>().sqrt();
2209 assert!((norm - 1.0).abs() <= 1e-5);
2210 }
2211 }
2212
2213 #[test]
2214 fn pooled_output_rows_are_normalized_and_extra_rows_dropped() {
2215 let dim = 4;
2216 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];
2217 let m = [1i64];
2218 let out = pool_batch(&[3, 4], &values, &[&m, &m], dim).unwrap();
2219 assert_eq!(out.len(), 2);
2220 assert_eq!(out[0], vec![0.6, 0.8, 0.0, 0.0]);
2221 assert_eq!(out[1], vec![0.0, 0.0, 1.0, 0.0]);
2222 }
2223
2224 #[test]
2225 fn pool_rejects_bad_shapes_and_empty_masks() {
2226 let m = [0i64, 0];
2227 assert!(pool_batch(&[1, 2, 2], &[1.0; 4], &[&m], 2).is_err());
2228 assert!(pool_batch(&[1, 3], &[1.0; 3], &[&[1]], 2).is_err());
2229 assert!(pool_batch(&[1, 2], &[1.0; 2], &[&[1], &[1]], 2).is_err());
2230 }
2231
2232 #[test]
2233 fn chunks_respect_row_cap_token_budget_and_cover_everything() {
2234 let lengths = [1, 2, 3, 4, 5, 6, 7];
2235 assert_eq!(plan_chunks(&lengths, 3, 1000), vec![0..3, 3..6, 6..7]);
2236 assert_eq!(plan_chunks(&[1, 3, 4, 4], 8, 10), vec![0..2, 2..4]);
2238 assert_eq!(plan_chunks(&[50, 60], 8, 10), vec![0..1, 1..2]);
2240 assert!(plan_chunks(&[], 4, 10).is_empty());
2241 }
2242
2243 #[test]
2244 fn length_order_is_stable_ascending() {
2245 assert_eq!(length_order(&[5, 2, 5, 1, 2]), vec![3, 1, 4, 0, 2]);
2246 }
2247
2248 #[test]
2249 fn accelerators_use_bucketed_shapes_cpu_uses_exact() {
2250 assert_eq!(padded_shape(3, 70, false), (3, 70));
2251 assert_eq!(padded_shape(3, 70, true), (4, 128));
2252 assert_eq!(padded_shape(1, 64, true), (1, 64));
2253 assert_eq!(padded_shape(16, 2048, true), (16, 2048));
2254 }
2255
2256 #[test]
2257 fn probe_rejects_zero_nonfinite_and_wrong_size() {
2258 assert!(probe_output_ok(&[0.0; 4], 4).is_err());
2259 assert!(probe_output_ok(&[f32::NAN, 1.0, 0.0, 0.0], 4).is_err());
2260 assert!(probe_output_ok(&[1.0; 3], 4).is_err());
2261 assert!(probe_output_ok(&[0.5, 0.5, 0.5, 0.5], 4).is_ok());
2262 }
2263
2264 #[test]
2265 fn parity_floor_is_pinned() {
2266 assert_eq!(ACCELERATOR_PARITY_MIN_COSINE, 0.995);
2267 }
2268
2269 #[test]
2270 fn identical_vectors_pass_parity_and_report_the_lowest_pair() {
2271 let reference = vec![vec![1.0, 0.0, 0.0], vec![0.0, 1.0, 0.0]];
2272 let lowest = compare_parity(&reference, &reference.clone(), 3, 0.995).unwrap();
2273 assert!((lowest.cosine - 1.0).abs() <= 1e-6);
2274 let candidate = vec![vec![1.0, 0.0, 0.0], vec![0.0, 0.6, 0.8]];
2276 let at_floor = cosine(&reference[1], &candidate[1]);
2277 let lowest = compare_parity(&reference, &candidate, 3, at_floor).unwrap();
2278 assert_eq!((lowest.index, lowest.cosine), (1, at_floor));
2279 }
2280
2281 #[test]
2282 fn a_low_cosine_pair_fails_and_reports_value_and_index() {
2283 let wrong = vec![(1.0f32 - 0.58 * 0.58).sqrt(), 0.58];
2285 let reference = vec![vec![1.0, 0.0], vec![0.0, 1.0], vec![1.0, 0.0]];
2286 let candidate = vec![vec![1.0, 0.0], wrong.clone(), vec![1.0, 0.0]];
2287 match compare_parity(&reference, &candidate, 2, 0.995) {
2288 Err(ParityMiss::Below(lowest)) => {
2289 assert_eq!(lowest.index, 1);
2290 assert!((lowest.cosine - 0.58).abs() <= 1e-6, "{lowest:?}");
2291 }
2292 other => panic!("expected a Below miss, got {other:?}"),
2293 }
2294 let single = compare_parity(&[vec![0.0, 1.0]], &[wrong], 2, 0.995);
2295 match single {
2296 Err(ParityMiss::Below(lowest)) => assert!((lowest.cosine - 0.58).abs() <= 1e-6),
2297 other => panic!("expected a Below miss, got {other:?}"),
2298 }
2299 }
2300
2301 #[test]
2302 fn nonfinite_zero_or_wrongly_sized_vectors_fail_parity() {
2303 let good = vec![vec![1.0, 0.0]];
2304 for bad in [
2305 vec![f32::NAN, 1.0],
2306 vec![f32::INFINITY, 0.0],
2307 vec![0.0, 0.0],
2308 vec![1.0, 0.0, 0.0],
2309 ] {
2310 assert!(
2311 matches!(
2312 compare_parity(&good, std::slice::from_ref(&bad), 2, 0.995),
2313 Err(ParityMiss::Invalid { index: 0, .. })
2314 ),
2315 "candidate {bad:?}"
2316 );
2317 assert!(
2318 matches!(
2319 compare_parity(std::slice::from_ref(&bad), &good, 2, 0.995),
2320 Err(ParityMiss::Invalid { index: 0, .. })
2321 ),
2322 "reference {bad:?}"
2323 );
2324 }
2325 }
2326
2327 #[test]
2328 fn mismatched_or_empty_parity_inputs_fail() {
2329 let one = vec![vec![1.0, 0.0]];
2330 let two = vec![vec![1.0, 0.0], vec![1.0, 0.0]];
2331 assert!(matches!(
2332 compare_parity(&one, &two, 2, 0.995),
2333 Err(ParityMiss::Invalid { .. })
2334 ));
2335 assert!(matches!(
2336 compare_parity(&two, &one, 2, 0.995),
2337 Err(ParityMiss::Invalid { .. })
2338 ));
2339 assert!(matches!(
2340 compare_parity(&[], &[], 2, 0.995),
2341 Err(ParityMiss::Invalid { .. })
2342 ));
2343 }
2344
2345 #[test]
2346 fn parity_miss_detail_names_cosine_length_and_floor() {
2347 let tokens = [20, 100, 600];
2348 let miss = ParityMiss::Below(ParityLowest {
2349 index: 0,
2350 cosine: 0.5791,
2351 });
2352 assert_eq!(
2353 parity_miss_detail(&miss, &tokens, 0.995),
2354 (
2355 Some(0.5791),
2356 "cosine 0.579 at ~20 tokens, floor 0.995".to_owned()
2357 )
2358 );
2359 let batched = ParityMiss::Below(ParityLowest {
2360 index: 5,
2361 cosine: 0.71,
2362 });
2363 assert_eq!(
2364 parity_miss_detail(&batched, &tokens, 0.995).1,
2365 "cosine 0.710 at ~600 tokens (batched), floor 0.995"
2366 );
2367 let invalid = ParityMiss::Invalid {
2368 index: 1,
2369 reason: "accelerator output: non-finite values".into(),
2370 };
2371 assert_eq!(
2372 parity_miss_detail(&invalid, &tokens, 0.995),
2373 (
2374 None,
2375 "accelerator output: non-finite values at ~100 tokens".to_owned()
2376 )
2377 );
2378 }
2379
2380 #[test]
2381 fn parity_long_document_is_deterministic_and_long() {
2382 let text = parity_long_document();
2383 assert_eq!(text, parity_long_document());
2384 assert!(text.split_whitespace().count() >= 400, "{}", text.len());
2385 assert!(PARITY_DOCUMENT.split_whitespace().count() >= 50);
2386 }
2387
2388 #[test]
2389 fn parity_error_names_provider_cosine_and_floor() {
2390 let error = EmbedError::AcceleratorParity {
2391 provider: "directml",
2392 lowest_cosine: Some(0.579),
2393 floor: ACCELERATOR_PARITY_MIN_COSINE,
2394 detail: "cosine 0.579 at ~20 tokens, floor 0.995".into(),
2395 };
2396 assert_eq!(
2397 error.to_string(),
2398 "directml failed the vector parity check against cpu: cosine 0.579 at ~20 tokens, floor 0.995"
2399 );
2400 }
2401
2402 fn parts<'a>(files: &'a [(&'a str, String)], truncate: usize) -> PipelineParts<'a> {
2403 PipelineParts {
2404 model_id: EMBEDDINGGEMMA_MODEL_ID,
2405 native_dim: 768,
2406 truncate_dim: truncate,
2407 max_tokens: 2048,
2408 scrubs_sentinels: false,
2409 files,
2410 }
2411 }
2412
2413 #[test]
2414 fn pipeline_fingerprint_is_stable_and_sensitive() {
2415 let files = [("model_q4.onnx", "aa".to_owned())];
2416 let base = pipeline_fingerprint(&parts(&files, 256));
2417 assert_eq!(base, pipeline_fingerprint(&parts(&files, 256)));
2418 assert!(base.starts_with("rkpipe1-") && base.len() == "rkpipe1-".len() + 32);
2419 assert_ne!(base, pipeline_fingerprint(&parts(&files, 128)));
2420 let changed = [("model_q4.onnx", "ab".to_owned())];
2421 assert_ne!(base, pipeline_fingerprint(&parts(&changed, 256)));
2422 let mut other = parts(&files, 256);
2423 other.scrubs_sentinels = true;
2424 assert_ne!(base, pipeline_fingerprint(&other));
2425 other = parts(&files, 256);
2426 other.max_tokens = 512;
2427 assert_ne!(base, pipeline_fingerprint(&other));
2428 }
2429
2430 #[test]
2431 fn hash_file_reports_absent_and_content() {
2432 let dir = tempfile::tempdir().unwrap();
2433 assert_eq!(hash_file(&dir.path().join("none")).unwrap(), "absent");
2434 let path = dir.path().join("f");
2435 std::fs::write(&path, b"abc").unwrap();
2436 assert_eq!(
2437 hash_file(&path).unwrap(),
2438 "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
2439 );
2440 }
2441
2442 #[test]
2443 fn config_defaults_and_builders() {
2444 let c = EmbedderConfig::new("m", "r");
2445 assert_eq!((c.max_batch, c.intra_threads), (16, 1));
2446 assert_eq!(c.ingest_batch, 8);
2447 assert_eq!(c.clone().with_ingest_batch(4).ingest_batch, 4);
2448 assert!(matches!(
2449 validate_config(&c.clone().with_ingest_batch(16)),
2450 Err(EmbedError::UnsupportedIngestBatch { requested: 16, .. })
2451 ));
2452 assert_eq!(c.provider, ExecutionProvider::Cpu);
2453 let c = c
2454 .with_max_batch(4)
2455 .with_intra_threads(3)
2456 .with_provider(ExecutionProvider::Auto);
2457 assert_eq!((c.max_batch, c.intra_threads), (4, 3));
2458 assert_eq!(c.provider, ExecutionProvider::Auto);
2459 let legacy =
2460 EmbedderConfig::new("m", "r").with_provider(rightkit_onnx::ExecutionProvider::DirectMl);
2461 assert_eq!(legacy.provider, ExecutionProvider::DirectMl);
2462 assert!(EmbedderConfig::new("m", "r").fallback_model_dir.is_none());
2463 let with_fallback = EmbedderConfig::new("m", "r").with_fallback_model_dir("q4");
2464 assert_eq!(with_fallback.fallback_model_dir, Some(PathBuf::from("q4")));
2465 assert!(EmbedderConfig::new("m", "r").coreml_cache_dir.is_none());
2466 let with_cache = EmbedderConfig::new("m", "r").with_coreml_cache_dir("cache");
2467 assert_eq!(with_cache.coreml_cache_dir, Some(PathBuf::from("cache")));
2468 assert!(EmbedderConfig::new("m", "r").coreml_units.is_none());
2469 let forced =
2470 EmbedderConfig::new("m", "r").with_coreml_units(crate::CoreMlComputeUnit::CpuAndGpu);
2471 assert_eq!(
2472 forced.coreml_units,
2473 Some(CoreMlUnits::uniform(crate::CoreMlComputeUnit::CpuAndGpu))
2474 );
2475 let custom = EmbedderConfig::new("m", "r")
2476 .with_coreml_units("256=cpu_only".parse::<CoreMlUnits>().unwrap());
2477 assert_eq!(
2478 custom.coreml_units.unwrap().unit_for(256),
2479 Some(crate::CoreMlComputeUnit::CpuOnly)
2480 );
2481 }
2482
2483 #[test]
2484 fn model_ids_are_distinct_so_pipeline_ids_differ_per_model() {
2485 let ids = [
2486 EMBEDDINGGEMMA_MODEL_ID,
2487 EMBEDDINGGEMMA2_MODEL_ID,
2488 EMBEDDINGGEMMA2_E40_MODEL_ID,
2489 EMBEDDINGGEMMA2_COREML_MODEL_ID,
2490 ];
2491 for (i, a) in ids.iter().enumerate() {
2492 for b in &ids[i + 1..] {
2493 assert_ne!(a, b);
2494 }
2495 }
2496 let files = [("model_32.onnx", "aa".to_owned())];
2497 let parts = |model_id| PipelineParts {
2498 model_id,
2499 native_dim: 768,
2500 truncate_dim: 256,
2501 max_tokens: 512,
2502 scrubs_sentinels: true,
2503 files: &files,
2504 };
2505 assert_ne!(
2506 pipeline_fingerprint(&parts(EMBEDDINGGEMMA2_E40_MODEL_ID)),
2507 pipeline_fingerprint(&parts(EMBEDDINGGEMMA2_MODEL_ID))
2508 );
2509 }
2510
2511 #[test]
2512 fn parity_sections_are_prefixes_of_the_long_document() {
2513 assert!(parity_long_document().starts_with(&parity_sections(4)));
2514 assert!(parity_sections(4).starts_with(&parity_sections(1)));
2515 assert!(
2516 parity_sections(1).split_whitespace().count()
2517 < parity_sections(4).split_whitespace().count()
2518 );
2519 }
2520}