第2428篇：AI工程的代码评审规范——AI项目特有的代码审查关注点
2026/4/30大约 6 分钟
第2428篇：AI工程的代码评审规范——AI项目特有的代码审查关注点
适读人群：AI工程师和技术负责人 | 阅读时长：约12分钟 | 核心价值：超越通用代码规范，专门针对AI项目的代码评审检查要点
我做过上百次AI项目的代码评审，发现AI代码有一批反复出现的问题，在传统软件的评审清单里完全找不到。
比如训练测试集污染——这在普通业务代码里根本不是概念，但在AI代码里可以导致整个模型的可信度崩塌，而且很难被发现。
再比如特征不一致性问题——训练时用了某个特征的计算方式，推理时用了另一个计算方式，模型在测试集上表现完美，上线就垮掉。
这类问题靠通用的代码规范抓不住。今天整理一套AI特有的代码评审检查框架。
一、数据处理代码的关键检查点
1.1 训练测试集污染检查
最严重、最隐蔽的问题之一
# 错误示范1：在划分测试集之前做了依赖整体数据的处理
from sklearn.preprocessing import StandardScaler
import pandas as pd
import numpy as np
# BAD: 这会造成数据泄露！
def bad_preprocessing(df):
scaler = StandardScaler()
# 对全部数据fit，再split，均值和标准差包含了测试集信息
df['feature_scaled'] = scaler.fit_transform(df[['feature']])
X_train, X_test = train_test_split(df) # 泄露！
return X_train, X_test
# GOOD: 先split，再在训练集上fit，用同一个scaler transform测试集
def good_preprocessing(df):
X_train_raw, X_test_raw = train_test_split(df)
scaler = StandardScaler()
X_train_raw['feature_scaled'] = scaler.fit_transform(X_train_raw[['feature']])
X_test_raw['feature_scaled'] = scaler.transform(X_test_raw[['feature']]) # 注意：transform不是fit_transform
return X_train_raw, X_test_raw
# 评审时问的关键问题：
# Q: 任何在fit/transform之前的操作，是否在整个数据集上进行了？
# Q: 测试集中的统计量（均值、方差、分布）是否被用于训练数据的处理？更隐蔽的污染形式：
# 错误示范2：目标编码时的污染
# BAD: 对全量数据计算目标均值，再split
def bad_target_encoding(df, cat_col, target_col):
target_mean = df.groupby(cat_col)[target_col].mean()
df[f'{cat_col}_encoded'] = df[cat_col].map(target_mean)
return df # 泄露！
# GOOD: 只用训练集的统计量做编码
def good_target_encoding(X_train, X_test, cat_col, target_col):
# 仅从训练集学习映射
target_mean = X_train.groupby(cat_col)[target_col].mean()
X_train[f'{cat_col}_encoded'] = X_train[cat_col].map(target_mean)
X_test[f'{cat_col}_encoded'] = X_test[cat_col].map(target_mean) # 用训练集的统计量
return X_train, X_test1.2 特征一致性检查
训练和推理的特征工程代码必须完全一致
# 评审时最应该寻找的危险模式
# BAD: 训练和推理用了不同的特征计算逻辑
class TrainingFeatureBuilder:
def build(self, data):
# 训练时用的是过去30天的均值
data['avg_purchase'] = data.groupby('user_id')['amount'].rolling(30).mean()
return data
class InferenceFeatureBuilder:
def build(self, data):
# 推理时用的是过去7天的均值（逻辑不一致！）
data['avg_purchase'] = data.groupby('user_id')['amount'].rolling(7).mean()
return data
# GOOD: 训练和推理使用完全相同的特征工程类
class FeatureBuilder:
"""一个类，同时用于训练和推理"""
def __init__(self, window_days: int = 30):
self.window_days = window_days
def build(self, data: pd.DataFrame) -> pd.DataFrame:
# 通过参数控制，保证一致性
data['avg_purchase'] = (
data.groupby('user_id')['amount']
.rolling(self.window_days)
.mean()
)
return data
def save(self, path: str):
"""序列化特征工程对象（包括所有参数）"""
import joblib
joblib.dump(self, path)
@staticmethod
def load(path: str):
import joblib
return joblib.load(path)
# 评审检查点：
# Q: 特征工程代码是否在训练和推理中共用同一份代码？
# Q: 是否有任何特征的计算逻辑在两处不同？
# Q: 模型保存时是否同时保存了特征工程的参数/对象？1.3 缺失值处理检查
# BAD: 训练时填充了NaN，但推理时没有处理
def train_model(X, y):
X_filled = X.fillna(X.mean()) # 训练时填充
model.fit(X_filled, y)
model.save("model.pkl")
def predict(X):
# 推理时忘记了填充，遇到NaN会报错或产生错误结果
return model.predict(X)
# GOOD: 把缺失值处理纳入pipeline
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
pipeline = Pipeline([
('imputer', SimpleImputer(strategy='mean')), # 这里的mean是从训练集计算的
('model', your_model)
])
pipeline.fit(X_train, y_train) # imputer从训练集学习均值
# 推理时pipeline会自动用训练集的均值填充
predictions = pipeline.predict(X_test)
# 评审检查点：
# Q: 所有数据转换操作是否都在sklearn pipeline或等效机制中封装？
# Q: 推理代码是否处理了训练时未见过的类别值？
# Q: 推理代码是否处理了NaN和无穷大值？二、模型代码的关键检查点
2.1 随机性控制
# BAD: 没有固定随机种子，每次运行结果不同
def train():
X_train, X_test = train_test_split(data, test_size=0.2) # 每次分割不同
model = RandomForestClassifier()
model.fit(X_train, y_train)
return evaluate(model, X_test)
# GOOD: 所有随机操作都固定种子
def train(random_seed: int = 42):
import numpy as np
import random
import torch
# 固定所有随机源
random.seed(random_seed)
np.random.seed(random_seed)
torch.manual_seed(random_seed)
if torch.cuda.is_available():
torch.cuda.manual_seed_all(random_seed)
X_train, X_test = train_test_split(data, test_size=0.2, random_state=random_seed)
model = RandomForestClassifier(random_state=random_seed)
model.fit(X_train, y_train)
return evaluate(model, X_test)
# 评审检查点：
# Q: 代码中是否有不可控的随机操作（没有固定seed）？
# Q: 随机种子是否作为参数传入，而不是硬编码？2.2 超参数管理
# BAD: 超参数散落在代码各处
class MyModel:
def train(self, data):
lr = 0.001 # 硬编码在方法里
model = nn.Sequential(
nn.Linear(128, 64), # 层大小硬编码
nn.ReLU(),
nn.Linear(64, 1)
)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001) # 另一处硬编码
...
# GOOD: 超参数集中管理
from dataclasses import dataclass
@dataclass
class ModelConfig:
"""所有超参数在一个地方定义"""
learning_rate: float = 0.001
hidden_dims: list = None
batch_size: int = 32
max_epochs: int = 100
dropout_rate: float = 0.1
random_seed: int = 42
def __post_init__(self):
if self.hidden_dims is None:
self.hidden_dims = [128, 64]
def to_dict(self) -> dict:
from dataclasses import asdict
return asdict(self)
class MyModel:
def __init__(self, config: ModelConfig):
self.config = config
def train(self, data):
model = self._build_model()
optimizer = torch.optim.Adam(
model.parameters(),
lr=self.config.learning_rate
)
# 所有超参数从config获取
...
def _build_model(self):
layers = []
input_dim = self.config.hidden_dims[0]
for dim in self.config.hidden_dims[1:]:
layers.extend([
nn.Linear(input_dim, dim),
nn.ReLU(),
nn.Dropout(self.config.dropout_rate)
])
input_dim = dim
return nn.Sequential(*layers)三、评估代码的关键检查点
3.1 评估指标的陷阱
# 评审时要问的问题：为什么用这个指标？
# 常见陷阱1：类别不平衡时用准确率
def bad_evaluation(model, X_test, y_test):
accuracy = model.score(X_test, y_test)
print(f"Accuracy: {accuracy:.3f}")
# 如果正例只占1%，全预测为负例准确率也有99%！
# GOOD: 不平衡数据应该用F1/AUC/PR曲线
from sklearn.metrics import (
f1_score, roc_auc_score, precision_recall_curve,
average_precision_score, confusion_matrix, classification_report
)
def good_evaluation(model, X_test, y_test):
y_pred = model.predict(X_test)
y_prob = model.predict_proba(X_test)[:, 1]
print("分类报告（包含每个类的详细指标）：")
print(classification_report(y_test, y_pred))
print(f"ROC-AUC: {roc_auc_score(y_test, y_prob):.3f}")
print(f"PR-AUC: {average_precision_score(y_test, y_prob):.3f}")
# 显示混淆矩阵帮助理解错误分布
print("混淆矩阵:")
print(confusion_matrix(y_test, y_pred))四、AI评审时常被忽视的问题
问题一：模型保存是否完整
# BAD: 只保存了模型权重，没有保存preprocessing
joblib.dump(model, 'model.pkl') # 只有模型
# GOOD: 保存完整的推理pipeline
artifact = {
'model': model,
'feature_builder': feature_builder, # 特征工程
'scaler': scaler, # 数据转换
'feature_names': feature_names, # 特征名列表（确保顺序一致）
'model_version': '1.2.0',
'trained_at': datetime.now().isoformat(),
'training_data_version': 'v3.1'
}
joblib.dump(artifact, 'model_artifact.pkl')问题二：推理时的输入校验
# BAD: 没有输入校验
def predict(features: dict) -> float:
x = np.array([features[f] for f in FEATURE_NAMES])
return model.predict(x.reshape(1, -1))[0]
# GOOD: 严格校验输入
def predict(features: dict) -> float:
# 检查特征是否齐全
missing = set(FEATURE_NAMES) - set(features.keys())
if missing:
raise ValueError(f"缺少必要特征: {missing}")
# 检查特征类型和范围
for feature_name in FEATURE_NAMES:
value = features[feature_name]
if not isinstance(value, (int, float)):
raise TypeError(f"特征 {feature_name} 应为数值类型")
if np.isnan(value) or np.isinf(value):
raise ValueError(f"特征 {feature_name} 包含无效值: {value}")
x = np.array([features[f] for f in FEATURE_NAMES])
return float(model.predict(x.reshape(1, -1))[0])好的AI代码评审，是保证系统在生产环境中不意外崩溃的最后一道关。