1+ # Copyright (c) 2026 BAAI. All rights reserved.
2+
3+ """
4+ ILUVATAR rotary embedding operator implementations.
5+
6+ NOTE: This is a template/stub implementation using PyTorch reference code.
7+ Replace with actual Iluvatar-optimized implementations when available.
8+ """
9+
10+ from __future__ import annotations
11+
12+ import torch
13+
14+
15+ def rotary_embedding_iluvatar (
16+ obj ,
17+ query : torch .Tensor ,
18+ key : torch .Tensor ,
19+ cos : torch .Tensor ,
20+ sin : torch .Tensor ,
21+ position_ids : torch .Tensor ,
22+ rotary_interleaved : bool = False ,
23+ inplace : bool = True ,
24+ ) -> tuple [torch .Tensor , torch .Tensor ]:
25+ """
26+ Apply rotary position embedding using Iluvatar.
27+
28+ This is a placeholder implementation using PyTorch reference code.
29+ TODO: Replace with actual Iluvatar GPU optimized implementation.
30+
31+ Args:
32+ obj: The calling obj (for interface consistency)
33+ query: Query tensor [batch, num_heads, seq_len, head_dim] or [seq_len, num_heads, head_dim]
34+ key: Key tensor [batch, num_heads, seq_len, head_dim] or [seq_len, num_heads, head_dim]
35+ cos: Cosine cache [max_seq_len, rotary_dim] where rotary_dim = head_dim or head_dim // 2
36+ sin: Sine cache [max_seq_len, rotary_dim] where rotary_dim = head_dim or head_dim // 2
37+ position_ids: Position indices [batch, seq_len] or [seq_len]
38+ rotary_interleaved: Whether to use interleaved rotary
39+ inplace: Whether to modify tensors in-place (ignored in reference impl)
40+
41+ Returns:
42+ Tuple of (embedded_query, embedded_key)
43+ """
44+ # Get cos/sin for the positions
45+ # position_ids can be [batch, seq_len] or [seq_len]
46+ if position_ids .dim () == 1 :
47+ # [seq_len] -> [seq_len, rotary_dim]
48+ cos_selected = cos [position_ids ]
49+ sin_selected = sin [position_ids ]
50+ else :
51+ # [batch, seq_len] -> [batch, seq_len, rotary_dim]
52+ cos_selected = cos [position_ids ]
53+ sin_selected = sin [position_ids ]
54+
55+ # Expand dimensions to match query/key shape
56+ # query/key: [batch, num_heads, seq_len, head_dim] or [seq_len, num_heads, head_dim]
57+ if query .dim () == 4 :
58+ # [batch, num_heads, seq_len, head_dim]
59+ # cos_selected: [batch, seq_len, rotary_dim] -> [batch, 1, seq_len, rotary_dim]
60+ cos_selected = cos_selected .unsqueeze (1 )
61+ sin_selected = sin_selected .unsqueeze (1 )
62+ elif query .dim () == 3 :
63+ # [seq_len, num_heads, head_dim]
64+ # cos_selected: [seq_len, rotary_dim] -> [seq_len, 1, rotary_dim]
65+ cos_selected = cos_selected .unsqueeze (1 )
66+ sin_selected = sin_selected .unsqueeze (1 )
67+
68+ # Check if we need to repeat cos/sin to match head_dim
69+ rotary_dim = cos_selected .shape [- 1 ]
70+ head_dim = query .shape [- 1 ]
71+
72+ if rotary_dim != head_dim :
73+ # cos/sin only covers half of head_dim, need to repeat
74+ # This handles the case where rotary is only applied to part of the dimensions
75+ cos_selected = torch .cat ([cos_selected , cos_selected ], dim = - 1 )
76+ sin_selected = torch .cat ([sin_selected , sin_selected ], dim = - 1 )
77+
78+ def rotate_half (x ):
79+ """Rotates half the hidden dims of the input."""
80+ x1 = x [..., : x .shape [- 1 ] // 2 ]
81+ x2 = x [..., x .shape [- 1 ] // 2 :]
82+ return torch .cat ((- x2 , x1 ), dim = - 1 )
83+
84+ if rotary_interleaved :
85+ # Interleaved rotary
86+ def rotate_interleaved (x ):
87+ x1 = x [..., ::2 ]
88+ x2 = x [..., 1 ::2 ]
89+ return torch .stack ((- x2 , x1 ), dim = - 1 ).flatten (- 2 )
90+
91+ q_embed = (query * cos_selected ) + (rotate_interleaved (query ) * sin_selected )
92+ k_embed = (key * cos_selected ) + (rotate_interleaved (key ) * sin_selected )
93+ else :
94+ # Standard rotary (neox style)
95+ q_embed = (query * cos_selected ) + (rotate_half (query ) * sin_selected )
96+ k_embed = (key * cos_selected ) + (rotate_half (key ) * sin_selected )
97+
98+ return q_embed , k_embed
0 commit comments