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

Jupyter 工程化:Kernel、Notebook 状态、复现、参数化和安全

Jupyter 最适合探索问题:输入一小段代码,立即观察结果,再根据结果修改下一段代码。但“能够交互运行”与“能够稳定复现”是两种不同的能力。

交互式工作依赖状态。变量、导入的模块、当前工作目录、随机数生成器、打开的文件、数据库连接,都会停留在一个持续运行的 Python 进程中。复现则要求把这些隐含状态尽量变成显式输入,并能够在新的进程、新的环境和明确的数据版本中重新得到相同或可解释的结果。

因此,Jupyter 工程化的核心问题不是“怎样把代码放进 Notebook”,而是:

  1. 代码究竟在哪个进程中执行;
  2. Notebook 文件保存了什么,没有保存什么;
  3. 单元格的执行顺序如何影响结果;
  4. 如何把环境、数据、随机性、参数和制品记录下来;
  5. 如何自动执行、验证和失败;
  6. 如何防止打开或运行 Notebook 时执行不可信代码。

一、先建立正确的执行模型:Notebook 不是 Python 进程

1.1 Notebook、前端、Server 和 Kernel 的职责

一个常见的 Jupyter 工作环境至少包含以下组件:

flowchart LR
    B[浏览器<br/>JupyterLab / Notebook 前端]
    S[Jupyter Server]
    K[Python Kernel<br/>通常是 ipykernel]
    F[文件系统]
    D[外部数据源]
    
    B <--> |HTTP / WebSocket| S
    S <--> |Jupyter 消息协议| K
    K --> |读写文件| F
    K --> |查询 / 下载| D

Notebook 是文档,通常以 .ipynb 文件保存。它包含代码单元格、Markdown、输出和元数据;它不是负责执行代码的进程。

前端是浏览器中的编辑与显示界面,例如 JupyterLab。它负责展示单元格、发送执行请求、接收输出。

Jupyter Server 负责提供文件服务、管理 Kernel,并在前端和 Kernel 之间转发通信。

Kernel 是真正执行代码的长期运行进程。对于 Python,常见实现是 ipykernel,其底层使用 IPython 的交互式解释器。Kernel 会保留用户创建的对象,因此后续单元格可以引用前面单元格建立的变量。(docs.jupyter.org)

Jupyter 的消息协议规定了前端与 Kernel 之间的通信方式。执行请求、执行结果、标准输出、错误信息、状态变化等都通过消息传递;一个 Kernel 还可以同时连接多个前端。(jupyter-client.readthedocs.io)

这一区分解释了一个常见现象:

关闭浏览器标签页,不一定会删除 Python 变量;重启 Kernel,才会清空该 Kernel 进程中的内存状态。

1.2 Kernel 的生命周期

一个简化的 Kernel 生命周期如下:

stateDiagram-v2
    [*] --> 未启动
    未启动 --> 启动中: 打开 Notebook / 选择 Kernel
    启动中 --> 空闲: 启动成功并完成握手
    空闲 --> 忙碌: 执行单元格
    忙碌 --> 空闲: 执行成功
    忙碌 --> 空闲: 执行失败
    忙碌 --> 中断中: Interrupt
    中断中 --> 空闲: 中断成功
    忙碌 --> 已崩溃: 进程异常退出
    空闲 --> 已重启: Restart
    已重启 --> 启动中
    空闲 --> 已关闭: Shutdown
    已关闭 --> [*]
    已崩溃 --> 启动中: 自动重启或手动重启

执行失败通常只表示当前单元格抛出了异常,并不表示 Kernel 已经清空。比如:

x = 10

raise RuntimeError("当前单元格失败")

执行后,x 仍然存在:

x
# 10

如果代码在失败前已经修改了对象,那么这些修改也可能保留:

items = []

items.append("before-error")
raise RuntimeError("失败")

此时:

items
# ['before-error']

所以,异常恢复不能简单理解为“重新执行失败的单元格”。需要先判断:

  • 失败前是否修改了可变对象;
  • 是否生成了部分文件;
  • 是否提交了数据库事务;
  • 是否改变了当前目录、环境变量或随机数状态;
  • 是否启动了线程、子进程或后台任务。

1.3 Python Kernel 与 Python 环境不是同一个概念

终端中的:

python --version

显示的是当前终端命令找到的 Python。

Notebook 中的:

import sys
print(sys.executable)
print(sys.version)

显示的是当前 Kernel 使用的 Python。

两者可能不同。典型原因是:

  • 终端激活了一个虚拟环境,但 Jupyter 使用了另一个环境;
  • Notebook 的 kernelspec 指向旧环境;
  • Jupyter Server 与 Kernel 不在同一个容器或主机;
  • 通过远程 JupyterHub 连接到了服务器上的环境。

诊断时应同时检查:

import os
import sys
import site

print("Python:", sys.executable)
print("Version:", sys.version)
print("Working directory:", os.getcwd())
print("Site packages:", site.getsitepackages())

安装依赖时,优先使用 Kernel 对应的解释器:

import sys
!{sys.executable} -m pip install pandas numpy

这里的 {sys.executable} 会展开为当前 Kernel 的 Python 路径,比直接执行 pip 更不容易装错环境。

但是,在生产或团队环境中,不建议把 pip install 作为普通分析单元格的一部分。安装动作应进入环境构建脚本、锁定文件或容器构建过程,否则 Notebook 的执行结果依赖于“某个用户曾经运行过哪些安装命令”。


二、Notebook 状态:文件状态不等于 Kernel 状态

2.1 .ipynb 保存的是文档,不是完整会话

.ipynb 本质上是 JSON 文档,顶层通常包括:

{
  "cells": [],
  "metadata": {},
  "nbformat": 4,
  "nbformat_minor": 5
}

单元格包含类型、源代码、元数据和输出。Notebook 格式由 JSON 结构描述,代码单元格的输出也可以被保存到文件中。(nbformat.readthedocs.io)

但 Notebook 文件通常不会完整保存以下状态:

  • Kernel 当前内存中的变量;
  • 变量指向的对象身份;
  • 已导入模块的内部状态;
  • 当前随机数生成器状态;
  • 已打开的文件句柄;
  • 数据库连接和事务状态;
  • 后台线程或子进程;
  • 外部服务中的变化;
  • 进程环境变量在运行期间的修改。

因此,Notebook 文件是“代码与结果的记录”,不是“可直接恢复的 Python 虚拟机快照”。

可以把系统状态拆成两个部分:

St+1=F(St,Ct,Et)S_{t+1} = F(S_t, C_t, E_t)

其中:

  • StS_t:Kernel 在时刻 tt 的内存状态;
  • CtC_t:当前执行的代码单元格;
  • EtE_t:外部状态,如文件、数据库、时间和网络响应;
  • FF:代码执行过程;
  • St+1S_{t+1}:执行后的新状态。

Notebook 主要保存 CtC_t 和部分输出,却没有完整保存 StS_tEtE_t。这就是为什么“看起来代码都在 Notebook 里”仍然不能保证复现。

2.2 执行计数不是执行顺序的证明

代码单元格左侧的:

In [7]:

表示该单元格在当前 Kernel 会话中获得了执行计数 7。它反映的是执行历史,不是 Notebook 中的物理位置。

例如,按以下顺序运行:

# 单元格 A
a = 1
# 单元格 B
b = a + 1

再回到 A,修改并重新执行:

a = 100

此时 B 的旧输出可能仍然显示为 2,但重新执行 B 得到的是 101。Notebook 中同时存在:

  • 代码:b = a + 1
  • 旧输出:2
  • 当前 Kernel 状态:a == 100

这三者可能不一致。

另一个反例是先运行依赖单元格:

# 先运行
result = df["amount"].sum()

如果 df 来自更早但尚未运行的单元格,代码可能直接失败;如果当前 Kernel 恰好残留了旧的 df,代码又可能“意外成功”。这类结果最危险,因为它把隐藏状态伪装成了稳定逻辑。

2.3 用最小例子观察状态污染

value = 10

执行后再运行:

value = value + 5
value

输出:

15

如果只重新执行第二个单元格,输出变成:

20

因为第二个单元格不是纯函数,它依赖了上一次执行留下的 value

把逻辑改成显式输入:

def add_five(value: int) -> int:
    return value + 5

value = 10
result = add_five(value)

result
# 15

此时,无论单元格执行多少次,只要输入相同,函数结果就相同。工程化并不要求 Notebook 中所有代码都变成纯函数,但应尽量把核心计算写成:

y=f(x,θ)y = f(x, \theta)

其中:

  • xx 是数据;
  • θ\theta 是参数;
  • yy 是结果;
  • ff 不依赖隐藏的 Kernel 历史。

三、如何恢复和验证一个干净状态

3.1 三种操作的语义不同

重新执行单元格:只执行选定代码,保留整个 Kernel 状态。

清除输出:删除 Notebook 文件中的显示结果,不会删除 Kernel 中的变量。

重启 Kernel:终止并重新启动 Python 进程,清除变量、导入模块和大部分内存状态。

IPython 提供了命名空间重置能力:

%reset -f

它会清理用户命名空间;在 Notebook 这类不提供标准输入确认的客户端中,通常需要使用 -f 强制执行。(ipython.readthedocs.io)

不过,%reset 不等价于操作系统级重启:

  • 已经产生的文件不会自动删除;
  • 外部数据库提交不会回滚;
  • 其他线程和子进程可能仍然存在;
  • 某些扩展或底层库状态不一定完全恢复;
  • 导入模块的缓存可能继续影响行为。

因此,验证 Notebook 的最低成本流程通常是:

  1. 保存 Notebook;
  2. 重启 Kernel;
  3. 选择“Run All”或使用命令行执行;
  4. 检查是否从第一个单元格开始成功;
  5. 确认输出与预期一致;
  6. 保存执行后的副本,而不是覆盖源 Notebook。

3.2 单元格应形成可解释的数据流

一个适合工程化的 Notebook 通常具有如下顺序:

flowchart TD
    A[导入依赖] --> B[读取参数]
    B --> C[读取输入数据]
    C --> D[校验数据模式与范围]
    D --> E[清洗与转换]
    E --> F[核心计算或训练]
    F --> G[评估与检查]
    G --> H[写出结果与制品]
    H --> I[记录运行信息]

顺序不是形式要求,而是数据依赖关系的表达。例如:

  • 读取数据必须先于使用 DataFrame
  • 清洗必须先于聚合;
  • 训练必须先于评估;
  • 评估通过后才能发布制品。

如果一个单元格同时完成“下载数据、修改全局变量、训练模型、写文件”,那么失败后很难判断系统处于什么状态。应拆分副作用,让每个单元格的输入和输出清楚可见。

3.3 用断言把隐含假设变成失败条件

import pandas as pd

df = pd.DataFrame(
    {
        "user_id": [1, 2, 3],
        "amount": [10.0, 20.0, 30.0],
    }
)

assert not df.empty
assert {"user_id", "amount"} <= set(df.columns)
assert df["user_id"].notna().all()
assert (df["amount"] >= 0).all()

这些断言把“我以为数据满足条件”改成了“程序明确验证条件”。如果条件不满足,流程应尽早失败,而不是继续生成看似合理的统计结果。


四、复现:不是只设置一个随机种子

4.1 复现的形式化条件

设一次 Notebook 运行的结果为:

Y=F(C,E,D,θ,R,T,N)Y = F(C, E, D, \theta, R, T, N)

其中:

  • CC:代码和 Notebook 内容;
  • EE:执行环境,包括 Python、依赖包和系统库;
  • DD:输入数据及其版本;
  • θ\theta:显式参数;
  • RR:随机性;
  • TT:时间、时区等时间状态;
  • NN:网络、数据库和其他外部服务状态。

要让两次运行结果一致,至少需要让这些影响结果的输入保持一致,或者明确允许某些维度存在差异:

(C1,E1,D1,θ1,R1,T1,N1)(C2,E2,D2,θ2,R2,T2,N2)(C_1,E_1,D_1,\theta_1,R_1,T_1,N_1) \equiv (C_2,E_2,D_2,\theta_2,R_2,T_2,N_2)

“等价”不一定要求每个字节相同。例如图表中的字体元数据可能变化,但统计数值应一致;浮点计算也可能因底层库或硬件而出现微小差异。

4.2 随机种子只控制随机性的一部分

简单例子:

import numpy as np

rng = np.random.default_rng(20260901)
values = rng.normal(size=5)
values

使用 default_rng 创建局部随机数生成器,比依赖全局随机状态更容易控制。函数可以显式接收生成器:

def sample_mean(size: int, rng: np.random.Generator) -> float:
    samples = rng.normal(loc=0.0, scale=1.0, size=size)
    return float(samples.mean())

rng = np.random.default_rng(20260901)
sample_mean(1000, rng)

但即使设置了种子,仍可能存在以下差异:

  • 使用了 Python 标准库 random
  • 机器学习框架有独立随机数生成器;
  • 多线程或 GPU 运算的执行顺序不同;
  • 算法本身不是确定性的;
  • 输入数据顺序不稳定;
  • 外部服务返回了不同内容;
  • 浮点并行归约顺序不同。

因此,种子是复现记录中的一个字段,而不是复现保证本身。

4.3 环境复现

至少记录:

import importlib.metadata as md
import platform
import sys

print("python:", sys.version)
print("executable:", sys.executable)
print("platform:", platform.platform())

for name in ["numpy", "pandas", "ipykernel", "nbformat", "nbclient"]:
    try:
        print(name, md.version(name))
    except md.PackageNotFoundError:
        print(name, "<not installed>")

开发阶段可以使用虚拟环境:

python3.14 -m venv .venv
source .venv/bin/activate       # Windows PowerShell: .venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install jupyterlab numpy pandas nbformat nbclient
python -m ipykernel install --user \
  --name wr-blog-py314 \
  --display-name "Python 3.14 (wr-blog)"

这里的 --name 是 Kernel 的内部标识,--display-name 是界面中显示的名称。安装后检查:

jupyter kernelspec list

然后在 Notebook 中验证 sys.executable,因为“列表里存在某个 kernelspec”并不证明当前 Notebook 正在使用它。

Python 3.14 项目中,不能只锁定 Python 主版本,还要验证每个依赖是否提供兼容发行版。若某个科学计算包或二进制扩展尚未支持 Python 3.14,正确的恢复方式是:

  • 选择该包支持的 Python 版本;
  • 或等待兼容版本;
  • 或替换依赖;
  • 而不是在 Notebook 中临时混装多个版本。

4.4 数据复现

对输入数据应记录:

  • 来源;
  • 获取时间;
  • 查询条件;
  • 文件路径或对象存储键;
  • 文件大小;
  • 内容哈希;
  • 数据模式;
  • 过滤条件;
  • 脱敏或清洗版本。

一个简单的 SHA-256 记录方式:

from hashlib import sha256
from pathlib import Path

def sha256_file(path: str, chunk_size: int = 1024 * 1024) -> str:
    digest = sha256()
    with Path(path).open("rb") as f:
        while chunk := f.read(chunk_size):
            digest.update(chunk)
    return digest.hexdigest()

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

哈希只能证明“当前文件内容与某次记录一致”,不能证明数据本身正确,也不能替代数据质量校验。应同时验证列名、类型、行数、主键唯一性和关键字段范围。

4.5 制品和运行记录

运行结果不应只留在输出区域。应把重要结果保存为明确制品:

from pathlib import Path
import json
import pandas as pd

output_dir = Path("artifacts")
output_dir.mkdir(exist_ok=True)

summary = pd.DataFrame(
    {
        "metric": ["row_count", "amount_sum"],
        "value": [len(df), float(df["amount"].sum())],
    }
)

summary.to_csv(output_dir / "summary.csv", index=False)

manifest = {
    "python": sys.version,
    "executable": sys.executable,
    "input_sha256": sha256_file("data/input.csv"),
    "seed": 20260901,
    "rows": int(len(df)),
}

(output_dir / "run.json").write_text(
    json.dumps(manifest, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

这里的 run.json 是运行清单,不是完整的实验追踪系统,但它已经把最关键的隐式信息变成了可审计记录。


五、参数化:把“修改代码”变成“提供输入”

5.1 参数与代码的边界

以下内容适合成为参数:

  • 输入文件;
  • 日期范围;
  • 分组字段;
  • 过滤阈值;
  • 随机种子;
  • 输出目录;
  • 模型超参数。

以下内容不应通过参数化绕过安全边界:

  • 任意 Python 代码;
  • 任意 Shell 命令;
  • 未校验的 SQL 片段;
  • 任意文件系统路径;
  • 任意导入模块名。

例如,不要这样写:

query = f"SELECT * FROM orders WHERE status = '{status}'"

如果 status 来自外部输入,可能导致 SQL 注入。应使用数据库驱动提供的参数绑定机制。

5.2 最简单、最透明的参数化方式

先把参数集中在一个代码单元格中:

from pathlib import Path

INPUT_PATH = Path("data/input.csv")
OUTPUT_DIR = Path("artifacts")
START_DATE = "2026-01-01"
RANDOM_SEED = 20260901

后续代码只读取这些参数:

import pandas as pd

df = pd.read_csv(INPUT_PATH)
df["date"] = pd.to_datetime(df["date"], errors="raise")

filtered = df[df["date"] >= START_DATE].copy()

这种方式适合交互探索,因为参数在哪里、默认值是什么都很清楚。

5.3 使用 Papermill 执行不同参数实例

Papermill 用于参数化和执行 Jupyter Notebook。其常见约定是:将一个代码单元格标记为 parameters,执行时由外部值覆盖其中的默认变量。(papermill.readthedocs.io)

参数单元格:

# 该单元格的 tags 包含: ["parameters"]
INPUT_PATH = "data/input.csv"
OUTPUT_DIR = "artifacts"
RANDOM_SEED = 20260901

命令行执行:

papermill \
  notebooks/report.ipynb \
  runs/report-2026-01.ipynb \
  -p INPUT_PATH data/input.csv \
  -p OUTPUT_DIR runs/report-2026-01 \
  -p RANDOM_SEED 20260901

输入参数应满足以下条件:

  1. 有默认值,便于交互运行;
  2. 在参数单元格集中声明;
  3. 在执行早期完成类型转换和合法性校验;
  4. 不把参数直接拼接到 SQL 或 Shell;
  5. 输出 Notebook 保存参数和执行结果,便于审阅。

参数化解决的是“同一套代码运行多组输入”,不自动解决环境、数据和随机性复现问题。


六、自动执行:把 Notebook 当作可测试工件

6.1 使用 nbclient 程序化执行

nbclient 可以在浏览器之外加载并执行 Notebook;典型流程是读取 Notebook,创建 NotebookClient,执行后写回新的 .ipynb 文件。(docs.jupyter.org)

from pathlib import Path

import nbformat
from nbclient import NotebookClient
from nbclient.exceptions import CellExecutionError

source = Path("notebooks/report.ipynb")
target = Path("runs/report-executed.ipynb")

nb = nbformat.read(source, as_version=4)

client = NotebookClient(
    nb,
    timeout=600,
    kernel_name="wr-blog-py314",
    resources={"metadata": {"path": str(source.parent)}},
    allow_errors=False,
)

try:
    client.execute()
except CellExecutionError as exc:
    target.parent.mkdir(exist_ok=True)
    nbformat.write(nb, target)
    raise RuntimeError(
        f"Notebook 执行失败,已保存失败现场: {target}"
    ) from exc

target.parent.mkdir(exist_ok=True)
nbformat.write(nb, target)
print(f"执行成功: {target}")

关键配置的含义:

  • timeout=600:单个单元格最长允许执行时间;
  • kernel_name:指定要启动的 Kernel;
  • resources["metadata"]["path"]:设置执行工作目录;
  • allow_errors=False:遇到异常时让整个流程失败;
  • 失败时保存 Notebook:保留错误单元格和 traceback,便于诊断。

如果 Notebook 依赖当前目录中的相对路径,而自动执行时工作目录不同,就会出现“交互运行成功、CI 运行失败”。这不是 pandas 或 Jupyter 的随机故障,而是执行上下文没有被显式固定。

6.2 执行前清理旧输出

为了避免旧输出误导审阅者,可以先清空输出并重置执行计数:

for cell in nb.cells:
    if cell.cell_type == "code":
        cell["execution_count"] = None
        cell["outputs"] = []

这一步只修改文档,不会启动 Kernel。随后再交给 NotebookClient 执行。

执行后应检查:

  • 是否存在错误输出;
  • 输出文件是否存在;
  • 输出文件哈希是否符合预期;
  • 行数和关键指标是否满足断言;
  • Notebook 是否使用了预期 Kernel;
  • 执行时间是否超出预算。

6.3 Notebook 适合什么,不适合什么

Notebook 适合:

  • 探索数据;
  • 展示推导和图表;
  • 记录实验过程;
  • 生成可审阅报告;
  • 作为端到端验证入口。

核心业务逻辑更适合放进 .py 模块:

project/
├── src/
│   └── wr_project/
│       ├── io.py
│       ├── transform.py
│       └── metrics.py
├── notebooks/
│   └── report.ipynb
├── tests/
├── data/
├── artifacts/
└── pyproject.toml

Notebook 负责组织输入、调用函数和展示结果:

from wr_project.transform import build_summary

summary = build_summary(df, min_amount=10.0)
summary

这样可以对 build_summary 编写普通单元测试,而不必通过“运行整个 Notebook”才能测试一条转换规则。


七、失败路径与诊断方法

7.1 NameError:变量不存在

NameError: name 'df' is not defined

常见原因:

  • 跳过了读取数据的单元格;
  • Kernel 刚重启;
  • 使用了错误的 Kernel;
  • 变量名被重命名;
  • 代码执行顺序与文档顺序不一致。

诊断:

"df" in globals()

修复不是简单地再运行一次报错单元格,而是从干净 Kernel 执行全部流程。

7.2 FileNotFoundError:工作目录不同

from pathlib import Path
print(Path.cwd())
print(Path("data/input.csv").resolve())

Notebook 中的相对路径相对于当前工作目录,而不是相对于当前 .ipynb 文件必然成立。自动执行时应显式设置工作目录,或者使用项目根目录解析路径。

7.3 输出与代码不一致

表现包括:

  • 代码显示使用阈值 0.5,输出却像是 0.3
  • Markdown 中写着一个日期,图表来自另一个日期;
  • 执行计数不是递增顺序;
  • 重新打开后输出消失或被标记为不可信。

诊断方式:

  1. 清空输出;
  2. 重启 Kernel;
  3. 从头执行;
  4. 保存新文件;
  5. 比较新旧 Notebook;
  6. 检查参数和数据哈希。

不要把执行计数改成顺序数字来“修复”问题。执行计数只是历史记录,不能补偿状态依赖。

7.4 超时、死循环和资源耗尽

Notebook 中的死循环、超大 DataFrame、无限等待的网络请求,都会让 Kernel 长时间处于 busy 状态。

应区分:

  • 中断:尝试向 Kernel 发送中断;
  • 重启:终止并重新启动 Kernel;
  • 杀进程:Kernel 无响应时由操作系统终止;
  • 恢复外部状态:检查部分写入文件、临时表和事务。

对外部 I/O 设置超时,对大数据处理设置内存预算,对输出目录使用临时路径并在成功后原子移动,避免失败过程留下“看似完整”的结果文件。


八、安全:Notebook 是可执行程序,不是普通文档

8.1 为什么打开 Notebook 也存在风险

Notebook 同时包含:

  • 可执行 Python 代码;
  • Markdown;
  • HTML;
  • JavaScript 输出;
  • 图片和富文本;
  • 可能包含敏感路径、令牌和查询结果。

Jupyter 的安全模型重点防止“打开一个别人提供的 Notebook 就自动执行其中的代码”。官方模型包括:不可信 HTML 会被清理,不可信 JavaScript 不执行,Markdown 中的 HTML 和 JavaScript 不被信任;用户自己生成的输出则被视为可信。(jupyter-server.readthedocs.io)

这不表示 Notebook 安全。只要用户主动运行代码,代码就可以:

import os
os.environ

读取环境变量,也可能访问文件、网络和数据库。Notebook 的核心能力就是任意代码执行,所以安全边界必须放在“谁能访问 Server、Kernel 运行在哪里、代码拥有什么权限”上。

8.2 Trust 机制解决的是什么问题

Jupyter 会为 Notebook 内容生成签名,并在打开时检查签名;未被信任的 Notebook 输出中的 HTML 和 JavaScript 不会直接按可信内容加载。可以使用:

jupyter trust notebook.ipynb

显式信任 Notebook。(jupyter-server.readthedocs.io)

但要正确理解:

Trust 不是杀毒软件,也不是代码审计。

信任 Notebook 主要影响输出内容是否被信任展示,不代表代码没有恶意,也不代表数据源安全。对外部 Notebook 的合理流程是:

  1. 以文本或 JSON 方式审阅源代码;
  2. 搜索 os, subprocess, socket, requests, open 等敏感操作;
  3. 检查 Markdown、HTML 和输出中的可疑内容;
  4. 在隔离环境中执行;
  5. 使用低权限账户和最小文件权限;
  6. 不把生产凭据注入测试 Kernel;
  7. 确认无风险后再决定是否信任。

8.3 Server 暴露风险

本地 Notebook Server 通常只监听本机地址;如果把它暴露到局域网或公网,风险会迅速增加。单用户 Server 不应被当作多用户隔离平台;多用户场景应使用具备认证和用户隔离能力的 JupyterHub 或其他受控部署方案。(jupyter-notebook.readthedocs.io)

部署时至少考虑:

  • 监听地址是否必要;
  • 是否启用认证;
  • 是否使用 HTTPS 或受保护的反向代理;
  • 是否限制可访问目录;
  • Kernel 是否运行在容器或受限用户中;
  • 是否限制 CPU、内存、进程数和网络出口;
  • 是否避免把访问令牌写入日志、Notebook 或输出;
  • 是否隔离不同用户的文件和 Kernel。

不要使用“关闭认证”来解决访问问题。访问不便应通过正确的认证、代理和网络策略解决,而不是移除安全边界。

8.4 参数化带来的注入风险

危险写法:

import subprocess

filename = "input.csv"
subprocess.run(f"cat {filename}", shell=True)

filename 来自外部参数时,字符串可能被解释为 Shell 命令。

更安全的写法:

from pathlib import Path

base_dir = Path("data").resolve()
candidate = (base_dir / filename).resolve()

if base_dir not in candidate.parents:
    raise ValueError("禁止访问 data 目录之外的路径")

content = candidate.read_text(encoding="utf-8")

这里同时处理了两个问题:

  1. 不通过 Shell 解释字符串;
  2. 校验规范化后的路径仍位于允许目录内。

对于 SQL,也应使用参数绑定,而不是字符串拼接。对于 Notebook 参数,不要把“可配置”误解成“允许任意代码”。


九、一个可落地的工程化模板

下面的结构把交互、测试、自动执行和审计分开:

project/
├── pyproject.toml
├── src/
│   └── report_project/
│       ├── __init__.py
│       ├── io.py
│       └── report.py
├── notebooks/
│   └── report.ipynb
├── scripts/
│   └── run_notebook.py
├── tests/
│   └── test_report.py
├── data/
│   └── input.csv
├── runs/
└── artifacts/

Notebook 的职责:

# 参数单元格
INPUT_PATH = "data/input.csv"
OUTPUT_DIR = "artifacts"
RANDOM_SEED = 20260901

# 导入项目代码
from report_project.io import load_input
from report_project.report import build_report

df = load_input(INPUT_PATH)
report = build_report(df, seed=RANDOM_SEED)
report

项目模块的职责:

# src/report_project/report.py
import numpy as np
import pandas as pd

def build_report(
    df: pd.DataFrame,
    *,
    seed: int,
) -> pd.DataFrame:
    required = {"user_id", "amount"}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"缺少列: {sorted(missing)}")

    if df["amount"].isna().any():
        raise ValueError("amount 不允许缺失")

    rng = np.random.default_rng(seed)

    result = (
        df.groupby("user_id", as_index=False)["amount"]
        .sum()
        .rename(columns={"amount": "total_amount"})
    )

    result["sample"] = rng.random(len(result))
    return result

测试职责:

# tests/test_report.py
import pandas as pd
from report_project.report import build_report

def test_build_report_is_reproducible():
    df = pd.DataFrame(
        {
            "user_id": [1, 1, 2],
            "amount": [10.0, 5.0, 7.0],
        }
    )

    first = build_report(df, seed=42)
    second = build_report(df, seed=42)

    pd.testing.assert_frame_equal(first, second)

这里的复现条件较清晰:

  • 输入 df 显式传入;
  • 种子显式传入;
  • 分组逻辑位于模块中;
  • Notebook 只负责展示和编排;
  • 测试可在没有浏览器的环境中执行。

十、最终检查:从“能运行”到“可交付”

一个 Notebook 交付前,应至少完成以下验证:

python --version
jupyter kernelspec list

在 Notebook 内:

import sys
print(sys.executable)
print(sys.version)

然后:

  1. 重启 Kernel;
  2. 从头执行全部单元格;
  3. 检查工作目录;
  4. 检查输入数据哈希;
  5. 检查关键列、类型和行数;
  6. 固定或记录随机种子;
  7. 检查输出制品是否完整;
  8. 保存执行后的 Notebook 副本;
  9. 在隔离环境中通过 nbclient 或其他自动化方式执行;
  10. 审阅代码、Markdown、HTML 输出和敏感信息;
  11. 不把令牌、密码和个人数据写入 Notebook;
  12. 对失败运行保留日志和失败现场,但清理不应暴露的机密。

最重要的判断标准不是“打开 Notebook 后输出看起来正确”,而是:

在新的 Kernel、明确的环境、明确的数据、明确的参数和可审计的执行记录下,流程是否仍然能够成功,并且失败时能解释原因。

Jupyter 的交互性来自持久状态;工程化则要求识别、限制并记录这些状态。Kernel 负责运行,Notebook 负责记录,参数负责表达输入,环境和数据负责确定运行条件,制品和日志负责留下证据,安全边界负责限制代码执行的影响范围。只有这些部分同时成立,Notebook 才能从一次性的实验草稿变成可复现、可验证、可维护的工程资产。


系列导航与关联阅读

官方资料

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