复杂系统的问题，通常不是某个函数特别长，也不只是代码量变多了。真正的困难在于：多个部分相互影响，系统行为会随着时间、数据和环境变化而变化，而且局部修改可能带来难以预料的全局结果。The Real Python Podcast 第 309 期围绕复杂系统、复杂编码问题，以及如何维护数据科学流水线展开讨论，这些主题对日常 Python 开发同样有直接价值。
复杂系统难在哪里
可以从几个特征识别复杂系统：
- 组件之间存在反馈：一个模块的输出会改变另一个模块的输入，甚至反过来影响原模块。
- 行为具有涌现性：单个组件看起来很简单，但组合后会出现难以从局部代码直接推断的行为。
- 状态会随时间积累：缓存、历史数据、模型版本、重试记录和用户行为都会影响下一次执行。
- 边界条件很多：输入数据可能缺失、重复、延迟或格式漂移，外部服务也可能超时或返回不完整结果。
- 系统目标不止一个：准确率、延迟、成本、可解释性和可恢复性之间往往需要权衡。
因此，面对复杂编码问题时，目标不应是一次性理解整个系统，而是建立一组能够单独验证的边界。一个好的边界需要明确输入、输出、失败方式和可观察信息。
用数据契约降低隐性耦合
数据科学流水线经常把读取数据、清洗、特征计算、模型预测和结果保存写在一个脚本里。这样的代码开始运行很快，但维护成本会随着数据源数量和实验次数上升：某个字段改名后，错误可能直到模型预测阶段才暴露。
一个更稳妥的实践是为阶段之间定义轻量的数据契约。契约不一定要引入大型框架，Python 标准库就能完成一组有用的检查：
from dataclasses import dataclass
from typing import Iterable
@dataclass(frozen=True)
class Record:
user_id: int
amount: float
def parse_records(rows: Iterable[dict]) -> list[Record]:
records: list[Record] = []
for index, row in enumerate(rows, start=1):
try:
user_id = int(row["user_id"])
amount = float(row["amount"])
except (KeyError, TypeError, ValueError) as exc:
raise ValueError(f"invalid row at line {index}: {row!r}") from exc
if user_id <= 0:
raise ValueError(f"user_id must be positive at line {index}")
if amount < 0:
raise ValueError(f"amount must not be negative at line {index}")
records.append(Record(user_id=user_id, amount=amount))
return records
if __name__ == "__main__":
raw_rows = [
{"user_id": "101", "amount": "19.50"},
{"user_id": 102, "amount": 7},
]
print(parse_records(raw_rows))
运行方式：
python pipeline_contract.py
这个例子的重点不是 dataclass 本身，而是把不可信的外部数据转换为内部稳定类型。后续特征计算只需要处理 Record，不必重复判断字段是否存在、数值是否可转换。验证失败也会带上行号和原始数据，便于定位问题。
在真实项目中，可以继续扩展这条边界：
- 为时间字段统一时区和格式。
- 明确缺失值是拒绝、填充还是单独标记。
- 为重复记录定义去重键。
- 保存输入数据版本和契约版本。
- 在流水线入口处记录被拒绝记录的数量和原因。
把流水线拆成可重试的阶段
可维护流水线通常具有清晰的阶段，例如：
extract -> validate -> transform -> train/predict -> publish
每个阶段最好满足几个条件：
- 输入和输出可描述：文件、表、对象或消息的格式要明确。
- 尽量幂等：同样的输入重复执行，不应产生不可控的重复副作用。
- 失败可重试：网络读取失败和数据契约失败需要区别处理；前者通常可以重试，后者通常应该停止并报警。
- 结果可追踪：记录运行 ID、代码版本、数据版本、参数和耗时。
- 中间结果可检查：不要让所有问题都只能通过最终模型指标发现。
例如，发布预测结果时，可以使用临时文件加原子替换，避免下游读到半成品：
from pathlib import Path
import csv
def publish_predictions(rows: list[dict], destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
temporary = destination.with_suffix(destination.suffix + ".tmp")
with temporary.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=["user_id", "score"])
writer.writeheader()
writer.writerows(rows)
temporary.replace(destination)
publish_predictions(
[{"user_id": 101, "score": 0.91}],
Path("artifacts/predictions.csv"),
)
这只是一个可以改造的最小模式。生产环境还需要考虑并发运行、文件系统限制、对象存储的一致性语义，以及发布失败后的清理策略。关键决策是：让“生成结果”和“让结果对下游可见”成为两个明确步骤。
用可观察性理解复杂行为
复杂系统不能只依赖日志中的一条异常信息。至少应该为每次流水线运行关联以下信息：
run_id：本次执行的唯一标识。dataset_version：输入数据或快照的版本。code_version：代码提交、包版本或容器镜像版本。stage：失败发生在哪个阶段。input_count、output_count、rejected_count：数据量变化。duration_seconds：阶段耗时。
这些字段能帮助回答具体问题：是数据量突然变化，还是代码版本变化？是清洗阶段丢弃了大量数据，还是模型服务变慢？没有这些信息，排查复杂问题很容易退化为猜测。
测试策略也应覆盖系统边界，而不只是纯函数：
- 用单元测试验证转换规则和异常信息。
- 用少量固定数据做流水线集成测试。
- 用脏数据、空数据、重复数据和字段漂移数据做契约测试。
- 对外部服务使用超时、重试和故障响应测试。
- 对关键运行保存可复现的输入样本和参数。
采用建议：先减少隐式状态
复杂系统不可能被完全简化，但可以让复杂性集中在可管理的位置。改造现有数据科学流水线时，可以按以下顺序推进：
- 先画出阶段、输入、输出和外部依赖，不急着重写代码。
- 找出最常见的失败点，为它们增加明确的验证和错误信息。
- 把外部数据转换为内部类型，减少下游重复判断。
- 给阶段增加运行 ID、数据版本和代码版本。
- 将有副作用的发布操作与纯计算步骤分开。
- 为一条真实的小数据路径补上端到端测试。
复杂系统的可维护性，往往来自这些朴素但持续有效的约束：边界清楚、状态可见、失败可解释、阶段可重试。它们不会消除所有复杂性，却能把一次难以定位的全局故障，转化为一个可以复现和修复的局部问题。