Compare commits

...

7 Commits

Author SHA1 Message Date
Neil Ge c6a855929d
Merge 9cb07dd1ec into cdf0a93e02 2026-09-14 23:35:08 +08:00
jingyaogong cdf0a93e02
Merge pull request #847 from a0917-cell/fix/safe-math-eval
[fix] 用 AST 求值器替换工具里的 eval(),并修复 Windows 上工具调用全部静默失败
2026-09-14 23:34:28 +08:00
jingyaogong c367572702
Merge pull request #854 from basil-k-aji-dev/fix/rl-ddp-grad-sync
[fix] route RL trainer forward through DDP so gradients all-reduce
2026-09-14 22:28:03 +08:00
basil-k-aji-dev 9c088dd3a5 [fix] route RL trainer forward through DDP so gradients all-reduce
train_grpo.py, train_ppo.py and train_agent.py wrap the policy in
DistributedDataParallel but run the loss-bearing forward through
.module, bypassing the wrapper. DDP arms its gradient all-reduce hooks
inside DistributedDataParallel.forward(), so calling .module directly
means backward() performs no all-reduce: every rank steps on purely
local gradients and the replicas silently diverge. Only rank 0 is
saved, so multi-GPU RL training quietly degrades to single-GPU
training on 1/N of the data with no error.

The supervised trainers are unaffected; they already call the wrapped
module. no_grad uses of .module (rollout generation, checkpoint
saving) are left untouched, where unwrapping is correct.

Single-GPU behaviour is unchanged: with no process group, model is the
bare module and these calls are identical to before.
2026-09-14 14:54:07 +05:30
tkgcc 0f6803e551 [fix] replace tool eval() with an AST math evaluator
execute_tool() relied on signal.SIGALRM, which does not exist on Windows.
The AttributeError was swallowed by a bare except, so every tool call
silently returned None and the model always observed
{"error": "tool not found"} (issue #846).

calculate_math also ran eval() on model-generated text. Two of the three
call sites (scripts/eval_toolcall.py, scripts/web_demo.py) passed no
globals at all, leaving full builtins reachable from sampled output.

safe_math_eval() in trainer_utils.py parses the expression with ast and
walks it, allowing only arithmetic operators and a math whitelist, with
caps on expression length and exponent size. It cannot loop forever, so
the signal-based timeout is no longer needed and is removed. Both sqrt(x)
and math.sqrt(x) spellings keep working.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-07 07:53:33 +08:00
neil.ge 9cb07dd1ec perf: enable fused AdamW in pretrain training
Same fused AdamW optimization as the full SFT script, using the
existing device_type variable so the CPU-only path is unaffected.

Co-authored-by: TRAE CLI <traecli@bytedance.com>
2026-09-05 19:16:51 +08:00
neil.ge 859c8ee2a3 perf: enable fused AdamW in full SFT training
Use PyTorch's fused AdamW implementation to reduce optimizer kernel
launch overhead and speed up the training step on CUDA devices.
Guard with device check so the CPU-only fallback path is unaffected.

Co-authored-by: TRAE CLI <traecli@bytedance.com>
2026-09-05 18:37:04 +08:00
8 changed files with 75 additions and 24 deletions

View File

@ -12,7 +12,7 @@ from datetime import datetime
from transformers import AutoTokenizer, AutoModelForCausalLM, TextStreamer from transformers import AutoTokenizer, AutoModelForCausalLM, TextStreamer
from openai import OpenAI from openai import OpenAI
from model.model_minimind import MiniMindConfig, MiniMindForCausalLM from model.model_minimind import MiniMindConfig, MiniMindForCausalLM
from trainer.trainer_utils import setup_seed, get_model_params from trainer.trainer_utils import setup_seed, get_model_params, safe_math_eval
warnings.filterwarnings('ignore') warnings.filterwarnings('ignore')
TOOLS = [ TOOLS = [
@ -27,7 +27,7 @@ TOOLS = [
] ]
MOCK_RESULTS = { MOCK_RESULTS = {
"calculate_math": lambda args: {"result": str(eval(str(args.get("expression", "0")).replace("^", "**").replace("×", "*").replace("÷", "/").replace("−", "-").replace("²", "**2").replace("³", "**3").replace("(", "(").replace(")", ")")))}, "calculate_math": lambda args: {"result": str(safe_math_eval(args.get("expression", "0")))},
"get_current_time": lambda args: {"datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "timezone": args.get("timezone", "Asia/Shanghai")}, "get_current_time": lambda args: {"datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "timezone": args.get("timezone", "Asia/Shanghai")},
"random_number": lambda args: {"result": random.randint(int(args.get("min", 0)), int(args.get("max", 100)))}, "random_number": lambda args: {"result": random.randint(int(args.get("min", 0)), int(args.get("max", 100)))},
"text_length": lambda args: {"characters": len(args.get("text", "")), "words": len(args.get("text", "").split())}, "text_length": lambda args: {"characters": len(args.get("text", "")), "words": len(args.get("text", "").split())},

View File

@ -2,6 +2,7 @@ import random
import re import re
import json import json
import os import os
import sys
from threading import Thread from threading import Thread
import torch import torch
@ -9,6 +10,9 @@ import numpy as np
import streamlit as st import streamlit as st
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from trainer.trainer_utils import safe_math_eval
st.set_page_config(page_title="MiniMind", initial_sidebar_state="collapsed") st.set_page_config(page_title="MiniMind", initial_sidebar_state="collapsed")
st.markdown(""" st.markdown("""
@ -125,7 +129,7 @@ def execute_tool(tool_name, args):
import datetime import datetime
try: try:
if tool_name == 'calculate_math': if tool_name == 'calculate_math':
return {"result": eval(args.get('expression', '0'))} return {"result": safe_math_eval(args.get('expression', '0'))}
elif tool_name == 'get_current_time': elif tool_name == 'get_current_time':
tz = args.get('timezone', 'Asia/Shanghai') tz = args.get('timezone', 'Asia/Shanghai')
return {"result": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} return {"result": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

View File

@ -10,7 +10,6 @@ import gc
import json import json
import math import math
import random import random
import signal
import argparse import argparse
import warnings import warnings
import torch import torch
@ -24,7 +23,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, safe_math_eval
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')
@ -55,7 +54,7 @@ UNIT_DATA = {"km_miles": 0.621371, "miles_km": 1.60934, "kg_pounds": 2.20462, "p
# ======== 模拟执行 ======== # ======== 模拟执行 ========
MOCK_RESULTS = { MOCK_RESULTS = {
"calculate_math": lambda args: {"result": str(eval(str(args.get("expression", "0")).replace("^", "**").replace("×", "*").replace("÷", "/").replace("−", "-").replace("(", "(").replace(")", ")"), {"__builtins__": {}, "math": math}))}, "calculate_math": lambda args: {"result": str(safe_math_eval(args.get("expression", "0")))},
"unit_converter": lambda args: {"result": round(float(args.get("value", 0)) * UNIT_DATA.get(f"{args.get('from_unit', '').lower()}_{args.get('to_unit', '').lower()}", 1), 4)}, "unit_converter": lambda args: {"result": round(float(args.get("value", 0)) * UNIT_DATA.get(f"{args.get('from_unit', '').lower()}_{args.get('to_unit', '').lower()}", 1), 4)},
"get_current_weather": lambda args: (lambda w: {"city": args.get("location"), "temperature": w[0], "humidity": "65%", "condition": w[1]})(WEATHER_DATA.get(args.get("location"), ("22°C", "晴"))), "get_current_weather": lambda args: (lambda w: {"city": args.get("location"), "temperature": w[0], "humidity": "65%", "condition": w[1]})(WEATHER_DATA.get(args.get("location"), ("22°C", "晴"))),
"get_current_time": lambda args: {"datetime": TIME_DATA.get(args.get("timezone", "Asia/Shanghai"), "2025-03-07 14:30:00"), "timezone": args.get("timezone", "Asia/Shanghai")}, "get_current_time": lambda args: {"datetime": TIME_DATA.get(args.get("timezone", "Asia/Shanghai"), "2025-03-07 14:30:00"), "timezone": args.get("timezone", "Asia/Shanghai")},
@ -85,14 +84,9 @@ def execute_tool(name, args):
fn = MOCK_RESULTS.get(name) fn = MOCK_RESULTS.get(name)
if not fn: return None if not fn: return None
try: try:
signal.signal(signal.SIGALRM, lambda *_: (_ for _ in ()).throw(TimeoutError()))
signal.alarm(1)
return fn(args) return fn(args)
except: except Exception:
return None return None
finally:
try: signal.alarm(0)
except: pass
# ======== 多轮 Rollout ======== # ======== 多轮 Rollout ========
def rollout_single(rollout_engine, tokenizer, messages, tools, max_turns=3, max_new_tokens=256, thinking_ratio=0.5, device="cuda"): def rollout_single(rollout_engine, tokenizer, messages, tools, max_turns=3, max_new_tokens=256, thinking_ratio=0.5, device="cuda"):
@ -278,9 +272,10 @@ def rl_train_epoch(epoch, loader, iters, rollout_engine, ref_model, reward_model
rewards = calculate_rewards(prompts, completions, gt_batch, tools_batch, args.num_generations, reward_model, device=args.device, turn_outputs_batch=turn_outputs_batch, unfinished_batch=unfinished_batch) rewards = calculate_rewards(prompts, completions, gt_batch, tools_batch, args.num_generations, reward_model, device=args.device, turn_outputs_batch=turn_outputs_batch, unfinished_batch=unfinished_batch)
model_unwrapped = model.module if isinstance(model, DistributedDataParallel) else model
with autocast_ctx: with autocast_ctx:
res = model_unwrapped(input_ids, attention_mask=full_mask) # 反向传播必须经过 DDP 包装后的模块:直接调用 .module 会跳过
# DDP 的 prepare_for_backward,梯度不会 all-reduce,各卡静默发散。
res = model(input_ids, attention_mask=full_mask)
aux_loss = res.aux_loss if lm_config.use_moe else torch.tensor(0.0, device=args.device) aux_loss = res.aux_loss if lm_config.use_moe else torch.tensor(0.0, device=args.device)
logits = res.logits[:, :-1, :] logits = res.logits[:, :-1, :]
per_token_logps = F.log_softmax(logits, dim=-1).gather(2, input_ids[:, 1:].unsqueeze(-1)).squeeze(-1) per_token_logps = F.log_softmax(logits, dim=-1).gather(2, input_ids[:, 1:].unsqueeze(-1)).squeeze(-1)

View File

@ -137,7 +137,7 @@ if __name__ == "__main__":
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=(args.dtype == 'float16'))
optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate) optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate, fused=('cuda' in args.device))
# ========== 6. 从ckp恢复状态 ========== # ========== 6. 从ckp恢复状态 ==========
start_epoch, start_step = 0, 0 start_epoch, start_step = 0, 0

View File

@ -95,9 +95,10 @@ def grpo_train_epoch(epoch, loader, iters, rollout_engine, ref_model, reward_mod
rewards = calculate_rewards(prompts, completions, reward_model).to(args.device) # [B*num_gen] rewards = calculate_rewards(prompts, completions, reward_model).to(args.device) # [B*num_gen]
model_unwrapped = model.module if isinstance(model, DistributedDataParallel) else model
with autocast_ctx: with autocast_ctx:
res = model_unwrapped(outputs, attention_mask=full_mask) # 反向传播必须经过 DDP 包装后的模块:直接调用 .module 会跳过
# DDP 的 prepare_for_backward,梯度不会 all-reduce,各卡静默发散。
res = model(outputs, attention_mask=full_mask)
aux_loss = res.aux_loss if lm_config.use_moe else torch.tensor(0.0, device=args.device) aux_loss = res.aux_loss if lm_config.use_moe else torch.tensor(0.0, device=args.device)
per_token_logps = F.log_softmax(res.logits[:, :-1, :], dim=-1).gather(2, outputs[:, 1:].unsqueeze(-1)).squeeze(-1).gather(1, logp_pos) per_token_logps = F.log_softmax(res.logits[:, :-1, :], dim=-1).gather(2, outputs[:, 1:].unsqueeze(-1)).squeeze(-1).gather(1, logp_pos)

View File

@ -159,8 +159,6 @@ def ppo_train_epoch(epoch, loader, iters, rollout_engine, ref_model, actor_sched
clipfrac_sum = 0.0 clipfrac_sum = 0.0
aux_loss_sum = 0.0 aux_loss_sum = 0.0
log_count = 0 log_count = 0
actor_unwrapped = actor_model.module if isinstance(actor_model, DistributedDataParallel) else actor_model
critic_unwrapped = critic_model.module if isinstance(critic_model, DistributedDataParallel) else critic_model
for ppo_epoch in range(args.ppo_update_iters): for ppo_epoch in range(args.ppo_update_iters):
if stop_ppo: if stop_ppo:
break break
@ -168,11 +166,13 @@ def ppo_train_epoch(epoch, loader, iters, rollout_engine, ref_model, actor_sched
for i in range(0, B, mb_size): for i in range(0, B, mb_size):
inds = b_inds[i:i + mb_size] inds = b_inds[i:i + mb_size]
mb_values_seq = critic_unwrapped(input_ids=gen_out[inds], attention_mask=full_mask[inds]) # 反向传播必须经过 DDP 包装后的模块:直接调用 .module 会跳过
# DDP 的 prepare_for_backward,梯度不会 all-reduce,各卡静默发散。
mb_values_seq = critic_model(input_ids=gen_out[inds], attention_mask=full_mask[inds])
mb_resp_values = mb_values_seq.gather(1, logp_pos[inds]) mb_resp_values = mb_values_seq.gather(1, logp_pos[inds])
with autocast_ctx: with autocast_ctx:
res = actor_unwrapped(input_ids=gen_out[inds], attention_mask=full_mask[inds]) res = actor_model(input_ids=gen_out[inds], attention_mask=full_mask[inds])
aux_loss = res.aux_loss if lm_config.use_moe else torch.tensor(0.0, device=args.device) aux_loss = res.aux_loss if lm_config.use_moe else torch.tensor(0.0, device=args.device)
# 在 autocast 内计算 log_softmax,避免直接对 fp16/bf16 logits # 在 autocast 内计算 log_softmax,避免直接对 fp16/bf16 logits
# 计算造成额外数值偏差。 # 计算造成额外数值偏差。
@ -191,7 +191,7 @@ def ppo_train_epoch(epoch, loader, iters, rollout_engine, ref_model, actor_sched
f"ratio_max={torch.exp(_lrv).max().item():.6f} " f"ratio_max={torch.exp(_lrv).max().item():.6f} "
f"ratio_min={torch.exp(_lrv).min().item():.6f} " f"ratio_min={torch.exp(_lrv).min().item():.6f} "
f"dropout={getattr(lm_config, 'dropout', None)} " f"dropout={getattr(lm_config, 'dropout', None)} "
f"training={actor_unwrapped.training}") f"training={actor_model.training}")
approx_kl = (0.5 * (log_ratio ** 2) * resp_policy_mask[inds]).sum() / resp_policy_mask[inds].sum().clamp(min=1) approx_kl = (0.5 * (log_ratio ** 2) * resp_policy_mask[inds]).sum() / resp_policy_mask[inds].sum().clamp(min=1)
# 同步各卡的 approx_kl,防止某卡 break 而其它卡继续导致 DDP 死锁 # 同步各卡的 approx_kl,防止某卡 break 而其它卡继续导致 DDP 死锁

View File

@ -136,7 +136,7 @@ if __name__ == "__main__":
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=(args.dtype == 'float16'))
optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate) optimizer = optim.AdamW(model.parameters(), lr=args.learning_rate, fused=(device_type == 'cuda'))
# ========== 6. 从ckp恢复状态 ========== # ========== 6. 从ckp恢复状态 ==========
start_epoch, start_step = 0, 0 start_epoch, start_step = 0, 0

View File

@ -5,6 +5,8 @@ import os
import sys import sys
__package__ = "trainer" __package__ = "trainer"
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
import ast
import operator
import random import random
import math import math
import numpy as np import numpy as np
@ -175,3 +177,52 @@ class LMForRewardModel:
] ]
score = self.model.get_score(self.tokenizer, eval_messages) score = self.model.get_score(self.tokenizer, eval_messages)
return max(min(score, 3.0), -3.0) return max(min(score, 3.0), -3.0)
# ===== 数学表达式安全求值(替代 eval,避免执行模型生成的任意代码) =====
_MATH_FUNCS = {name: getattr(math, name) for name in (
'sqrt', 'pow', 'exp', 'log', 'log2', 'log10', 'sin', 'cos', 'tan', 'asin', 'acos', 'atan', 'atan2',
'sinh', 'cosh', 'tanh', 'floor', 'ceil', 'trunc', 'fabs', 'fmod', 'hypot', 'gcd', 'degrees', 'radians'
)}
_MATH_CONSTS = {'pi': math.pi, 'e': math.e, 'tau': math.tau}
_BIN_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv,
ast.FloorDiv: operator.floordiv, ast.Mod: operator.mod, ast.Pow: operator.pow}
_UNARY_OPS = {ast.UAdd: operator.pos, ast.USub: operator.neg}
_MAX_EXPR_LEN, _MAX_EXP, _MAX_DIGITS = 512, 1e4, 100 # 表达式长度 / 指数 / 结果位数上限
def _check_pow(base, exp): # 防止 9**9**9 这类表达式把进程算死
if abs(exp) > _MAX_EXP or (abs(base) > 1 and abs(exp) * math.log10(abs(base)) > _MAX_DIGITS):
raise ValueError('幂运算结果过大')
def _resolve_name(node): # 同时兼容 sqrt(4) 与 math.sqrt(4) 两种写法
if isinstance(node, ast.Name): return node.id
if isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name) and node.value.id == 'math': return node.attr
return None
def _eval_ast(node):
if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)) and not isinstance(node.value, bool):
return node.value
if isinstance(node, (ast.Name, ast.Attribute)) and _resolve_name(node) in _MATH_CONSTS:
return _MATH_CONSTS[_resolve_name(node)]
if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPS:
return _UNARY_OPS[type(node.op)](_eval_ast(node.operand))
if isinstance(node, ast.BinOp) and type(node.op) in _BIN_OPS:
left, right = _eval_ast(node.left), _eval_ast(node.right)
if isinstance(node.op, ast.Pow): _check_pow(left, right)
return _BIN_OPS[type(node.op)](left, right)
if isinstance(node, ast.Call) and _resolve_name(node.func) in _MATH_FUNCS and not node.keywords:
name, args = _resolve_name(node.func), [_eval_ast(arg) for arg in node.args]
if name == 'pow' and len(args) == 2: _check_pow(*args)
return _MATH_FUNCS[name](*args)
raise ValueError(f'不支持的表达式语法: {type(node).__name__}')
def safe_math_eval(expression):
"""对模型生成的数学表达式求值:只放行算术运算与 math 白名单函数,不执行任意代码。"""
expr = str(expression).replace('^', '**').replace('×', '*').replace('÷', '/').replace('−', '-')
expr = expr.replace('²', '**2').replace('³', '**3').replace('(', '(').replace(')', ')').strip()
if not expr or len(expr) > _MAX_EXPR_LEN: raise ValueError('表达式为空或过长')
return _eval_ast(ast.parse(expr, mode='eval').body)