Compare commits

...

4 Commits

Author SHA1 Message Date
Walle1999 7feae82da3
Merge f6f56f0a7d into 512eed0b65 2026-06-28 18:03:49 +08:00
jingyaogong 512eed0b65 [fix] robustness 2026-06-28 18:03:34 +08:00
Walle1999 f6f56f0a7d
Merge pull request #1 from shiguol/develop/mac
feat: 添加 macOS Metal (MPS) 设备支持
2026-05-25 10:15:06 +08:00
shiguoliang 2d829fe29f feat: 添加 macOS Metal (MPS) 设备支持
通过条件分支实现 CUDA/MPS/CPU 三端兼容:
- 新增 get_default_device() 自动检测最优设备
- 混合精度、GradScaler、DDP 按设备类型条件启用
- CUDA 环境行为完全不变,MPS 作为新增分支

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-24 10:52:56 +08:00
15 changed files with 155 additions and 80 deletions

View File

@ -765,16 +765,16 @@ LoRA 是一种常见的参数高效微调(Parameter-Efficient Fine-Tuning, PEF
```bash
# train_lora.py 在 CPU 上通常也能比较轻快地完成
# 方式1
torchrun --nproc_per_node 1 train_lora.py
cd trainer && torchrun --nproc_per_node 1 train_lora.py
# 方式2
python train_lora.py
cd trainer && python train_lora.py
```
> 训练后的模型权重文件默认每隔`save_interval步`保存为: `lora_xxx_*.pth`(*为模型具体dimension,每次保存时新文件会覆盖旧文件)
LoRA 很适合处理“如何在尽量保留通用能力的前提下,让模型快速适应私有领域或垂直场景”这类问题。例如基础模型医学知识不足时,就可以在原有模型之上叠加一层面向医疗场景的 LoRA 权重,以较小代价获得更好的领域表现。
通常只需要准备同样的多轮对话格式数据,放置到 `lora_xxx.jsonl`,再执行 `python train_lora.py`,即可得到新的 `LoRA` 模型权重。
通常只需要准备同样的多轮对话格式数据,放置到 `lora_xxx.jsonl`,再从仓库根目录执行 `cd trainer && python train_lora.py`,即可得到新的 `LoRA` 模型权重。
例1:垂域数据

View File

@ -765,16 +765,16 @@ Its core idea is to introduce low-rank incremental branches alongside the origin
```bash
# train_lora.py can usually be completed fairly quickly even on CPU
# Method 1
torchrun --nproc_per_node 1 train_lora.py
cd trainer && torchrun --nproc_per_node 1 train_lora.py
# Method 2
python train_lora.py
cd trainer && python train_lora.py
```
> The trained model weight files are saved by default every `save_interval steps` as: `lora_xxx_*.pth` (* is the specific model dimension, each save overwrites the previous file)
LoRA is well-suited for handling problems like "how to let the model quickly adapt to private domains or vertical scenarios while preserving general capabilities as much as possible." For example, when the base model lacks medical knowledge, a medical-oriented LoRA weight layer can be stacked on top of the original model to achieve better domain performance at relatively small cost.
Usually you only need to prepare multi-turn dialogue format data in the same way, place it in `lora_xxx.jsonl`, and then run `python train_lora.py` to obtain new `LoRA` model weights.
Usually you only need to prepare multi-turn dialogue format data in the same way, place it in `lora_xxx.jsonl`, and then run `cd trainer && python train_lora.py` from the repository root to obtain new `LoRA` model weights.
Example 1: Vertical domain data

View File

@ -6,7 +6,7 @@ import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, TextStreamer
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
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')
def init_model(args):
@ -27,7 +27,9 @@ def init_model(args):
else:
model = AutoModelForCausalLM.from_pretrained(args.load_from, trust_remote_code=True)
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():
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('--historys', default=0, type=int, help="携带历史对话轮数(需为偶数,0表示不携带历史)")
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()
prompts = [

View File

@ -21,13 +21,19 @@ while True:
extra_body={"chat_template_kwargs": {"open_thinking": True}, "reasoning_effort": "medium"} # 思考开关
)
if not stream:
if not response.choices or response.choices[0].message is None:
raise ValueError("LLM returned empty or filtered response")
assistant_res = response.choices[0].message.content
print('[A]: ', assistant_res)
else:
print('[A]: ', end='', flush=True)
assistant_res = ''
for chunk in response:
if not chunk.choices:
continue
delta = chunk.choices[0].delta
if delta is None:
continue
r = getattr(delta, 'reasoning_content', None) or ""
c = delta.content or ""
if r:

View File

@ -15,7 +15,7 @@ from threading import Thread
from queue import Queue
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from pydantic import BaseModel, Field
from transformers import AutoTokenizer, AutoModelForCausalLM, TextStreamer
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
from model.model_lora import apply_lora, load_lora
@ -54,7 +54,7 @@ class ChatRequest(BaseModel):
top_p: float = 0.92
max_tokens: int = 8192
stream: bool = True
tools: list = []
tools: list = Field(default_factory=list)
open_thinking: bool = False
chat_template_kwargs: dict = None
@ -104,13 +104,14 @@ def parse_response(text):
def generate_stream_response(messages, temperature, top_p, max_tokens, tools=None, open_thinking=False):
try:
new_prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True, tools=tools or None, open_thinking=open_thinking)[-max_tokens:]
new_prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True, tools=tools or None, open_thinking=open_thinking)
inputs = tokenizer(new_prompt, return_tensors="pt", truncation=True).to(device)
queue = Queue()
streamer = CustomStreamer(tokenizer, queue)
def _generate():
try:
model.generate(
inputs.input_ids,
max_new_tokens=max_tokens,
@ -122,6 +123,9 @@ def generate_stream_response(messages, temperature, top_p, max_tokens, tools=Non
eos_token_id=tokenizer.eos_token_id,
streamer=streamer
)
except Exception as e:
queue.put({"error": str(e)})
queue.put(None)
Thread(target=_generate).start()
@ -133,6 +137,9 @@ def generate_stream_response(messages, temperature, top_p, max_tokens, tools=Non
text = queue.get()
if text is None:
break
if isinstance(text, dict):
yield json.dumps(text, ensure_ascii=False)
continue
full_text += text
if not thinking_ended:
@ -190,7 +197,7 @@ async def chat_completions(request: ChatRequest):
add_generation_prompt=True,
tools=request.tools or None,
open_thinking=request.get_open_thinking()
)[-request.max_tokens:]
)
inputs = tokenizer(new_prompt, return_tensors="pt", truncation=True).to(device)
with torch.no_grad():
generated_ids = model.generate(

View File

@ -339,8 +339,8 @@ def main():
st.markdown(
f'<div style="display: flex; justify-content: flex-end;"><div style="display: inline-block; margin: 10px 0; padding: 8px 12px 8px 12px; background-color: #3d4450; border-radius: 22px; color: white;">{prompt}</div></div>',
unsafe_allow_html=True)
messages.append({"role": "user", "content": prompt[-st.session_state.max_new_tokens:]})
st.session_state.chat_messages.append({"role": "user", "content": prompt[-st.session_state.max_new_tokens:]})
messages.append({"role": "user", "content": prompt})
st.session_state.chat_messages.append({"role": "user", "content": prompt})
placeholder = st.empty()

View File

@ -24,7 +24,7 @@ from torch.optim.lr_scheduler import CosineAnnealingLR
from transformers import AutoTokenizer
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
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
warnings.filterwarnings('ignore')
@ -378,7 +378,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=1, 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("--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("--num_workers", type=int, default=8, 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))
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
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
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()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
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"))

View File

@ -17,7 +17,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig
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')
@ -151,7 +151,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=6, help="训练轮数")
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("--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("--num_workers", type=int, default=8, 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
# ========== 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
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 ==========
wandb = None
@ -210,7 +215,7 @@ if __name__ == "__main__":
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_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)
# ========== 6. 从ckp恢复状态 ==========
@ -235,7 +240,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
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)

View File

@ -17,7 +17,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig
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')
@ -135,7 +135,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=1, help="训练轮数")
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("--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("--num_workers", type=int, default=8, 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
# ========== 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
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 ==========
wandb = None
@ -190,7 +195,7 @@ if __name__ == "__main__":
train_ds = DPODataset(args.data_path, tokenizer, max_length=args.max_seq_len)
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)
# ========== 6. 从ckp恢复状态 ==========
@ -215,7 +220,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
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)

View File

@ -16,7 +16,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig
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')
@ -88,7 +88,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=2, help="训练轮数")
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("--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("--num_workers", type=int, default=8, 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
# ========== 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
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 ==========
wandb = None
@ -135,7 +140,7 @@ if __name__ == "__main__":
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_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)
# ========== 6. 从ckp恢复状态 ==========
@ -160,7 +165,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
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 model.model_minimind import MiniMindConfig, MiniMindForCausalLM
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
warnings.filterwarnings('ignore')
@ -210,7 +210,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=1, help="训练轮数")
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("--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("--num_workers", type=int, default=8, 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
# ========== 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
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 ==========
wandb = None
@ -321,7 +326,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
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"))

View File

@ -17,7 +17,7 @@ from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig
from dataset.lm_dataset import SFTDataset
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')
@ -82,7 +82,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=10, help="训练轮数")
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("--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("--num_workers", type=int, default=8, 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
# ========== 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
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 ==========
wandb = None
@ -148,7 +153,7 @@ if __name__ == "__main__":
# ========== 6. 定义数据和优化器 ==========
train_ds = SFTDataset(args.data_path, tokenizer, max_length=args.max_seq_len)
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)
# ========== 7. 从ckp恢复状态 ==========
@ -173,7 +178,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
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)

View File

@ -21,7 +21,7 @@ from torch.nn.utils import clip_grad_norm_
from torch.optim.lr_scheduler import CosineAnnealingLR
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
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
warnings.filterwarnings('ignore')
@ -301,7 +301,7 @@ if __name__ == "__main__":
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("--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("--num_workers", type=int, default=8, help="数据加载线程数")
parser.add_argument("--accumulation_steps", type=int, default=1, help="梯度累积步数")
@ -349,9 +349,14 @@ if __name__ == "__main__":
ckp_data = lm_checkpoint(lm_config, weight=args.save_weight, save_dir='../checkpoints') if args.from_resume==1 else None
# ========== 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
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 ==========
wandb = None
@ -424,7 +429,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
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, use_sglang = (args.rollout_engine == "sglang"))

View File

@ -16,7 +16,7 @@ from torch.nn.parallel import DistributedDataParallel
from torch.utils.data import DataLoader, DistributedSampler
from model.model_minimind import MiniMindConfig
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')
@ -87,7 +87,7 @@ if __name__ == "__main__":
parser.add_argument("--epochs", type=int, default=2, help="训练轮数")
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("--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("--num_workers", 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
# ========== 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
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 ==========
wandb = None
@ -134,7 +139,7 @@ if __name__ == "__main__":
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_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)
# ========== 6. 从ckp恢复状态 ==========
@ -159,7 +164,7 @@ if __name__ == "__main__":
setup_seed(42 + epoch); indices = torch.randperm(len(train_ds)).tolist()
skip = start_step if (epoch == start_epoch and start_step > 0) else 0
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:
Logger(f'Epoch [{epoch + 1}/{args.epochs}]: 跳过前{start_step}个step,从step {start_step + 1}开始')
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 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):
total = sum(p.numel() for p in model.parameters()) / 1e6
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:
return 0 # 非DDP模式
if not torch.cuda.is_available():
return 0 # MPS/CPU 不支持 NCCL DDP,静默跳过
dist.init_process_group(backend="nccl")
local_rank = int(os.environ["LOCAL_RANK"])
torch.cuda.set_device(local_rank)
@ -55,6 +67,7 @@ def setup_seed(seed: int):
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
if torch.cuda.is_available():
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
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)
os.replace(resume_tmp, resume_path)
del state_dict, resume_data
if torch.cuda.is_available():
torch.cuda.empty_cache()
elif hasattr(torch, "mps") and torch.backends.mps.is_available():
torch.mps.empty_cache()
else: # 加载模式
if os.path.exists(resume_path):
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
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)
model = MiniMindForCausalLM(lm_config)
@ -158,7 +176,9 @@ class SkipBatchSampler(Sampler):
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.model = AutoModel.from_pretrained(model_path, torch_dtype=dtype, trust_remote_code=True)
self.model = self.model.to(device).eval()