1use crate::kv_cache::LayerKvCache;
15use crate::linear_core::{GdnCfg, GdnWeights, VmfPhaseCfg, VmfPhaseWeights};
16use crate::pipeline::{
17 AttnKind, DenseFfn, FfnKind, LayerWeights, MoeFfn, MtpModule, Pipeline, PipelineWeights,
18};
19use crate::qtensor::QTensor;
20use crate::sampler::SamplerConfig;
21use crate::tokenizer::Tokenizer;
22use cortiq_core::quant::dequant_tensor;
23use cortiq_core::{CmfError, CmfModel, LayerType, ModelArch};
24use std::sync::Arc;
25
26pub enum Overlay<'a> {
29 None,
30 One(&'a str),
31 Blend(&'a [(String, f32)]),
33}
34
35impl Overlay<'_> {
36 fn blend_touches(&self, model: &CmfModel, name: &str) -> bool {
37 match self {
38 Overlay::Blend(list) => list
39 .iter()
40 .any(|(sid, _)| model.tensor(&format!("skill.{sid}.{name}")).is_some()),
41 _ => false,
42 }
43 }
44}
45
46fn dequant_by_name(model: &CmfModel, name: &str) -> Result<Vec<f32>, String> {
47 let entry = model
48 .tensor(name)
49 .ok_or_else(|| format!("tensor '{name}' not found in CMF directory"))?;
50 let mut out = vec![0.0f32; entry.n_elems()];
51 dequant_tensor(entry, model.entry_bytes(entry), &mut out)?;
52 Ok(out)
53}
54
55fn blend_f32(model: &CmfModel, name: &str, list: &[(String, f32)]) -> Result<Vec<f32>, String> {
58 let mut acc: Option<Vec<f32>> = None;
59 for (sid, w) in list {
60 let sname = format!("skill.{sid}.{name}");
61 let src = if model.tensor(&sname).is_some() { &sname } else { name };
62 let t = dequant_by_name(model, src)?;
63 match &mut acc {
64 None => {
65 let mut t = t;
66 for v in t.iter_mut() {
67 *v *= w;
68 }
69 acc = Some(t);
70 }
71 Some(a) => {
72 for (av, tv) in a.iter_mut().zip(&t) {
73 *av += w * tv;
74 }
75 }
76 }
77 }
78 acc.ok_or_else(|| "empty blend".into())
79}
80
81fn load_f32(model: &CmfModel, name: &str, ov: &Overlay) -> Result<Vec<f32>, String> {
83 if ov.blend_touches(model, name) {
84 if let Overlay::Blend(list) = ov {
85 return blend_f32(model, name, list);
86 }
87 }
88 let skill = match ov {
89 Overlay::One(s) => Some(*s),
90 _ => None,
91 };
92 let entry = model
93 .resolve_tensor(name, skill)
94 .ok_or_else(|| format!("tensor '{name}' not found in CMF directory"))?;
95 let bytes = model.entry_bytes(entry);
96 let mut out = vec![0.0f32; entry.n_elems()];
97 dequant_tensor(entry, bytes, &mut out)?;
98 Ok(out)
99}
100
101pub(crate) fn build_layer_ffn(
107 model: &Arc<CmfModel>,
108 arch: &ModelArch,
109 li: usize,
110 force_f32: bool,
111 ov: &Overlay,
112) -> Result<FfnKind, CmfError> {
113 let prefix = format!("model.layers.{li}.");
114 let load_dense = |p: &str| -> Result<DenseFfn, CmfError> {
115 let gate_proj = load_matrix(model, &format!("{p}gate_proj.weight"), force_f32, ov)?;
116 let up_proj = load_matrix(model, &format!("{p}up_proj.weight"), force_f32, ov)?;
117 let down_proj = load_matrix(model, &format!("{p}down_proj.weight"), force_f32, ov)?;
118 let inter = gate_proj.rows();
122 if up_proj.rows() != inter || down_proj.cols() != inter {
123 return Err(CmfError::Parse(format!(
124 "{p}: FFN dims disagree (gate.rows={inter}, up.rows={}, \
125 down.cols={}); all three must equal inter'",
126 up_proj.rows(),
127 down_proj.cols()
128 )));
129 }
130 if down_proj.rows() != arch.hidden_size {
131 return Err(CmfError::Parse(format!(
132 "{p}: down_proj.rows={} != hidden_size={}",
133 down_proj.rows(),
134 arch.hidden_size
135 )));
136 }
137 Ok(DenseFfn {
138 gate_proj,
139 up_proj,
140 down_proj,
141 })
142 };
143 let router_name = format!("{prefix}mlp.gate.weight");
144 if model.tensor(&router_name).is_none() {
145 return Ok(FfnKind::Dense(load_dense(&format!("{prefix}mlp."))?));
146 }
147 let cfg = arch.moe.as_ref().ok_or_else(|| {
148 CmfError::Parse(format!("{router_name} present but header has no arch.moe block"))
149 })?;
150 let experts = (0..cfg.num_experts)
151 .map(|e| load_dense(&format!("{prefix}mlp.experts.{e}.")))
152 .collect::<Result<Vec<_>, _>>()?;
153 let shared = if model
154 .tensor(&format!("{prefix}mlp.shared_expert.gate_proj.weight"))
155 .is_some()
156 {
157 Some((
158 load_dense(&format!("{prefix}mlp.shared_expert."))?,
159 load_matrix(model, &format!("{prefix}mlp.shared_expert_gate.weight"), force_f32, ov)?,
160 ))
161 } else {
162 None
163 };
164 Ok(FfnKind::Moe(MoeFfn {
165 router: load_matrix(model, &router_name, force_f32, ov)?,
166 experts,
167 top_k: cfg.top_k,
168 norm_topk_prob: cfg.norm_topk_prob,
169 shared,
170 stats: std::cell::RefCell::new(Vec::new()),
171 }))
172}
173
174fn load_matrix(
175 model: &Arc<CmfModel>,
176 name: &str,
177 force_f32: bool,
178 ov: &Overlay,
179) -> Result<QTensor, CmfError> {
180 if ov.blend_touches(model, name) {
184 if let Overlay::Blend(list) = ov {
185 let entry = model
186 .tensor(name)
187 .ok_or_else(|| CmfError::MissingTensor(name.to_string()))?;
188 let data = blend_f32(model, name, list)
189 .map_err(|e| CmfError::Parse(format!("blend: {e}")))?;
190 return Ok(QTensor::from_f32(data, entry.shape[0], entry.shape[1]));
191 }
192 }
193 let skill = match ov {
194 Overlay::One(s) => Some(*s),
195 _ => None,
196 };
197 let name: &str = &match skill {
200 Some(sid) if model.tensor(&format!("skill.{sid}.{name}")).is_some() => {
201 format!("skill.{sid}.{name}")
202 }
203 _ => name.to_string(),
204 };
205 let err = |e: String| CmfError::Parse(format!("weight loading: {e}"));
206 if force_f32 {
207 let entry = model
208 .tensor(name)
209 .ok_or_else(|| CmfError::MissingTensor(name.to_string()))?;
210 if entry.shape.len() != 2 {
211 return Err(err(format!("'{name}' is not 2-D")));
212 }
213 let data = load_f32(model, name, &Overlay::None).map_err(err)?;
214 Ok(QTensor::from_f32(data, entry.shape[0], entry.shape[1]))
215 } else {
216 QTensor::from_model(model, name).map_err(err)
217 }
218}
219
220impl Pipeline {
221 pub fn from_model(model: &Arc<CmfModel>, sampler_config: SamplerConfig) -> Result<Self, CmfError> {
223 Self::from_model_with_skill(model, sampler_config, None)
224 }
225
226 pub fn from_model_with_skill(
232 model: &Arc<CmfModel>,
233 sampler_config: SamplerConfig,
234 skill: Option<&str>,
235 ) -> Result<Self, CmfError> {
236 match skill {
237 Some(s) => Self::from_model_with_overlay(model, sampler_config, &Overlay::One(s)),
238 None => Self::from_model_with_overlay(model, sampler_config, &Overlay::None),
239 }
240 }
241
242 pub fn from_model_with_blend(
245 model: &Arc<CmfModel>,
246 sampler_config: SamplerConfig,
247 blend: &[(String, f32)],
248 ) -> Result<Self, CmfError> {
249 Self::from_model_with_overlay(model, sampler_config, &Overlay::Blend(blend))
250 }
251
252 fn from_model_with_overlay(
253 model: &Arc<CmfModel>,
254 sampler_config: SamplerConfig,
255 ov: &Overlay,
256 ) -> Result<Self, CmfError> {
257 let skill = match ov {
258 Overlay::One(s) => Some(*s),
259 _ => None,
260 };
261 if let Some(sid) = skill {
262 let known = model.header.skills.iter().any(|s| s.id == sid)
263 || model.skill_tensors(sid).next().is_some();
264 if !known {
265 return Err(CmfError::Parse(format!(
266 "skill '{sid}' not in this container (header.skills: {:?})",
267 model.header.skills.iter().map(|s| &s.id).collect::<Vec<_>>()
268 )));
269 }
270 tracing::info!(
271 "skill '{sid}': {} replacement tensors overlaid",
272 model.skill_tensors(sid).count()
273 );
274 }
275 let arch = model.arch().clone();
276 let err = |e: String| CmfError::Parse(format!("weight loading: {e}"));
277
278 let masks_present = !model.masks.masks.is_empty();
284 let force_f32 = masks_present; let mut tokenizer = if let Some(vocab_bytes) = &model.vocab {
288 Tokenizer::from_bytes(vocab_bytes)
289 .map_err(|e| CmfError::Parse(format!("embedded tokenizer: {e}")))?
290 } else {
291 let sidecar = model.path.with_file_name("tokenizer.json");
292 if sidecar.exists() {
293 Tokenizer::from_file(&sidecar)
294 .map_err(|e| CmfError::Parse(format!("sidecar tokenizer: {e}")))?
295 } else {
296 tracing::warn!("no tokenizer in file or sidecar — using byte-level fallback");
297 Tokenizer::byte_level()
298 }
299 };
300 if let Some(tc) = &model.header.tokenizer_config {
302 tokenizer.chat_template = tc.chat_template.clone();
303 tokenizer.extra_eos.extend(tc.eos_token_ids.iter().copied());
304 if tokenizer.bos_token_id.is_none() {
305 tokenizer.bos_token_id = tc.bos_token_id;
306 }
307 tracing::info!(
308 "chat bundle: template {} chars, {} stop ids",
309 tc.chat_template.as_deref().map(str::len).unwrap_or(0),
310 tc.eos_token_ids.len()
311 );
312 }
313
314 let embed_tokens = load_matrix(model, "model.embed_tokens.weight", false, ov)?;
316 let final_norm = load_f32(model, "model.norm.weight", ov).map_err(err)?;
317 let lm_head = if model.tensor("lm_head.weight").is_some() {
318 load_matrix(model, "lm_head.weight", false, ov)?
319 } else if arch.tie_word_embeddings {
320 load_matrix(model, "model.embed_tokens.weight", false, ov)?
322 } else {
323 return Err(CmfError::MissingTensor(
324 "lm_head.weight (and tie_word_embeddings is false)".into(),
325 ));
326 };
327
328 let has_linear = arch
330 .layer_types
331 .iter()
332 .any(|t| matches!(t, LayerType::LinearAttention));
333 let mut vmf_cfg = None;
334 let mut gdn_cfg = None;
335 if has_linear {
336 let lc = arch.linear_core.as_ref().ok_or_else(|| {
337 CmfError::Parse(
338 "model has LinearAttention layers but no arch.linear_core — \
339 reconvert with the current converter"
340 .into(),
341 )
342 })?;
343 let need = |v: Option<usize>, name: &str| {
344 v.ok_or_else(|| CmfError::Parse(format!("linear core needs arch.{name}")))
345 };
346 match lc.kind.as_str() {
347 "vmf_phase" => {
348 vmf_cfg = Some(VmfPhaseCfg {
349 num_heads: lc.num_heads,
350 nphase: need(lc.nphase, "linear_core.nphase")?,
351 value_head_dim: lc.value_head_dim,
352 hidden_size: arch.hidden_size,
353 phase_mass: std::env::var("CMF_PHASE_MASS")
356 .ok()
357 .and_then(|v| v.parse().ok())
358 .unwrap_or(0.0),
359 });
360 }
361 "gated_delta_net" => {
362 gdn_cfg = Some(GdnCfg {
363 num_v_heads: lc.num_heads,
364 num_k_heads: need(arch.linear_num_key_heads, "linear_num_key_heads")?,
365 key_head_dim: need(arch.linear_key_head_dim, "linear_key_head_dim")?,
366 value_head_dim: lc.value_head_dim,
367 conv_kernel: need(
368 arch.linear_conv_kernel_dim,
369 "linear_conv_kernel_dim",
370 )?,
371 hidden_size: arch.hidden_size,
372 rms_eps: arch.rms_norm_eps as f64,
373 });
374 }
375 other => {
376 return Err(CmfError::Parse(format!(
377 "unknown linear core '{other}' (this runtime executes: \
378 gated_delta_net, vmf_phase)"
379 )));
380 }
381 }
382 }
383
384 let load_full_attn = |prefix: &str| -> Result<AttnKind, CmfError> {
386 let t = |suffix: &str| load_matrix(model, &format!("{prefix}{suffix}"), force_f32, ov);
387 let n = |suffix: &str| -> Option<Vec<f32>> {
388 model
389 .tensor(&format!("{prefix}{suffix}"))
390 .and_then(|_| load_f32(model, &format!("{prefix}{suffix}"), ov).ok())
391 };
392 let wq = t("self_attn.q_proj.weight")?;
393 let output_gate = wq.rows() == 2 * arch.num_attention_heads * arch.head_dim;
395 let bias = match (
397 n("self_attn.q_proj.bias"),
398 n("self_attn.k_proj.bias"),
399 n("self_attn.v_proj.bias"),
400 ) {
401 (Some(a), Some(b), Some(c)) => Some((a, b, c)),
402 _ => None,
403 };
404 Ok(AttnKind::Full {
405 wq,
406 wk: t("self_attn.k_proj.weight")?,
407 wv: t("self_attn.v_proj.weight")?,
408 wo: t("self_attn.o_proj.weight")?,
409 q_norm: n("self_attn.q_norm.weight"),
410 k_norm: n("self_attn.k_norm.weight"),
411 output_gate,
412 bias,
413 })
414 };
415
416 let load_linear_attn = |prefix: &str| -> Result<AttnKind, CmfError> {
417 if gdn_cfg.is_some() {
418 let t = |suffix: &str| {
420 load_matrix(model, &format!("{prefix}linear_attn.{suffix}"), force_f32, ov)
421 };
422 let f = |suffix: &str| {
423 load_f32(model, &format!("{prefix}linear_attn.{suffix}"), ov).map_err(err)
424 };
425 return Ok(AttnKind::LinearGdn(GdnWeights {
426 in_proj_qkv: t("in_proj_qkv.weight")?,
427 in_proj_z: t("in_proj_z.weight")?,
428 in_proj_a: t("in_proj_a.weight")?,
429 in_proj_b: t("in_proj_b.weight")?,
430 conv1d: f("conv1d.weight")?,
431 a_log: f("A_log")?,
432 dt_bias: f("dt_bias")?,
433 norm: f("norm.weight")?,
434 out_proj: t("out_proj.weight")?,
435 }));
436 }
437 let t = |suffix: &str| load_matrix(model, &format!("{prefix}vmf_attn.{suffix}"), force_f32, ov);
438 let a_log = load_f32(model, &format!("{prefix}vmf_attn.A_log"), ov).map_err(err)?;
439 let k_gate = if model
443 .tensor(&format!("{prefix}vmf_attn.k_gate.weight"))
444 .is_some()
445 {
446 Some((
447 t("k_gate.weight")?,
448 load_f32(model, &format!("{prefix}vmf_attn.k_gate.bias"), ov).map_err(err)?,
449 ))
450 } else {
451 None
452 };
453 Ok(AttnKind::Linear(VmfPhaseWeights {
454 thq: t("thq.weight")?,
455 thk: t("thk.weight")?,
456 v_proj: t("v_proj.weight")?,
457 out_proj: t("out_proj.weight")?,
458 decay: a_log
459 .iter()
460 .map(|&a| (-(a as f64).exp()).exp())
461 .collect(),
462 k_gate,
463 }))
464 };
465
466 let mut layers = Vec::with_capacity(arch.num_layers);
467 for li in 0..arch.num_layers {
468 let prefix = format!("model.layers.{li}.");
469 let attn = match arch.layer_types.get(li) {
470 Some(LayerType::LinearAttention) => load_linear_attn(&prefix)?,
471 _ => load_full_attn(&prefix)?,
472 };
473 layers.push(LayerWeights {
474 input_norm: load_f32(model, &format!("{prefix}input_layernorm.weight"), ov).map_err(err)?,
475 post_norm: load_f32(model, &format!("{prefix}post_attention_layernorm.weight"), ov)
476 .map_err(err)?,
477 ffn: build_layer_ffn(model, &arch, li, false, ov)?,
479 attn,
480 });
481 }
482
483 let mtp = if let Some(cfg) = &arch.mtp {
485 if cfg.num_layers != 1 {
486 return Err(CmfError::Parse(format!(
487 "MTP with {} blocks not supported yet (only 1)",
488 cfg.num_layers
489 )));
490 }
491 let p = "model.mtp.";
492 let attn = load_full_attn("model.mtp.layers.0.")?;
493 Some(MtpModule {
494 enorm: load_f32(model, &format!("{p}enorm.weight"), ov).map_err(err)?,
495 hnorm: load_f32(model, &format!("{p}hnorm.weight"), ov).map_err(err)?,
496 eh_proj: load_matrix(model, &format!("{p}eh_proj.weight"), false, ov)?,
497 layer: LayerWeights {
498 input_norm: load_f32(model, &format!("{p}layers.0.input_layernorm.weight"), ov)
499 .map_err(err)?,
500 post_norm: load_f32(
501 model,
502 &format!("{p}layers.0.post_attention_layernorm.weight"),
503 ov,
504 )
505 .map_err(err)?,
506 ffn: FfnKind::Dense(DenseFfn {
507 gate_proj: load_matrix(model, &format!("{p}layers.0.mlp.gate_proj.weight"), false, ov)?,
508 up_proj: load_matrix(model, &format!("{p}layers.0.mlp.up_proj.weight"), false, ov)?,
509 down_proj: load_matrix(model, &format!("{p}layers.0.mlp.down_proj.weight"), false, ov)?,
510 }),
511 attn,
512 },
513 final_norm: load_f32(model, &format!("{p}norm.weight"), ov).map_err(err)?,
514 kv: LayerKvCache::new(arch.num_kv_heads, arch.head_dim),
515 })
516 } else {
517 None
518 };
519
520 tracing::info!(
521 "Pipeline loaded: {} | {}L ({} linear) | {:.2}B params | storage: {} | MTP: {}",
522 arch.arch_name,
523 arch.num_layers,
524 arch.layer_types
525 .iter()
526 .filter(|t| matches!(t, LayerType::LinearAttention))
527 .count(),
528 model.total_param_count() as f64 / 1e9,
529 if force_f32 { "f32 (masked)" } else { "quantized mmap" },
530 if mtp.is_some() { "yes" } else { "no" }
531 );
532
533 let cap = std::env::var("CMF_MAX_SEQ")
536 .ok()
537 .and_then(|v| v.parse::<usize>().ok())
538 .unwrap_or(8192);
539 let max_seq_len = arch.max_position_embeddings.min(cap);
540
541 let mut pipeline = Pipeline::new(
542 tokenizer,
543 PipelineWeights {
544 embed_tokens,
545 layers,
546 lm_head,
547 final_norm,
548 },
549 arch.hidden_size,
550 arch.intermediate_size,
551 arch.num_attention_heads,
552 arch.num_kv_heads,
553 arch.head_dim,
554 arch.num_layers,
555 arch.vocab_size,
556 arch.rms_norm_eps,
557 arch.rope_theta as f32,
558 arch.norm_style,
559 max_seq_len,
560 sampler_config,
561 );
562 let rotary = ((arch.head_dim as f32 * arch.partial_rotary_factor) as usize).max(2);
563 pipeline.set_rotary(rotary, arch.rope_theta as f32);
564 pipeline.vmf_cfg = vmf_cfg;
565 pipeline.gdn_cfg = gdn_cfg;
566 pipeline.mtp = mtp;
567 pipeline.install_dynamic_routing(model, false);
568 match ov {
572 Overlay::One(sid) => {
573 pipeline.dyn_active = model.header.skills.iter().position(|s| &s.id == sid);
574 }
575 Overlay::Blend(_) => pipeline.dyn_blend_loaded = true,
576 Overlay::None => {}
577 }
578 if let Some(c) = &model.header.calibration {
581 pipeline.set_calib_temp(c.temperature);
582 }
583 let o1 = match crate::nystrom::o1_from_env() {
589 crate::nystrom::O1Env::Off => None,
590 crate::nystrom::O1Env::On(cfg) => Some(cfg),
591 crate::nystrom::O1Env::Unset => model
592 .header
593 .provenance
594 .as_ref()
595 .and_then(|p| p.get("o1_attn"))
596 .and_then(crate::nystrom::O1Cfg::from_json),
597 };
598 if o1.is_some() {
599 pipeline.set_o1(o1);
600 }
601 Ok(pipeline)
602 }
603
604 pub(crate) fn install_dynamic_routing(
609 &mut self,
610 model: &Arc<CmfModel>,
611 force_f32: bool,
612 ) {
613 self.model = Some(model.clone());
614 self.dyn_force_f32 = force_f32;
615 let mut per_skill = Vec::with_capacity(model.header.skills.len());
616 for sk in &model.header.skills {
617 let mut ffn_layers = std::collections::BTreeSet::new();
618 let mut non_ffn = false;
619 let prefix = format!("skill.{}.", sk.id);
620 for t in model.skill_tensors(&sk.id) {
621 let rel = &t.name[prefix.len()..]; let toks: Vec<&str> = rel.split('.').collect();
623 if toks.len() >= 5
624 && toks[0] == "model"
625 && toks[1] == "layers"
626 && toks[3] == "mlp"
627 {
628 if let Ok(li) = toks[2].parse::<usize>() {
629 ffn_layers.insert(li);
630 continue;
631 }
632 }
633 non_ffn = true; }
635 if non_ffn {
636 tracing::warn!(
637 "skill '{}' replaces non-FFN tensors — excluded from dynamic \
638 routing (static overlay still works)",
639 sk.id
640 );
641 per_skill.push(None);
642 } else {
643 per_skill.push(Some(ffn_layers.into_iter().collect::<Vec<_>>()));
644 }
645 }
646 self.dyn_skill_layers = per_skill;
647 }
648
649 pub fn set_active_skill(&mut self, idx: Option<usize>) -> Result<(), CmfError> {
656 if self.dyn_active == idx {
657 return Ok(());
658 }
659 let model = self
660 .model
661 .clone()
662 .ok_or_else(|| CmfError::Parse("dynamic routing needs a model-backed pipeline".into()))?;
663 let mut union: std::collections::BTreeSet<usize> = std::collections::BTreeSet::new();
664 if let Some(old) = self.dyn_active {
665 if let Some(Some(ls)) = self.dyn_skill_layers.get(old) {
666 union.extend(ls.iter().copied());
667 }
668 }
669 let new_id: Option<String> = match idx {
670 Some(n) => match self.dyn_skill_layers.get(n) {
671 Some(Some(ls)) => {
672 union.extend(ls.iter().copied());
673 Some(model.header.skills[n].id.clone())
674 }
675 _ => {
676 return Err(CmfError::Parse(format!(
677 "skill index {n} not dynamic-eligible"
678 )))
679 }
680 },
681 None => None,
682 };
683 let ov = match &new_id {
684 Some(s) => Overlay::One(s),
685 None => Overlay::None,
686 };
687 let arch = model.arch();
688 for li in union {
689 self.weights.layers[li].ffn =
690 build_layer_ffn(&model, arch, li, self.dyn_force_f32, &ov)?;
691 }
692 self.dyn_active = idx;
693 Ok(())
694 }
695}