哈喽,大家好~
今儿和大家把基础中的随机森林回归,再做一个总结~
一般来说,随机森林回归是机器学习中非常常用、稳定且“好用”的方法。很多时候你只需要把数据喂进去,模型就能跑出不错的结果,并且对特征尺度、异常值、非线性等具有较强的鲁棒性。
核心逻辑
比如说,我们要预测房价,但我们不知道哪个特征很重要,也不知道特征和目标之间的关系是不是线性的。
我们可以用一棵决策树(CART)把数据分成很多“格子”,每个终端格子的预测就是该格子里样本目标值的平均值。单棵树能拟合复杂关系,但容易“过拟合”且不稳定。
随机森林的想法是:别只用一棵树,用很多棵树,把每棵树训练在数据子集(Bootstrap 采样)上,并且在每个节点分裂时随机只考虑一部分特征(随机子空间),然后把多棵树的预测平均起来(回归任务)。
这样做有两个好处:
-
Bootstrap + 随机特征能降低树与树之间的相关性; -
平均多个弱/强的预测器可以降低方差,提高泛化能力。
通俗地讲,随机森林就是“让很多个(不太一样的)树一起投票/平均”的办法:如果每棵树有自己的小偏差,但偏差方向不完全相同,平均后就能把噪音抵消掉。
数学表达
均方误差(MSE)作为回归树分裂评价和模型评估常用指标:
-
单个样本误差: -
样本集 MSE:
分裂点选择(回归树用方差或 MSE 减少作为准则):对于候选分裂把样本分为左右两部分,目标是最大化方差减少:
其中 是左右子集样本数, 表示目标变量的方差。
随机森林的最终预测(回归)是各个树预测的平均:
完整案例
我们生成一个二维特征,方便可视化的数据,构建单颗回归树(CART)并训练。
构建随机森林,包含 Bootstrap、随机特征,训练过程中的 OOB 误差随树数量变化曲线。
数据集
我们构造一个包含两个输入特征 x1, x2 的数据集,目标由非线性函数生成,加上一点噪声,使得任务既非线性又有噪声。
import torch
import torch.nn.functional as F
import numpy as np
import matplotlib.pyplot as plt
from matplotlib import cm
from mpl_toolkits.mplot3d import Axes3D
import random
from tqdm import trange
seed = 2026
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
def make_synthetic_data(n_samples=800, noise_std=0.15):
# x1 在 [0,1], x2 在 [-1,1]
x1 = np.random.rand(n_samples)
x2 = np.random.rand(n_samples) * 2 - 1
# 非线性关系:混合了周期、指数与多项式
y = np.sin(2 * np.pi * x1) * (1 + x2**2) + 0.5 * x2 + 0.3 * np.exp(-x1*3)
# 添加噪声
y = y + np.random.randn(n_samples) * noise_std
X = np.stack([x1, x2], axis=1)
return X.astype(np.float32), y.astype(np.float32)
X_np, y_np = make_synthetic_data(800, noise_std=0.12)
# 转成 PyTorch 张量
X = torch.from_numpy(X_np) # (n_samples, 2)
y = torch.from_numpy(y_np).unsqueeze(1) # (n_samples, 1)
n_samples, n_features = X.shape
print("样本数:", n_samples, " 特征数:", n_features)
-
我们选了二维特征方便在平面/曲面上可视化真值与预测。 -
噪声使问题更接近真实世界场景。
实现单棵 CART 回归树
为了讲清楚流程,这里实现一个简单的回归树构建器。
关键思路:
-
对当前节点的样本集合,遍历每个特征和若干候选分割点,计算分裂后的 MSE(或方差)减少,选取最优分裂; -
递归构建直到达到最大深度或样本数小于阈值或分裂收益不足; -
叶节点的预测是该叶样本目标的均值。
但是要注意,为简单易懂,我们用较直观但不一定是最高效的候选分裂枚举(基于排序的中间点)。
class TreeNode:
def __init__(self):
self.is_leaf = False
self.pred = None
self.feature = None
self.threshold = None
self.left = None
self.right = None
def mse_of_tensor(y):
# y: (n,1) tensor
if y.numel() == 0:
return0.0
mean = y.mean()
return ((y - mean)**2).mean().item()
class CARTRegressor:
def __init__(self, max_depth=8, min_samples_leaf=5, min_impurity_decrease=1e-6, max_features=None):
self.max_depth = max_depth
self.min_samples_leaf = min_samples_leaf
self.min_impurity_decrease = min_impurity_decrease
self.max_features = max_features # 随机特征子集大小
self.root = None
def fit(self, X, y):
# X: tensor (n, d), y: tensor (n,1)
self.n_features = X.shape[1]
if self.max_features isNone:
self.max_features = self.n_features
self.root = self._build_tree(X, y, depth=0)
return self
def _build_tree(self, X, y, depth):
node = TreeNode()
n_samples = X.shape[0]
# 叶节点条件
if depth >= self.max_depth or n_samples <= self.min_samples_leaf:
node.is_leaf = True
node.pred = y.mean().item()
return node
parent_mse = mse_of_tensor(y)
best_impurity = 0.0
best_feat, best_thresh = None, None
best_left_idx, best_right_idx = None, None
# 随机选择特征子集(如果指定)
feat_indices = list(range(self.n_features))
if self.max_features < self.n_features:
feat_indices = random.sample(feat_indices, self.max_features)
# 对每个特征枚举候选分割点(排除重复值)
for feat in feat_indices:
col = X[:, feat]
# 候选阈值为排序后相邻值的中点
unique_vals, _ = torch.sort(torch.unique(col))
if unique_vals.numel() <= 1:
continue
thresholds = (unique_vals[:-1] + unique_vals[1:]) / 2.0
for thresh in thresholds:
left_mask = (col <= thresh)
right_mask = ~left_mask
nL = left_mask.sum().item()
nR = right_mask.sum().item()
# 叶子最小样本数检查
if nL < self.min_samples_leaf or nR < self.min_samples_leaf:
continue
yL = y[left_mask]
yR = y[right_mask]
mseL = mse_of_tensor(yL)
mseR = mse_of_tensor(yR)
weighted_mse = (nL * mseL + nR * mseR) / (nL + nR)
impurity_decrease = parent_mse - weighted_mse
if impurity_decrease > best_impurity:
best_impurity = impurity_decrease
best_feat = feat
best_thresh = float(thresh.item())
best_left_idx = left_mask
best_right_idx = right_mask
# 如果无法分裂或收益小,则设为叶子
if best_feat isNoneor best_impurity < self.min_impurity_decrease:
node.is_leaf = True
node.pred = y.mean().item()
return node
# 否则继续递归构建
node.feature = best_feat
node.threshold = best_thresh
node.left = self._build_tree(X[best_left_idx], y[best_left_idx], depth + 1)
node.right = self._build_tree(X[best_right_idx], y[best_right_idx], depth + 1)
return node
def _predict_one(self, x, node):
if node.is_leaf:
return node.pred
if x[node.feature] <= node.threshold:
return self._predict_one(x, node.left)
else:
return self._predict_one(x, node.right)
def predict(self, X):
# X: tensor (n, d)
preds = []
for i in range(X.shape[0]):
preds.append(self._predict_one(X[i], self.root))
return torch.tensor(preds).unsqueeze(1).float()
-
_build_tree:核心是枚举特征与阈值,选择使得 MSE 最大降低的分裂。 -
为了简洁,阈值候选使用了排序后相邻两个不同值的中点(常见做法)。 -
叶节点预测是该叶上 y 的均值。 -
max_features 支持在分裂时只考虑一部分特征(这是随机森林的关键)。
随机森林实现
随机森林由多棵树组成。每棵树用 Bootstrap 抽样(有放回)来选择训练样本,并记录哪些样本没有被抽中(OOB),便于计算 OOB 估计。每棵树在分裂时只考虑随机选择的 max_features 个特征。
下面的 RandomForestRegressor 类实现了:
-
n_estimators:树的数量 -
max_features:每次分裂时考虑的特征数(可传 int 或 'sqrt') -
支持 OOB 误差随树数变化的计算
class RandomForestRegressorTorch:
def __init__(self, n_estimators=50, max_depth=8, min_samples_leaf=5, max_features='sqrt', bootstrap=True, min_impurity_decrease=1e-6):
self.n_estimators = n_estimators
self.max_depth = max_depth
self.min_samples_leaf = min_samples_leaf
self.bootstrap = bootstrap
self.min_impurity_decrease = min_impurity_decrease
self.trees = []
self.oob_indices = [] # 每棵树的 OOB 样本索引
self.max_features = max_features
def _resolve_max_features(self, n_features):
if isinstance(self.max_features, int):
return self.max_features
elif self.max_features == 'sqrt':
return max(1, int(np.sqrt(n_features)))
elif self.max_features == 'log2':
return max(1, int(np.log2(n_features)))
else:
# 默认使用全部
return n_features
def fit(self, X, y, verbose=False):
n_samples, n_features = X.shape
self.n_features_ = n_features
self.trees = []
self.oob_indices = []
# 用于 OOB 计算的累计预测与计数
oob_pred_sum = torch.zeros_like(y)
oob_counts = torch.zeros((n_samples, 1), dtype=torch.int32)
oob_mse_list = []
for t in trange(self.n_estimators, desc='训练树'):
# Bootstrap 采样(有放回)
if self.bootstrap:
indices = np.random.randint(0, n_samples, size=n_samples)
oob_mask = np.ones(n_samples, dtype=bool)
oob_mask[indices] = False
oob_idx = np.where(oob_mask)[0]
else:
indices = np.arange(n_samples)
oob_idx = np.array([], dtype=int)
X_sample = X[indices]
y_sample = y[indices]
# 决定每次分裂考虑多少个特征
max_feat = self._resolve_max_features(n_features)
tree = CARTRegressor(max_depth=self.max_depth, min_samples_leaf=self.min_samples_leaf,
min_impurity_decrease=self.min_impurity_decrease,
max_features=max_feat)
tree.fit(X_sample, y_sample)
self.trees.append(tree)
self.oob_indices.append(oob_idx)
# 更新 OOB 预测累加
if len(oob_idx) > 0:
X_oob = X[oob_idx]
preds_oob = tree.predict(X_oob) # (k,1)
oob_pred_sum[oob_idx] += preds_oob
oob_counts[oob_idx] += 1
# 计算当前的 OOB MSE(只对那些至少被 OOB 过一次的样本)
mask = (oob_counts.squeeze() > 0).numpy()
if mask.sum() > 0:
avg_preds = oob_pred_sum[mask] / oob_counts[mask].float()
mse_val = ((avg_preds - y[mask])**2).mean().item()
else:
mse_val = None
oob_mse_list.append(mse_val)
if verbose:
print(f"树 {t+1}/{self.n_estimators}, 当前 OOB MSE: {mse_val}")
self.oob_mse_curve = oob_mse_list
return self
def predict(self, X):
# 对所有树取平均预测
preds_sum = torch.zeros((X.shape[0], 1), dtype=torch.float32)
for tree in self.trees:
preds_sum += tree.predict(X)
return preds_sum / len(self.trees)
def oob_score(self, y_true):
# 根据训练时记录的 oob_counts 求 OOB MSE(需要在 fit 后)
raise NotImplementedError("我们在 fit 中已经计算了 oob_mse_curve,可直接查看最后一项。")
-
每棵树构建完后,我们把该树对其 OOB 样本的预测累加到 oob_pred_sum,并对每个样本维护计数器 oob_counts(记录被多少棵树视为 OOB)。OOB 误差可以逐棵树计算跟踪。 -
最终 predict 对所有树预测取平均。
训练模型并进行预测
# 转成训练使用的 tensor(已在前面定义)
# X, y 已经是 torch tensors
# 划分训练/测试(用于最终验证)。我们训练森林时使用全部训练集来演示 OOB。
# 这里还是划分一下:80% 训练(用于训练与 OOB),20% 测试(独立用于最终评估)
from sklearn.model_selection import train_test_split
X_all = X
y_all = y
X_train_np, X_test_np, y_train_np, y_test_np = train_test_split(X_np, y_np, test_size=0.2, random_state=seed)
X_train = torch.from_numpy(X_train_np)
y_train = torch.from_numpy(y_train_np).unsqueeze(1)
X_test = torch.from_numpy(X_test_np)
y_test = torch.from_numpy(y_test_np).unsqueeze(1)
rf = RandomForestRegressorTorch(n_estimators=60, max_depth=10, min_samples_leaf=5, max_features='sqrt')
rf.fit(X_train, y_train, verbose=False)
# 预测训练集与测试集
pred_train = rf.predict(X_train)
pred_test = rf.predict(X_test)
from sklearn.metrics import mean_squared_error
mse_train = mean_squared_error(y_train.numpy(), pred_train.numpy())
mse_test = mean_squared_error(y_test.numpy(), pred_test.numpy())
print("训练集 MSE:", mse_train)
print("测试集 MSE:", mse_test)
-
我们把数据划分为训练集(用于训练 + OOB)和测试集(独立评估)。 -
rf.fit 会打印或返回 OOB MSE 曲线(每增加一棵树的 OOB 估计)。 -
训练完成后用 predict 对测试集评估最终泛化能力。
可视化分析
为了便于大家理解,我们绘制一下图形,方便学习~
-
数据散点与真实函数曲面(二维 → 3D 曲面) -
单棵树预测曲面(展示单树的过拟合/阶梯形结构) -
随机森林预测曲面(展示 ensemble 平滑效果) -
偏依赖图(Partial Dependence)— 对 x1 与 x2 分别绘制 -
特征重要性柱状图(通过树分裂时的方差减少累计估算) -
OOB MSE 随树数变化曲线(展示收敛稳定性) -
残差分布 & 预测 vs 实际 散点(用于诊断偏差/方差)
先写辅助函数来构建网格并绘图:
def make_grid(x1_min=0, x1_max=1, x2_min=-1, x2_max=1, steps=80):
g1 = np.linspace(x1_min, x1_max, steps)
g2 = np.linspace(x2_min, x2_max, steps)
G1, G2 = np.meshgrid(g1, g2)
grid = np.stack([G1.ravel(), G2.ravel()], axis=1).astype(np.float32)
return G1, G2, grid
# 真值函数(与数据生成时一致,但不含噪声)
def true_function(x):
# x: (n,2) numpy
x1 = x[:,0]
x2 = x[:,1]
y = np.sin(2 * np.pi * x1) * (1 + x2**2) + 0.5 * x2 + 0.3 * np.exp(-x1*3)
return y
G1, G2, grid = make_grid(0,1,-1,1,steps=80)
grid_t = torch.from_numpy(grid)
真值曲面 + 训练数据彩色散点:
y_true_grid = true_function(grid) # (steps*steps,)
fig = plt.figure(figsize=(10,8))
ax = fig.add_subplot(111, projection='3d')
surf = ax.plot_surface(G1, G2, y_true_grid.reshape(G1.shape), cmap=cm.plasma, alpha=0.9)
# 散点(训练数据)
ax.scatter(X_train[:,0].numpy(), X_train[:,1].numpy(), y_train.squeeze().numpy(), c='cyan', edgecolor='k', s=25)
ax.set_title("图1:真实函数曲面与训练数据散点", fontsize=12)
ax.set_xlabel("x1")
ax.set_ylabel("x2")
ax.set_zlabel("y")
fig.colorbar(surf, shrink=0.5, aspect=10)
plt.tight_layout()
plt.show()
-
展示我们要拟合的真实函数(无噪声曲面)与训练样本点(带噪声)。 -
你可以直观观察到目标函数对 x1 有周期性变化,对 x2 有调制效应。
一棵单独的树的预测曲面:
我们取训练好的随机森林中的第一棵树来绘制其预测曲面(单树往往有阶梯形决策边界)。
single_tree = rf.trees[0]
pred_single_grid = single_tree.predict(grid_t).numpy().reshape(G1.shape)
fig = plt.figure(figsize=(10,8))
ax = fig.add_subplot(111, projection='3d')
surf = ax.plot_surface(G1, G2, pred_single_grid, cmap=cm.jet, alpha=0.85)
ax.set_title("图2:第一棵树的预测曲面", fontsize=12)
ax.set_xlabel("x1"); ax.set_ylabel("x2"); ax.set_zlabel("pred")
fig.colorbar(surf, shrink=0.5, aspect=10)
plt.tight_layout()
plt.show()
单树通常分裂形成“阶梯化(piecewise constant)”的预测面,看起来很不平滑,容易过拟合细节和噪声。
随机森林预测曲面:
pred_rf_grid = rf.predict(grid_t).numpy().reshape(G1.shape)
fig = plt.figure(figsize=(10,8))
ax = fig.add_subplot(111, projection='3d')
surf = ax.plot_surface(G1, G2, pred_rf_grid, cmap=cm.viridis, alpha=0.9)
ax.set_title("图3:随机森林预测曲面", fontsize=12)
ax.set_xlabel("x1"); ax.set_ylabel("x2"); ax.set_zlabel("pred")
fig.colorbar(surf, shrink=0.5, aspect=10)
plt.tight_layout()
plt.show()
多棵树平均后,模型输出变得更平滑、更接近真实函数,和训练数据一起看,这就是集成降低方差的直观体现。
偏依赖图对 x1 和 x2 各一张:
偏依赖图能够展示单个特征对预测的边际影响,在所有其它特征值平均的前提下。
def partial_dependence(rf_model, X_ref, feature_index, grid_vals):
# 对于每个候选值,把该特征设置为这个值,其他保持原样,取预测平均
preds = []
for val in grid_vals:
X_tmp = X_ref.clone()
X_tmp[:, feature_index] = val
preds.append(rf_model.predict(X_tmp).mean().item())
return np.array(preds)
# 生成用于偏依赖的参考样本(使用训练集)
X_ref = X_train.clone()
grid_x1 = np.linspace(0,1,80, dtype=np.float32)
grid_x2 = np.linspace(-1,1,80, dtype=np.float32)
pd_x1 = partial_dependence(rf, X_ref, 0, torch.from_numpy(grid_x1))
pd_x2 = partial_dependence(rf, X_ref, 1, torch.from_numpy(grid_x2))
plt.figure(figsize=(12,5))
plt.subplot(1,2,1)
plt.plot(grid_x1, pd_x1, color='orangered', linewidth=2)
plt.scatter(X_train[:,0].numpy(), y_train.squeeze().numpy(), alpha=0.15, s=10, color='lightcoral')
plt.title("图4a:x1 的偏依赖曲线")
plt.xlabel("x1")
plt.ylabel("predicted y")
plt.subplot(1,2,2)
plt.plot(grid_x2, pd_x2, color='dodgerblue', linewidth=2)
plt.scatter(X_train[:,1].numpy(), y_train.squeeze().numpy(), alpha=0.15, s=10, color='lightblue')
plt.title("图4b:x2 的偏依赖曲线")
plt.xlabel("x2")
plt.ylabel("predicted y")
plt.tight_layout()
plt.show()
-
通过偏依赖图可以看到 x1 对预测的周期性影响(应当类似于 sin(2πx1) 的形状),x2 对预测有调制趋势与线性成分(在我们的真函数中 x2 存在线性项 0.5 x2 和乘以 (1+x2^2) 的调制)。 -
偏依赖可帮助解释模型:哪些特征在整体上推动预测上升或下降。
特征重要性:
我们可以通过统计每棵树每次分裂带来的方差减少并累加,得到特征的重要性估计。
# 简单通过遍历树结构,累加每次分裂带来的 impurity decrease(近似)
def compute_feature_importances(rf_model, X_train, y_train):
n_features = X_train.shape[1]
importances = np.zeros(n_features, dtype=np.float64)
# 对每棵树递归统计
def traverse(node, X_node_idx):
# X_node_idx: indices of samples in this node (numpy array)
if node.is_leaf:
return
# 当前节点的 MSE
y_node = y_train[X_node_idx]
parent_mse = ((y_node - y_node.mean())**2).mean()
# 左/右 masks
feat = node.feature
thresh = node.threshold
col = X_train[X_node_idx, feat]
left_idx = X_node_idx[col <= thresh]
right_idx = X_node_idx[col > thresh]
if len(left_idx) == 0or len(right_idx) == 0:
return
mse_left = ((y_train[left_idx] - y_train[left_idx].mean())**2).mean() if len(left_idx)>0else0.0
mse_right = ((y_train[right_idx] - y_train[right_idx].mean())**2).mean() if len(right_idx)>0else0.0
weighted_child_mse = (len(left_idx) * mse_left + len(right_idx) * mse_right) / (len(left_idx)+len(right_idx))
imp_decrease = parent_mse - weighted_child_mse
importances[feat] += imp_decrease
traverse(node.left, left_idx)
traverse(node.right, right_idx)
# 初始索引(训练集索引)
train_idx_all = np.arange(X_train.shape[0])
for tree in rf_model.trees:
traverse(tree.root, train_idx_all)
# 归一化
importances = importances / importances.sum()
return importances
# 转回 numpy 方便计算
X_train_np2 = X_train.numpy()
y_train_np2 = y_train.numpy().squeeze()
feat_imp = compute_feature_importances(rf, X_train_np2, y_train_np2)
plt.figure(figsize=(6,4))
bars = plt.bar(['x1', 'x2'], feat_imp, color=['magenta','limegreen'])
plt.title("图5:特征重要性(累计方差减少,归一化)")
for i, v in enumerate(feat_imp):
plt.text(i, v+0.01, f"{v:.3f}", ha='center')
plt.ylim(0, max(feat_imp)*1.4)
plt.show()
特征重要性反映了模型在训练时倾向于使用哪些特征来分裂,从而影响模型输出。通常会看到 x1 与 x2 的相对贡献——理论上 x1 的周期影响在我们的生成函数中很显著,x2 也重要但作用不同。
OOB MSE 随树数变化:
展示模型随着树数增加的稳定性。
oob_mse = np.array([v if v is not None else np.nan for v in rf.oob_mse_curve])
plt.figure(figsize=(8,4))
plt.plot(np.arange(1, len(oob_mse)+1), oob_mse, color='orange', marker='o')
plt.title("图6:OOB MSE 随树数的变化")
plt.xlabel("树的数量")
plt.ylabel("OOB MSE")
plt.grid(alpha=0.3)
plt.show()
OOB MSE 是训练过程中的一种内置交叉验证(可替代验证集)评估方式。通常 OOB MSE 会在增加树数时先下降然后趋于平稳。它可以帮助选择合适的树数以及判断是否过拟合/欠拟合。
残差直方图 & 预测 vs 实际 散点:
residuals = (pred_test - y_test).numpy().squeeze()
plt.figure(figsize=(12,5))
plt.subplot(1,2,1)
plt.hist(residuals, bins=30, color='fuchsia', alpha=0.85)
plt.title("图7a:测试集残差分布")
plt.xlabel("残差 (pred - actual)")
plt.ylabel("频数")
plt.subplot(1,2,2)
plt.scatter(y_test.numpy(), pred_test.numpy(), c=pred_test.numpy().squeeze(), cmap='coolwarm', edgecolor='k', s=40)
minv = min(y_test.min().item(), pred_test.min().item())
maxv = max(y_test.max().item(), pred_test.max().item())
plt.plot([minv, maxv], [minv, maxv], '--', color='gray')
plt.title("图7b:预测 vs 实际(测试集)")
plt.xlabel("真实 y")
plt.ylabel("预测 y")
plt.colorbar(label='pred value')
plt.tight_layout()
plt.show()
-
残差直方图显示残差是否近似均值为 0 的对称分布;若有偏则说明模型存在偏差(bias)。 -
预测 vs 实际 散点图可以直观看到模型是否在某些区间系统性高估或低估。
总结
总的来说,这类基于树的模型用起来特别省心,基本不用做特征缩放,靠默认参数往往就能跑出不错的效果。
还能通过特征重要性、偏依赖图这些方式简单解释模型,对缺失值和异常点也有一定容忍度,抗噪声能力不错,训练时不同的树还能并行跑,效率也有保障。
不过它也有明显短板,处理高维稀疏数据比如文本的高维表示时,表现不如一些线性模型,外推能力也很差,没法像线性模型那样推断趋势,而且树的数量一多,模型体积就会变大,占用内存也多。
如果追求更高的准确率和更快的训练速度,直接换成XGBoost、LightGBM、CatBoost这类梯度提升树模型会更合适。
最后
最近准备了16大块的内容,124个算法问题的总结,完整的机器学习小册,免费领取~

