1. 从零手搓AI工程为什么我不建议你直接调包第一次看到ai-engineering-from-scratch这个项目名的时候我脑子里蹦出来的画面是一个人坐在黑漆漆的终端前拒绝所有现成的框架从矩阵乘法开始一行一行地把一个能跑起来的AI系统给搭出来。这个直觉基本是对的但又不完全对。因为“from scratch”这个词在AI工程语境里其实有两层意思一层是算法层面的从零实现比如不用PyTorch的nn.Linear自己拿NumPy写全连接层的前向和反向传播另一层是工程层面的从零搭建比如不用现成的推理服务框架自己写一个带批处理、带缓存、带降级策略的推理网关。这个项目标题真正戳中的痛点是现在绝大多数AI应用开发者都患上的“框架依赖症”。你打开任何一个招聘JD看到的都是“熟悉PyTorch/TensorFlow”“了解LangChain/LlamaIndex”但很少有人问一句如果这些框架明天不维护了你还能不能把模型跑起来如果线上推理延迟突然飙到2秒你能不能在不看框架源码的情况下定位到是哪个环节出了问题ai-engineering-from-scratch适合三类人第一类是刚入门AI、被各种框架的抽象层绕晕的新手想通过手写一遍来建立直觉第二类是有一定经验、但只会调包、遇到性能瓶颈就束手无策的中级工程师第三类是想做技术选型、需要评估“自研”和“用现成”之间成本差异的架构师。这篇文章我会按照一个完整的AI工程链路来拆解从数据管道、模型实现、训练循环到推理服务、监控告警每一层都讲清楚“从零做”和“用现成”的边界在哪里以及我踩过的那些坑。2. 整体设计思路从零不是目的可控才是2.1 为什么选择“手搓”而不是“调包”先说一个反直觉的观点从零实现AI工程不是为了替代框架而是为了在关键时刻能看懂框架在干什么。我见过太多团队模型训练loss不下降第一反应是换优化器、换学习率、换batch size但从来没人去检查数据管道里是不是有脏数据、梯度裁剪是不是写错了位置。这些问题的根源都是因为对底层机制缺乏感知。这个项目的整体设计思路我把它归纳为“三层递进”第一层用NumPy实现核心算子。包括矩阵乘法、卷积、注意力机制、反向传播。这一层的目标是建立数学直觉知道每个算子在内存里是怎么排布的、计算复杂度是多少。第二层用纯Python搭建训练框架。包括数据加载器、损失函数、优化器、学习率调度、模型保存与加载。这一层的目标是理解训练循环的每一个环节知道什么时候该同步、什么时候该异步。第三层用标准库搭建推理服务。包括HTTP服务、请求队列、批处理、缓存、限流、降级。这一层的目标是理解线上服务的瓶颈在哪里知道怎么在不引入重型依赖的情况下保证可用性。选择这种设计的原因很简单依赖越少可控性越强。你调一个model.fit()背后可能跑了几万行代码出了问题你只能看日志。但你自己写的训练循环每一行你都知道它在干什么出了问题你能直接定位到行号。当然代价是开发效率低、性能可能不如优化过的框架。所以我的建议是学习阶段从零写生产阶段用框架但心里要有一张从零到框架的映射图。2.2 技术选型的取舍逻辑在具体技术选型上我做了几个关键决策每个决策背后都有明确的理由环节选择放弃的方案理由数值计算NumPyPyTorch/TensorFlowNumPy的API更底层没有自动求导逼你手写反向传播理解更深刻训练框架纯Python NumPyPyTorch Lightning不引入额外抽象层训练循环完全透明推理服务Python标准库http.serverFastAPI/Flask标准库零依赖适合理解HTTP协议和并发模型数据存储本地文件 内存数据库/消息队列减少外部依赖聚焦核心逻辑监控标准库logging 自定义指标Prometheus/Grafana先理解指标采集原理再上工具这个选型表里最容易被质疑的是“为什么不用FastAPI”。我的理由是FastAPI确实好用但它的异步模型、依赖注入、中间件机制对于理解“一个推理请求从进来到出去到底经历了什么”这件事反而是一种干扰。你用http.server写一遍就会明白什么是阻塞IO、什么是线程池、什么是请求队列这些概念在FastAPI里被封装得太好了好到你根本注意不到它们的存在。注意从零实现不等于拒绝工具。我的原则是“能用标准库解决的不引入第三方库必须用第三方库的先读源码再决定”。比如NumPy是必须的因为纯Python的列表运算太慢了但Flask是不必要的因为http.server足够表达核心逻辑。2.3 项目目录结构与模块划分一个清晰的目录结构能让“从零”这件事变得有章可循。我建议的目录结构是这样的ai-engineering-from-scratch/ ├── core/ # 核心算子层 │ ├── tensor.py # 张量基础类 │ ├── ops.py # 基础运算加减乘除、矩阵乘法 │ ├── activations.py # 激活函数 │ └── losses.py # 损失函数 ├── nn/ # 神经网络层 │ ├── linear.py # 全连接层 │ ├── conv.py # 卷积层 │ ├── attention.py # 注意力机制 │ └── module.py # 模块基类 ├── train/ # 训练框架 │ ├── dataloader.py # 数据加载器 │ ├── optimizer.py # 优化器 │ ├── scheduler.py # 学习率调度 │ └── loop.py # 训练循环 ├── serve/ # 推理服务 │ ├── server.py # HTTP服务 │ ├── batching.py # 批处理 │ ├── cache.py # 缓存 │ └── guard.py # 限流降级 ├── monitor/ # 监控 │ ├── metrics.py # 指标采集 │ └── logger.py # 日志 └── examples/ # 示例 ├── mnist_from_scratch.py └── serve_demo.py这个结构的核心思想是分层解耦core层不依赖任何其他层nn层依赖coretrain层依赖nn和coreserve层依赖train和nn。每一层都可以单独测试、单独替换。比如你想把NumPy换成JAX只需要改core层上层代码基本不动。这种设计在真实项目里非常重要因为AI工程的变化速度太快了今天用PyTorch明天可能就要换ONNX Runtime分层解耦能让你把迁移成本降到最低。3. 核心细节解析手写算子的那些坑3.1 张量类的设计与内存布局手写AI工程的第一步是设计一个张量类。很多人觉得张量就是多维数组用NumPy的ndarray不就行了但问题是ndarray没有梯度信息也没有计算图。你需要自己包一层。我设计的Tensor类核心字段只有三个class Tensor: def __init__(self, data, requires_gradFalse): self.data np.array(data, dtypenp.float32) self.requires_grad requires_grad self.grad None self._backward lambda: None self._prev set()这里有几个关键决策。第一dtype固定为float32因为这是大多数AI框架的默认精度也是GPU计算最友好的精度。第二grad初始为None而不是全零数组这样可以区分“没有梯度”和“梯度为零”两种情况。第三_backward是一个闭包用来存储反向传播的逻辑这是参考micrograd的设计非常优雅。内存布局方面NumPy默认是行优先C order这意味着对于一个(batch, seq, hidden)的张量最后一个维度是连续存储的。这个细节在实现注意力机制的时候特别重要因为Q K.T这个操作如果K的转置不是视图而是拷贝内存开销会翻倍。我的做法是尽量用transposereshape的组合来避免显式拷贝只有在必须连续内存的时候才调np.ascontiguousarray。实操心得在实现Linear层的时候权重矩阵的shape是(in_features, out_features)输入是(batch, in_features)所以前向是x W。但如果你从PyTorch转过来可能会习惯性地写成W x然后发现shape对不上。记住NumPy的是矩阵乘法不是点积(batch, in) (in, out)得到(batch, out)这是正确的。3.2 反向传播的手写实现与梯度检查反向传播是“从零”这件事里最容易出错的地方。我见过太多人前向传播写得飞快一到反向就各种shape不匹配、梯度爆炸。我的经验是每实现一个算子立刻写梯度检查。梯度检查的原理很简单用数值微分近似解析梯度。对于函数f(x)数值梯度是(f(xeps) - f(x-eps)) / (2*eps)解析梯度是反向传播算出来的。如果两者差距在1e-5以内说明反向传播写对了。def grad_check(f, x, eps1e-5): # 数值梯度 num_grad np.zeros_like(x) for i in range(x.size): x_flat x.flatten() x_flat[i] eps f_plus f(x_flat.reshape(x.shape)) x_flat[i] - 2 * eps f_minus f(x_flat.reshape(x.shape)) num_grad.flat[i] (f_plus - f_minus) / (2 * eps) # 解析梯度 x_tensor Tensor(x, requires_gradTrue) out f(x_tensor) out.backward() ana_grad x_tensor.grad # 比较 diff np.abs(num_grad - ana_grad).max() print(f最大梯度误差: {diff}) return diff 1e-5这个检查函数我建议你每写一个新算子就跑一遍。我当初写Softmax的反向传播时就是靠梯度检查发现了一个bug我在计算dL/dx的时候忘记减去sum(dL/dy * y)这一项导致梯度整体偏大。这个bug在训练初期看不出来因为Softmax的输出都在0到1之间梯度偏大一点也能收敛但训练到后期就会导致loss震荡。3.3 训练循环的每个环节拆解训练循环看起来简单不就是“前向、反向、更新”三步吗但真正手写一遍你会发现每个环节都有讲究。数据加载我实现了一个简单的DataLoader支持batch、shuffle、多进程预取。这里的关键是预取如果不用预取GPU或者CPU在等数据的时候会空闲利用率可能只有50%。预取的实现方式是用一个后台线程提前把下一个batch的数据准备好放在队列里。class DataLoader: def __init__(self, dataset, batch_size, shuffleTrue, num_workers2): self.dataset dataset self.batch_size batch_size self.shuffle shuffle self.num_workers num_workers self.queue queue.Queue(maxsizenum_workers * 2) def _worker(self): indices np.arange(len(self.dataset)) if self.shuffle: np.random.shuffle(indices) for i in range(0, len(indices), self.batch_size): batch_idx indices[i:iself.batch_size] batch_data [self.dataset[j] for j in batch_idx] self.queue.put(batch_data) def __iter__(self): threads [threading.Thread(targetself._worker) for _ in range(self.num_workers)] for t in threads: t.start() while True: try: yield self.queue.get(timeout1) except queue.Empty: break损失函数交叉熵损失是最常用的但手写的时候要注意数值稳定性。直接算-log(softmax(x))会溢出因为exp(x)可能非常大。正确的做法是先减去最大值x x - x.max(axis-1, keepdimsTrue)然后再算exp和log。优化器SGD是最简单的但实际用的时候一般会用Adam。Adam的核心是维护一阶矩和二阶矩的指数移动平均然后做偏差修正。手写Adam的时候最容易错的是偏差修正的步数t这个t是全局步数不是epoch数。学习率调度我实现了三种调度器StepLR每N个epoch降一次、CosineAnnealing余弦退火、Warmup预热。Warmup在Transformer训练里几乎是必须的因为一开始梯度很大直接上大学习率会炸。常见问题训练loss不下降怎么排查我的排查顺序是先看数据有没有脏数据、标签对不对再看梯度有没有梯度消失或爆炸再看学习率是不是太大或太小最后看模型结构是不是太浅或太深。这个顺序不能乱因为数据问题是最常见的模型问题是最少见的。4. 实操过程从MNIST到推理服务4.1 手写MNIST训练全流程光说不练假把式。我用自己写的框架在MNIST上跑了一个完整的训练流程。MNIST虽然简单但包含了AI工程的所有核心环节数据加载、模型定义、训练循环、评估、保存。数据准备MNIST的原始格式是IDX需要自己解析。IDX的格式是前4个字节是magic number接着4个字节是图像数量再4个字节是行数再4个字节是列数然后就是像素数据。解析代码如下def load_mnist_images(path): with open(path, rb) as f: magic int.from_bytes(f.read(4), big) num_images int.from_bytes(f.read(4), big) rows int.from_bytes(f.read(4), big) cols int.from_bytes(f.read(4), big) data np.frombuffer(f.read(), dtypenp.uint8) data data.reshape(num_images, rows * cols) return data.astype(np.float32) / 255.0模型定义一个两层全连接网络输入784隐藏层128输出10。激活函数用ReLU最后一层不加激活因为交叉熵损失里包含了Softmax。class MLP(Module): def __init__(self): self.fc1 Linear(784, 128) self.fc2 Linear(128, 10) def forward(self, x): x self.fc1(x) x relu(x) x self.fc2(x) return x训练循环batch size设为64学习率0.001优化器用Adam训练10个epoch。每个epoch打印一次训练loss和验证准确率。model MLP() optimizer Adam(model.parameters(), lr0.001) criterion CrossEntropyLoss() for epoch in range(10): model.train() for batch_x, batch_y in train_loader: optimizer.zero_grad() logits model(batch_x) loss criterion(logits, batch_y) loss.backward() optimizer.step() model.eval() correct 0 total 0 for batch_x, batch_y in val_loader: logits model(batch_x) pred logits.data.argmax(axis-1) correct (pred batch_y).sum() total len(batch_y) print(fEpoch {epoch}, Val Acc: {correct / total:.4f})实测下来这个简单的MLP在MNIST上能跑到97%左右的准确率训练时间在CPU上大概2分钟。这个结果不算好用CNN能到99%以上但足以验证框架的正确性。4.2 推理服务的搭建与批处理优化训练完之后下一步是把模型部署成服务。我用http.server写了一个最简单的推理服务核心逻辑是接收POST请求解析JSON跑前向传播返回结果。class InferenceHandler(BaseHTTPRequestHandler): def do_POST(self): content_length int(self.headers[Content-Length]) body self.rfile.read(content_length) data json.loads(body) # 预处理 x np.array(data[input], dtypenp.float32) x x.reshape(1, -1) # 推理 with torch.no_grad(): logits model(Tensor(x)) pred logits.data.argmax(axis-1)[0] # 返回 response json.dumps({prediction: int(pred)}) self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() self.wfile.write(response.encode())这个服务能跑但性能很差。我实测了一下单请求延迟大概20msQPS只有50左右。瓶颈在于每个请求都要重新做一遍前向传播没有批处理没有缓存。批处理优化批处理的核心思想是把多个请求攒在一起一次性跑前向传播。这样能充分利用矩阵运算的并行性。实现方式是维护一个请求队列后台线程每隔一段时间比如10ms从队列里取一批请求拼成一个batch跑一次推理然后把结果分发给各个请求。class BatchProcessor: def __init__(self, model, max_batch_size32, max_wait0.01): self.model model self.max_batch_size max_batch_size self.max_wait max_wait self.queue queue.Queue() self.thread threading.Thread(targetself._process, daemonTrue) self.thread.start() def _process(self): while True: batch [] start time.time() while len(batch) self.max_batch_size: timeout self.max_wait - (time.time() - start) if timeout 0: break try: item self.queue.get(timeouttimeout) batch.append(item) except queue.Empty: break if batch: inputs np.stack([item[input] for item in batch]) with torch.no_grad(): logits self.model(Tensor(inputs)) preds logits.data.argmax(axis-1) for item, pred in zip(batch, preds): item[future].set_result(int(pred))加了批处理之后QPS从50提升到了300左右延迟从20ms降到了8ms。这个提升非常明显因为矩阵运算的并行性被充分利用了。缓存优化如果有些请求的输入是重复的可以用缓存来避免重复计算。我用了一个简单的LRU缓存key是输入的hashvalue是预测结果。实测下来如果请求有20%的重复率QPS能再提升15%左右。4.3 监控指标的采集与告警服务上线之后必须要有监控。我采集的指标包括QPS、延迟P50/P95/P99、错误率、CPU/内存使用率。采集方式是在请求处理的前后打时间戳然后定期汇总。class Metrics: def __init__(self): self.latencies [] self.errors 0 self.total 0 self.lock threading.Lock() def record(self, latency, is_error): with self.lock: self.latencies.append(latency) self.total 1 if is_error: self.errors 1 def report(self): with self.lock: if not self.latencies: return latencies sorted(self.latencies) p50 latencies[int(len(latencies) * 0.5)] p95 latencies[int(len(latencies) * 0.95)] p99 latencies[int(len(latencies) * 0.99)] error_rate self.errors / self.total if self.total 0 else 0 print(fQPS: {self.total}, P50: {p50:.4f}, P95: {p95:.4f}, P99: {p99:.4f}, Error: {error_rate:.4f}) self.latencies [] self.errors 0 self.total 0告警规则我设了三条P99延迟超过100ms、错误率超过1%、QPS超过500。触发告警的时候会打印日志并发送通知。这里的关键是告警阈值不能拍脑袋定要根据历史数据来定。我的做法是先跑一周收集P99延迟的分布然后取99.9分位数作为阈值。实操心得监控指标不要贪多先盯住三个核心指标延迟、错误率、吞吐量。这三个指标能覆盖80%的问题。其他的指标比如CPU、内存、GPU利用率是辅助定位问题的不是用来告警的。5. 常见问题与排查技巧实录5.1 训练阶段的典型问题问题一Loss变成NaN。这是最常见的问题原因通常是梯度爆炸或者学习率太大。排查步骤先检查数据里有没有NaN或Inf再检查学习率是不是太大试试除以10再检查有没有做梯度裁剪。梯度裁剪的代码很简单def clip_gradients(model, max_norm1.0): total_norm 0 for p in model.parameters(): if p.grad is not None: total_norm (p.grad ** 2).sum() total_norm np.sqrt(total_norm) if total_norm max_norm: scale max_norm / (total_norm 1e-6) for p in model.parameters(): if p.grad is not None: p.grad * scale问题二训练loss下降但验证loss上升。这是过拟合的典型表现。解决方法加Dropout、加L2正则化、减小模型规模、增加数据量。我一般先试加Dropout因为最简单效果也最明显。问题三训练速度慢。排查顺序先看数据加载是不是瓶颈用time.time()打点再看前向传播是不是瓶颈再看反向传播是不是瓶颈。如果是数据加载慢加预取如果是计算慢检查有没有用向量化操作有没有不必要的循环。5.2 推理阶段的典型问题问题一延迟忽高忽低。原因通常是批处理的大小不稳定。如果请求量波动大批处理的大小就会波动导致延迟不稳定。解决方法是设置一个最小批处理大小或者用动态批处理根据队列长度调整等待时间。问题二内存泄漏。Python的垃圾回收机制有时候会漏掉一些循环引用。排查方法是定期打印gc.get_objects()的数量如果持续增长说明有泄漏。解决方法是手动断开循环引用或者用weakref。问题三服务崩溃。原因可能是未捕获的异常、内存溢出、或者死锁。解决方法是加全局异常捕获加内存限制加超时机制。问题可能原因排查方法解决方案Loss NaN梯度爆炸/学习率大检查梯度范数梯度裁剪/降学习率过拟合模型太复杂对比训练/验证曲线Dropout/L2/减模型训练慢数据加载瓶颈打点计时预取/向量化延迟波动批处理不稳定监控批大小动态批处理内存泄漏循环引用gc对象计数weakref/手动断开服务崩溃未捕获异常看日志全局异常捕获5.3 独家避坑技巧技巧一用np.seterr捕获数值异常。NumPy默认对除零、溢出等操作只是警告不会报错。设置np.seterr(allraise)可以让这些操作直接抛异常方便定位问题。技巧二用tracemalloc定位内存泄漏。tracemalloc是Python标准库里的内存追踪工具能告诉你哪一行代码分配了最多内存。用法很简单import tracemalloc tracemalloc.start() # ... 跑一段时间 ... snapshot tracemalloc.take_snapshot() top_stats snapshot.statistics(lineno) for stat in top_stats[:10]: print(stat)技巧三用cProfile定位性能瓶颈。cProfile是Python标准库里的性能分析工具能告诉你每个函数调用了多少次、耗时多少。用法import cProfile cProfile.run(train_one_epoch(), sortcumtime)技巧四推理服务一定要加超时。不管你的模型多快都要加一个超时机制防止某个请求卡死整个服务。我的做法是用signal.alarm或者threading.Timer超时之后直接返回错误。技巧五模型保存要保存完整状态。不仅要保存权重还要保存优化器状态、epoch数、学习率调度器状态。否则恢复训练的时候学习率会从头开始导致loss震荡。6. 从零到生产的扩展路径6.1 性能优化的几个方向当你把基础版本跑通之后下一步就是优化性能。我总结了几个方向方向一向量化。把Python循环换成NumPy的向量化操作。比如计算准确率不要用for循环逐个比较直接用(pred label).mean()。这个提升通常是10倍到100倍。方向二并行化。数据加载用多进程推理用多线程训练用多GPU。Python的GIL限制了多线程的计算并行所以计算密集型的任务要用多进程。方向三量化。把float32换成float16或者int8能减少一半到四分之三的内存占用推理速度也能提升。但量化会损失精度需要做校准。方向四算子融合。把多个小算子合并成一个大算子减少内存访问次数。比如把Linear ReLU融合成一个算子能减少一次内存读写。6.2 从单机到分布式的演进单机跑通之后如果数据量或者模型规模上来了就要考虑分布式。分布式的核心问题是通信。数据并行的时候每个GPU算完梯度之后要和其他GPU做all-reduce把梯度平均一下。这个通信开销可能占到总时间的30%以上。我的建议是先做数据并行再做模型并行。数据并行简单改几行代码就行模型并行复杂需要切分模型、处理跨设备的依赖关系。如果数据并行还不够再考虑流水线并行。6.3 持续迭代的工程实践AI工程不是一次性的而是一个持续迭代的过程。我建议建立以下几个实践版本管理模型、数据、代码都要版本化。模型用model_v1.pkl、model_v2.pkl数据用DVC或者类似的工具代码用Git。实验追踪每次训练都要记录超参数、指标、模型文件。可以用MLflow或者TensorBoard也可以自己写一个简单的JSON日志。A/B测试新模型上线之前先跑A/B测试对比新旧模型的指标。不要直接全量替换。回滚机制新模型出问题的时候要能快速回滚到旧模型。所以模型文件要保留最近几个版本。我个人在实际操作中的体会是从零实现的最大价值不是让你真的在生产环境里手搓一切而是让你在调包的时候心里有一张清晰的地图。你知道model.fit()背后发生了什么你知道optimizer.step()到底更新了哪些参数你知道DataLoader的num_workers为什么会影响训练速度。这种“知道”在遇到问题的时候能让你少走很多弯路。最后再分享一个小技巧如果你觉得从零写整个框架太耗时可以只写你最不熟悉的那一部分。比如你不懂反向传播就只手写反向传播你不懂批处理就只手写批处理。聚焦一个点写透它比泛泛地写一遍整个框架收获更大。这个项目后续还可以这样扩展加一个简单的自动微分引擎支持动态计算图或者加一个模型编译器把Python代码编译成C或者加一个分布式训练模块支持多机多卡。每一个方向都够你写好几篇文章了。