跳转至

代码库设计与模式

一句话总结：代码库设计把散乱的实验代码整理成可长期维护的工程，涵盖目录布局、整洁代码、ML 设计模式、配置管理、日志、API 与打包。

好的代码库设计，是研究原型与生产软件之间的分水岭。 本节涵盖项目结构、整洁代码原则、与 ML 相关的设计模式、配置管理、日志、API 设计与打包.

  • 大多数 ML 代码起步于一个 Jupyter notebook. Notebook 越写越长，被复制、修改、分享，最后变成一坨全局变量、废弃 cell 与魔法数字混在一起的、根本没法维护的乱麻。 代码库设计 (codebase design) 就是把代码组织好，让项目越长越大时仍然读得懂、改得动的那门功夫。

  • 这不是为了循规蹈矩。 它的真正意义在于，缩短"我想改 X"到"X 已经改好并跑通了"之间的时间。 设计良好的代码库里，这个时间是分钟级。 设计糟糕的代码库里，这个时间是"考古挖掘几天没文档的意大利面条式代码".

项目结构

  • 一致的项目布局能让任何人 (包括未来的你自己) 秒级定位代码。
my_project/
├── src/my_project/       # 源代码 (可导入的 package)
│   ├── __init__.py
│   ├── data/             # 数据加载与预处理
│   │   ├── __init__.py
│   │   ├── dataset.py
│   │   └── transforms.py
│   ├── models/           # 模型架构
│   │   ├── __init__.py
│   │   ├── transformer.py
│   │   └── layers.py
│   ├── training/         # 训练循环、优化器
│   │   ├── __init__.py
│   │   ├── trainer.py
│   │   └── losses.py
│   └── utils/            # 共享工具
│       ├── __init__.py
│       └── logging.py
├── configs/              # 配置文件
│   ├── base.yaml
│   └── experiment_1.yaml
├── scripts/              # 入口 (训练、评估、服务)
│   ├── train.py
│   ├── evaluate.py
│   └── serve.py
├── tests/                # 测试文件 (镜像 src/ 结构)
│   ├── test_dataset.py
│   ├── test_model.py
│   └── test_trainer.py
├── notebooks/            # 只做探索 (不是生产代码)
├── pyproject.toml        # 项目元数据与依赖
├── README.md
├── .gitignore
└── Dockerfile

  • src/ 布局：把源码放在 src/my_project/ 之下，可以避免从当前目录意外导入 (那种误导入在开发时能跑通，但在生产环境里就会暴露 import 错误). 开发时用 pip install -e . 以可编辑模式安装。

  • Monorepo vs multi-repo: monorepo 把所有相关项目放在一个仓库里 (跨项目改动更方便，CI 可共享). multi-repo 让每个项目各自有仓库 (边界更清晰，版本独立管理). 大多数 ML 团队一开始都用 monorepo，有必要再拆分。

  • Scripts vs library：入口脚本 (train.py、evaluate.py) 放在 scripts/ 里。 可复用的逻辑放在 src/ 里。 一个训练脚本理想状态大概 50 行：解析配置、构建数据集、构建模型、构建 trainer、开训。 所有复杂度都住在 library 里。

整洁代码原则

  • 命名：单个最能立竿见影的改进。 名叫 x 的变量要求你去读周围的代码才能理解它，名叫 learning_rate 的变量自己就说明了自己。
# BAD
def proc(d, n, lr):
    for i in range(n):
        for k, v in d.items():
            v -= lr * g[k]

# GOOD
def update_parameters(parameters, num_steps, learning_rate):
    for step in range(num_steps):
        for name, param in parameters.items():
            param -= learning_rate * gradients[name]

  • 单一职责原则 (Single Responsibility Principle)：每个函数/类只做一件事。 名叫 load_data_and_train_model 的函数干了两件事，应该拆开。 拆开后每部分都能独立测试、独立复用、独立看懂。

  • DRY (Don't Repeat Yourself，不要重复自己) —— 但别过早抽象。 如果你已经复制粘贴过三次，那就提取成一个函数。 但不要为只用过一次的代码造抽象。 过早抽象比重复更糟：它把复杂度提了上去，却没有拿到经过验证的收益。

# 过早抽象 (只有一处用到, 却过度工程化)
class AbstractDataTransformPipelineFactory:
    ...

# 刚刚好 (直接、清楚, 在三处被使用)
def normalise_image(image, mean, std):
    return (image - mean) / std
  • 魔法数字：永远不要用没有解释的字面量数值。
# BAD
if len(batch) > 32:
    split_batch(batch, 32)

# GOOD
MAX_BATCH_SIZE = 32
if len(batch) > MAX_BATCH_SIZE:
    split_batch(batch, MAX_BATCH_SIZE)
  • 函数要短：如果一个函数一屏 (约 30 行) 装不下，那它大概在做的事情太多。 把逻辑块提取成描述性命名的辅助函数，函数体就会读起来像一份高层次的摘要。

ML 里的设计模式

  • 设计模式 (design pattern) 是常见问题的可复用解法。 下面几种在 ML 代码库里最常用:

  • 工厂模式 (Factory pattern)：无需指定具体类就能创建对象。 当配置里写着 model: "transformer"，你需要根据它去实例化对应类时特别有用:

MODEL_REGISTRY = {
    "transformer": TransformerModel,
    "cnn": CNNModel,
    "mlp": MLPModel,
}

def build_model(config):
    model_cls = MODEL_REGISTRY[config["model"]]
    return model_cls(**config["model_params"])

  • 这样一来，训练脚本就与具体模型实现解耦了。 加一个新模型只需要在 registry 里加一行，而不必修改训练循环。

  • 策略模式 (Strategy pattern)：在运行时切换算法。 对损失函数、优化器、调度器都很好用:

LOSS_FUNCTIONS = {
    "mse": nn.MSELoss,
    "cross_entropy": nn.CrossEntropyLoss,
    "focal": FocalLoss,
}

loss_fn = LOSS_FUNCTIONS[config["loss"]]()
  • 观察者模式 (Observer pattern)，也叫 callbacks / hooks：让模块能对事件做出反应，又不必与事件源紧耦合。 训练框架 (PyTorch Lightning、Keras) 广泛使用回调:
class EarlyStopping:
    def __init__(self, patience=5):
        self.patience = patience
        self.best_loss = float('inf')
        self.counter = 0

    def on_epoch_end(self, epoch, val_loss):
        if val_loss < self.best_loss:
            self.best_loss = val_loss
            self.counter = 0
        else:
            self.counter += 1
            if self.counter >= self.patience:
                return "stop"
  • 依赖注入 (dependency injection)：把依赖从外部传入函数/类，而不是在内部创建。 这让测试变得简单 (注入一个 mock 就行)，配置也更灵活:
# BAD: 硬编码依赖
class Trainer:
    def __init__(self):
        self.logger = WandbLogger()  # 没有 W&B 就没法测试

# GOOD: 注入依赖
class Trainer:
    def __init__(self, logger):
        self.logger = logger  # 可以注入任何 logger, 包括 mock

配置管理

  • 把超参数、文件路径、模型设置硬编码在代码里，会让实验没法复现，修改起来也痛苦。 请把配置外部化到文件里。

  • YAML 是 ML 配置最常见的格式:

# configs/experiment_1.yaml
model:
  name: transformer
  d_model: 512
  n_heads: 8
  n_layers: 6

training:
  batch_size: 64
  learning_rate: 3e-4
  max_epochs: 100
  early_stopping_patience: 10

data:
  train_path: /data/train.parquet
  val_path: /data/val.parquet
  max_seq_length: 512

  • Hydra (Facebook 开源) 是一个配置框架，支持组合 (把基础配置与实验专用的 override 合并)、命令行覆盖 (python train.py training.lr=1e-3)，以及多轮运行 (对超参数扫参).

  • argparse 对参数不多的脚本更简单:

import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--lr", type=float, default=3e-4)
parser.add_argument("--batch-size", type=int, default=64)
parser.add_argument("--config", type=str, default="configs/base.yaml")
args = parser.parse_args()
  • 最佳实践：一份基础配置，把所有默认值都写好；每个实验配置只 override 变动的部分。 每次实验都把配置和结果一起归档。

日志与可观测性

  • print 语句是给 debug 用的。 日志 (logging) 才是给生产用的:
import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

logger.debug("Batch loaded: %d samples", len(batch))     # 噪声大, 用于调试
logger.info("Epoch %d: loss=%.4f, lr=%.6f", epoch, loss, lr)  # 正常运行
logger.warning("GPU memory >90%%, consider reducing batch size")
logger.error("Failed to load checkpoint: %s", path)       # 可恢复错误
logger.critical("CUDA out of memory, aborting")            # 致命错误

  • 为什么不用 print: logging 支持日志级别 (生产里可以过滤掉 debug)，支持格式化 (时间戳、模块名)，也支持不同 handler (写文件、发送到监控系统) —— 而这一切都不需要改动调用 log 的代码。

  • 结构化日志 (structured logging) 在人类可读消息旁边额外输出机器可解析的格式 (JSON). 这样就能对特定字段做搜索和告警:

logger.info("training_step", extra={
    "epoch": 5, "step": 1200, "loss": 0.0342, "lr": 2.1e-4
})

API 设计

  • 如果你的模型会被别的服务 (Web app、移动端、另一条 ML 流水线) 调用，它就需要一个应用程序接口 (Application Programming Interface, API).

  • REST API 使用 HTTP 方法：GET 读，POST 创建/预测，PUT 更新，DELETE 删除。 端点按资源命名:

POST /api/v1/predict          # 发送输入, 得到预测
GET  /api/v1/models           # 列出可用模型
GET  /api/v1/models/{id}      # 获取某个模型的详情
POST /api/v1/models/{id}/predict  # 用指定模型做预测
  • FastAPI 是 Python 生态里做 ML 服务的首选框架:
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class PredictRequest(BaseModel):
    text: str

class PredictResponse(BaseModel):
    label: str
    confidence: float

@app.post("/predict", response_model=PredictResponse)
async def predict(request: PredictRequest):
    result = model.predict(request.text)
    return PredictResponse(label=result.label, confidence=result.score)

  • FastAPI 会自动生成 API 文档 (/docs 下的 Swagger UI)，用 Pydantic 模型校验输入输出，还支持异步以获得高吞吐。

  • gRPC 在服务与服务之间的内部通信里比 REST 更快。 它用 Protocol Buffers (二进制序列化，比 JSON 更小更快)，并支持流式传输。 TensorFlow Serving、Triton Inference Server 以及很多微服务架构都用它。

打包与分发

  • 把代码打成可安装的 package，别人 (以及你自己写的脚本) 才能干净地 import 它:
# pyproject.toml
[project]
name = "my-ml-project"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
    "torch>=2.0",
    "jax>=0.4",
    "pydantic>=2.0",
]

[project.optional-dependencies]
dev = ["pytest", "ruff", "mypy"]

[build-system]
requires = ["setuptools>=64"]
build-backend = "setuptools.backends._legacy:_Backend"

pip install -e ".[dev]"    # 以可编辑模式安装, 同时装上 dev 依赖

  • 可编辑安装 (editable install, -e)：源码的变更立即生效，不用重新安装。 开发阶段必备。

  • 锁定依赖版本: requirements.txt 里写精确版本 (torch==2.2.1，而不是 torch>=2.0)，才能保证可复现。 用 pip freeze > requirements.txt 抓当前环境快照。 想要更精细的依赖管理，用 uv、poetry 或 pip-tools.

与 AI 编码智能体协作

  • AI 编码智能体 (Claude Code、GitHub Copilot、Cursor 等) 现在已经是专业工程工作流的一部分。 用得好，开发速度会显著提升；用得差，就会引入难以察觉的 bug，侵蚀你对自家代码库的理解，制造一种虚假的效率感。

  • 正确的心智模型：智能体是一个反应飞快但经验尚浅的结对程序员. 它能快速写代码，熟悉语法与标准模式，读过的文档比你这辈子读过的都多。 但它并不理解你的具体系统、你的业务约束、你的边界情况，也不了解你那些设计决策背后的"为什么". 你才是资深工程师，智能体是初级工程师。 你负责指挥、review，并承担最终责任。

智能体擅长什么

  • 样板代码与脚手架：生成 Dockerfile、CI 配置、测试 fixture、数据类定义、argparse 设置。 这些都遵循熟知模式，手写又枯燥。 让智能体去生成，你再 review 正确性即可。

  • 写测试：描述函数的行为，智能体就能生成测试用例。 它往往能想到你会漏掉的边界情况 (空输入、负值、Unicode). 生成的测试一定要读 —— 它们校验的是你的假设，而不仅仅是你的代码。

  • 重构: "把这段代码抽成函数", "把这个类改成 dataclass", "给这个模块加类型注解". 意图明确、出错风险低的机械式改造。

  • 探索与原型: "写个脚本快速跑一下推理延迟" 或者 "教我怎么用 HuggingFace 的 tokeniser API". 智能体能比翻文档更快给你一个能跑的起点。

  • 文档与 docstring：智能体能从代码结构里生成文档。 Review 一下准确性，苦力活儿就自动化了。

  • 调试辅助：把报错的 traceback 贴给它，请它诊断。 对常见问题 (shape 不匹配、import 错误、CUDA 显存不足) 它经常能一针见血指出根因并给出修复建议。

什么时候不该依赖智能体

  • 新颖的架构决策：如果你在设计一条全新的训练流水线，智能体会给你一个通用的、平庸的方案。 它不了解你的数据约束、延迟要求或团队专长。 让智能体去实现你已经想清楚的设计，而不是让它替你想。

  • 安全关键代码：身份认证、加密、输入净化。 智能体生成的代码看起来对，却可能有微妙漏洞 (SQL 注入、不安全的默认值、时序攻击). 安全代码应由懂威胁模型的人写，由另一个人 review.

  • 性能关键的内层循环：智能体会写出正确但幼稚的代码。 GPU 核函数、内存敏感的数据结构、延迟敏感的服务路径，你都需要理解硬件约束 (见第 13 章、第 16 章)，有意识地去优化。

  • 你看不懂的代码：如果智能体生成了 200 行代码而你无法逐行解释它在干嘛，别把它 commit 进去。 你现在维护的是一份自己都不理解的代码，哪天它坏了 (一定会坏)，你连 debug 都无从下手。 这是最常见、也最危险的失败模式。

Review 的纪律

  • 每一行都要读，再 commit. 这不是可选项。 智能体的代码是草稿，不是成品。 把它当作同事发来的 pull request，用批判的眼光过一遍。

  • 重点查什么:

    • 正确性：它真的做了你要它做的事吗？智能体经常在解一个"与你意图非常相似但不完全一样"的问题。
    • 边界情况：空输入、None、负数、超大输入，处理了吗？智能体经常漏边界处理。
    • 幻觉 API：智能体可能调用根本不存在的函数，或者用了不存在的参数，尤其是较新或较冷门的库。 每个 API 调用都要核实是否真存在。
    • 过度工程：智能体倾向于写比必要更多的代码。 用 50 行解一个 10 行能搞定的问题，只是徒增复杂度。 该砍就狠狠地砍。
    • 安全：硬编码的密钥、未净化的用户输入、不安全的默认值。 智能体不会用"对抗性思维"看问题。
    • 风格一致性：生成的代码是否符合你们项目的命名、模式、错误处理约定?

怎么写出好 prompt

  • 智能体输出的质量，与你指令的质量成正比。 模糊的 prompt 只会得到模糊的代码。

  • 差的写法: "写一个 data loader"

  • 好的写法: "写一个 PyTorch DataLoader，读取一个 CSV 文件，列名为 'text' 和 'label'. 用 HuggingFace 的 'bert-base-uncased' tokeniser 分词，max_length=512. 返回 input_ids、attention_mask 与 label，都是 tensor. 如果 CSV 里 label 列有缺失值，跳过那些行."

  • 提供上下文：告诉智能体项目结构、已有代码、约束条件、命名习惯。 上下文越充分，输出越贴切。

  • 指定约束: "只用标准库", "必须兼容 Python 3.10", "不要用全局变量", "按照 src/models/transformer.py 里已有的模式来".

  • 要求解释: "实现 X，并解释关键设计决策". 逼智能体把自己的推理讲出来，你就更容易看出它哪个假设有问题。

用质量门捕获智能体的错误

  • 你已有的质量基础设施 (第 04 节) 抓智能体的错和抓人写的错一样有效:

    • 类型检查 (mypy)：抓幻觉 API 的签名不一致和类型不匹配。
    • Lint (ruff)：抓未使用的 import、未定义的变量、风格违规。
    • 测试 (pytest)：如果智能体的代码能通过你的测试，那更可能是对的。 如果你还没写测试，那就在让智能体实现功能之前先写测试 (测试驱动开发 (Test-Driven Development, TDD) 和智能体配合得特别好).
    • CI 流水线：每次 commit 自动跑上面这些。

  • "智能体写代码" + "质量门校验它" 的组合，比任何一方单独出场都更高产。 智能体又快又粗，门又严又不写代码，合起来就同时兼得速度与正确性。

效率幻觉陷阱

  • 编码智能体最大的风险，是效率幻觉. 10 分钟就能生成 500 行代码。 但如果你花 2 小时去 debug 那 500 行你没搞懂的代码，你的整体速度反而慢于自己写 200 行、用 30 分钟搞定。

  • 用智能体真正提高生产力的方式:

    1. 保持掌控：你决定架构，智能体填实现。
    2. 理解生成的代码：讲不清就重写，或者让智能体简化。
    3. 投资质量门：测试、类型、lint 的成本会摊到每一次与智能体的交互里。
    4. 让智能体补你的短板：你算法强、写测试慢？让智能体写测试。 你 UI 熟，数据库查询陌生？让智能体起草 SQL. 扬长，让智能体补短。

  • 真正把智能体用好的工程师，本来就已经很会写代码。 智能体放大你已有的能力，但不能替代它。 理解数据结构、算法、系统设计与软件工程 (也就是本章全部内容)，才是让你有能力指挥智能体、也有能力批判性评估其输出的前提。

📌 本节要点

  • 代码库设计：目标是把"想改 X"到"X 改好并跑通"的时间压缩到分钟级。
  • src/ 布局：把源码放在 src/my_project/ 下，避免当前目录误导入，用 pip install -e . 开发。
  • 单一职责与 DRY：每个函数只做一件事，复制三次再抽象，别过早抽象。
  • 命名与魔法数字：名字自解释，字面量提取为具名常量。
  • ML 设计模式：工厂模式解耦模型选择，策略模式切算法，观察者模式做回调，依赖注入方便测试。
  • 配置外部化: YAML/Hydra/argparse 分层管理，基础配置 + 实验 override，与结果一同归档。
  • 日志分级：用 logging 替代 print，结构化日志便于搜索与告警。
  • API 与打包: FastAPI 做 REST 服务并自动出 Swagger, gRPC 走内部高性能通信，pyproject.toml + 精确版本保证可复现。
  • AI 智能体协作：智能体是快而经验浅的结对程序员，擅长样板/测试/重构，不适合新架构决策、安全代码与你看不懂的代码。
  • Review 与质量门：每行必读，靠 mypy / ruff / pytest / CI 兜底智能体的错误，警惕"生成即高产"的效率幻觉。