4 Commits
Author SHA1 Message Date
ViperEkura 19532440b4 chore: 版本号升至 1.3.5 2026-05-15 18:23:27 +08:00
ViperEkura 9096e413c3 refactor: RotaryEmbedding 合并 cos/sin 为单一复数缓存
- get_rotary_emb() 返回复数张量替代 Tuple[cos, sin]
- RotaryEmbedding 存储单一 freqs_cis buffer 替代分离的 cos_cached/sin_cached
- forward 中 view_as_complex 重建复数
2026-05-15 18:03:59 +08:00
ViperEkura 9d5e9fa6c4 perf: DDP 加 gradient_as_bucket_view/static_graph/broadcast_buffers,AdamW fused
- gradient_as_bucket_view=True 零拷贝梯度归并
- static_graph=True 跳过每轮 bucket 重建
- broadcast_buffers=False 省 buffer 广播
- AdamW fused=True 融合优化器 kernel
2026-05-15 15:30:24 +08:00
ViperEkura 08dde46778 fix: 修复训练循环 step/backward 顺序,重构为三重循环嵌套
- 训练循环改用 itertools.batched 实现 epoch→step→batch 三重嵌套
- on_step_begin 包裹 batch 循环,on_step_end 后接 optimizer.step/scheduler.step
- 修复首次 iteration=0 时 optimizer.step() 在 backward 之前触发的 bug
- GradientClippingCallback 改为 on_step_end(梯度已累积,step 前裁剪)
- SchedulerCallback 移除,schduler.step 由 trainer 在 optimizer.step 后直接调用
- metric_util 提取 _grad_stat 公共 helper,if param.grad: 修正为 is not None
2026-05-15 14:44:44 +08:00
10 changed files with 102 additions and 138 deletions
+10
View File
@@ -65,6 +65,16 @@ For development dependencies:
pip install -e ".[dev]" pip install -e ".[dev]"
``` ```
#### Download Pre-trained Model
Download pre-trained model weights (1B bilingual checkpoint) to `params/`:
```bash
python scripts/demo/download.py
```
Or download manually from [HuggingFace](https://huggingface.co/ViperEk/KHAOSZ) into `params/`.
#### Train a Model #### Train a Model
```bash ```bash
+10
View File
@@ -71,6 +71,16 @@ pip install -e .
pip install -e ".[dev]" pip install -e ".[dev]"
``` ```
#### 下载预训练模型
下载预训练模型权重(1B 双语检查点)到 `params/` 目录:
```bash
python scripts/demo/download.py
```
或从 [HuggingFace](https://huggingface.co/ViperEk/KHAOSZ) 手动下载放入 `params/`
#### 训练模型 #### 训练模型
```bash ```bash
+11 -11
View File
@@ -88,7 +88,7 @@ flowchart LR
- **`DecoderBlock`**: GQA attention + residual + MLP + RMSNorm - **`DecoderBlock`**: GQA attention + residual + MLP + RMSNorm
- **`GQA`**: Grouped Query Attention (also `MLA` for multi-latent attention) - **`GQA`**: Grouped Query Attention (also `MLA` for multi-latent attention)
- **`MLP`**: `SiLU(gate(x)) * up(x)` → down projection - **`MLP`**: `SiLU(gate(x)) * up(x)` → down projection
- **`RotaryEmbedding`**: RoPE cos/sin cache - **`RotaryEmbedding`**: RoPE complex cache (freqs_cis)
- **`RMSNorm`**: Layer normalization - **`RMSNorm`**: Layer normalization
### 4. Training Module ### 4. Training Module
@@ -104,22 +104,23 @@ The training loop is nested: **epoch** → **batch** (with step phase interspers
``` ```
on_train_begin on_train_begin
on_epoch_begin on_epoch_begin
for each batch: for each accumulation window of batches: ← step phase
if iteration % accumulation_steps == 0: ← step phase on_step_begin
on_step_begin → optimizer.step() → zero_grad → on_step_end for each batch in window: ← batch phase
← batch phase
on_batch_begin → strategy(batch) → loss → backward → on_batch_end on_batch_begin → strategy(batch) → loss → backward → on_batch_end
iteration += 1 iteration += 1
on_step_end
optimizer.step() → zero_grad
on_epoch_end on_epoch_end
on_train_end on_train_end
``` ```
Key points: Key points:
- `on_step_*` wraps optimizer step (fires every `accumulation_steps` batches) - `on_step_*` fires every `accumulation_steps` batches, wrapping optimizer step AFTER the hook
- `on_batch_*` wraps loss computation (fires every batch) - `on_batch_*` fires every batch, wrapping loss computation
- `SchedulerCallback` fires on `on_batch_end` — LR scheduler steps every batch - `GradientClippingCallback` fires on `on_step_end`
- `GradientClippingCallback` fires on `on_step_begin` - LR scheduler steps inline (no `SchedulerCallback` class)
#### 4.3 Strategy (`strategy.py`) #### 4.3 Strategy (`strategy.py`)
- **`SEQStrategy`**: Next-token prediction, cross-entropy with label smoothing - **`SEQStrategy`**: Next-token prediction, cross-entropy with label smoothing
@@ -136,8 +137,7 @@ Key points:
- **`CheckpointCallback`**: Saves safetensors at `ckpt_interval` iterations - **`CheckpointCallback`**: Saves safetensors at `ckpt_interval` iterations
- **`ProgressBarCallback`**: tqdm progress display - **`ProgressBarCallback`**: tqdm progress display
- **`MetricLoggerCallback`**: Writes JSONL metrics to `{ckpt_dir}/logs/` - **`MetricLoggerCallback`**: Writes JSONL metrics to `{ckpt_dir}/logs/`
- **`GradientClippingCallback`**: `clip_grad_norm_` on `on_step_begin` - **`GradientClippingCallback`**: `clip_grad_norm_` on `on_step_end`
- **`SchedulerCallback`**: `scheduler.step()` on `on_batch_end`
### 5. Inference Module ### 5. Inference Module
+10 -20
View File
@@ -91,8 +91,8 @@ classDiagram
} }
class BaseStorage { class BaseStorage {
+Dict segments +MultiSegmentFetcher _fetcher
+List keys +keys (property)
+load(load_path, tokenizer) +load(load_path, tokenizer)
+fetch(begin, end, keys) +fetch(begin, end, keys)
+__len__() +__len__()
@@ -145,7 +145,7 @@ classDiagram
+ModelConfig config +ModelConfig config
+Registry _registry +Registry _registry
+register(model_type) decorator +register(model_type) decorator
+get_model_class(model_type) Type +get_component_class(model_type) Type
+from_pretrained(path, disable_random_init) nn.Module +from_pretrained(path, disable_random_init) nn.Module
+save_pretrained(save_directory) +save_pretrained(save_directory)
+to(*args, **kwargs) Self +to(*args, **kwargs) Self
@@ -214,7 +214,7 @@ classDiagram
+int dim +int dim
+int max_len +int max_len
+float base +float base
+forward(x, position_ids=None) Tuple[Tensor, Tensor] +forward(x, position_ids=None) Tensor
} }
class Embedding { class Embedding {
@@ -225,13 +225,10 @@ classDiagram
namespace tokenize { namespace tokenize {
class AutoTokenizer { class AutoTokenizer {
+List[int] stop_ids
+int bos_id
+int eos_id
+int pad_id
+vocab_size int +vocab_size int
+encode(tokens, out_ids, add_special_tokens) List[int] +encode(tokens, out_ids, add_special_tokens) List[int]
+decode(tokens, skip_special_tokens) str +decode(tokens, skip_special_tokens) str
+__getattr__(name) Any (bos_id, eos_id, pad_id, stop_ids)
+apply_chat_template(messages, tokenize) Union[str, List[int]] +apply_chat_template(messages, tokenize) Union[str, List[int]]
+set_chat_template(template) +set_chat_template(template)
+load(path) +load(path)
@@ -325,6 +322,8 @@ classDiagram
+float clip_eps +float clip_eps
+float kl_coef +float kl_coef
+int group_size +int group_size
+str reduction
+int sync_interval
+compute_loss(batch) Tensor +compute_loss(batch) Tensor
} }
@@ -369,11 +368,6 @@ classDiagram
+on_step_begin(context) +on_step_begin(context)
} }
class SchedulerCallback {
+on_train_begin(context)
+on_batch_end(context)
}
class CheckpointCallback { class CheckpointCallback {
+str save_dir +str save_dir
+int interval +int interval
@@ -409,8 +403,6 @@ classDiagram
+nn.Module model +nn.Module model
+AutoTokenizer tokenizer +AutoTokenizer tokenizer
+InferenceScheduler scheduler +InferenceScheduler scheduler
+int max_batch_size
+Optional int max_seq_len
+generate(prompt, stream, max_tokens, temperature, top_p, top_k) Union[Generator, str, List[str]] +generate(prompt, stream, max_tokens, temperature, top_p, top_k) Union[Generator, str, List[str]]
+generate_with_request(request) Union[Generator, str, List[str]] +generate_with_request(request) Union[Generator, str, List[str]]
+generate_async(prompt, max_tokens, temperature, top_p, top_k) AsyncGenerator +generate_async(prompt, max_tokens, temperature, top_p, top_k) AsyncGenerator
@@ -421,13 +413,12 @@ classDiagram
class InferenceScheduler { class InferenceScheduler {
+nn.Module model +nn.Module model
+AutoTokenizer tokenizer +AutoTokenizer tokenizer
+KVCache page_cache +KVCache _page_cache
+int max_batch_size +int max_batch_size
+int max_seq_len +int max_seq_len
+int max_prompt_len +int max_prompt_len
+int page_size +int page_size
+List waiting_queue +TaskManager _task_mgr
+List active_tasks
+add_task(prompt, max_tokens, temperature, top_p, top_k, stream_callback) str +add_task(prompt, max_tokens, temperature, top_p, top_k, stream_callback) str
+remove_task(task_id) +remove_task(task_id)
+start() +start()
@@ -568,7 +559,7 @@ classDiagram
} }
class GenerateResult { class GenerateResult {
+List[str] tokens +List[Tuple[int, str]] tokens
+List[str] results +List[str] results
+List[bool] _done +List[bool] _done
+append(token, idx) +append(token, idx)
@@ -643,7 +634,6 @@ classDiagram
BaseScheduler <|-- SGDRScheduler BaseScheduler <|-- SGDRScheduler
CallbackFactory ..> TrainCallback : creates CallbackFactory ..> TrainCallback : creates
TrainCallback <|-- GradientClippingCallback TrainCallback <|-- GradientClippingCallback
TrainCallback <|-- SchedulerCallback
TrainCallback <|-- CheckpointCallback TrainCallback <|-- CheckpointCallback
TrainCallback <|-- ProgressBarCallback TrainCallback <|-- ProgressBarCallback
TrainCallback <|-- MetricLoggerCallback TrainCallback <|-- MetricLoggerCallback
+1 -1
View File
@@ -1,4 +1,4 @@
__version__ = "1.3.4" __version__ = "1.3.5"
__author__ = "ViperEkura" __author__ = "ViperEkura"
from astrai.config import ( from astrai.config import (
+12 -11
View File
@@ -1,4 +1,4 @@
from typing import Optional, Tuple from typing import Optional
import torch import torch
import torch.nn as nn import torch.nn as nn
@@ -25,11 +25,13 @@ def get_rotary_emb(
max_len: int, max_len: int,
base: float = 10000, base: float = 10000,
device: Optional[torch.device] = None, device: Optional[torch.device] = None,
) -> Tuple[Tensor, Tensor]: ) -> Tensor:
theta = base ** (-torch.arange(0, dim, 2, dtype=torch.float64, device=device) / dim) theta = base ** (-torch.arange(0, dim, 2, dtype=torch.float64, device=device) / dim)
t = torch.arange(0, max_len, dtype=torch.float64, device=device) t = torch.arange(0, max_len, dtype=torch.float64, device=device)
freqs = torch.outer(t, theta) freqs = torch.outer(t, theta).float()
return torch.cos(freqs).float(), torch.sin(freqs).float() cos = torch.cos(freqs)
sin = torch.sin(freqs)
return torch.complex(cos, sin)
def apply_rotary_emb(x: torch.Tensor, freqs_cis: Tensor) -> Tensor: def apply_rotary_emb(x: torch.Tensor, freqs_cis: Tensor) -> Tensor:
@@ -50,10 +52,10 @@ class RotaryEmbedding(nn.Module):
self.base = base self.base = base
self._set_rotary_buffer(self.max_len) self._set_rotary_buffer(self.max_len)
def _set_rotary_buffer(self, max_len: int, device: Optional[torch.device] = None): def _set_rotary_buffer(self, max_len: int):
cos_cached, sin_cached = get_rotary_emb(self.dim, max_len, self.base, device) rotary_emb = get_rotary_emb(self.dim, max_len, self.base)
self.register_buffer("cos_cached", cos_cached, persistent=False) freqs_cis = torch.view_as_real(rotary_emb)
self.register_buffer("sin_cached", sin_cached, persistent=False) self.register_buffer("freqs_cis", freqs_cis, persistent=False)
def forward(self, x: Tensor, position_ids: Optional[Tensor] = None) -> Tensor: def forward(self, x: Tensor, position_ids: Optional[Tensor] = None) -> Tensor:
if position_ids is None: if position_ids is None:
@@ -62,9 +64,8 @@ class RotaryEmbedding(nn.Module):
.unsqueeze(0) .unsqueeze(0)
.expand(x.size(0), -1) .expand(x.size(0), -1)
) )
cos = self.cos_cached[position_ids].float() position_freq_cis = self.freqs_cis[position_ids].float()
sin = self.sin_cached[position_ids].float() return torch.view_as_complex(position_freq_cis)
return torch.complex(cos, sin)
class Linear(nn.Module): class Linear(nn.Module):
+19 -52
View File
@@ -1,75 +1,42 @@
from typing import Dict from typing import Any, Callable, Dict
import torch
import torch.nn as nn import torch.nn as nn
def grad_norm(model: nn.Module, norm_type: int = 2) -> Dict[str, float]: def _grad_stat(
"""Compute gradient norm for each parameter in the model.""" model: nn.Module, fn: Callable[[torch.Tensor], Any], default: Any
norms = {} ) -> dict:
results = {}
for name, param in model.named_parameters(): for name, param in model.named_parameters():
norms[name] = 0.0 results[name] = default
if param.grad: if param.grad is not None:
norm = param.grad.data.norm(norm_type).item() results[name] = fn(param.grad.data)
norms[name] = norm return results
return norms
def grad_norm(model: nn.Module, norm_type: int = 2) -> Dict[str, float]:
return _grad_stat(model, lambda g: g.norm(norm_type).item(), 0.0)
def grad_std(model: nn.Module) -> Dict[str, float]: def grad_std(model: nn.Module) -> Dict[str, float]:
"""Compute standard deviation of gradients for each parameter.""" return _grad_stat(model, lambda g: g.std().item(), 0.0)
stds = {}
for name, param in model.named_parameters():
stds[name] = 0.0
if param.grad:
std = param.grad.data.std().item()
stds[name] = std
return stds
def grad_max(model: nn.Module) -> Dict[str, float]: def grad_max(model: nn.Module) -> Dict[str, float]:
"""Find the maximum absolute gradient value for each parameter.""" return _grad_stat(model, lambda g: g.max().item(), -float("inf"))
max_vals = {}
for name, param in model.named_parameters():
max_vals[name] = -float("inf")
if param.grad:
max_val = param.grad.data.max().item()
max_vals[name] = max_val
return max_vals
def grad_min(model: nn.Module) -> Dict[str, float]: def grad_min(model: nn.Module) -> Dict[str, float]:
"""Find the minimum absolute gradient value for each parameter.""" return _grad_stat(model, lambda g: g.min().item(), float("inf"))
min_vals = {}
for name, param in model.named_parameters():
min_vals[name] = float("inf")
if param.grad:
min_val = param.grad.data.min().item()
min_vals[name] = min_val
return min_vals
def grad_mean(model: nn.Module) -> Dict[str, float]: def grad_mean(model: nn.Module) -> Dict[str, float]:
"""Compute mean of gradients for each parameter.""" return _grad_stat(model, lambda g: g.mean().item(), 0.0)
means = {}
for name, param in model.named_parameters():
means[name] = 0.0
if param.grad:
mean = param.grad.data.mean().item()
means[name] = mean
return means
def grad_nan_num(model: nn.Module) -> Dict[str, int]: def grad_nan_num(model: nn.Module) -> Dict[str, int]:
"""Count the number of NaNs in gradients for each parameter.""" return _grad_stat(model, lambda g: g.isnan().sum().item(), 0)
nan_nums = {}
for name, param in model.named_parameters():
nan_nums[name] = 0
if param.grad:
nan_num = param.grad.isnan().sum().item()
nan_nums[name] = nan_num
return nan_nums
def ctx_get_loss(ctx): def ctx_get_loss(ctx):
+1 -20
View File
@@ -79,30 +79,11 @@ class GradientClippingCallback(TrainCallback):
def __init__(self, max_grad_norm: float): def __init__(self, max_grad_norm: float):
self.max_grad_norm = max_grad_norm self.max_grad_norm = max_grad_norm
def on_step_begin(self, context: TrainContext): def on_step_end(self, context: TrainContext):
_ = context _ = context
clip_grad_norm_(context.model.parameters(), self.max_grad_norm) clip_grad_norm_(context.model.parameters(), self.max_grad_norm)
@CallbackFactory.register("scheduler")
class SchedulerCallback(TrainCallback):
"""
Scheduler callback for trainer.
"""
def __init__(self):
pass
def on_train_begin(self, context: TrainContext):
for group in context.optimizer.param_groups:
if "initial_lr" not in group:
group["initial_lr"] = group["lr"]
def on_batch_end(self, context: TrainContext):
if context.scheduler:
context.scheduler.step()
@CallbackFactory.register("checkpoint") @CallbackFactory.register("checkpoint")
class CheckpointCallback(TrainCallback): class CheckpointCallback(TrainCallback):
""" """
+14 -13
View File
@@ -1,4 +1,5 @@
import logging import logging
from itertools import batched
from typing import List, Optional from typing import List, Optional
from astrai.config import TrainConfig from astrai.config import TrainConfig
@@ -30,7 +31,6 @@ class Trainer:
CallbackFactory.create("checkpoint", cfg.ckpt_dir, cfg.ckpt_interval), CallbackFactory.create("checkpoint", cfg.ckpt_dir, cfg.ckpt_interval),
CallbackFactory.create("metric_logger", cfg.ckpt_dir, cfg.ckpt_interval), CallbackFactory.create("metric_logger", cfg.ckpt_dir, cfg.ckpt_interval),
CallbackFactory.create("gradient_clipping", cfg.max_grad_norm), CallbackFactory.create("gradient_clipping", cfg.max_grad_norm),
CallbackFactory.create("scheduler"),
] ]
def _build_context(self, checkpoint: Optional[Checkpoint]) -> TrainContext: def _build_context(self, checkpoint: Optional[Checkpoint]) -> TrainContext:
@@ -62,32 +62,33 @@ class Trainer:
try: try:
context.model.train() context.model.train()
# 1.epoch accumulation_steps = max(self.train_config.accumulation_steps, 1)
for epoch in range(context.epoch, self.train_config.n_epoch): for epoch in range(context.epoch, self.train_config.n_epoch):
context.epoch = epoch context.epoch = epoch
self._call_callbacks("on_epoch_begin", context) self._call_callbacks("on_epoch_begin", context)
accumulation_steps = max(self.train_config.accumulation_steps, 1) for steps in batched(context.dataloader, accumulation_steps):
for batch in context.dataloader:
if context.iteration % accumulation_steps == 0:
# 2. step
self._call_callbacks("on_step_begin", context) self._call_callbacks("on_step_begin", context)
context.optimizer.step()
context.optimizer.zero_grad()
self._call_callbacks("on_step_end", context)
# 3. batch step_batch_nums = len(steps)
for batch in steps:
self._call_callbacks("on_batch_begin", context) self._call_callbacks("on_batch_begin", context)
loss = context.strategy(batch) loss = context.strategy(batch)
context.loss = loss.item() context.loss = loss.item()
context.iteration += 1 context.iteration += 1
# to make the loss normalized by accumulation steps stand_loss = loss / step_batch_nums
stand_loss = loss / accumulation_steps
stand_loss.backward() stand_loss.backward()
self._call_callbacks("on_batch_end", context) self._call_callbacks("on_batch_end", context)
self._call_callbacks("on_step_end", context)
context.optimizer.step()
context.optimizer.zero_grad()
if context.scheduler:
context.scheduler.step()
self._call_callbacks("on_epoch_end", context) self._call_callbacks("on_epoch_end", context)
except Exception as e: except Exception as e:
+6 -2
View File
@@ -155,18 +155,20 @@ def parse_args() -> argparse.Namespace:
def ddp_wrap(model: nn.Module): def ddp_wrap(model: nn.Module):
local_rank = get_rank() local_rank = get_rank()
model = model.to(dtype=torch.bfloat16)
ddp_model = DDP( ddp_model = DDP(
model, model,
device_ids=[local_rank], device_ids=[local_rank],
output_device=local_rank, output_device=local_rank,
static_graph=True,
find_unused_parameters=False, find_unused_parameters=False,
gradient_as_bucket_view=True,
broadcast_buffers=False,
) )
return ddp_model return ddp_model
def create_optimizer(model: nn.Module, **kwargs) -> optim.Optimizer: def create_optimizer(model: nn.Module, **kwargs) -> optim.Optimizer:
return optim.AdamW(model.parameters(), **kwargs) return optim.AdamW(model.parameters(), fused=True, **kwargs)
def create_scheduler( def create_scheduler(
@@ -231,6 +233,8 @@ def train(
state_dict = st.load_file(weights_path) state_dict = st.load_file(weights_path)
model.load_state_dict(state_dict, strict=False) model.load_state_dict(state_dict, strict=False)
model = model.to(dtype=torch.bfloat16)
strategy_kwargs = { strategy_kwargs = {
"dpo_beta": dpo_beta, "dpo_beta": dpo_beta,
"label_smoothing": label_smoothing, "label_smoothing": label_smoothing,