Implement domain expert model fusion. Given predictions and confidence scores from multiple domain-specific expert models, combine them using a weighted fusion strategy based on each expert's confidence and domain relevance.
import numpy as np
from typing import Dict, List
def expert_fusion(
expert_predictions: List[np.ndarray],
expert_confidences: List[np.ndarray],
domain_weights: List[float] = None
) -> Dict:
n_experts = len(expert_predictions)
n_samples = len(expert_predictions[0])
if domain_weights is None:
domain_weights = [1.0] * n_experts
# Normalize domain weights
total_dw = sum(domain_weights)
domain_weights = [w / total_dw for w in domain_weights]
# Compute combined weights: domain_weight * confidence
combined_weights = np.zeros((n_samples, n_experts))
for i in range(n_experts):
combined_weights[:, i] = domain_weights[i] * expert_confidences[i]
# Normalize weights per sample
weight_sums = combined_weights.sum(axis=1, keepdims=True)
weight_sums = np.maximum(weight_sums, 1e-10)
normalized_weights = combined_weights / weight_sums
# Weighted fusion of predictions
fused = np.zeros(n_samples)
for i in range(n_experts):
fused += normalized_weights[:, i] * expert_predictions[i]
# Compute fusion confidence
fusion_confidence = np.zeros(n_samples)
for i in range(n_experts):
fusion_confidence += normalized_weights[:, i] * expert_confidences[i]
return {
"fused_predictions": fused.tolist(),
"fusion_confidence": fusion_confidence.tolist(),
"expert_weights": normalized_weights.tolist(),
"n_experts": n_experts
}