Python 基础体系 · 第 110/112 篇。示例统一以 Python 3.14 为语言基线;第三方库使用与其兼容的现代稳定版本,版本敏感行为会单独说明。

Python 实验可复现:随机种子、环境、数据、制品和运行记录

“同样的代码再次运行,应该得到同样的结果”是实验可复现的直觉,但它并不是一个只靠 random.seed(42) 就能满足的条件。

一个实验通常可以抽象为:

Y=F(C,E,D,R,I,T)Y = F(C, E, D, R, I, T)

其中:

  • CC:代码(Code),包括源码、Notebook 和配置;
  • EE:环境(Environment),包括 Python、依赖包、操作系统、CPU、BLAS 等;
  • DD:输入数据(Data),包括原始数据、清洗结果和外部数据版本;
  • RR:随机状态(Random state),包括 Python、NumPy 以及其他库的随机数生成器;
  • II:运行输入(Inputs),包括命令行参数、环境变量和配置文件;
  • TT:运行时上下文(Trace),包括时间、时区、区域设置、并发度和硬件行为;
  • YY:输出,包括指标、模型、图表、日志和中间制品。

只有当再次运行时这些影响因素都相同,或者差异被明确限制在允许范围内,才可能得到相同的结果。

因此,“实验可复现”至少有三种强度:

  1. 运行可复现:别人可以按照记录重新运行程序。
  2. 结果可复现:重新运行得到相同或在容差内相同的指标。
  3. 位级可复现:输出文件的每一个字节都相同。

这三者不是同一个目标。浮点并行计算可能让指标相同,但模型文件的字节不同;CSV 的行顺序可能变化,但统计结论不变;更换 NumPy 版本后,随机数序列甚至可能发生变化。


一、先区分“确定性”“随机性”和“可复现性”

1. 确定性不是没有随机数

确定性指的是:给定相同的完整输入和状态,计算过程产生唯一输出。

随机算法通常不是“真正不可预测”的算法,而是伪随机算法(Pseudo-Random Number Generator,PRNG)。它可以表示为:

Sn+1=G(Sn)S_{n+1} = G(S_n)

Xn=H(Sn)X_n = H(S_n)

其中:

  • SnS_n 是第 nn 步的内部状态;
  • GG 是状态转移函数;
  • HH 是从状态生成随机值的函数;
  • 初始状态 S0S_0 由种子(seed)决定。

如果初始状态、生成器算法和调用顺序都相同,就可以产生相同的伪随机序列。

种子并不是“随机数本身”,而是初始化随机状态的输入:

S0=Init(seed,algorithm)S_0 = \operatorname{Init}(\text{seed}, \text{algorithm})

所以仅记录整数 42 并不足以描述实验。还必须知道:

  • 使用的是哪个随机库;
  • 使用的是哪个生成器算法;
  • 生成器被调用了多少次;
  • 每次调用的参数是否相同;
  • 调用顺序是否发生变化。

Python 标准库的 random 模块使用伪随机算法,并且模块级函数实际上绑定到一个隐藏的 Random 实例。也可以显式创建多个互不共享状态的 random.Random 实例。Python 文档只对部分跨版本行为作出保证:兼容的播种方式配合同一种子时,random() 的序列应保持一致;其他算法和播种细节可能随 Python 版本变化。(docs.python.org)

2. “固定种子”只约束了一部分输入

下面的代码并不一定能让整个实验可复现:

import random
import numpy as np

random.seed(42)

values = np.random.random(3)
print(values)

它只固定了 Python 标准库 random 的状态,却没有固定 NumPy 的随机状态。NumPy 当前推荐使用 Generatordefault_rng(),而不是依赖全局随机状态。default_rng(seed) 会根据种子创建一个生成器;省略种子时,NumPy 会从操作系统获取不可预测熵。(numpy.org)

正确的局部写法是:

import random
import numpy as np

py_rng = random.Random(42)
np_rng = np.random.default_rng(42)

print(py_rng.random())
print(np_rng.random())

这里有两个独立的随机状态:

Python Random(42) ──> Python 逻辑中的随机值
NumPy Generator    ──> 数组、采样和分布计算中的随机值

一个生成器消耗状态后,另一个生成器不会受到影响。这比在程序入口处修改全局状态更容易测试和审计。


二、随机种子:从全局状态转向显式随机流

1. 为什么全局 seed() 容易失控

下面的函数看似简单:

import numpy as np

def choose_rows(df, n):
    np.random.seed(42)
    return df.sample(n=n)

它存在两个问题。

第一,函数修改了进程级的全局 NumPy 随机状态。其他函数如果在它前后调用 np.random,行为就会发生变化。

第二,df.sample() 默认会使用当前的 NumPy 随机状态。调用方看不到随机状态从哪里来,也无法将多个实验步骤拆分成独立的随机流。

更清晰的方式是把生成器作为依赖传入:

import numpy as np
import pandas as pd

def choose_rows(df: pd.DataFrame, n: int, rng: np.random.Generator):
    indices = rng.choice(df.index.to_numpy(), size=n, replace=False)
    return df.loc[indices]

df = pd.DataFrame({"value": range(10)})
rng = np.random.default_rng(42)

sample = choose_rows(df, n=3, rng=rng)
print(sample)

这里的因果关系是显式的:

seed=42
   │
   ▼
Generator
   │
   ├── choose_rows
   ├── shuffle
   └── augmentation

函数不再偷偷读取某个全局状态,而是明确消耗传入的随机流。

pandas 的抽样 API 支持通过 random_state 传入整数、NumPy 的 GeneratorBitGenerator 或旧式 RandomState;如果传入 None,则会使用当前的 NumPy 随机状态。(pandas.pydata.org)

例如:

import numpy as np
import pandas as pd

df = pd.DataFrame({
    "label": ["a", "a", "b", "b", "c", "c"],
    "value": [0, 1, 2, 3, 4, 5],
})

rng = np.random.default_rng(123)

result = df.groupby("label").sample(
    n=1,
    random_state=rng,
)

print(result.sort_values("label"))

需要注意:一个 Generator 是有状态的。下面两次调用通常不会得到同一个样本:

rng = np.random.default_rng(123)

first = df.sample(n=3, random_state=rng)
second = df.sample(n=3, random_state=rng)

因为第二次调用读取的是第一次调用之后的状态。若两个步骤需要独立且可重复的结果,应为它们分配不同的随机流,而不是反复复用同一个全局生成器。

2. 用命名随机流避免“加一行代码就全变”

实验代码经常会发生这种情况:

rng = np.random.default_rng(42)

train = rng.permutation(data)
noise = rng.normal(size=len(data))

如果以后在 permutation() 之前新增一次随机调用,noise 的全部结果都会改变。这种变化可能是合理的,但会让局部调试困难。

可以为不同任务派生独立的随机流:

from numpy.random import SeedSequence, default_rng

root = SeedSequence(42)
train_ss, noise_ss, eval_ss = root.spawn(3)

train_rng = default_rng(train_ss)
noise_rng = default_rng(noise_ss)
eval_rng = default_rng(eval_ss)

此时:

  • train_rng 只负责训练集打乱;
  • noise_rng 只负责噪声;
  • eval_rng 只负责评估抽样。

NumPy 的 SeedSequence.spawn() 用于从一个根熵派生多个独立生成器;官方文档将其作为并行和多任务随机流的重要机制。(numpy.org)

这并不意味着数学上绝对证明了所有流永不重叠,而是通过设计使它们在实际用途下具有极高概率的独立性。工程上仍应记录:

record = {
    "root_entropy": root.entropy,
    "streams": {
        "train": train_ss.spawn_key,
        "noise": noise_ss.spawn_key,
        "eval": eval_ss.spawn_key,
    },
}

3. 并发场景不能简单地给每个进程使用同一个种子

错误写法:

def worker(_):
    rng = np.random.default_rng(42)
    return rng.random(3)

如果每个进程都执行同样的代码,每个进程都会生成相同的三组随机数。

更好的方式是为每个 worker 派生子种子:

from numpy.random import SeedSequence, default_rng

root = SeedSequence(20260101)
children = root.spawn(4)

def worker(worker_id: int):
    rng = default_rng(children[worker_id])
    return rng.random(3)

for i in range(4):
    print(i, worker(i))

如果任务调度顺序可能变化,还应使用稳定的任务标识,而不是使用“第几个被调度”的位置作为种子。否则同一任务可能因为调度顺序不同而得到不同随机流。

4. 随机种子不是安全凭据

NumPy 的随机数生成器面向统计模拟和建模,不适合密码学用途。需要生成令牌、密码重置链接或安全随机标识时,应使用 secrets,而不是固定种子或普通 PRNG。(numpy.org)


三、环境:代码相同不等于运行环境相同

1. 环境包含哪些内容

实验环境至少包括:

Python 解释器
├── Python 主版本、次版本、补丁版本
├── Python 实现:CPython、PyPy 等
├── 操作系统和架构
├── 直接依赖的包及其版本
├── 间接依赖及其版本
├── 原生动态库:OpenBLAS、MKL、libstdc++ 等
├── CPU 指令集和浮点行为
├── 时区、区域设置和编码
├── 线程数、进程数和并发后端
└── 外部服务、数据库和数据源版本

例如,两个环境都显示“Python 3.14”,但一个是 CPython 3.14.0,另一个是 CPython 3.14.7;一个使用 OpenBLAS,另一个使用 MKL。它们不一定产生完全相同的浮点结果。

Python 的 platform.python_version() 会返回完整的主版本、次版本和补丁版本;platform.uname() 可获取系统、发行版和机器架构等信息。但 platform.platform() 主要面向人阅读,不应当直接作为机器解析的唯一格式。(docs.python.org)

可以在每次运行开始时记录环境:

from __future__ import annotations

import importlib.metadata as metadata
import json
import os
import platform
import sys

def collect_environment() -> dict:
    package_names = ["numpy", "pandas"]

    packages = {}
    for name in package_names:
        try:
            packages[name] = metadata.version(name)
        except metadata.PackageNotFoundError:
            packages[name] = None

    return {
        "python": sys.version,
        "python_executable": sys.executable,
        "implementation": platform.python_implementation(),
        "platform": platform.platform(),
        "machine": platform.machine(),
        "processor": platform.processor(),
        "packages": packages,
        "timezone": os.environ.get("TZ"),
        "locale": os.environ.get("LC_ALL") or os.environ.get("LANG"),
    }

print(json.dumps(collect_environment(), indent=2, ensure_ascii=False))

这段代码记录的是“当前观察到的环境”,不是环境本身。它不能替代依赖锁定,也不能保证其他机器能够安装出同样的环境。

2. 版本约束和精确锁定不是一回事

以下依赖声明表达的是“允许的范围”:

dependencies = [
    "numpy>=2.0,<3",
    "pandas>=2.2,<4",
]

它解决的是兼容范围问题,却没有确定某一次安装最终选中了哪个版本。

而锁定结果需要更具体,例如:

numpy==2.x.y
pandas==2.x.y
...

实际项目还应记录间接依赖、分发文件和哈希。原因是:

可安装可重复安装\text{可安装} \neq \text{可重复安装}

依赖解析器通常会根据版本约束、当前平台、Python 版本和可用发行包选择结果。因此同一个宽松约束,在不同日期或不同平台上可能解析出不同依赖树。

锁文件描述一次解析后的依赖图;哈希则进一步约束实际下载的分发文件。二者解决的问题不同:

  • 版本锁定:防止解析结果漂移;
  • 哈希校验:防止相同版本对应的文件被替换或下载到非预期文件;
  • 私有索引或制品仓库:控制依赖来源;
  • 软件物料清单(SBOM):记录最终构建中包含了哪些组件。

不要把 pip freeze 当成完整的供应链证明。它通常能列出当前环境中的已安装版本,但不能自动说明:

  • 这些包为什么被安装;
  • 它们来自哪个索引;
  • 安装文件的哈希是什么;
  • 是否包含开发环境遗留包;
  • 当前环境是否已经被手工修改。

3. 解释器和依赖必须一起验证

一个锁文件如果只声明包版本,而没有声明 Python 版本范围,仍可能不能复现。依赖包可能根据 Python 版本选择不同 wheel,也可能因为某个版本没有适配 Python 3.14 而触发源码构建。

运行前至少验证:

python --version
python -c "import sys; print(sys.executable)"
python -c "import numpy, pandas; print(numpy.__version__, pandas.__version__)"

如果项目要求 Python 3.14,应在项目元数据中表达,而不是只写在 README:

[project]
requires-python = ">=3.14,<3.15"

这里的约束含义是:项目不接受 Python 3.13,也不接受 Python 3.15 及更高版本。它不能保证操作系统和原生库相同,但能避免最明显的解释器漂移。


四、数据:文件路径不是数据身份

1. 数据版本必须可识别

以下代码不能证明读入的是同一份数据:

import pandas as pd

df = pd.read_csv("data/input.csv")

路径只表示“从哪里读”,不表示“读到的字节是什么”。文件可能被覆盖、重新导出、改变编码,或者由上游任务生成了不同内容。

数据身份至少应包括:

数据集名称
数据来源
获取时间
文件格式
文件字节哈希
行数与列数
列名和类型
过滤条件
脱敏或清洗版本

对原始文件计算 SHA-256:

from hashlib import file_digest
from pathlib import Path

def sha256_file(path: str | Path) -> str:
    with Path(path).open("rb") as f:
        return file_digest(f, "sha256").hexdigest()

print(sha256_file("data/input.csv"))

hashlib.file_digest() 接收以二进制方式打开的文件,并返回文件内容的摘要对象;调用后应由调用方负责关闭文件。SHA-256 的摘要可用于验证文件字节是否发生变化,但哈希相同并不代表数据一定符合业务语义。(docs.python.org)

2. 内容哈希和语义哈希不同

两个 CSV 文件可能包含完全相同的数据,但因为:

  • 行顺序不同;
  • 列顺序不同;
  • 换行符不同;
  • 编码不同;
  • 浮点格式不同;

导致文件哈希不同。

反过来,一个文件字节没有变化,也不代表外部含义没有变化。例如“2025 年销售额”对应的业务定义可能被修改,但文件内容暂时没变。

因此可以同时记录两类信息:

import json
import pandas as pd

df = pd.read_csv("data/input.csv")

schema = {
    "columns": [
        {"name": str(column), "dtype": str(dtype)}
        for column, dtype in df.dtypes.items()
    ],
    "rows": len(df),
    "columns_count": len(df.columns),
    "null_counts": df.isna().sum().astype(int).to_dict(),
}

print(json.dumps(schema, indent=2, ensure_ascii=False))

其中:

  • 文件哈希验证字节级一致性;
  • schema 验证结构一致性;
  • 行数、空值数和关键字段统计验证基本语义;
  • 数据版本号和上游快照标识验证业务来源。

3. 读取数据本身也可能改变结果

以下因素都可能造成差异:

  • CSV 编码或换行方式;
  • 日期解析规则;
  • 时区;
  • 缺失值识别;
  • 浮点解析;
  • 推断出的列类型;
  • 无序目录中的文件合并顺序;
  • 外部数据库查询没有 ORDER BY
  • API 返回结果分页顺序变化。

例如:

from pathlib import Path
import pandas as pd

paths = sorted(Path("data/parts").glob("*.csv"))

frames = [
    pd.read_csv(path, dtype={"id": "string"})
    for path in paths
]

df = pd.concat(frames, ignore_index=True)
df = df.sort_values(["id"], kind="stable").reset_index(drop=True)

这里的 sorted() 固定了文件拼接顺序,dtype 固定了关键字段类型,sort_values() 固定了后续处理的行顺序。

如果业务允许多行拥有相同的排序键,仅按 id 排序仍可能留下顺序歧义。应提供足够的稳定排序键,例如:

df = df.sort_values(
    ["id", "event_time", "source_row_number"],
    kind="stable",
).reset_index(drop=True)

五、浮点数和并发:为什么同样输入仍可能有微小差异

1. 浮点加法不满足结合律

IEEE 754 浮点运算存在舍入。一般而言:

round(a+b)+ca+round(b+c)\operatorname{round}(a+b)+c \neq a+\operatorname{round}(b+c)

一个具体例子:

a = 1e16
b = -1e16
c = 1.0

print((a + b) + c)
print(a + (b + c))

可能输出:

1.0
0.0

差异来自中间结果的舍入,而不是 Python 逻辑错误。

当矩阵乘法、归约或统计计算使用多线程时,元素可能被分块并以不同顺序相加。不同线程数、不同 BLAS 实现或不同 CPU 指令集都可能改变归约顺序。

因此需要预先定义验证标准:

  • 整数、字符串和索引:要求精确相等;
  • 浮点数组:使用绝对误差和相对误差;
  • 指标:规定业务容差;
  • 模型文件:决定是否要求字节相同;
  • 图像:规定像素误差或感知误差。

例如:

import numpy as np

expected = np.array([0.1, 0.2, 0.3])
actual = np.array([0.1, 0.2, 0.30000000000000004])

np.testing.assert_allclose(
    actual,
    expected,
    rtol=1e-12,
    atol=1e-12,
)

rtol 是相对误差阈值,atol 是绝对误差阈值。不能机械地使用 round(),因为不同数量级的数值需要不同的误差模型。

2. “固定线程数”只能减少一个变量

可以在运行前设置:

OMP_NUM_THREADS=1 \
OPENBLAS_NUM_THREADS=1 \
MKL_NUM_THREADS=1 \
python run_experiment.py

但这不是跨平台位级一致性的规范保证。它只是减少并行归约顺序变化,并且必须在相关原生库初始化前生效。某些库还会使用自己的线程池或调度器。

生产记录中应保存:

线程环境变量
实际 CPU 信息
NumPy 配置
BLAS 实现
进程并发度

NumPy 的随机数兼容性文档也明确指出,严格的随机流复现需要相同的 BitGenerator、种子、调用序列、NumPy 构建、环境和机器;不同 CPU 的浮点行为可能导致差异并逐步传播。(numpy.org)


六、制品:结果不仅是一个浮点数

1. 什么是实验制品

**制品(Artifact)**是运行产生并需要保存、比较或交付的对象,例如:

指标 JSON
预测结果 CSV
模型文件
特征表
图表 PNG/SVG
中间数据集
日志
环境清单
运行配置

只保存终端上的:

accuracy = 0.8734

通常不够。这个数无法回答:

  • 使用了哪份数据;
  • 哪个模型参数;
  • 哪个 Python 和依赖版本;
  • 随机种子是什么;
  • 训练是否出现警告;
  • 结果能否被重新验证。

2. 制品应具有稳定的序列化方式

如果需要比较 JSON 文件,应固定:

  • 字段名称;
  • 编码;
  • 是否排序键;
  • 浮点表示;
  • 换行方式。

示例:

import json
from pathlib import Path

def write_json(path: str, value: dict) -> None:
    text = json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        indent=2,
    ) + "\n"
    Path(path).write_text(text, encoding="utf-8")

write_json(
    "artifacts/metrics.json",
    {
        "accuracy": 0.8734,
        "samples": 1000,
    },
)

sort_keys=True 只能固定字典键的输出顺序,不能自动固定列表顺序,也不能解决浮点计算本身的差异。

写入 CSV 时也应显式指定列顺序、索引策略和浮点格式:

result.to_csv(
    "artifacts/predictions.csv",
    index=False,
    columns=["id", "prediction"],
    float_format="%.17g",
    lineterminator="\n",
)

风险在于:过度固定格式会增加文件体积,或者让不同版本的软件难以读取;格式选择应服务于验证目标,而不是盲目追求字节相同。

3. 制品哈希应进入运行记录

from hashlib import file_digest
from pathlib import Path

def digest(path: Path) -> str:
    with path.open("rb") as f:
        return file_digest(f, "sha256").hexdigest()

artifacts = {}
for path in sorted(Path("artifacts").glob("*")):
    if path.is_file():
        artifacts[str(path)] = digest(path)

print(artifacts)

这样可以在后续验证时区分:

代码没变,数据变了
数据没变,环境变了
输入没变,制品序列化变了
所有记录相同,但算法实现变了

哈希只证明内容一致或不一致,不解释差异原因。因此哈希必须和元数据、日志及配置一起保存。


七、运行记录:让一次实验成为可审计事件

1. 运行记录的最小结构

一次运行可以记录为如下对象:

{
  "run_id": "2026-09-01T10:20:30Z-abc123",
  "code_revision": "git-commit-id",
  "python": "3.14.x",
  "dependencies": {
    "numpy": "版本",
    "pandas": "版本"
  },
  "random": {
    "python_seed": 42,
    "numpy_root_entropy": 42
  },
  "inputs": {
    "data/input.csv": "sha256:..."
  },
  "parameters": {
    "test_size": 0.2,
    "threshold": 0.5
  },
  "outputs": {
    "metrics.json": "sha256:..."
  },
  "status": "success"
}

其中 run_id 是这一次运行的身份;它不能只由时间组成,因为同一时间可能有并发运行。可以将代码提交哈希、时间和随机标识组合起来。

运行记录应在成功和失败时都写入。失败记录尤其重要,因为它能说明:

  • 实验是否已经生成部分制品;
  • 使用了哪份输入;
  • 失败发生在哪一步;
  • 是否允许从中间状态恢复。

2. 一个可运行的最小实验

目录结构:

repro-demo/
├── data/
│   └── input.csv
├── artifacts/
├── run_experiment.py
└── run.json

data/input.csv

id,value,label
1,10.0,a
2,20.0,a
3,30.0,b
4,40.0,b
5,50.0,c
6,60.0,c

run_experiment.py

from __future__ import annotations

import hashlib
import importlib.metadata as metadata
import json
import os
import platform
import random
import sys
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd


ROOT = Path(__file__).resolve().parent
DATA_PATH = ROOT / "data" / "input.csv"
ARTIFACT_DIR = ROOT / "artifacts"
RUN_PATH = ROOT / "run.json"

PYTHON_SEED = 42
NUMPY_SEED = 42


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for block in iter(lambda: f.read(1024 * 1024), b""):
            h.update(block)
    return h.hexdigest()


def package_version(name: str) -> str | None:
    try:
        return metadata.version(name)
    except metadata.PackageNotFoundError:
        return None


def environment_record() -> dict:
    return {
        "python": sys.version,
        "executable": sys.executable,
        "implementation": platform.python_implementation(),
        "platform": platform.platform(),
        "machine": platform.machine(),
        "packages": {
            "numpy": package_version("numpy"),
            "pandas": package_version("pandas"),
        },
        "env": {
            "TZ": os.environ.get("TZ"),
            "OMP_NUM_THREADS": os.environ.get("OMP_NUM_THREADS"),
            "OPENBLAS_NUM_THREADS": os.environ.get("OPENBLAS_NUM_THREADS"),
            "MKL_NUM_THREADS": os.environ.get("MKL_NUM_THREADS"),
        },
    }


def write_json(path: Path, value: dict) -> None:
    path.write_text(
        json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n",
        encoding="utf-8",
    )


def main() -> None:
    ARTIFACT_DIR.mkdir(exist_ok=True)

    py_rng = random.Random(PYTHON_SEED)
    np_rng = np.random.default_rng(NUMPY_SEED)

    df = pd.read_csv(
        DATA_PATH,
        dtype={"id": "int64", "value": "float64", "label": "string"},
    )

    # 用显式 NumPy 生成器产生一列可重复噪声。
    noise = np_rng.normal(loc=0.0, scale=1.0, size=len(df))
    result = df.assign(noise=noise)
    result = result.sort_values("id", kind="stable").reset_index(drop=True)

    predictions = result.assign(
        score=result["value"] + result["noise"]
    )[["id", "score"]]

    predictions_path = ARTIFACT_DIR / "predictions.csv"
    metrics_path = ARTIFACT_DIR / "metrics.json"

    predictions.to_csv(
        predictions_path,
        index=False,
        columns=["id", "score"],
        float_format="%.17g",
        lineterminator="\n",
    )

    metrics = {
        "rows": int(len(result)),
        "score_mean": float(predictions["score"].mean()),
        "python_random_probe": py_rng.random(),
    }
    write_json(metrics_path, metrics)

    run_record = {
        "created_at": datetime.now(timezone.utc).isoformat(),
        "status": "success",
        "random": {
            "python_seed": PYTHON_SEED,
            "numpy_seed": NUMPY_SEED,
        },
        "environment": environment_record(),
        "inputs": {
            str(DATA_PATH.relative_to(ROOT)): sha256_file(DATA_PATH),
        },
        "parameters": {
            "noise_loc": 0.0,
            "noise_scale": 1.0,
        },
        "outputs": {
            str(predictions_path.relative_to(ROOT)): sha256_file(predictions_path),
            str(metrics_path.relative_to(ROOT)): sha256_file(metrics_path),
        },
    }

    write_json(RUN_PATH, run_record)
    print(json.dumps(run_record, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

运行前提:

python --version
python -m pip install numpy pandas
python run_experiment.py

程序的数据流是:

flowchart LR
    A[data/input.csv] --> B[读取并固定 dtype]
    S[Python seed] --> C[Python Random]
    N[NumPy seed] --> D[NumPy Generator]
    D --> E[生成 noise]
    B --> F[合并并稳定排序]
    E --> F
    F --> G[predictions.csv]
    F --> H[metrics.json]
    A --> I[输入 SHA-256]
    G --> J[输出 SHA-256]
    H --> J
    B --> K[环境与参数记录]
    I --> L[run.json]
    J --> L
    K --> L

关键路径如下:

  1. 读取数据时显式指定类型,避免自动推断变化;
  2. 使用局部 Random 和 NumPy Generator,避免隐式共享全局状态;
  3. 生成结果后稳定排序,避免输入顺序影响制品;
  4. 固定 CSV 的列顺序、换行符和浮点格式;
  5. 记录输入、输出、参数、环境和随机种子;
  6. 无论最终指标是多少,都可以根据 run.json 进行检查。

注意,created_at 会让 run.json 每次运行的字节哈希不同。这是有意的:运行时间是运行事实的一部分。如果需要比较实验内容,应将动态元数据和确定性结果分开保存,例如:

run.json       # 含时间、主机、状态
metrics.json   # 只含确定性指标
predictions.csv

八、Notebook:隐藏状态是可复现性的主要破坏源之一

Jupyter Notebook 不只是一个按文件顺序执行的 Python 脚本。它还包含:

  • Kernel 中已经存在的变量;
  • 已导入的模块;
  • 隐藏的随机状态;
  • 修改过但尚未保存的对象;
  • 单元格的执行编号;
  • 外部文件和环境变量;
  • Notebook 元数据。

例如,按顺序执行:

import numpy as np

rng = np.random.default_rng(42)

然后执行:

rng.random()

如果再次执行第二个单元格,却不重新执行第一个单元格,得到的就是随机流中的下一个值,而不是首次运行的值。

因此 Notebook 复现必须明确生命周期:

启动 Kernel
   │
   ▼
固定环境变量与随机状态
   │
   ▼
从头按顺序执行所有单元格
   │
   ▼
保存输出和运行记录
   │
   ▼
关闭 Kernel

调试时可以在 Notebook 中自由执行单元格,但交付时应使用“从头执行”验证,而不是依赖当前 Kernel 的状态。

一个简单的 Notebook 检查方法是:

jupyter nbconvert \
  --to notebook \
  --execute notebook.ipynb \
  --output executed.ipynb

命令的实际可用性取决于环境中是否安装了 Jupyter 和对应执行器。它也不能自动解决:

  • 外部数据变更;
  • 密钥泄露;
  • 不安全的 Notebook 输出;
  • 依赖未锁定;
  • 代码依赖执行顺序之外的隐藏状态。

Notebook 中如果包含 pickle、HTML、JavaScript 或自定义输出,应将其视为不可信制品。不要在生产环境或敏感机器上直接打开来源不明的 Notebook。

更稳妥的工程边界是:

Notebook:探索、解释和展示
Python 模块:核心逻辑
命令行入口:参数化运行
运行记录:统一写入

例如核心逻辑放在:

# experiment.py
def run_experiment(
    input_path: str,
    output_path: str,
    seed: int,
) -> dict:
    ...

Notebook 只负责调用:

from experiment import run_experiment

metrics = run_experiment(
    input_path="data/input.csv",
    output_path="artifacts/predictions.csv",
    seed=42,
)

这样可以用普通测试、命令行和 CI 复用相同逻辑。


九、失败路径:复现失败时先定位哪一层发生了漂移

不要看到最终指标不同就立即修改随机种子。应按数据流逐层比较。

1. 第一步:输入哈希不同

如果输入数据哈希不同,后续差异是预期的。继续比较:

数据文件哈希
数据 schema
行数
关键字段统计
文件来源和获取时间

常见原因是数据被覆盖、API 返回了新快照,或者数据库查询没有排序。

2. 第二步:环境不同

如果输入相同但环境不同,比较:

Python 完整版本
NumPy、pandas 版本
依赖锁文件
操作系统和架构
BLAS 实现
线程环境变量

不要只比较:

pip list

还要验证运行时实际导入的包:

python -c "import numpy, pandas; print(numpy.__file__); print(pandas.__file__)"

这可以发现当前 Shell 激活的环境与实际解释器并不是同一个环境。

3. 第三步:随机流不同

如果环境和输入相同,再检查:

  • 是否使用了多个随机库;
  • 是否某处调用了模块级 np.random
  • 是否在循环中增加了随机调用;
  • 是否并发任务使用了相同种子;
  • 是否复用了已经消耗过的 Generator
  • 是否在 Notebook 中没有重启 Kernel;
  • 是否调用了依赖全局状态的 pandas API。

NumPy 的 Generator 不提供跨版本的随机流兼容保证;算法改进可能导致比特流变化。若必须长期保存并逐位重现,应显式记录 NumPy 版本、BitGenerator、种子和调用协议,而不能只记录 default_rng()。(numpy.org)

4. 第四步:数值或序列化差异

如果中间数组已经相同,但最终制品不同,检查:

  • 排序是否稳定;
  • 字典键是否排序;
  • CSV 是否包含索引;
  • 浮点格式是否固定;
  • JSON 是否包含时间或主机字段;
  • 压缩包是否写入了当前时间;
  • 模型序列化是否包含非确定性元数据。

这类问题说明算法结果可能相同,但制品生成过程不是确定性的。


十、常见误解和它们的反例

误解一:设置一次 random.seed() 就够了

反例:

import random
import numpy as np

random.seed(1)

print(random.random())
print(np.random.random())

第一行随机值受 Python 随机状态影响,第二行通常不受 random.seed(1) 控制。

修正方法是分别管理随机源,或由实验框架统一创建并传递:

py_rng = random.Random(1)
np_rng = np.random.default_rng(1)

误解二:种子相同,任何机器都应得到完全相同的浮点结果

反例是多线程矩阵运算。随机状态可能完全相同,但浮点归约顺序、CPU 指令和底层数学库不同,仍可能产生微小差异。

修正方法是区分:

随机流复现
数值结果复现
字节级制品复现

并分别设计验证标准。

误解三:pip freeze 就是锁文件

pip freeze 描述当前环境中的安装结果,却不必然描述依赖来源、解析原因、哈希和可重建过程。

真正的交付记录应至少包括:

项目直接依赖约束
解析后的完整依赖树
Python 版本
平台标识
分发文件哈希
安装来源

误解四:Notebook 保存了,所以实验就保存了

Notebook 文件没有保存 Kernel 内存、外部文件内容和环境。一个保存的 Notebook 可能依赖:

df  # 变量来自之前执行过但未保存的单元格

也可能依赖已经变化的:

本地 CSV
数据库表
环境变量
当前工作目录

因此 Notebook 必须从干净 Kernel 从头执行,并配套保存输入、环境和输出记录。

误解五:文件哈希相同,就证明实验正确

哈希只能回答:

这次读取的字节是否等于上次记录的字节?\text{这次读取的字节是否等于上次记录的字节?}

它不能回答:

  • 数据是否来自正确业务源;
  • 标签是否泄漏;
  • 清洗逻辑是否正确;
  • 结果是否统计显著;
  • 模型是否符合业务约束。

完整复现需要完整记录,但完整记录也不等于结果正确性证明。


十一、如何设计可复现等级

不同用途不应强行采用同一成本。

1. 探索性分析

目标是让本人或同事能够重跑:

记录随机种子
保存输入快照
记录 Python 和包版本
Notebook 从头执行
保存主要输出

允许小幅浮点差异,但应保存执行日志。

2. 研究结果或模型训练

目标是让独立环境验证结论:

锁定解释器和依赖
记录完整数据版本
使用命名随机流
固定关键排序
保存模型和指标
定义数值容差
执行自动复现测试

此时仅有 Notebook 通常不够,应将训练逻辑模块化。

3. 生产交付或监管审计

目标是回答“这份结果是如何产生的”:

代码提交版本
依赖锁定与哈希
输入数据哈希
参数快照
随机状态或种子
运行主机与资源
输出制品哈希
成功或失败状态
日志与异常
审批或发布记录

如果数据涉及个人信息,运行记录不应直接保存敏感原始数据。应保存数据标识、哈希、访问位置和权限范围;哈希不是脱敏措施,低熵的敏感字段仍可能被猜测。


十二、一个可执行的复现验收标准

可以把一次复现检查写成以下顺序:

1. 创建干净 Python 3.14 环境
2. 按锁定文件安装依赖
3. 验证依赖哈希
4. 验证输入数据 SHA-256
5. 从代码提交版本运行
6. 使用记录中的参数和随机流
7. 比较中间数据的 schema 和摘要
8. 比较最终指标
9. 按约定比较制品哈希或数值容差
10. 保存本次验证记录

比较结果时,不要只有一个布尔值。应输出差异位置:

import numpy as np

np.testing.assert_allclose(
    actual_scores,
    expected_scores,
    rtol=1e-10,
    atol=1e-12,
)

if not np.array_equal(actual_ids, expected_ids):
    raise AssertionError("id 顺序或内容发生变化")

对表格数据,应分别验证:

列名
列顺序
数据类型
行数
主键唯一性
缺失值统计
排序键
数值列误差

对文件制品,应分别验证:

字节哈希
文件格式
元数据
业务内容

结语

实验可复现不是“把种子写成 42”,而是把实验的隐含输入变成显式记录。

随机种子解决的是随机状态初始化;随机流解决的是多个步骤和并发任务之间的状态隔离;环境记录和依赖锁定解决的是执行平台漂移;数据哈希和版本解决的是输入变化;制品哈希解决的是输出身份;运行记录则把这些对象连接成一次可审计的执行事件。

更准确的目标是:

可复现性=代码+环境+数据+随机状态+参数+执行协议+验证标准\text{可复现性} = \text{代码} + \text{环境} + \text{数据} + \text{随机状态} + \text{参数} + \text{执行协议} + \text{验证标准}

当这些因素被分别定义、记录并验证时,实验才不再依赖某台机器、某个 Kernel 或某次“刚好成功”的运行状态。


系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理;正文、示例与工程清单由 WR BLOG 编写。