optirs_learned/transformer/architecture/

positional_encoding.rs

use std::fmt::Debug;
// Positional encoding mechanisms for transformer optimization
//
// This module implements various positional encoding strategies used in the
// transformer optimizer to provide position information to the attention mechanisms.

#[allow(dead_code)]
use scirs2_core::ndarray::{s, Array1, Array2};
use scirs2_core::numeric::Float;
use scirs2_core::random::{Random, Rng as SCRRng};

use super::super::TransformerOptimizerConfig;
use crate::error::{OptimError, Result};

/// Types of positional encoding
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PositionalEncodingType {
    /// Sinusoidal position encoding
    Sinusoidal,
    /// Learned position embedding
    Learned,
    /// Rotary position embedding (RoPE)
    Rotary,
    /// Relative position encoding
    Relative,
    /// ALiBi (Attention with Linear Biases)
    ALiBi,
}

/// Positional encoder for transformer inputs
#[derive(Debug, Clone)]
pub struct PositionalEncoder<T: Float + Debug + Send + Sync + 'static> {
    /// Encoding type
    encoding_type: PositionalEncodingType,

    /// Cached encodings
    cached_encodings: Option<Array2<T>>,

    /// Maximum sequence length
    max_seqlen: usize,

    /// Model dimension
    modeldim: usize,

    /// Learned position embeddings (if applicable)
    position_embeddings: Option<Array2<T>>,

    /// ALiBi slopes (if applicable)
    alibi_slopes: Option<Array1<T>>,
}

impl<T: Float + Debug + Default + Clone + Send + Sync + 'static> PositionalEncoder<T> {
    /// Create new positional encoder
    pub fn new(config: &TransformerOptimizerConfig) -> Result<Self> {
        let max_seqlen = config.max_sequence_length;
        let modeldim = config.modeldim;

        let mut cached_encodings = None;
        let mut position_embeddings = None;
        let mut alibi_slopes = None;

        match config.pos_encoding_type {
            PositionalEncodingType::Sinusoidal => {
                // Precompute sinusoidal encodings
                let mut encodings = Array2::zeros((max_seqlen, modeldim));

                for pos in 0..max_seqlen {
                    for i in 0..modeldim {
                        let angle = scirs2_core::numeric::NumCast::from(pos)
                            .unwrap_or_else(|| T::zero())
                            / T::from(10000.0_f64.powf(2.0 * (i as f64) / modeldim as f64))
                                .expect("unwrap failed");

                        if i % 2 == 0 {
                            encodings[[pos, i]] = angle.sin();
                        } else {
                            encodings[[pos, i]] = angle.cos();
                        }
                    }
                }
                cached_encodings = Some(encodings);
            }
            PositionalEncodingType::Learned => {
                // Initialize learnable position embeddings
                let mut rng = scirs2_core::random::thread_rng();
                let mut embeddings = Array2::zeros((max_seqlen, modeldim));

                // Xavier initialization
                let bound = (6.0 / (max_seqlen + modeldim) as f64).sqrt();
                for elem in embeddings.iter_mut() {
                    *elem =
                        T::from((rng.random::<f64>() - 0.5) * 2.0 * bound).expect("unwrap failed");
                }
                position_embeddings = Some(embeddings);
            }
            PositionalEncodingType::ALiBi => {
                // Initialize ALiBi slopes
                let numheads = config.numheads;
                let mut slopes = Array1::zeros(numheads);

                for h in 0..numheads {
                    let slope = T::from(2.0_f64.powf(-8.0 * (h + 1) as f64 / numheads as f64))
                        .expect("unwrap failed");
                    slopes[h] = slope;
                }
                alibi_slopes = Some(slopes);
            }
            _ => {
                // Default to sinusoidal for other types
                let mut encodings = Array2::zeros((max_seqlen, modeldim));

                for pos in 0..max_seqlen {
                    for i in 0..modeldim {
                        let angle = scirs2_core::numeric::NumCast::from(pos)
                            .unwrap_or_else(|| T::zero())
                            / T::from(10000.0_f64.powf(2.0 * (i as f64) / modeldim as f64))
                                .expect("unwrap failed");

                        if i % 2 == 0 {
                            encodings[[pos, i]] = angle.sin();
                        } else {
                            encodings[[pos, i]] = angle.cos();
                        }
                    }
                }
                cached_encodings = Some(encodings);
            }
        }

        Ok(Self {
            encoding_type: config.pos_encoding_type,
            cached_encodings,
            max_seqlen,
            modeldim,
            position_embeddings,
            alibi_slopes,
        })
    }

    /// Encode input with positional information
    pub fn encode(&self, input: &Array2<T>) -> Result<Array2<T>> {
        let (seq_len, modeldim) = input.dim();

        if seq_len > self.max_seqlen {
            return Err(OptimError::InvalidConfig(format!(
                "Sequence length {} exceeds maximum {}",
                seq_len, self.max_seqlen
            )));
        }

        if modeldim != self.modeldim {
            return Err(OptimError::InvalidConfig(format!(
                "Model dimension {} doesn't match expected {}",
                modeldim, self.modeldim
            )));
        }

        let mut output = input.clone();

        match self.encoding_type {
            PositionalEncodingType::Sinusoidal => {
                if let Some(ref encodings) = self.cached_encodings {
                    let pos_enc = encodings.slice(s![..seq_len, ..]);
                    output = output + pos_enc;
                }
            }
            PositionalEncodingType::Learned => {
                if let Some(ref embeddings) = self.position_embeddings {
                    let pos_emb = embeddings.slice(s![..seq_len, ..]);
                    output = output + pos_emb;
                }
            }
            PositionalEncodingType::Rotary => {
                // Rotary position embedding (RoPE) doesn't add to input,
                // it modifies attention computation
                // For now, just return input unchanged
            }
            PositionalEncodingType::Relative => {
                // Relative position encoding doesn't add to input,
                // it modifies attention computation
                // For now, just return input unchanged
            }
            PositionalEncodingType::ALiBi => {
                // ALiBi doesn't add to input, it modifies attention scores
                // For now, just return input unchanged
            }
        }

        Ok(output)
    }

    /// Get ALiBi slopes for attention bias calculation
    pub fn get_alibi_slopes(&self) -> Option<&Array1<T>> {
        self.alibi_slopes.as_ref()
    }

    /// Get encoding type
    pub fn encoding_type(&self) -> PositionalEncodingType {
        self.encoding_type
    }

    /// Get maximum sequence length
    pub fn max_sequence_length(&self) -> usize {
        self.max_seqlen
    }

    /// Get model dimension
    pub fn model_dimension(&self) -> usize {
        self.modeldim
    }

    /// Update position embeddings (for learned encoding)
    pub fn update_embeddings(&mut self, new_embeddings: Array2<T>) -> Result<()> {
        match self.encoding_type {
            PositionalEncodingType::Learned => {
                let (pos_len, model_dim) = new_embeddings.dim();
                if pos_len != self.max_seqlen || model_dim != self.modeldim {
                    return Err(OptimError::InvalidConfig(
                        "New embeddings dimensions don't match encoder configuration".to_string(),
                    ));
                }
                self.position_embeddings = Some(new_embeddings);
                Ok(())
            }
            _ => Err(OptimError::InvalidConfig(
                "Position embeddings can only be updated for learned encoding type".to_string(),
            )),
        }
    }

    /// Compute sinusoidal encoding for a specific position
    pub fn compute_sinusoidal_position(&self, position: usize) -> Result<Array1<T>> {
        if position >= self.max_seqlen {
            return Err(OptimError::InvalidConfig(
                "Position exceeds maximum sequence length".to_string(),
            ));
        }

        let mut encoding = Array1::zeros(self.modeldim);
        for i in 0..self.modeldim {
            let angle = scirs2_core::numeric::NumCast::from(position).unwrap_or_else(|| T::zero())
                / T::from(10000.0_f64.powf(2.0 * (i as f64) / self.modeldim as f64))
                    .expect("unwrap failed");

            if i % 2 == 0 {
                encoding[i] = angle.sin();
            } else {
                encoding[i] = angle.cos();
            }
        }

        Ok(encoding)
    }

    /// Apply ALiBi bias to attention scores
    pub fn apply_alibi_bias(
        &self,
        attention_scores: &mut Array2<T>,
        head_idx: usize,
    ) -> Result<()> {
        if self.encoding_type != PositionalEncodingType::ALiBi {
            return Ok(()); // No-op for non-ALiBi encoding
        }

        if let Some(ref slopes) = self.alibi_slopes {
            if head_idx >= slopes.len() {
                return Err(OptimError::InvalidConfig(
                    "Head index exceeds number of ALiBi slopes".to_string(),
                ));
            }

            let slope = slopes[head_idx];
            let (seq_len, _) = attention_scores.dim();

            for i in 0..seq_len {
                for j in 0..seq_len {
                    let distance = T::from((i as i32 - j as i32).abs()).expect("unwrap failed");
                    attention_scores[[i, j]] = attention_scores[[i, j]] - slope * distance;
                }
            }
        }

        Ok(())
    }
}