Python多进程场景下单元测试如何设置一致随机种子？

我有一段代码使用multiprocessing.Pool创建工作进程，每个进程都会调用seed.py中的generate_sensor函数——这个函数多次用到random.randint和random.choice。我想设置统一的随机种子来保证单元测试的可复现性，但尝试了两种方法都失败了，推测是多进程导致随机调用顺序不一致，或者种子设置方式有误。

相关代码（my_module.py的main函数） #

num_processes = multiprocessing.cpu_count() -1
pool = multiprocessing.Pool(num_processes)
tasks: list[tuple[seed.Seed, str, int, int]] = []
sensors: dict[str, sensor.Sensor] = {}

for i in range(num_processes):
    tasks.extend([(s, self.recSeq, seedSensPerProcess, i) for s in seeds])

for t in tasks:
    pool.apply_async(
        generate_sensor, t, callback=lambda result: sensors.update(
            [(r.seq, r) for r in result]
        ),
    )
pool.close()
pool.join()
return sensors

尝试过的失败方法 #

失败方法1 #

def test_my_function() -> None:
    random.seed(0)
    out_sensors = my_module.main()
    print(out_sensors)
    assert False # 让pytest无论如何都显示打印内容

每次输出结果都不一样，说明seed.py中导入的random模块没受到这个种子设置的影响。

失败方法2 #

def test_my_function() -> None:
    with patch("my_package.seed.random") as mock_random:
        mock_random = random.Random(0)

        out_sensors = my_module.main()
        print(out_sensors)
        assert False # 让pytest无论如何都显示打印内容

没有输出，推测seed.py中导入的random无法正常工作。

正确解决方案 #

多进程环境下，主进程设置的随机种子不会自动传递给子进程——每个子进程启动时都会重新初始化random模块。要实现可复现性，有两种可靠的思路：

方案1：让generate_sensor使用独立的随机生成器 #

修改seed.py中的函数，让它接收种子参数，创建独立的Random实例来处理所有随机操作，避免依赖全局random状态：

# seed.py
import random

def generate_sensor(seed_val, s, rec_seq, sens_per_process, proc_idx):
    # 用传入的种子初始化独立随机生成器
    rng = random.Random(seed_val)
    # 后续所有随机操作都用rng，比如：
    # random_num = rng.randint(1, 100)
    # random_item = rng.choice(items)
    # ... 原函数逻辑 ...

然后在主进程的任务生成阶段，为每个任务分配固定的可复现种子：

# my_module.py的main函数中修改任务逻辑
base_seed = 0 # 测试时固定这个值
for i in range(num_processes):
    # 为每个任务生成唯一且可复现的种子
    tasks.extend([
        (hash((base_seed, i, seed_idx)) % (2**32), s, self.recSeq, seedSensPerProcess, i)
        for seed_idx, s in enumerate(seeds)
    ])

# 调用apply_async时传递种子参数
for t in tasks:
    pool.apply_async(
        generate_sensor, t, callback=lambda result: sensors.update(
            [(r.seq, r) for r in result]
        ),
    )

方案2：初始化进程池时为子进程设置种子 #

利用multiprocessing.Pool的initializer参数，在每个子进程启动时直接设置种子：

# my_module.py中创建进程池的代码
def init_worker(seed):
    import random
    random.seed(seed)

base_seed = 0 # 测试时固定这个值
num_processes = multiprocessing.cpu_count() -1
# 给每个子进程传递不同但可复现的种子（避免所有进程随机序列完全一致）
pool = multiprocessing.Pool(
    num_processes,
    initializer=init_worker,
    initargs=(base_seed,) # 若要每个进程序列不同，可传base_seed + i（i为进程索引）
)

测试代码修改 #

无论用哪种方案，测试时只需固定基础种子即可：

def test_my_function() -> None:
    base_seed = 0
    # 将种子传递给main函数，让内部使用
    out_sensors = my_module.main(base_seed)
    # 替换为实际的断言逻辑
    assert out_sensors == expected_sensors

内容的提问来源于stack exchange，提问作者Andrew J Bonham