Compare commits

...

3 Commits

Author SHA1 Message Date
Walle1999 701601681a
Merge 450885c0d5 into eabad706da 2026-08-21 16:26:34 +08:00
jingyaogong eabad706da [update] readme content 2026-08-21 16:25:20 +08:00
shiguoliang 450885c0d5 feat: 添加 macOS Metal (MPS) 设备支持
通过条件分支实现 CUDA/MPS/CPU 三端兼容:
- 新增 get_default_device() 自动检测最优设备
- 混合精度、GradScaler、DDP 按设备类型条件启用
- CUDA 环境行为完全不变,MPS 作为新增分支

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-06 10:54:20 +08:00
12 changed files with 535 additions and 414 deletions

657
README.md

File diff suppressed because it is too large Load Diff

View File

@ -913,7 +913,17 @@ Before introducing the implementation of specific algorithms, let me first descr
The essence of all RL algorithms is just optimizing an expectation: The essence of all RL algorithms is just optimizing an expectation:
$$\mathcal{J}_{PO} = \mathbb{E}_{q \sim P(Q), o \sim \pi(O|q)} \left[ \underbrace{f(r_t)}_{\text{policy term}} \cdot \underbrace{g(A_t)}_{\text{advantage term}} - \underbrace{h(\text{KL}_t)}_{\text{regularization term}} \right]$$ $$
\mathcal{J}_{PO} =
\mathbb{E}_{q \sim P(Q),\, o \sim \pi(O \mid q)}
\left[
\underbrace{f(r_t)}_{\text{policy term}}
\cdot
\underbrace{g(A_t)}_{\text{advantage term}}
-
\underbrace{h(\text{KL}_t)}_{\text{regularization term}}
\right]
$$
During training, one only needs to **minimize the negative objective function**, i.e.: $\mathcal{L}_{PO} = -\mathcal{J}_{PO}$ During training, one only needs to **minimize the negative objective function**, i.e.: $\mathcal{L}_{PO} = -\mathcal{J}_{PO}$
@ -929,7 +939,7 @@ This framework contains only three core components:
|--------|---------|-------------|-------| |--------|---------|-------------|-------|
| $q$ | Question/Prompt | Sampled from dataset $P(Q)$ | - | | $q$ | Question/Prompt | Sampled from dataset $P(Q)$ | - |
| $o$ | Model output sequence | Generated by policy $\pi$ | - | | $o$ | Model output sequence | Generated by policy $\pi$ | - |
| $r_t$ | Probability ratio | $r_t = \frac{\pi_\theta(o_t \mid q, o_{<t})}{\pi_{ref}(o_t \mid q, o_{<t})}$ | $(0, +\infty)$ | | $r_t$ | Probability ratio | $r_t(\theta) = \frac{\pi_\theta(o_t \mid q, o_{<t})}{\pi_{\mathrm{old}}(o_t \mid q, o_{<t})}$ | $(0, +\infty)$ |
| $A_t$ | Advantage function | Measures how much better a certain action is compared to the baseline | $(-\infty, +\infty)$ | | $A_t$ | Advantage function | Measures how much better a certain action is compared to the baseline | $(-\infty, +\infty)$ |
| $\text{KL}_t$ | KL divergence | Prevents the policy from drifting too far from the reference model | $[0, +\infty)$ | | $\text{KL}_t$ | KL divergence | Prevents the policy from drifting too far from the reference model | $[0, +\infty)$ |
@ -1084,9 +1094,9 @@ PPO is a very classic reinforcement learning algorithm proposed by OpenAI in 201
$$\mathcal{L}_{PPO} = -\mathbb{E}\left[\min(r_t \cdot A_t, \text{clip}(r_t, 1-\varepsilon, 1+\varepsilon) \cdot A_t)\right] + \beta \cdot \mathbb{E}[\text{KL}]$$ $$\mathcal{L}_{PPO} = -\mathbb{E}\left[\min(r_t \cdot A_t, \text{clip}(r_t, 1-\varepsilon, 1+\varepsilon) \cdot A_t)\right] + \beta \cdot \mathbb{E}[\text{KL}]$$
Where: Where:
- **Policy term**: $f(r_t) = \min(r_t, \text{clip}(r_t, 1-\varepsilon, 1+\varepsilon))$ (clips probability ratio to prevent overly aggressive updates) - **Policy term**: $\min\left(r_t A_t,\ \mathrm{clip}(r_t, 1-\varepsilon, 1+\varepsilon)A_t\right)$ (clips the probability ratio and advantage together to prevent overly aggressive updates)
- **Advantage term**: $g(A_t) = R - V(s)$ (estimates value function through Critic network) - **Advantage term**: $A_t$ (usually estimated by the Critic, or computed with GAE)
- **Regularization term**: $h(\text{KL}_t) = \beta \cdot \mathbb{E}[\text{KL}]$ (global KL divergence constraint) - **Regularization term**: $\beta \cdot \mathbb{E}[\text{KL}]$ (global KL divergence constraint)
Compared to DPO, Compared to DPO,
- DPO (Off-Policy): Training data consists of static preference pairs (chosen vs rejected), which can be reused across multiple training epochs, like traditional supervised learning. High data efficiency, low cost, and no Reward Model needed. - DPO (Off-Policy): Training data consists of static preference pairs (chosen vs rejected), which can be reused across multiple training epochs, like traditional supervised learning. High data efficiency, low cost, and no Reward Model needed.
@ -1126,9 +1136,9 @@ In early 2025, as DeepSeek-R1 exploded in popularity, GRPO from the DeepSeekMath
$$\mathcal{L}_{GRPO} = -\mathbb{E}\left[\min(r_t \cdot A_t, \mathrm{clip}(r_t, 1-\varepsilon, 1+\varepsilon) \cdot A_t) - \beta \cdot \text{KL}_t\right]$$ $$\mathcal{L}_{GRPO} = -\mathbb{E}\left[\min(r_t \cdot A_t, \mathrm{clip}(r_t, 1-\varepsilon, 1+\varepsilon) \cdot A_t) - \beta \cdot \text{KL}_t\right]$$
Where: Where:
- **Policy term**: $f(r_t) = \min(r_t, \mathrm{clip}(r_t, 1-\varepsilon, 1+\varepsilon))$ (uses symmetric clip on the probability ratio) - **Policy term**: $\min\left(r_t A_t,\ \mathrm{clip}(r_t, 1-\varepsilon, 1+\varepsilon)A_t\right)$ (uses symmetric clip on the probability ratio)
- **Advantage term**: $g(A_t) = \frac{R - \mu_{group}}{\sigma_{group}}$ (intra-group normalization, eliminating the Critic network) - **Advantage term**: $A_{i,j} = \frac{R_{i,j} - \mu_i}{\sigma_i + \epsilon}$ (intra-group normalization, without training an additional Critic)
- **Regularization term**: $h(\text{KL}_t) = \beta \cdot \text{KL}_t$ (token-level KL divergence constraint) - **Regularization term**: $\beta \cdot \text{KL}_t$ (token-level KL divergence constraint)
For the same question, the model generates N responses and computes their respective rewards, then uses the intra-group average reward as a baseline. Responses above the baseline are encouraged, and responses below the baseline are suppressed, thus no additional critic network needs to be trained. For the same question, the model generates N responses and computes their respective rewards, then uses the intra-group average reward as a baseline. Responses above the baseline are encouraged, and responses below the baseline are suppressed, thus no additional critic network needs to be trained.
@ -1160,11 +1170,11 @@ CISPO does not focus on redesigning the group baseline. Instead, it uses a very
**CISPO Loss**: **CISPO Loss**:
$$\mathcal{L}_{CISPO} = -\mathbb{E}\left[\min(r_t, \varepsilon_{max}) \cdot A_t \cdot \log \pi_\theta(a_t|s) - \beta \cdot \text{KL}_t\right]$$ $$\mathcal{L}_{CISPO} = -\mathbb{E}\left[\min(r_t, \varepsilon_{\mathrm{high}}) \cdot A_t \cdot \log \pi_\theta(a_t|s) - \beta \cdot \text{KL}_t\right]$$
Where: Where:
- **Policy term**: $f(r_t) = \min(r_t, \varepsilon_{max}) \cdot \log \pi_\theta(a_t|s)$ (ratio serves only as a clipped weight) - **Policy term**: $f(r_t) = \min(r_t, \varepsilon_{\mathrm{high}}) \cdot \log \pi_\theta(a_t|s)$ (the ratio serves only as a clipped weight)
- **Advantage term**: $g(A_t) = \frac{R - \mu_{group}}{\sigma_{group}}$ (can directly reuse GRPO's intra-group relative advantage) - **Advantage term**: $g(A_t) = \frac{R - \mu_{group}}{\sigma_{group} + \epsilon}$ (can directly reuse GRPO's intra-group relative advantage)
- **Regularization term**: $h(\text{KL}_t) = \beta \cdot \text{KL}_t$ (token-level KL divergence constraint) - **Regularization term**: $h(\text{KL}_t) = \beta \cdot \text{KL}_t$ (token-level KL divergence constraint)
CISPO, building on GRPO, rewrites the policy term that was easily clipped into a constant into the form "clipped weight × log probability". This way, even if the ratio is truncated, the gradient path is not truncated along with it. Therefore, CISPO can be directly viewed as a loss variant of GRPO to implement, rather than maintaining a separate standalone script. No separate experiment is listed here. One only needs to set `loss_type` to `cispo` in `train_grpo.py`; the rest of the training process still follows GRPO's group sampling, reward computation, and advantage construction logic. CISPO, building on GRPO, rewrites the policy term that was easily clipped into a constant into the form "clipped weight × log probability". This way, even if the ratio is truncated, the gradient path is not truncated along with it. Therefore, CISPO can be directly viewed as a loss variant of GRPO to implement, rather than maintaining a separate standalone script. No separate experiment is listed here. One only needs to set `loss_type` to `cispo` in `train_grpo.py`; the rest of the training process still follows GRPO's group sampling, reward computation, and advantage construction logic.
@ -1269,9 +1279,9 @@ Returning to the "**unified framework**", the table below summarizes how differe
| Algorithm | Policy term $f(r_t)$ | Advantage term $g(A_t)$ | Regularization term $h(\text{KL}_t)$ | Number of training models | | Algorithm | Policy term $f(r_t)$ | Advantage term $g(A_t)$ | Regularization term $h(\text{KL}_t)$ | Number of training models |
|-----------|---------------------|------------------------|-------------------------------------|--------------------------| |-----------|---------------------|------------------------|-------------------------------------|--------------------------|
| **DPO** | $\log r_w - \log r_l$ | No explicit advantage term | Implicit in $\beta$ | 1 (2 participate in forward) | | **DPO** | $\log r_w - \log r_l$ | No explicit advantage term | Implicit in $\beta$ | 1 (2 participate in forward) |
| **PPO** | $\min(r, \text{clip}(r))$ | $R - V(s)$ | $\beta \cdot \mathbb{E}[\text{KL}]$ | 2 | | **PPO** | $\min(r_t A_t, \mathrm{clip}(r_t, 1-\varepsilon, 1+\varepsilon)A_t)$ | $A_t$ (typically estimated by Critic / GAE) | $\beta \cdot \mathbb{E}[\text{KL}]$ | 2 |
| **GRPO** | $\min(r, \text{clip}(r))$ | $\frac{R - \mu}{\sigma}$ | $\beta \cdot \text{KL}_t$ | 1 | | **GRPO** | $\min(r_t A_t, \mathrm{clip}(r_t, 1-\varepsilon, 1+\varepsilon)A_t)$ | $\frac{R - \mu}{\sigma + \epsilon}$ | $\beta \cdot \text{KL}_t$ | 1 |
| **CISPO** | $\mathrm{clip}(r, 0, \varepsilon_{max}) \cdot A_t \cdot \log \pi_\theta$ | $\frac{R - \mu}{\sigma}$ | $\beta \cdot \text{KL}_t$ | 1 | | **CISPO** | $\min(r_t, \varepsilon_{\mathrm{high}}) \cdot A_t \cdot \log \pi_\theta$ | $\frac{R - \mu}{\sigma + \epsilon}$ | $\beta \cdot \text{KL}_t$ | 1 |
**To put it plainly, these RL algorithms are not isolated from one another. From a unified optimization perspective, they are natural variants formed by making different design trade-offs on the same objective function, presenting a beautifully self-consistent unity.** **To put it plainly, these RL algorithms are not isolated from one another. From a unified optimization perspective, they are natural variants formed by making different design trade-offs on the same objective function, presenting a beautifully self-consistent unity.**

View File

@ -6,7 +6,7 @@ import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, TextStreamer from transformers import AutoTokenizer, AutoModelForCausalLM, TextStreamer
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
from model.model_lora import * from model.model_lora import *
from trainer.trainer_utils import setup_seed, get_model_params from trainer.trainer_utils import setup_seed, get_model_params, get_default_device
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
def init_model(args): def init_model(args):
@ -27,7 +27,9 @@ def init_model(args):
else: else:
model = AutoModelForCausalLM.from_pretrained(args.load_from, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained(args.load_from, trust_remote_code=True)
get_model_params(model, model.config) get_model_params(model, model.config)
return model.half().eval().to(args.device), tokenizer if 'cuda' in args.device:
model = model.half()
return model.eval().to(args.device), tokenizer
def main(): def main():
parser = argparse.ArgumentParser(description="MiniMind模型推理与对话") parser = argparse.ArgumentParser(description="MiniMind模型推理与对话")
@ -45,7 +47,7 @@ def main():
parser.add_argument('--open_thinking', default=0, type=int, help="是否开启自适应思考(0=否,1=是)") parser.add_argument('--open_thinking', default=0, type=int, help="是否开启自适应思考(0=否,1=是)")
parser.add_argument('--historys', default=0, type=int, help="携带历史对话轮数(需为偶数,0表示不携带历史)") parser.add_argument('--historys', default=0, type=int, help="携带历史对话轮数(需为偶数,0表示不携带历史)")
parser.add_argument('--show_speed', default=1, type=int, help="显示decode速度(tokens/s)") parser.add_argument('--show_speed', default=1, type=int, help="显示decode速度(tokens/s)")
parser.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu', type=str, help="运行设备") parser.add_argument('--device', default=get_default_device(), type=str, help="运行设备")
args = parser.parse_args() args = parser.parse_args()
prompts = [ prompts = [

View File

@ -24,7 +24,7 @@ from torch.optim.lr_scheduler import CosineAnnealingLR
from transformers import AutoTokenizer from transformers import AutoTokenizer
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
from dataset.lm_dataset import AgentRLDataset from dataset.lm_dataset import AgentRLDataset
from trainer.trainer_utils import Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, SkipBatchSampler, init_model, LMForRewardModel from trainer.trainer_utils import Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, SkipBatchSampler, init_model, LMForRewardModel, get_default_device
from trainer.rollout_engine import create_rollout_engine, compute_per_token_logps from trainer.rollout_engine import create_rollout_engine, compute_per_token_logps
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -378,7 +378,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=1, help="训练轮数") parser.add_argument("--epochs", type=int, default=1, help="训练轮数")
parser.add_argument("--batch_size", type=int, default=2, help="批次大小") parser.add_argument("--batch_size", type=int, default=2, help="批次大小")
parser.add_argument("--learning_rate", type=float, default=3e-7, help="学习率") parser.add_argument("--learning_rate", type=float, default=3e-7, help="学习率")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="数据类型 bfloat16/float16") parser.add_argument("--dtype", type=str, default="bfloat16", help="数据类型 bfloat16/float16")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -421,9 +421,14 @@ if __name__ == "__main__":
max_seq_len=args.max_seq_len + args.max_gen_len, use_moe=bool(args.use_moe)) max_seq_len=args.max_seq_len + args.max_gen_len, use_moe=bool(args.use_moe))
ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume == 1 else None ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume == 1 else None
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
wandb = None wandb = None
if args.use_wandb and is_main_process(): if args.use_wandb and is_main_process():
@ -480,7 +485,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True, collate_fn=collate_fn) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"), collate_fn=collate_fn)
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch+1}/{args.epochs}]: skip {start_step} steps') Logger(f'Epoch [{epoch+1}/{args.epochs}]: skip {start_step} steps')
rl_train_epoch(epoch, loader, len(loader) + skip, rollout_engine, ref_model, reward_model, start_step, wandb, use_sglang = (args.rollout_engine == "sglang")) rl_train_epoch(epoch, loader, len(loader) + skip, rollout_engine, ref_model, reward_model, start_step, wandb, use_sglang = (args.rollout_engine == "sglang"))

View File

@ -17,7 +17,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig from model.model_minimind import MiniMindConfig
from dataset.lm_dataset import SFTDataset from dataset.lm_dataset import SFTDataset
from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler, get_default_device
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -151,7 +151,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=6, help="训练轮数") parser.add_argument("--epochs", type=int, default=6, help="训练轮数")
parser.add_argument("--batch_size", type=int, default=32, help="batch size") parser.add_argument("--batch_size", type=int, default=32, help="batch size")
parser.add_argument("--learning_rate", type=float, default=5e-6, help="初始学习率") parser.add_argument("--learning_rate", type=float, default=5e-6, help="初始学习率")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型") parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -188,9 +188,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config_student, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None ckp_data = lm_checkpoint(lm_config_student, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 3. 设置混合精度 ========== # ========== 3. 设置混合精度 ==========
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
# ========== 4. 配wandb ========== # ========== 4. 配wandb ==========
wandb = None wandb = None
@ -210,7 +215,7 @@ if __name__ == "__main__":
Logger(f'教师模型总参数量:{sum(p.numel() for p in teacher_model.parameters()) / 1e6:.3f} M') Logger(f'教师模型总参数量:{sum(p.numel() for p in teacher_model.parameters()) / 1e6:.3f} M')
train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len) train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len)
train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None
scaler = torch.cuda.amp.GradScaler(enabled=(args.dtype == 'float16')) scaler = torch.cuda.amp.GradScaler(enabled=(device_type == "cuda" and args.dtype == "float16"))
optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate) optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate)
# ========== 6. 从ckp恢复状态 ========== # ========== 6. 从ckp恢复状态 ==========
@ -235,7 +240,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"))
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始') Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
train_epoch(epoch, loader, len(loader) + skip, teacher_model, lm_config_student, start_step, wandb, args.alpha, args.temperature) train_epoch(epoch, loader, len(loader) + skip, teacher_model, lm_config_student, start_step, wandb, args.alpha, args.temperature)

View File

@ -17,7 +17,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig from model.model_minimind import MiniMindConfig
from dataset.lm_dataset import DPODataset from dataset.lm_dataset import DPODataset
from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler, get_default_device
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -135,7 +135,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=1, help="训练轮数") parser.add_argument("--epochs", type=int, default=1, help="训练轮数")
parser.add_argument("--batch_size", type=int, default=4, help="batch size") parser.add_argument("--batch_size", type=int, default=4, help="batch size")
parser.add_argument("--learning_rate", type=float, default=4e-8, help="初始学习率(建议<=5e-8避免遗忘)") parser.add_argument("--learning_rate", type=float, default=4e-8, help="初始学习率(建议<=5e-8避免遗忘)")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型") parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -166,9 +166,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 3. 设置混合精度 ========== # ========== 3. 设置混合精度 ==========
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
# ========== 4. 配wandb ========== # ========== 4. 配wandb ==========
wandb = None wandb = None
@ -190,7 +195,7 @@ if __name__ == "__main__":
train_ds = DPODataset(args.data_path, tokenizer, max_length=args.max_seq_len) train_ds = DPODataset(args.data_path, tokenizer, max_length=args.max_seq_len)
train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None
scaler = torch.cuda.amp.GradScaler(enabled=(args.dtype == 'float16')) scaler = torch.cuda.amp.GradScaler(enabled=(device_type == "cuda" and args.dtype == "float16"))
optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate) optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate)
# ========== 6. 从ckp恢复状态 ========== # ========== 6. 从ckp恢复状态 ==========
@ -215,7 +220,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"))
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始') Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
train_epoch(epoch, loader, len(loader) + skip, ref_model, lm_config, start_step, wandb, args.beta) train_epoch(epoch, loader, len(loader) + skip, ref_model, lm_config, start_step, wandb, args.beta)

View File

@ -16,7 +16,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig from model.model_minimind import MiniMindConfig
from dataset.lm_dataset import SFTDataset from dataset.lm_dataset import SFTDataset
from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler, get_default_device
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -88,7 +88,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=2, help="训练轮数") parser.add_argument("--epochs", type=int, default=2, help="训练轮数")
parser.add_argument("--batch_size", type=int, default=16, help="batch size") parser.add_argument("--batch_size", type=int, default=16, help="batch size")
parser.add_argument("--learning_rate", type=float, default=1e-5, help="初始学习率") parser.add_argument("--learning_rate", type=float, default=1e-5, help="初始学习率")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型") parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -118,9 +118,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 3. 设置混合精度 ========== # ========== 3. 设置混合精度 ==========
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
# ========== 4. 配wandb ========== # ========== 4. 配wandb ==========
wandb = None wandb = None
@ -135,7 +140,7 @@ if __name__ == "__main__":
model, tokenizer = init_model(lm_config, args.from_weight, device=args.device) model, tokenizer = init_model(lm_config, args.from_weight, device=args.device)
train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len) train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len)
train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None
scaler = torch.cuda.amp.GradScaler(enabled=(args.dtype == 'float16')) scaler = torch.cuda.amp.GradScaler(enabled=(device_type == "cuda" and args.dtype == "float16"))
optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate) optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate)
# ========== 6. 从ckp恢复状态 ========== # ========== 6. 从ckp恢复状态 ==========
@ -160,7 +165,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"))
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始') Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
train_epoch(epoch, loader, len(loader) + skip, start_step, wandb) train_epoch(epoch, loader, len(loader) + skip, start_step, wandb)

View File

@ -22,7 +22,7 @@ from torch.optim.lr_scheduler import CosineAnnealingLR
from transformers import AutoModel from transformers import AutoModel
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
from dataset.lm_dataset import RLAIFDataset from dataset.lm_dataset import RLAIFDataset
from trainer.trainer_utils import Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, SkipBatchSampler, init_model, LMForRewardModel from trainer.trainer_utils import Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, SkipBatchSampler, init_model, LMForRewardModel, get_default_device
from trainer.rollout_engine import create_rollout_engine from trainer.rollout_engine import create_rollout_engine
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -210,7 +210,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=1, help="训练轮数") parser.add_argument("--epochs", type=int, default=1, help="训练轮数")
parser.add_argument("--batch_size", type=int, default=2, help="batch size") parser.add_argument("--batch_size", type=int, default=2, help="batch size")
parser.add_argument("--learning_rate", type=float, default=3e-7, help="初始学习率") parser.add_argument("--learning_rate", type=float, default=3e-7, help="初始学习率")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型") parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -255,9 +255,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 3. 设置混合精度 ========== # ========== 3. 设置混合精度 ==========
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
# ========== 4. 配wandb ========== # ========== 4. 配wandb ==========
wandb = None wandb = None
@ -321,7 +326,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"))
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始') Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
grpo_train_epoch(epoch, loader, len(loader) + skip, rollout_engine, ref_model, reward_model, start_step, wandb, use_sglang = (args.rollout_engine == "sglang")) grpo_train_epoch(epoch, loader, len(loader) + skip, rollout_engine, ref_model, reward_model, start_step, wandb, use_sglang = (args.rollout_engine == "sglang"))

View File

@ -17,7 +17,7 @@ from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig from model.model_minimind import MiniMindConfig
from dataset.lm_dataset import SFTDataset from dataset.lm_dataset import SFTDataset
from model.model_lora import save_lora, apply_lora from model.model_lora import save_lora, apply_lora
from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler, get_default_device
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -82,7 +82,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=10, help="训练轮数") parser.add_argument("--epochs", type=int, default=10, help="训练轮数")
parser.add_argument("--batch_size", type=int, default=32, help="batch size") parser.add_argument("--batch_size", type=int, default=32, help="batch size")
parser.add_argument("--learning_rate", type=float, default=1e-4, help="初始学习率") parser.add_argument("--learning_rate", type=float, default=1e-4, help="初始学习率")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型") parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -112,9 +112,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config, weight=args.lora_name, save_dir='../checkpoints') if args.from_resume==1 else None ckp_data = lm_checkpoint(lm_config, weight=args.lora_name, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 3. 设置混合精度 ========== # ========== 3. 设置混合精度 ==========
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
# ========== 4. 配wandb ========== # ========== 4. 配wandb ==========
wandb = None wandb = None
@ -148,7 +153,7 @@ if __name__ == "__main__":
# ========== 6. 定义数据和优化器 ========== # ========== 6. 定义数据和优化器 ==========
train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len) train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len)
train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None
scaler = torch.cuda.amp.GradScaler(enabled=(args.dtype == 'float16')) scaler = torch.cuda.amp.GradScaler(enabled=(device_type == "cuda" and args.dtype == "float16"))
optimizer = optim.AdamW(lora_params, lr=args.learning_rate) optimizer = optim.AdamW(lora_params, lr=args.learning_rate)
# ========== 7. 从ckp恢复状态 ========== # ========== 7. 从ckp恢复状态 ==========
@ -173,7 +178,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"))
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始') Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
train_epoch(epoch, loader, len(loader) + skip, lora_params, start_step, wandb) train_epoch(epoch, loader, len(loader) + skip, lora_params, start_step, wandb)

View File

@ -20,7 +20,7 @@ from torch.nn.utils import clip_grad_norm_
from torch.optim.lr_scheduler import CosineAnnealingLR from torch.optim.lr_scheduler import CosineAnnealingLR
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
from dataset.lm_dataset import RLAIFDataset from dataset.lm_dataset import RLAIFDataset
from trainer.trainer_utils import Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, SkipBatchSampler, init_model, LMForRewardModel from trainer.trainer_utils import Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, SkipBatchSampler, init_model, LMForRewardModel, get_default_device
from trainer.rollout_engine import create_rollout_engine from trainer.rollout_engine import create_rollout_engine
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -314,7 +314,7 @@ if __name__ == "__main__":
parser.add_argument("--batch_size", type=int, default=2, help="batch size") parser.add_argument("--batch_size", type=int, default=2, help="batch size")
parser.add_argument("--learning_rate", type=float, default=3e-7, help="Actor学习率") parser.add_argument("--learning_rate", type=float, default=3e-7, help="Actor学习率")
parser.add_argument("--critic_learning_rate", type=float, default=5e-7, help="Critic学习率") parser.add_argument("--critic_learning_rate", type=float, default=5e-7, help="Critic学习率")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型") parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -363,9 +363,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 3. 设置混合精度 ========== # ========== 3. 设置混合精度 ==========
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
# ========== 4. 配wandb ========== # ========== 4. 配wandb ==========
wandb = None wandb = None
@ -438,7 +443,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"))
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始') Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
ppo_train_epoch(epoch, loader, len(loader) + skip, rollout_engine, ref_model, actor_scheduler, critic_scheduler, reward_model, start_step, wandb) ppo_train_epoch(epoch, loader, len(loader) + skip, rollout_engine, ref_model, actor_scheduler, critic_scheduler, reward_model, start_step, wandb)

View File

@ -16,7 +16,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig from model.model_minimind import MiniMindConfig
from dataset.lm_dataset import PretrainDataset from dataset.lm_dataset import PretrainDataset
from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler from trainer.trainer_utils import get_lr, Logger, is_main_process, lm_checkpoint, init_distributed_mode, setup_seed, init_model, SkipBatchSampler, get_default_device
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
@ -87,7 +87,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=2, help="训练轮数") parser.add_argument("--epochs", type=int, default=2, help="训练轮数")
parser.add_argument("--batch_size", type=int, default=32, help="batch size") parser.add_argument("--batch_size", type=int, default=32, help="batch size")
parser.add_argument("--learning_rate", type=float, default=5e-4, help="初始学习率") parser.add_argument("--learning_rate", type=float, default=5e-4, help="初始学习率")
parser.add_argument("--device", type=str, default="cuda:0" if torch.cuda.is_available() else "cpu", help="训练设备") parser.add_argument("--device", type=str, default=get_default_device(), help="训练设备")
parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型") parser.add_argument("--dtype", type=str, default="bfloat16", help="混合精度类型")
parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数") parser.add_argument("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=8, help="梯度累积步数") parser.add_argument("--accumulation_steps", type=int, default=8, help="梯度累积步数")
@ -117,9 +117,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 3. 设置混合精度 ========== # ========== 3. 设置混合精度 ==========
device_type = "cuda" if "cuda" in args.device else "cpu" device_type = "cuda" if "cuda" in args.device else ("mps" if "mps" in args.device else "cpu")
dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16 dtype = torch.bfloat16 if args.dtype == "bfloat16" else torch.float16
autocast_ctx = nullcontext() if device_type == "cpu" else torch.cuda.amp.autocast(dtype=dtype) if device_type == "cuda":
autocast_ctx = torch.cuda.amp.autocast(dtype=dtype)
elif device_type == "mps":
autocast_ctx = torch.amp.autocast("mps", dtype=dtype)
else:
autocast_ctx = nullcontext()
# ========== 4. 配wandb ========== # ========== 4. 配wandb ==========
wandb = None wandb = None
@ -134,7 +139,7 @@ if __name__ == "__main__":
model, tokenizer = init_model(lm_config, args.from_weight, device=args.device) model, tokenizer = init_model(lm_config, args.from_weight, device=args.device)
train_ds = PretrainDataset(args.data_path, tokenizer, max_length=args.max_seq_len) train_ds = PretrainDataset(args.data_path, tokenizer, max_length=args.max_seq_len)
train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None train_sampler = DistributedSampler(train_ds) if dist.is_initialized() else None
scaler = torch.cuda.amp.GradScaler(enabled=(args.dtype == 'float16')) scaler = torch.cuda.amp.GradScaler(enabled=(device_type == "cuda" and args.dtype == "float16"))
optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate) optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate)
# ========== 6. 从ckp恢复状态 ========== # ========== 6. 从ckp恢复状态 ==========
@ -159,7 +164,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist() setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0 skip = start_step if (epoch == start_epoch and start_step > 0) else 0
batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip) batch_sampler = SkipBatchSampler(train_sampler or indices, args.batch_size, skip)
loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=True) loader = DataLoader(train_ds, batch_sampler=batch_sampler, num_workers=args.num_workers, pin_memory=(device_type == "cuda"))
if skip > 0: if skip > 0:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始') Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
train_epoch(epoch, loader, len(loader) + skip, start_step, wandb) train_epoch(epoch, loader, len(loader) + skip, start_step, wandb)

View File

@ -15,6 +15,15 @@ from torch.utils.data import Sampler
from transformers import AutoTokenizer, AutoModel, AutoModelForSequenceClassification from transformers import AutoTokenizer, AutoModel, AutoModelForSequenceClassification
from model.model_minimind import MiniMindForCausalLM from model.model_minimind import MiniMindForCausalLM
def get_default_device():
"""运行时自动选择最优设备,CUDA > MPS > CPU"""
if torch.cuda.is_available():
return "cuda"
elif hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
return "mps"
return "cpu"
def get_model_params(model, config): def get_model_params(model, config):
total = sum(p.numel() for p in model.parameters()) / 1e6 total = sum(p.numel() for p in model.parameters()) / 1e6
n_routed = getattr(config, 'n_routed_experts', getattr(config, 'num_experts', 0)) n_routed = getattr(config, 'n_routed_experts', getattr(config, 'num_experts', 0))
@ -45,6 +54,9 @@ def init_distributed_mode():
if int(os.environ.get("RANK", -1)) == -1: if int(os.environ.get("RANK", -1)) == -1:
return 0 # 非DDP模式 return 0 # 非DDP模式
if not torch.cuda.is_available():
return 0 # MPS/CPU 不支持 NCCL DDP,静默跳过
dist.init_process_group(backend="nccl") dist.init_process_group(backend="nccl")
local_rank = int(os.environ["LOCAL_RANK"]) local_rank = int(os.environ["LOCAL_RANK"])
torch.cuda.set_device(local_rank) torch.cuda.set_device(local_rank)
@ -55,6 +67,7 @@ def setup_seed(seed: int):
random.seed(seed) random.seed(seed)
np.random.seed(seed) np.random.seed(seed)
torch.manual_seed(seed) torch.manual_seed(seed)
if torch.cuda.is_available():
torch.cuda.manual_seed(seed) torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed) torch.cuda.manual_seed_all(seed)
torch.backends.cudnn.deterministic = True torch.backends.cudnn.deterministic = True
@ -103,7 +116,10 @@ def lm_checkpoint(lm_config, weight='full_sft', model=None, optimizer=None, epoc
torch.save(resume_data, resume_tmp) torch.save(resume_data, resume_tmp)
os.replace(resume_tmp, resume_path) os.replace(resume_tmp, resume_path)
del state_dict, resume_data del state_dict, resume_data
if torch.cuda.is_available():
torch.cuda.empty_cache() torch.cuda.empty_cache()
elif hasattr(torch, "mps") and torch.backends.mps.is_available():
torch.mps.empty_cache()
else: # 加载模式 else: # 加载模式
if os.path.exists(resume_path): if os.path.exists(resume_path):
ckp_data = torch.load(resume_path, map_location='cpu') ckp_data = torch.load(resume_path, map_location='cpu')
@ -116,7 +132,9 @@ def lm_checkpoint(lm_config, weight='full_sft', model=None, optimizer=None, epoc
return None return None
def init_model(lm_config, from_weight='pretrain', tokenizer_path='../model', save_dir='../out', device='cuda'): def init_model(lm_config, from_weight='pretrain', tokenizer_path='../model', save_dir='../out', device=None):
if device is None:
device = get_default_device()
tokenizer = AutoTokenizer.from_pretrained(tokenizer_path) tokenizer = AutoTokenizer.from_pretrained(tokenizer_path)
model = MiniMindForCausalLM(lm_config) model = MiniMindForCausalLM(lm_config)
@ -158,7 +176,9 @@ class SkipBatchSampler(Sampler):
class LMForRewardModel: class LMForRewardModel:
def __init__(self, model_path, device="cuda", dtype=torch.float16): def __init__(self, model_path, device=None, dtype=torch.float16):
if device is None:
device = get_default_device()
self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True) self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
self.model = AutoModel.from_pretrained(model_path, torch_dtype=dtype, trust_remote_code=True) self.model = AutoModel.from_pretrained(model_path, torch_dtype=dtype, trust_remote_code=True)
self.model = self.model.to(device).eval() self.model = self.model.to(device).eval()