第十章面向复用的工具库：utils 的设计与拆分

构建可复用工具库的最佳实践本文探讨如何设计可持续维护的Python工具库(utils)，重点解决数据分析/AI项目中工具函数失控的问题。核心观点包括：明确边界：工具库应是基础设施层，提供跨模块复用的通用能力，不包含业务逻辑或环境依赖。结构化拆分：按变化频率和依赖方向组织工具库，保持"稳定层"(如路径处理)与"易变层"分离，避免反向依赖业务模块。工程化

赛博罗宾

297人浏览 · 2026-01-14 18:52:34

赛博罗宾 · 2026-01-14 18:52:34 发布

在这里插入图片描述

第十章面向复用的工具库：utils 的设计与拆分

你有没有遇到过这种时刻：

项目刚开始很清爽，三周后 utils.py 变成“万能垃圾桶”？
同一个 read_json() 在 A 脚本能用，在 B 脚本就炸（编码/路径/异常全不一致）？
新同事问：“这个函数能不能复用？”你犹豫半天，只能说“先别动”？

如果你做的是数据分析 + AI 工程，工具函数往往会在三个方向上失控：
迭代快、入口多、边界杂。
你不拆 utils，它不会立刻爆炸，但会逐渐变成“谁都不敢改”的技术债核心。

本章我们把 utils 讲清楚：
不是讲“文件怎么分”，而是讲怎么把工具库做成可长期复用、可测试、可演进的基础设施层。

0. 本章目标与适用场景

学完你应该能做到：

判断一个函数该不该进 utils（边界清晰）
把“一个 utils.py”拆成“一个 utils 包”，并保持依赖方向正确
设计可复用函数的签名：输入/输出/异常/默认值
为工具库建立最小测试集（pytest 能覆盖关键边界）
在数据工程/AI 工程/RAG 工程里，让工具库真正“搬得走”

1. utils 到底是什么？先把边界画出来

很多人的误区是：
“utils 就是放常用函数的地方。”

更准确的定义是：

utils 是项目的基础设施层：提供跨模块复用、稳定、可测试的通用能力。
它应该被业务依赖，但不反向依赖业务。

1.1 utils 应该收什么？

满足三条就可以收：

跨模块复用：不属于某个业务域（domain）
纯度高：输入输出明确，副作用可控
可测试：能写测试覆盖边界，不靠“跑一遍看看”

典型例子：

IO：读写 JSON/YAML/CSV、文件校验、压缩解压
路径：目录创建、文件枚举、相对/绝对路径规范化
运行时：计时器、重试、退避、并发小封装
配置：加载配置、默认值合并、schema 校验
日志：统一 logger、结构化字段（trace_id / run_id）
校验：参数检查、类型断言、数据范围验证

1.2 utils 不该收什么？

三类最常把工具库“污染”：

业务逻辑：比如“论文解析规则”“用户画像特征工程”
强依赖环境：读环境变量决定行为、写死目录/URL/集群地址
重框架封装：把 pandas/torch/llm 框架整体再包一层，维护成本极高

一句话：
工具库是稳定的通用能力，不是业务快捷方式。

2. 为什么数据/AI项目更需要“可复用 utils”？

Web 项目里，工具函数多半是“锦上添花”。
但在数据/AI工程里，工具函数往往是“主干血管”：

数据清洗需要统一编码、缺失值策略
特征/指标要处理 NaN、空集合、分母为 0
训练与推理要保证一致性（同样的预处理同样的结果）
评测、实验记录、追踪（trace/run_id）要可追溯

你不把工具库做稳，会出现一种很隐蔽的失控：

代码能跑，但每个脚本都“各自为政”；
结果能出，但没有一致性；
出问题时，没人敢改“那坨 utils”。

3. 建一个最小可维护的 utils 结构（建议）

先给你一套“中型项目”够用的结构（数据分析 + AI 工程通用）：

myproj/
  src/
    myproj/
      __init__.py
      utils/
        __init__.py
        paths.py
        io.py
        config.py
        logging.py
        timing.py
        retry.py
        text.py
        validate.py
      features/
      metrics/
      pipelines/
  tests/
    test_utils_io.py
    test_utils_retry.py
    test_utils_validate.py

你会发现它的拆分逻辑不是“随便分类”，而是：

paths/io/config/logging：基础能力、变化频率低
timing/retry/validate：工程兜底能力，跨项目可复用
text：数据项目最常见的通用模块（清洗/规范化/编码）

4. 拆分原则：按“变化频率”与“依赖方向”拆

4.1 按变化频率拆（非常实用）

稳定层（几乎不变）：paths / validate / retry
半稳定层（偶尔变）：io / config / logging
易变层（经常变）：这类通常不该进 utils（应放业务模块）

拆分的目标不是“更细”，而是“更稳定”。

4.2 按依赖方向拆（避免工具库被绑架）

工具库要遵守一个硬规则：

utils 只能被依赖，不能反向依赖业务模块。

坏例子：

utils/io.py 里 import features/
utils/config.py 里 import pipelines/

一旦这样做，utils 就失去复用性，变成“项目私货”。

5. 先做一个“可复用 IO 工具”：读写 JSON 的正确姿势

很多人写 IO 的方式是：

能读就行
能写就行

但复用失败往往发生在：编码、异常、路径、默认值不一致。

5.1 一个更可复用的 read_json / write_json

# src/myproj/utils/io.py
from __future__ import annotations
import json
from pathlib import Path
from typing import Any

def read_json(path: str | Path, *, encoding: str = "utf-8") -> Any:
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"json not found: {p}")
    with p.open("r", encoding=encoding) as f:
        return json.load(f)

def write_json(
    path: str | Path,
    obj: Any,
    *,
    encoding: str = "utf-8",
    indent: int = 2,
    ensure_ascii: bool = False,
    overwrite: bool = True,
) -> Path:
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    if p.exists() and not overwrite:
        raise FileExistsError(f"json exists: {p}")
    with p.open("w", encoding=encoding) as f:
        json.dump(obj, f, indent=indent, ensure_ascii=ensure_ascii)
    return p

你会注意到这几个“工程化细节”：

输入支持 str | Path（更友好）
默认 utf-8，并可覆盖
目录自动创建（减少样板代码）
overwrite 是显式策略（避免误覆盖）
异常不吞，错误可定位

6. 工具函数如何“可长期维护”？看 4 个关键点

6.1 函数签名要像文档一样清晰

不要写“万能 *args/**kwargs”。
在工具库里，签名不清晰等价于不可复用。

6.2 默认值要保守（安全优先）

工具默认策略应“可预测、少惊喜”。
例如写文件默认可覆盖与否，你必须显式选择。

6.3 异常要可控（不要吞）

工具库吞异常，会把问题延迟到线上爆炸。

6.4 副作用要可见（不要暗中改全局）

例如不要在工具函数里 os.chdir()、不要偷偷读环境变量改变行为。

7. 给 utils 配上“最小测试集”：否则它永远不稳

工具库最怕“看起来能用”，一复用就崩。
解决办法不是写一堆测试，而是写关键边界测试。

7.1 用 pytest 测 IO 的边界

# tests/test_utils_io.py
import pytest
from myproj.utils.io import read_json, write_json

def test_write_and_read_json(tmp_path):
    p = tmp_path / "a.json"
    write_json(p, {"x": 1})
    assert read_json(p) == {"x": 1}

def test_read_json_not_found(tmp_path):
    with pytest.raises(FileNotFoundError):
        read_json(tmp_path / "missing.json")

def test_write_json_no_overwrite(tmp_path):
    p = tmp_path / "a.json"
    write_json(p, {"x": 1})
    with pytest.raises(FileExistsError):
        write_json(p, {"x": 2}, overwrite=False)