gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(ml): parallel ensemble inference via rayon — sub-ms multi-model predictions

Browse Source 
Replace sequential for-loop over ModelInferenceAdapters with rayon
par_iter(). Each adapter's predict() runs on a separate thread,
then results are aggregated sequentially (fast arithmetic).

ModelInferenceAdapter: Send + Sync makes this safe for parallel execution.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-01 21:09:10 +01:00
parent 33a0656248
commit 1a6834fff2
1 changed files with 48 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
crates/ml/src/ensemble/inference_ensemble.rs
View File

@@ -6,6 +6,8 @@
use std::collections::HashMap;
use rayon::prelude::*;
use crate::ensemble::inference_adapter::{
EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta,
};
@@ -66,50 +68,59 @@ impl InferenceEnsemble {
));
}
// Run inference in parallel across all ready adapters
let results: Vec<(String, EnsemblePrediction)> = ready_adapters
.par_iter()
.filter_map(|adapter| {
let model_name = adapter.model_name().to_string();
match adapter.predict(features) {
Ok(pred) => {
if !pred.direction.is_finite() || !pred.confidence.is_finite() {
tracing::warn!(
model = %model_name,
direction = %pred.direction,
confidence = %pred.confidence,
"Model returned NaN/Inf prediction, skipping (circuit breaker)"
);
None
} else {
Some((model_name, pred))
}
}
Err(e) => {
tracing::warn!(
model = %model_name,
error = %e,
"Model prediction failed, skipping"
);
None
}
}
})
.collect();
// Aggregate results sequentially (fast arithmetic)
let mut weighted_direction_sum = 0.0_f64;
let mut weight_confidence_sum = 0.0_f64;
let mut confidence_sum = 0.0_f64;
let mut successful_count = 0_usize;
let mut model_names: Vec<String> = Vec::new();
for adapter in &ready_adapters {
let model_name = adapter.model_name().to_string();
match adapter.predict(features) {
Ok(pred) => {
// Circuit breaker: skip NaN/Inf predictions
if !pred.direction.is_finite() || !pred.confidence.is_finite() {
tracing::warn!(
model = %model_name,
direction = %pred.direction,
confidence = %pred.confidence,
"Model returned NaN/Inf prediction, skipping (circuit breaker)"
);
continue;
}
// Clamp confidence to valid range
let confidence = pred.confidence.clamp(0.0, 1.0);
let w = self
.weights
.get(&model_name)
.copied()
.unwrap_or(1.0);
let wc = w * confidence;
weighted_direction_sum += pred.direction * wc;
weight_confidence_sum += wc;
confidence_sum += confidence;
successful_count += 1;
model_names.push(model_name);
}
Err(e) => {
tracing::warn!(
model = %model_name,
error = %e,
"Model prediction failed, skipping"
);
}
}
for (model_name, pred) in &results {
let confidence = pred.confidence.clamp(0.0, 1.0);
let w = self
.weights
.get(model_name)
.copied()
.unwrap_or(1.0);
let wc = w * confidence;
weighted_direction_sum += pred.direction * wc;
weight_confidence_sum += wc;
confidence_sum += confidence;
model_names.push(model_name.clone());
}
let successful_count = results.len();
if successful_count == 0 {
return Err(MLError::InferenceError(
"All ready models failed during inference".to_owned(),