这份文档面向研究生日常科研开发,目标不是“把代码写得很花”,而是让代码真正具备科研价值。
一份合格的科研代码,至少要满足以下六个要求:
很多论文难以复现、结果前后不一致、后续学生无法接手,往往不是算法本身的问题,而是代码管理、实验记录和自动化流程做得不够规范。
版本管理的核心目的不是“备份代码”,而是:
错误示例:
final_code.py
final_code2.py
final_code_final.py
final_code_final_new.py
final_code_final_new2_真的最终版.py
这种做法最大的问题是:
正确做法是:
建议一个科研项目至少组织成下面这样:
project/
│
├── configs/ # 配置文件
│ ├── default.json
│ ├── noise_sweep.json
│ ├── field_data.json
│ └── ablation_hash.json
│
├── data/ # 数据或数据说明
│ ├── raw/
│ ├── processed/
│ └── README.md
│
├── scripts/ # Bash / 调度脚本
│ ├── run_train.sh
│ ├── run_batch.sh
│ ├── run_parallel.sh
│ └── summarize_results.sh
│
├── src/ # 核心源码
│ ├── train.py
│ ├── test.py
│ ├── dataset.py
│ ├── models/
│ ├── losses.py
│ ├── trainer.py
│ ├── metrics.py
│ ├── utils/
│ └── visualization/
│
├── experiments/ # 每次实验独立输出目录
│ ├── exp_20260416_101500_snr10_seed42/
│ ├── exp_20260416_102300_snr12_seed43/
│ └── summary/
│
├── notebooks/ # 仅用于分析和临时探索
├── tests/ # 单元测试或简单检查脚本
├── README.md
├── requirements.txt
├── environment.yml
└── .gitignore
优点:
参考文档:
a = np.zeros((9, 20))
tmp = model(x)
dn = data_norm
dr = result
问题:
num_noise_levels = 9
num_trials = 20
snr_results = np.zeros((num_noise_levels, num_trials))
predicted_wavefield = model(input_coordinates)
normalized_seismic_data = normalize(seismic_data)
reconstructed_section = reconstruction_result
推荐:
train_inr.pytest_interpolation.pyplot_snr_curve.pysummarize_experiments.pyvisualize_dashboard.py避免:
test.pynew.pyaaa.py临时代码.py错误示例:
git commit -m "update"
git commit -m "fix"
git commit -m "修改"
这种 commit 几乎没有任何价值。
推荐写法:
git commit -m "add configurable noise sweep for SNR experiments"
git commit -m "fix receiver coordinate normalization bug"
git commit -m "refactor training loop into Trainer class"
git commit -m "save final config and command to output directory"
git commit -m "add interactive experiment dashboard"
好的 commit 信息至少要说明:
提交格式:<type>: <subject>(类型+简短描述),类型可选:
feat:新功能(如添加维度解耦模块)fix:修复bug(如修正SNR计算错误)docs:文档更新(如补充README参数说明)refactor:代码重构(不改变功能,如封装函数)正反示例:
❌ 坏提交:git commit -m "改了点代码"
✅ 好提交:git commit -m "feat: 添加五维地震波场重建的张量隐式网络模块"
最简单实用的策略:
main:稳定可复现版本dev:日常开发feature/...:新功能exp/...:临时实验思路例如:
feature/add-hash-encoding
feature/add-result-dashboard
exp/test-lora-adaptation
exp/try-lower-learning-rate
例如:
git tag v1.0-submitted
git tag v1.1-camera-ready
git tag exp-best-synthetic
git tag exp-field-data-final
这样后续:
都会方便很多。
.gitignore 核心作用:忽略临时文件、依赖包、大体积实验结果,避免仓库臃肿。否则会把缓存、结果、模型权重和临时文件全提交上去。
科研场景示例(.gitignore 文件内容):
__pycache__/
*.pyc
*.pkl
*.npy
*.npz
*.mat
*.h5
*.log
*.ckpt
*.pth
*.pt
experiments/
wandb/
.vscode/
.ipynb_checkpoints/
实验记录的核心原则:
任何一张图、一个表、一个数值结论,都必须能追溯到具体代码版本、参数配置和运行环境。
如果做不到,实验基本上就不算可复现。
错误示例:
lr = 1e-3
epochs = 3000
batch_size = 4096
rank = 16
hidden_dim = 128
问题:
正确做法:
例如 configs/default.json:
{
"experiment_name": "default_inr",
"dataset_name": "synthetic_5d",
"snr": 10,
"seed": 42,
"epochs": 3000,
"learning_rate": 0.001,
"batch_size": 4096,
"hidden_dim": 128,
"num_layers": 4,
"activation": "sine",
"model_type": "inr",
"save_root": "experiments",
"num_trials": 20,
"noise_levels": [5, 7, 9, 11, 13, 15]
}
建议:
experiments/
└── exp_20260416_153000_snr10_seed42/
├── config_original.json
├── config_used.json
├── run_command.sh
├── launcher.sh
├── git_commit.txt
├── env_info.txt
├── train.log
├── metrics.json
├── checkpoint_best.pth
├── checkpoint_last.pth
├── curves/
└── figures/
这样以后找结果非常方便。
除了保存原始配置文件 config_original.json,还必须保存:
config_used.json:最终生效参数run_command.sh:本次实验启动命令launcher.sh:发起实验的 Bash 脚本副本因为真正重要的不只是“原始配置长什么样”,而是:
import os
import json
import argparse
from datetime import datetime
def load_config(config_path):
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
def save_json(data, path):
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=4, ensure_ascii=False)
parser = argparse.ArgumentParser()
parser.add_argument("--config", type=str, required=True)
parser.add_argument("--snr", type=int, default=None)
parser.add_argument("--seed", type=int, default=None)
parser.add_argument("--output_dir", type=str, default=None)
args = parser.parse_args()
config = load_config(args.config)
config["config_path"] = args.config
if args.snr is not None:
config["snr"] = args.snr
if args.seed is not None:
config["seed"] = args.seed
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if args.output_dir is None:
exp_name = f"exp_{timestamp}_snr{config['snr']}_seed{config['seed']}"
output_dir = os.path.join(config["save_root"], exp_name)
else:
output_dir = args.output_dir
os.makedirs(output_dir, exist_ok=True)
save_json(config, os.path.join(output_dir, "config_used.json"))
import os
import sys
import shlex
import platform
import subprocess
command_str = "python " + " ".join(shlex.quote(arg) for arg in sys.argv)
with open(os.path.join(output_dir, "run_command.sh"), "w", encoding="utf-8") as f:
f.write("#!/bin/bash\n")
f.write(command_str + "\n")
try:
git_commit = subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip()
except Exception:
git_commit = "unknown"
with open(os.path.join(output_dir, "git_commit.txt"), "w", encoding="utf-8") as f:
f.write(git_commit + "\n")
env_lines = [f"python_version: {platform.python_version()}",
f"platform: {platform.platform()}"]
with open(os.path.join(output_dir, "env_info.txt"), "w", encoding="utf-8") as f:
f.write("\n".join(env_lines) + "\n")
错误做法:
推荐:
例如:
import logging
import os
logging.basicConfig(
filename=os.path.join(output_dir, "train.log"),
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
logging.info("Training started")
logging.info(f"snr={config['snr']}, seed={config['seed']}")
可扩展性的核心:
今天代码能跑一个实验,明天来了新的数据或者任务也能方便地扩展成十个、一百个实验,而不是每次都推倒重来。
错误示例:
snr_list_noise = np.zeros((9, 20))
这里的 9 和 20 是典型魔法数字。别人看不出来是什么意思,自己以后也容易忘。
正确做法:
num_noise_levels = config["num_noise_levels"]
num_trials = config["num_trials"]
snr_list_noise = np.zeros((num_noise_levels, num_trials))
配置文件中写:
{
"num_noise_levels": 9,
"num_trials": 20
}
错误做法:
noise_levels = [5, 7, 9, 11, 13, 15]
num_trials = 20
hidden_dim = 128
num_layers = 4
lr = 1e-3
推荐:
{
"noise_levels": [5, 7, 9, 11, 13, 15],
"num_trials": 20,
"hidden_dim": 128,
"num_layers": 4,
"learning_rate": 0.001
}
错误写法:
for epoch in range(epochs):
# 读数据
# 前向传播
# 计算loss
# backward
# 更新参数
# 验证
# 保存模型
# 打印日志
# 画图
这会让主程序越来越长,最终难以维护。
推荐拆分:
def train_one_epoch(model, dataloader, optimizer, criterion, device):
...
def validate(model, dataloader, criterion, device):
...
def save_checkpoint(model, optimizer, epoch, save_path):
...
def log_metrics(metrics, output_dir):
...
主流程就会更清晰:
for epoch in range(config["epochs"]):
train_loss = train_one_epoch(model, train_loader, optimizer, criterion, device)
val_loss = validate(model, val_loader, criterion, device)
log_metrics({"epoch": epoch, "train_loss": train_loss, "val_loss": val_loss}, output_dir)
推荐拆分:
dataset.py:数据读入与预处理model.py:模型结构losses.py:损失函数trainer.py:训练逻辑metrics.py:评价指标summary.py:汇总实验结果dashboard.py:可视化交互界面utils.py:通用工具函数错误示例:
if mode == "train":
# 一套数据处理
elif mode == "val":
# 基本重复一遍
elif mode == "test":
# 再复制一遍
更好的做法:
def prepare_batch(batch, device):
...
return inputs, targets
训练、验证、测试都调用它。
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--config", type=str, required=True)
parser.add_argument("--seed", type=int, default=42)
parser.add_argument("--output_dir", type=str, required=True)
args = parser.parse_args()
这样便于脚本化、自动化、批量运行。
自动化的核心思想:
不要让人去手工重复做本来可以由代码自动完成的事情。
建议自动化的内容包括:
错误流程:
train.pysnr = 10snr = 12seed = 43这种流程非常低效,而且容易出错。
推荐 Bash 脚本:
#!/bin/bash
CONFIG=configs/default.json
SAVE_ROOT=experiments
mkdir -p "$SAVE_ROOT"
for snr in {5..15..2}; do
for seed in 42 43 44; do
timestamp=$(date +"%Y%m%d_%H%M%S")
exp_name="exp_${timestamp}_snr${snr}_seed${seed}"
output_dir="${SAVE_ROOT}/${exp_name}"
mkdir -p "$output_dir"
cmd="python train.py --config ${CONFIG} --snr ${snr} --seed ${seed} --output_dir ${output_dir}"
echo "#!/bin/bash" > "${output_dir}/run_command.sh"
echo "$cmd" >> "${output_dir}/run_command.sh"
cp "$CONFIG" "${output_dir}/config_original.json"
cp "$0" "${output_dir}/launcher.sh"
bash "${output_dir}/run_command.sh" 2>&1 | tee "${output_dir}/train.log"
done
done
很多学生的常见问题是:
metrics.json正确做法是再写一个汇总脚本,例如 summarize_experiments.py,自动扫描所有实验目录并生成:
summary.csvsummary.json示例:
import os
import json
import pandas as pd
def collect_metrics(experiments_root):
records = []
for exp_name in os.listdir(experiments_root):
exp_dir = os.path.join(experiments_root, exp_name)
metrics_path = os.path.join(exp_dir, "metrics.json")
config_path = os.path.join(exp_dir, "config_used.json")
if os.path.exists(metrics_path) and os.path.exists(config_path):
with open(metrics_path, "r", encoding="utf-8") as f:
metrics = json.load(f)
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
record = {}
record.update(config)
record.update(metrics)
record["experiment_dir"] = exp_dir
records.append(record)
return pd.DataFrame(records)
df = collect_metrics("experiments")
df.to_csv("experiments/summary/summary.csv", index=False)
推荐形成如下流程:
也就是说,尽量做到:
从启动实验到看到总结结论,中间少依赖手工操作。
在科研中,很多实验彼此独立,例如:
这种实验非常适合并行。
但并行不是简单加一个 & 就完事了,而是要做到:
& bash run_command.sh > train.log 2>&1 &
优点:
缺点:
例如同时最多跑 2 个:
MAX_JOBS=2
job_count=0
for snr in {5..15..2}; do
for seed in 42 43 44; do
run_job "$snr" "$seed" &
((job_count++))
if (( job_count % MAX_JOBS == 0 )); then
wait
fi
done
done
wait
例如:
CUDA_VISIBLE_DEVICES=0 python train.py ...
CUDA_VISIBLE_DEVICES=1 python train.py ...
或者在脚本中轮换设备:
gpus=(0 1)
gpu_index=0
for snr in {5..15..2}; do
for seed in 42 43 44; do
gpu=${gpus[$gpu_index]}
gpu_index=$(( (gpu_index + 1) % ${#gpus[@]} ))
CUDA_VISIBLE_DEVICES=$gpu python train.py --snr $snr --seed $seed &
done
done
wait
但必须考虑:
很多科研代码的问题不是“永远不能跑”,而是:
所以科研代码一定要有异常处理意识。
错误写法:
with open(config_path, "r") as f:
config = json.load(f)
更稳妥的写法:
import os
import json
if not os.path.exists(config_path):
raise FileNotFoundError(f"Config file not found: {config_path}")
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
required_keys = ["snr", "seed", "epochs", "learning_rate", "save_root"]
for key in required_keys:
if key not in config:
raise KeyError(f"Missing required config key: {key}")
if not os.path.exists(data_path):
raise FileNotFoundError(f"Data file not found: {data_path}")
data = np.load(data_path)
if data.ndim != 5:
raise ValueError(f"Expected 5D seismic data, but got shape {data.shape}")
例如在 Bash 中,可以写:
if bash "${output_dir}/run_command.sh" > "${output_dir}/train.log" 2>&1; then
echo "SUCCESS" > "${output_dir}/status.txt"
else
echo "FAILED" > "${output_dir}/status.txt"
echo "Experiment failed: ${output_dir}"
fi
这样即使某个实验失败了,其他实验还能继续。
例如:
try:
train_loss = train_one_epoch(model, train_loader, optimizer, criterion, device)
except RuntimeError as e:
logging.exception("Training failed due to runtime error")
raise e
或者在外层统一处理:
try:
main()
except Exception as e:
logging.exception("Experiment failed")
raise
这样日志里至少能保留堆栈信息,便于定位问题。
assert 的作用不是替代异常处理,而是:
在程序假定某些条件必然成立时,尽早检查这些条件。
如果条件不成立,就立即终止,而不是让错误继续传递下去。
assert config["epochs"] > 0, "epochs must be positive"
assert config["learning_rate"] > 0, "learning rate must be positive"
assert config["batch_size"] > 0, "batch size must be positive"
assert config["snr"] >= 0, "snr should be non-negative"
assert data.ndim == 5, f"Expected 5D data, got {data.ndim}D"
assert data.shape[0] > 0, "First dimension must be non-empty"
prediction = model(inputs)
assert prediction.shape == targets.shape, \
f"Prediction shape {prediction.shape} does not match target shape {targets.shape}"
loss = criterion(prediction, targets)
assert torch.isfinite(loss).all(), "Loss contains NaN or Inf"
也可以检查梯度:
for name, param in model.named_parameters():
if param.grad is not None:
assert torch.isfinite(param.grad).all(), f"Gradient has NaN or Inf in {name}"
assert os.path.exists(metrics_path), f"Missing metrics file: {metrics_path}"
assert os.path.exists(config_path), f"Missing config file: {config_path}"
适合 assert 的场景:
不适合只用 assert 的场景:
这些情况仍然应该用明确异常处理。
这是很多学生最容易忽略、但实际上非常重要的一点。
科研代码的目标不只是“输出很多实验目录”,而是:
实验做完后,能够高效总结规律,支持决策和写论文。
如果没有一个好的可视化总结流程,最后就会出现:
metrics.jsontrain.logpng所以建议增加一个交互式实验总结界面。
建议至少支持下面几个功能:
可以分为两层:
例如:
例如使用:
其中对研究生最友好的通常是 Streamlit,因为简单直接。
例如 dashboard.py:
import os
import pandas as pd
import streamlit as st
summary_path = "experiments/summary/summary.csv"
df = pd.read_csv(summary_path)
st.title("Experiment Summary Dashboard")
selected_model = st.selectbox("Model Type", sorted(df["model_type"].dropna().unique()))
selected_snr = st.multiselect("SNR", sorted(df["snr"].dropna().unique()), default=sorted(df["snr"].dropna().unique()))
filtered_df = df[(df["model_type"] == selected_model) & (df["snr"].isin(selected_snr))]
st.write("Filtered Results")
st.dataframe(filtered_df)
st.line_chart(filtered_df.groupby("snr")["test_snr"].mean())
st.bar_chart(filtered_df.groupby("seed")["test_snr"].mean())
运行:
streamlit run dashboard.py
这类界面非常适合:
整个交互界面不应该直接去逐个读原始实验目录,而应该基于统一的 summary.csv。
推荐流程:
config_used.json 和 metrics.jsonsummarize_experiments.py 扫描全部实验目录summary.csvdashboard.py 直接读取 summary.csv这样好处是:
除了图表和表格,也可以自动生成一些文本统计信息,例如:
例如:
best_row = df.loc[df["test_snr"].idxmax()]
print("Best experiment:")
print(best_row[["experiment_dir", "snr", "seed", "test_snr"]])
进一步还可以自动输出一段总结:
在当前实验中,SNR=11、seed=43 的模型表现最好,test_snr 达到 12.53 dB。
从整体趋势来看,随着噪声水平增加,重建性能逐渐下降,但在 SNR=9 到 11 区间模型表现相对稳定。
不同随机种子之间存在一定波动,建议最终报告结果时同时汇报均值与标准差。
这种自动总结对写周报、组会和论文都很有帮助。
建议大家逐步把项目组织成下面这条完整流水线:
summary.csv 和静态图下面这段可以直接发给学生:
科研代码最重要的不是“先跑出来”,而是做到:
参数留痕、命令留痕、版本留痕、日志留痕、异常可查、流程自动、结论可视、结果可复现。
只有做到这些,代码才真正具备科研积累价值,而不只是一次性的“能跑脚本”。