1. 从零搭建AI工程能力为什么我劝你别一上来就啃论文ai-engineering-from-scratch这个标题第一次看到的时候我愣了一下。不是因为它有多高深恰恰相反——它太直白了直白到像一句废话。但仔细想想这四个词组合在一起其实精准地戳中了当下很多人的困境AI相关的资料铺天盖地论文、课程、开源项目、付费专栏但真正能让人从零开始、一步步把AI工程能力搭起来的东西少得可怜。我自己在这个领域摸爬滚打了几年带过团队也面试过不少人。一个很普遍的现象是很多人对Transformer的注意力机制能说得头头是道但你让他从零搭一个能跑通的推理服务他卡在环境配置上就出不来了。这不是个例这是系统性的问题——我们太习惯自上而下地学AI了先学理论再学框架最后才碰工程。但真正做AI工程的人都知道这个顺序是反的。所以这篇内容我想聊的是ai-engineering-from-scratch这件事本身——从零开始构建AI工程能力到底应该怎么走。不是那种先学Python再学PyTorch的泛泛之谈而是从工程视角出发把这条路上真正关键的节点、容易踩的坑、以及我自己的实操经验尽可能完整地拆开来讲。适合谁看如果你已经会写一点代码对AI有基本认知但不知道如何把会调包变成能落地那这篇就是写给你的。如果你已经是资深工程师也可以看看我在工具选型和工程决策上的思路或许有能对上的地方。2. 整体思路拆解为什么从零不等于从理论开始2.1 先搞清楚AI工程到底在工程什么很多人把AI工程和机器学习研究混为一谈这是第一个要掰扯清楚的事。机器学习研究的核心是发现新方法AI工程的核心是让已有方法稳定、高效、可维护地跑起来。这两个目标的差异决定了它们对能力的要求完全不同。我见过不少从研究转工程的人最大的不适应在于研究追求的是最好结果工程追求的是可预期结果。你用一个模型在测试集上刷到95%的准确率和研究里发一篇论文是两码事。工程上你要考虑的是这个95%在线上环境能不能复现推理延迟能不能接受模型更新了怎么回滚数据分布漂移了怎么监控这些问题论文里不会告诉你。所以ai-engineering-from-scratch的第一步不是去补数学而是先建立工程思维。具体来说你需要理解三个核心概念可复现性、可观测性、可扩展性。这三个词听起来像口号但每一个都对应着具体的工程实践。可复现性意味着你的实验环境、数据版本、模型权重、超参数都要有记录和版本控制可观测性意味着你的服务要有日志、指标、追踪出问题能定位可扩展性意味着你的架构要能应对流量增长和模型迭代而不是每次都要推倒重来。2.2 工具选型的底层逻辑别被生态绑架AI工程领域有个很尴尬的现实工具迭代太快了。你今天选的框架可能半年后就没人维护了。我经历过从Theano到TensorFlow到PyTorch的迁移也见过无数团队在工具选型上反复横跳。所以从零构建能力时工具选型的逻辑应该是优先选生态成熟、社区活跃、迁移成本低的工具而不是选功能最炫的。具体到实操层面我的建议是分三层来看。底层计算框架PyTorch目前是事实标准不用犹豫中间层服务框架FastAPI加Uvicorn的组合足够覆盖大多数场景除非你有极端的性能需求才考虑Triton或TorchServe上层编排和监控初期用Docker Compose加Prometheus加Grafana就够了别一上来就上Kubernetes那是给自己找麻烦。这个选型逻辑背后的考量是从零构建能力时你的认知带宽是有限的。如果一开始就陷入复杂工具的配置泥潭你根本没精力去理解AI工程本身的核心问题。我见过太多人花两周时间搭了一个完美的MLOps平台结果连一个最简单的推理服务都没跑通。这是典型的工具先行、问题后置本末倒置。2.3 学习路径的设计以可运行的最小系统为锚点传统的学习路径是线性的先学Python再学NumPy再学PyTorch再学模型部署。这条路径的问题在于反馈周期太长学到后面忘了前面而且每个环节都是孤立的不知道为什么要学。我的建议是以可运行的最小系统为锚点反向驱动学习。什么意思你先定一个极简目标比如用预训练的模型做一个图片分类的API然后围绕这个目标去补所需的知识。你会发现为了跑通这个API你需要学Python的Web框架、需要理解模型加载和推理、需要处理输入输出格式、需要做基本的错误处理。这些知识是在解决问题的过程中自然习得的记忆更牢理解更深。这个思路和ai-engineering-from-scratch的内核是一致的从零不是从理论零开始而是从工程实践的零开始。你先有一个能跑的东西哪怕它很粗糙然后在这个基础上迭代、优化、扩展。这比先啃三个月论文再动手效率高得多。3. 核心细节解析从零搭建AI工程能力的四个关键节点3.1 环境管理别让你的项目死在在我机器上能跑环境问题是AI工程里最不起眼但最致命的问题。我敢说每个AI工程师都有过被环境依赖折磨的经历。CUDA版本不匹配、Python包冲突、系统库缺失这些问题看起来是小事但能消耗你大量时间。从零开始我强烈建议你从第一天就用容器化。不是说要你精通Docker而是说你要养成环境即代码的习惯。具体操作上一个最简的Dockerfile就能解决大部分问题FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]这个Dockerfile很简单但它解决了三个关键问题Python版本固定、依赖隔离、运行环境一致。你可能会说我用conda也能做到。没错但conda的环境在跨机器迁移时经常出问题而Docker镜像可以在任何支持Docker的机器上运行这是本质区别。注意如果你要用GPU基础镜像要换成带CUDA的版本比如nvidia/cuda:12.1-runtime-ubuntu22.04然后在里面装Python。别用python:3.10-slim再自己装CUDA那个坑我踩过驱动和运行时版本对不上排查起来很痛苦。还有一个细节requirements.txt要锁版本。不要写torch2.0要写torch2.1.0。我见过太多因为自动升级导致的线上事故。锁版本虽然看起来不够灵活但工程上确定性比灵活性重要得多。3.2 模型加载与推理从能跑到跑得稳模型加载和推理是AI工程的核心环节但很多人只关注能不能出结果忽略了出结果的过程是否可靠。从零构建能力时这个环节要重点关注三个问题加载策略、批处理、错误处理。加载策略上最简单的是每次请求都加载模型但这在生产环境是不可接受的因为加载模型可能耗时几秒到几十秒。正确的做法是服务启动时加载一次常驻内存。用FastAPI的话可以放在startup事件里from fastapi import FastAPI import torch app FastAPI() model None app.on_event(startup) async def load_model(): global model model torch.load(model.pth, map_locationcpu) model.eval()这里有个细节map_locationcpu。如果你在GPU机器上训练在CPU机器上推理不加这个参数会报错。这个坑很常见但文档里往往不会强调。批处理是提升推理吞吐的关键。单条推理的GPU利用率可能只有10%但批处理之后能到80%以上。实现上你可以用一个简单的队列来攒批import asyncio from collections import deque batch_queue deque() batch_size 8 async def process_batch(): while True: if len(batch_queue) batch_size: batch [batch_queue.popleft() for _ in range(batch_size)] inputs torch.stack([b[input] for b in batch]) with torch.no_grad(): outputs model(inputs) for b, out in zip(batch, outputs): b[future].set_result(out) await asyncio.sleep(0.01)这个实现很粗糙但思路是对的攒够一批再推理减少GPU的空转。生产环境可以用更成熟的方案比如Triton的dynamic batching但理解这个原理很重要。错误处理是很多人忽略的。模型推理可能因为各种原因失败输入格式不对、显存不足、模型文件损坏。如果不做错误处理一个请求失败可能导致整个服务崩溃。基本的做法是用try-except包裹推理逻辑返回有意义的错误信息同时记录日志。别让用户看到一堆堆栈信息那既不专业也不安全。3.3 数据管道AI工程里最容易被低估的部分如果说模型是AI工程的心脏那数据管道就是血管。但奇怪的是大部分AI工程的学习资料都在讲模型很少有人认真讲数据管道。我从零做项目时最大的时间消耗就在数据管道上。数据管道要解决的核心问题是数据从哪来、怎么处理、怎么喂给模型。从零开始你不需要一上来就搞Kafka加Flink那套流式架构但你需要理解几个基本原则。第一数据版本控制。你的模型是用哪个版本的数据训练的如果数据更新了模型要不要重新训练这些问题需要数据版本管理来回答。最简单的做法是用DVCData Version Control它能把数据文件和Git提交关联起来。别用data_final_v2.csv这种命名方式那是灾难的开始。第二数据校验。喂给模型的数据必须符合预期格式。我见过因为一个空值导致整个批次推理失败的案例。基本的校验包括字段是否存在、类型是否正确、数值范围是否合理。可以用Pydantic来做from pydantic import BaseModel, validator class InferenceRequest(BaseModel): text: str validator(text) def text_not_empty(cls, v): if not v.strip(): raise ValueError(text cannot be empty) return v第三预处理和后处理的一致性。训练时的预处理逻辑和推理时的预处理逻辑必须完全一致否则结果会莫名其妙地差。我的做法是把预处理逻辑封装成独立的模块训练和推理共用同一份代码。这个原则听起来简单但实际操作中很容易因为训练时用pandas推理时用numpy这种细节导致不一致。3.4 监控与日志让你的系统会说话AI系统上线后最怕的是什么是它悄悄坏了但你不知道。模型可能因为数据漂移导致准确率下降服务可能因为内存泄漏导致响应变慢这些都不会主动告诉你。所以监控和日志是AI工程的必备能力。从零开始你不需要复杂的监控体系但需要覆盖三个基本维度系统指标、业务指标、模型指标。系统指标包括CPU、内存、GPU利用率、请求延迟、错误率。这些可以用Prometheus加Grafana来采集和展示。业务指标取决于你的应用场景比如推荐系统的点击率、搜索系统的召回率。模型指标包括预测分布、置信度分布、特征重要性变化。日志方面我建议用结构化日志而不是简单的print。Python的logging模块配合JSON格式输出能让日志更容易被检索和分析import logging import json logger logging.getLogger(__name__) def log_inference(request_id, input_data, output_data, latency): logger.info(json.dumps({ request_id: request_id, input_shape: str(input_data.shape), output_shape: str(output_data.shape), latency_ms: latency }))这个日志格式看起来简单但它能让你在出问题时快速定位是输入有问题还是模型输出异常还是延迟突然升高。没有日志的AI系统就像没有仪表盘的飞机你只能凭感觉飞迟早出事。4. 实操过程从零搭一个可用的AI推理服务4.1 项目结构设计别把所有代码堆在一个文件里从零开始做项目最容易犯的错误是把所有代码写在一个main.py里。刚开始可能只有几十行但随着功能增加很快会变成几百上千行维护成本急剧上升。我的建议是从一开始就按职责拆分模块。一个合理的项目结构大概是这样ai-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI入口 │ ├── model.py # 模型加载和推理 │ ├── schemas.py # 请求/响应数据结构 │ ├── preprocess.py # 预处理逻辑 │ └── config.py # 配置管理 ├── tests/ │ └── test_api.py ├── Dockerfile ├── requirements.txt └── README.md这个结构的好处是职责清晰。model.py只管模型相关的事preprocess.py只管数据预处理main.py只管API路由。改一个地方不会影响其他部分。我见过太多项目因为结构混乱改一个bug引入三个新bug。配置管理也值得单独说。不要把配置硬编码在代码里用环境变量或配置文件。config.py可以这样写import os from pydantic import BaseSettings class Settings(BaseSettings): model_path: str os.getenv(MODEL_PATH, model.pth) batch_size: int int(os.getenv(BATCH_SIZE, 8)) max_length: int int(os.getenv(MAX_LENGTH, 512)) settings Settings()这样在不同环境开发、测试、生产部署时只需要改环境变量不用改代码。这是十二要素应用的基本原则但在AI项目里经常被忽略。4.2 核心代码实现一个完整的推理服务下面是一个完整的推理服务实现我尽量把关键细节都标注出来。这个服务接收文本输入用预训练的模型做分类返回类别和置信度。# app/main.py from fastapi import FastAPI, HTTPException from contextlib import asynccontextmanager import torch import time import logging from app.schemas import InferenceRequest, InferenceResponse from app.model import load_model, predict from app.config import settings logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) model None tokenizer None asynccontextmanager async def lifespan(app: FastAPI): global model, tokenizer logger.info(Loading model...) model, tokenizer load_model(settings.model_path) logger.info(Model loaded successfully) yield logger.info(Shutting down...) app FastAPI(lifespanlifespan) app.post(/predict, response_modelInferenceResponse) async def predict_endpoint(request: InferenceRequest): start_time time.time() try: result predict(model, tokenizer, request.text, settings.max_length) latency (time.time() - start_time) * 1000 logger.info(fPrediction completed in {latency:.2f}ms) return InferenceResponse( labelresult[label], confidenceresult[confidence], latency_mslatency ) except Exception as e: logger.error(fPrediction failed: {str(e)}) raise HTTPException(status_code500, detailPrediction failed)这里有几个关键点值得展开。第一用lifespan而不是on_event因为FastAPI的新版本已经推荐用lifespan了on_event将来会被废弃。第二predict函数里要做torch.no_grad()否则会构建计算图浪费显存。第三异常处理要记录日志但返回给用户的错误信息要模糊化不要暴露内部细节。model.py的实现# app/model.py import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification def load_model(model_path): tokenizer AutoTokenizer.from_pretrained(model_path) model AutoModelForSequenceClassification.from_pretrained(model_path) model.eval() if torch.cuda.is_available(): model model.cuda() return model, tokenizer def predict(model, tokenizer, text, max_length): inputs tokenizer( text, return_tensorspt, truncationTrue, max_lengthmax_length, paddingTrue ) if torch.cuda.is_available(): inputs {k: v.cuda() for k, v in inputs.items()} with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) confidence, predicted torch.max(probs, dim-1) return { label: model.config.id2label[predicted.item()], confidence: confidence.item() }这段代码里truncationTrue和paddingTrue是必须的否则不同长度的输入会报错。max_length要根据模型的最大支持长度来设超了会截断短了会浪费计算。这些细节看起来琐碎但每一个都可能导致线上问题。4.3 测试与验证别等上线了才发现问题从零做项目测试经常被忽略。但我的经验是测试不是可选项是必选项。没有测试的AI服务就像没有刹车的车跑得越快越危险。最基本的测试包括接口能正常响应、输入格式错误时返回合理错误、模型输出符合预期。用pytest写起来很简单# tests/test_api.py from fastapi.testclient import TestClient from app.main import app client TestClient(app) def test_predict_success(): response client.post(/predict, json{text: This is a test}) assert response.status_code 200 data response.json() assert label in data assert confidence in data assert 0 data[confidence] 1 def test_predict_empty_text(): response client.post(/predict, json{text: }) assert response.status_code 422 def test_predict_missing_field(): response client.post(/predict, json{}) assert response.status_code 422这三个测试覆盖了正常路径和两个异常路径。别小看这几个测试它们能在你改代码时快速告诉你有没有破坏原有功能。我见过太多改了一个小地方结果整个服务挂了的情况如果有测试这种问题在提交前就能发现。除了单元测试性能测试也很重要。用locust或wrk压一下看看QPS和延迟。我一般会关注P99延迟而不是平均延迟因为平均延迟会掩盖长尾问题。如果P99延迟超过1秒用户体验就会明显下降。实操心得测试的时候一定要用和线上一致的模型和配置。我见过测试用小的蒸馏模型线上用大模型测试通过但线上直接OOM的情况。测试环境可以缩规模但关键参数要一致。5. 常见问题与排查技巧实录5.1 环境与依赖问题速查环境问题是AI工程里最高频的故障源。我整理了一个速查表覆盖了大部分常见情况问题现象可能原因排查方法解决方案ImportError: libcudart.soCUDA运行时缺失ldd检查依赖安装对应CUDA版本或换CPU推理CUDA out of memory显存不足nvidia-smi查看占用减小batch size或清理缓存包版本冲突依赖不兼容pip check锁版本用虚拟环境隔离推理结果不一致预处理不一致对比训练和推理代码统一预处理模块服务启动慢模型加载耗时打时间戳模型预热或异步加载这个表里的每一条我都在实际项目中遇到过。最坑的是推理结果不一致排查了很久才发现是训练时用了tokenizer.encode推理时用了tokenizer.__call__两者对特殊字符的处理不同。这种问题没有日志很难发现所以日志里要记录输入的原始文本和tokenize后的结果。5.2 性能问题的排查思路性能问题通常表现为延迟高或吞吐低。排查思路是先定位瓶颈再针对性优化。定位瓶颈的方法在代码的关键路径上打时间戳看时间花在哪里。是数据预处理慢还是模型推理慢还是后处理慢。我见过一个案例推理只花了10ms但预处理花了200ms因为每次都在做正则匹配。这种问题不看时间戳根本发现不了。如果瓶颈在模型推理优化方向有几个量化把FP32转成FP16或INT8、剪枝去掉不重要的权重、蒸馏用大模型教小模型、批处理攒批推理。量化是最容易见效的PyTorch的torch.quantization或者ONNX Runtime都能做。但要注意量化可能带来精度损失需要评估。如果瓶颈在数据预处理优化方向是缓存和并行。比如tokenizer的结果可以缓存相同的输入不用重复tokenize。预处理可以用多进程并行Python的GIL在IO密集型任务上不是问题。5.3 模型效果下降的排查模型上线后效果下降是最让人头疼的问题。原因可能有很多数据漂移、特征变化、上游系统改动。排查思路是从数据入手逐层往上查。第一步检查输入数据的分布有没有变化。比如原来输入都是短文本现在突然来了很多长文本模型的截断策略可能导致信息丢失。第二步检查预处理逻辑有没有改动。有时候上游系统改了一个字段格式预处理没跟上数据就错了。第三步检查模型本身有没有变化。如果模型更新了新模型的效果可能不如旧模型。我的经验是建立一个基线定期对比。比如每周跑一次固定的测试集看准确率有没有下降。如果下降了再逐层排查。没有基线你连效果下降了都发现不了。避坑技巧模型更新一定要有回滚机制。新模型上线后如果效果不好要能快速切回旧模型。最简单的做法是保留旧模型的权重文件用配置切换。别把旧模型删了那是自断后路。6. 工具选型与扩展从能跑到好用6.1 推理框架的选择逻辑当你的服务从能跑进入要跑得好的阶段推理框架的选择就变得重要了。PyTorch原生推理适合原型和小规模场景但生产环境可能需要更专业的方案。ONNX Runtime的优势是跨平台和优化好能把PyTorch模型导出成ONNX格式然后在各种硬件上高效运行。TensorRT是NVIDIA的推理优化框架在NVIDIA GPU上性能最好但绑定硬件。Triton是NVIDIA的推理服务框架支持动态批处理和模型集成适合大规模部署。我的建议是先用PyTorch原生推理跑通有性能瓶颈再考虑迁移。迁移是有成本的模型导出、精度验证、服务改造都需要时间。别为了先进而迁移要为需要而迁移。6.2 从单模型到多模型架构的演进当你的服务需要支持多个模型时架构就要考虑扩展性了。最简单的做法是每个模型一个服务但这样资源利用率低。更好的做法是用模型路由层根据请求参数分发到不同的模型。Triton在这方面做得很好它支持多模型加载和动态批处理。但如果你不想引入Triton也可以用FastAPI加模型注册表来实现class ModelRegistry: def __init__(self): self.models {} def register(self, name, model, tokenizer): self.models[name] {model: model, tokenizer: tokenizer} def get(self, name): if name not in self.models: raise ValueError(fModel {name} not found) return self.models[name]这个注册表很简单但能让你的服务支持多模型。每个模型在启动时注册请求时根据参数选择。这种设计在模型数量不多时够用数量多了再考虑更复杂的方案。6.3 持续迭代AI工程没有终点最后想说的是AI工程是一个持续迭代的过程。模型会更新数据会变化需求会演进。从零搭建的能力不是一次性的而是需要持续维护和升级的。我的做法是保持一个技术雷达定期关注新工具和新方法但不盲目跟风。每引入一个新东西都要问它解决了什么问题迁移成本多大收益是否值得这三个问题能过滤掉大部分看起来很美的方案。从零开始做AI工程最难的不是技术本身而是在信息过载中保持判断力。你知道的越多越容易陷入这个也要学那个也要用的焦虑。但工程的核心是解决问题不是堆砌工具。回到ai-engineering-from-scratch这个标题从零不是从零开始学所有东西而是从零开始解决一个具体问题在解决问题的过程中能力自然就长出来了。我在实际项目中的体会是先跑通再优化最后才考虑扩展。这个顺序不能乱。跑通让你有反馈优化让你有提升扩展让你有规模。跳过任何一步都会在后面付出代价。踩过几次坑之后我越来越相信AI工程的能力不是学出来的是做出来的。
