1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年“AI工程”这个词被说得太多了多到有点变味。招聘JD上写着“熟悉AI工程化落地”培训班广告里喊着“三个月转型AI工程师”打开技术社区满屏都是“大模型应用实战”。但真到了要自己动手做一个能跑起来、能上线、能维护的AI项目时很多人会突然发现——自己连数据怎么清洗、特征怎么存、模型怎么打包、推理服务怎么压测都说不清楚。ai-engineering-from-scratch这个标题我第一次看到的时候就觉得挺有意思。它说的不是“从零训练一个大模型”也不是“从零学深度学习理论”而是从零构建AI工程能力。这两者差别很大。前者是研究导向后者是工程导向。研究导向关心的是模型结构、损失函数、SOTA指标工程导向关心的是数据管道稳不稳、推理延迟能不能压到200ms以内、模型更新会不会把线上服务搞崩、GPU利用率能不能再提10个百分点。我做了十多年一线开发带过不少从算法岗转工程岗、或者从后端岗转AI方向的同事。最常见的误区就是把AI工程等同于调包。pip install transformers写几行推理代码跑通一个demo就觉得自己会AI工程了。但真到了生产环境问题全来了模型加载要40秒怎么办并发一上来显存就爆怎么办输入长度忽长忽短导致batch效率极低怎么办模型版本更新后A/B测试怎么做日志里全是乱码怎么排查所以这篇内容我想认真聊聊“从零搭建AI工程能力”这件事。它适合谁看适合那些已经会写Python、懂一点机器学习基础、但没真正做过完整AI项目落地的开发者也适合后端工程师想往AI方向靠、或者算法同学想补工程短板的读者。我会从整体设计思路、核心细节、实操过程、常见问题四个大块展开尽量把每个“为什么”讲清楚把每个“怎么做”写明白。你不需要有GPU集群一台带显卡的笔记本或者一台云主机就能跟着走。2. 整体设计与思路拆解AI工程到底在工程什么2.1 先搞清楚AI工程和研究的分界线很多人一上来就想搞模型这是典型的算法思维。AI工程的核心不是模型本身而是围绕模型构建一套可重复、可监控、可扩展的系统。我习惯把AI工程分成四层数据层、模型层、服务层、运维层。每一层都有独立的工程问题而且层与层之间的接口设计往往比单层实现更重要。数据层要解决的是数据从哪来、怎么存、怎么版本化、怎么保证训练和推理时特征一致。模型层要解决的是模型怎么选、怎么训练、怎么评估、怎么导出成通用格式。服务层要解决的是模型怎么加载、怎么批处理、怎么限流、怎么降级。运维层要解决的是怎么监控、怎么告警、怎么灰度、怎么回滚。为什么这么分因为实际项目里出问题最多的往往不是模型精度不够而是数据管道断了、特征对不上、服务超时了、监控没覆盖到。我见过一个推荐系统项目离线AUC 0.85上线后CTR跌了30%最后查出来是训练时用的特征版本和线上服务用的特征版本差了两天。这种问题模型再强也救不了。2.2 技术选型的核心逻辑别追新追稳从零搭建AI工程能力技术选型是第一个坎。我的原则很简单优先选社区活跃、文档齐全、有生产案例的方案而不是最新最炫的。比如推理框架TensorFlow Serving、TorchServe、Triton Inference Server、ONNX Runtime这几个都成熟。选哪个看你的模型格式和团队技术栈。如果你用PyTorch训练那TorchServe或者Triton都行。Triton的优势是支持多框架、动态批处理做得好、GPU利用率高但学习曲线陡一点。TorchServe更贴近PyTorch生态上手快但批处理和并发性能要自己调。ONNX Runtime适合模型已经转成ONNX格式、追求轻量部署的场景。我一般建议新手从FastAPI ONNX Runtime或者FastAPI PyTorch开始。为什么因为FastAPI写服务简单ONNX Runtime部署轻量PyTorch直接加载方便调试。等业务量上来了再考虑上Triton或者TensorFlow Serving。别一上来就搞Kubernetes Istio Triton全套那是给自己找麻烦。数据存储也是同理。小规模用SQLite或者PostgreSQL存元数据特征用Parquet或者Feather文件存够用了。别一上来就上Feature Store那东西维护成本高小团队根本扛不住。等特征数量超过几百个、团队超过五个人再考虑Feast或者Tecton这类方案。2.3 项目目录结构一开始就要定好规矩从零做项目最容易犯的错是目录结构乱。今天写个train.py明天写个inference.py后天加个utils.py最后整个项目根目录几十个文件谁都不敢动。我建议一开始就按功能分层ai-engineering-from-scratch/ ├── configs/ # 配置文件 │ ├── train.yaml │ └── serve.yaml ├── data/ # 数据相关 │ ├── raw/ │ ├── processed/ │ └── features/ ├── src/ # 源代码 │ ├── data/ # 数据加载与预处理 │ ├── model/ # 模型定义与训练 │ ├── serve/ # 推理服务 │ └── utils/ # 通用工具 ├── tests/ # 测试 ├── scripts/ # 运维脚本 ├── notebooks/ # 实验记录 └── requirements.txt这个结构的好处是数据、代码、配置、脚本分离谁该放哪一目了然。configs用YAML管理超参和路径避免硬编码。notebooks只用来做探索性分析不参与生产。scripts放部署、压测、数据同步的脚本。tests至少覆盖数据加载、模型前向、服务接口三个关键路径。注意别把模型权重文件直接提交到Git。用DVC或者Git LFS管理或者干脆存在对象存储里代码里只留路径和版本号。2.4 环境隔离别让依赖打架AI项目的依赖冲突是出了名的。PyTorch、CUDA、cuDNN、Python版本这四个东西的兼容矩阵能让人崩溃。我的做法是用Conda管理Python环境用pip-tools锁定依赖版本用Docker做最终交付。Conda的好处是能装CUDA和cuDNN的预编译版本省去手动配置。requirements.in里写顶层依赖pip-compile生成requirements.txt锁定所有传递依赖。Docker镜像里把Conda环境复制进去保证开发、测试、生产环境一致。# 创建环境 conda create -n ai-eng python3.10 conda activate ai-eng # 安装PyTorch根据CUDA版本选 pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118 # 锁定依赖 pip install pip-tools pip-compile requirements.in -o requirements.txt pip-sync requirements.txt为什么这么麻烦因为AI项目依赖多、版本敏感今天能跑的代码明天换个环境就可能报CUDA error: no kernel image is available。锁定版本、容器化交付能省掉大量“在我机器上是好的”这类扯皮。3. 核心细节解析与实操要点数据、模型、服务三件套3.1 数据管道AI工程的隐形地基数据管道是AI工程里最不起眼但最重要的部分。我见过太多项目模型代码写得漂漂亮亮数据加载却是一团乱麻路径硬编码、格式不统一、缺失值处理逻辑散落在各处。结果就是每次换数据集都要改代码每次上线都要手动传文件。从零搭建我建议先定义一个数据契约。简单说就是明确输入数据的schema哪些列、什么类型、允许哪些缺失、取值范围是多少。用Pydantic或者Great Expectations都能做。比如from pydantic import BaseModel, Field from typing import Optional class InferenceRequest(BaseModel): text: str Field(..., min_length1, max_length512) user_id: Optional[int] None timestamp: Optional[float] None这个契约既是服务接口的校验规则也是数据管道的清洗标准。训练时用同样的schema校验数据推理时用同样的schema校验请求保证线上线下一致。数据版本化也很关键。我习惯用时间戳哈希的方式命名数据集版本比如20240115_a3f8c2。每次训练前记录用了哪个版本推理服务记录加载了哪个版本的模型和特征。出问题时能快速定位是数据变了还是模型变了。实操心得数据清洗逻辑一定要写成纯函数放在src/data/下训练和推理共用。别在notebook里写一堆df.dropna()然后复制到服务代码里那是给自己埋雷。3.2 模型训练别只盯着精度模型训练阶段新手最容易犯的错是只看验证集精度。但AI工程要考虑的是训练可复现、模型可导出、推理可加速。可复现性方面随机种子要固定数据划分要固定超参要记录。我用MLflow或者Weights Biases做实验跟踪每次训练记录git commit、数据版本、超参、指标。这样过两周回头看能清楚知道哪个模型是怎么来的。模型导出方面PyTorch用torch.onnx.export转ONNX或者用torch.jit.trace转TorchScript。ONNX的好处是跨框架、跨平台推理时可以用ONNX Runtime也可以用TensorRT加速。TorchScript的好处是保留PyTorch的动态图特性调试方便。import torch import torch.onnx # 假设model是训练好的PyTorch模型 dummy_input torch.randn(1, 3, 224, 224) torch.onnx.export( model, dummy_input, model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}}, opset_version13 )dynamic_axes这个参数很关键。不设置的话导出的ONNX模型只支持固定batch size线上并发一上来就废了。设置之后batch维度动态可变推理服务才能做动态批处理。推理加速方面量化是最直接的手段。PyTorch支持动态量化和静态量化ONNX Runtime也支持量化。动态量化对LSTM和Transformer类模型效果不错静态量化对CNN类模型更合适。量化后模型体积能缩小到1/4推理速度提升2-3倍精度损失通常控制在1%以内。3.3 推理服务从demo到生产的距离推理服务是AI工程的脸面。demo阶段一个Flask接口加model.predict()就完事了。生产阶段要考虑的东西多得多并发、批处理、超时、限流、降级、监控。我推荐用FastAPI Uvicorn做服务框架原因是异步支持好、自动生成OpenAPI文档、Pydantic校验开箱即用。核心接口设计如下from fastapi import FastAPI, HTTPException from pydantic import BaseModel import onnxruntime as ort import numpy as np app FastAPI() session ort.InferenceSession(model.onnx) class PredictRequest(BaseModel): inputs: list[float] class PredictResponse(BaseModel): outputs: list[float] model_version: str app.post(/predict, response_modelPredictResponse) async def predict(req: PredictRequest): try: input_array np.array([req.inputs], dtypenp.float32) outputs session.run(None, {input: input_array}) return PredictResponse( outputsoutputs[0].flatten().tolist(), model_version20240115_a3f8c2 ) except Exception as e: raise HTTPException(status_code500, detailstr(e))这个代码看起来简单但有几个工程细节model_version要返回方便排查问题异常要捕获别让服务直接崩输入要校验防止恶意请求打爆显存。动态批处理是提升GPU利用率的关键。思路是请求先进入队列服务端攒够一定数量或者等待一定时间后合并成一个batch推理。Triton自带这个功能自己实现的话可以用asyncio.Queue加后台任务。import asyncio from collections import deque class BatchProcessor: def __init__(self, max_batch_size32, max_wait_ms10): self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue deque() self.lock asyncio.Lock() async def add_request(self, input_data): async with self.lock: self.queue.append(input_data) if len(self.queue) self.max_batch_size: return await self._process_batch() await asyncio.sleep(self.max_wait_ms / 1000) async with self.lock: if self.queue: return await self._process_batch()这个实现简化了但核心逻辑在攒批、等待、合并推理。实际生产里还要考虑超时、优先级、错误隔离。3.4 监控与日志上线只是开始服务上线不是终点而是起点。没有监控的AI服务就像没有仪表盘的飞机。我至少要监控四类指标请求指标、性能指标、模型指标、资源指标。请求指标包括QPS、成功率、错误码分布。性能指标包括P50/P95/P99延迟、批处理大小分布。模型指标包括预测分布、置信度分布、特征漂移。资源指标包括GPU利用率、显存占用、CPU、内存。用Prometheus Grafana做监控面板用ELK或者Loki做日志收集。日志里要记录请求ID、模型版本、输入摘要、输出摘要、耗时。别记录完整输入输出一是隐私问题二是日志量太大。import time import logging from prometheus_client import Histogram, Counter REQUEST_LATENCY Histogram(request_latency_seconds, Request latency) REQUEST_COUNT Counter(request_count, Total requests, [status]) app.post(/predict) async def predict(req: PredictRequest): start time.time() try: result await process(req) REQUEST_COUNT.labels(statussuccess).inc() return result except Exception as e: REQUEST_COUNT.labels(statuserror).inc() raise finally: REQUEST_LATENCY.observe(time.time() - start)注意监控指标别只放在内存里要暴露/metrics接口给Prometheus抓取。Grafana面板至少要有QPS、延迟、错误率、GPU利用率四个图。4. 实操过程与核心环节实现从零跑通一个完整项目4.1 环境准备与依赖安装假设你有一台Ubuntu 22.04的机器带一张NVIDIA显卡。第一步是装驱动和CUDA。我习惯用Conda装CUDA Toolkit省去系统级配置。# 安装Miniconda wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh bash Miniconda3-latest-Linux-x86_64.sh # 创建环境 conda create -n ai-eng python3.10 -y conda activate ai-eng # 安装PyTorchCUDA 11.8版本 pip install torch2.1.0 torchvision0.16.0 --index-url https://download.pytorch.org/whl/cu118 # 安装其他依赖 pip install fastapi uvicorn onnx onnxruntime-gpu pydantic prometheus-client验证GPU是否可用import torch print(torch.cuda.is_available()) # 应该输出True print(torch.cuda.get_device_name(0)) # 输出显卡型号如果输出False检查驱动版本和CUDA版本是否匹配。nvidia-smi看驱动支持的CUDA版本nvcc --version看当前CUDA版本两者要兼容。4.2 数据准备与特征工程我用一个文本分类任务做示例。数据是CSV格式两列text和label。第一步是划分训练集、验证集、测试集比例7:1.5:1.5。import pandas as pd from sklearn.model_selection import train_test_split df pd.read_csv(data/raw/dataset.csv) train_df, temp_df train_test_split(df, test_size0.3, random_state42, stratifydf[label]) val_df, test_df train_test_split(temp_df, test_size0.5, random_state42, stratifytemp_df[label]) train_df.to_parquet(data/processed/train.parquet) val_df.to_parquet(data/processed/val.parquet) test_df.to_parquet(data/processed/test.parquet)用Parquet而不是CSV原因是Parquet列式存储、压缩率高、读取快而且保留schema。特征工程方面文本分类用tokenizer做分词和ID化。这里用HuggingFace的tokenizer但只做tokenize不涉及模型下载。from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(bert-base-chinese) def tokenize(text): return tokenizer(text, paddingmax_length, truncationTrue, max_length128, return_tensorspt) train_encodings [tokenize(t) for t in train_df[text].tolist()]实际项目里tokenize结果要缓存下来别每次训练都重新算。用datasets库的map函数加cache_file_name参数或者自己写个缓存逻辑。4.3 模型训练与导出模型定义用PyTorch结构简单点一个Embedding加一个线性层做分类。import torch.nn as nn class TextClassifier(nn.Module): def __init__(self, vocab_size, embed_dim, num_classes): super().__init__() self.embedding nn.Embedding(vocab_size, embed_dim) self.fc nn.Linear(embed_dim, num_classes) def forward(self, input_ids): x self.embedding(input_ids).mean(dim1) return self.fc(x)训练循环里记录loss和accuracy每个epoch结束在验证集上评估。训练完保存PyTorch权重然后导出ONNX。# 训练 model TextClassifier(vocab_size21128, embed_dim128, num_classes2) optimizer torch.optim.Adam(model.parameters(), lr1e-3) criterion nn.CrossEntropyLoss() for epoch in range(5): model.train() for batch in train_loader: optimizer.zero_grad() outputs model(batch[input_ids]) loss criterion(outputs, batch[label]) loss.backward() optimizer.step() # 导出ONNX model.eval() dummy_input torch.randint(0, 21128, (1, 128)) torch.onnx.export( model, dummy_input, model.onnx, input_names[input_ids], output_names[logits], dynamic_axes{input_ids: {0: batch_size}, logits: {0: batch_size}}, opset_version13 )导出后验证ONNX模型import onnxruntime as ort import numpy as np session ort.InferenceSession(model.onnx) test_input np.random.randint(0, 21128, (2, 128)).astype(np.int64) outputs session.run(None, {input_ids: test_input}) print(outputs[0].shape) # 应该是(2, 2)4.4 推理服务部署与压测服务代码用FastAPI写加载ONNX模型提供/predict接口。启动命令uvicorn src.serve.app:app --host 0.0.0.0 --port 8000 --workers 2workers数量根据CPU核数定一般是核数的一半。GPU推理时多个worker会竞争显存建议用1个worker加异步批处理。压测用locust或者wrk。我习惯用locust因为能模拟复杂请求。from locust import HttpUser, task, between class PredictUser(HttpUser): wait_time between(0.01, 0.05) task def predict(self): self.client.post(/predict, json{inputs: [1.0] * 128})压测时观察QPS、P99延迟、GPU利用率。如果GPU利用率低于50%说明批处理没做好或者请求量不够。如果P99延迟超过500ms检查是不是模型太大或者批处理等待时间太长。实操心得压测前先把日志级别调到WARNING不然日志IO会成为瓶颈。压测时用nvidia-smi -l 1实时看GPU利用率用htop看CPU和内存。5. 常见问题与排查技巧实录5.1 模型加载慢怎么办模型加载慢是常见问题。ONNX模型加载通常比PyTorch快但如果模型很大比如超过1GB加载时间可能超过10秒。解决方案有三个一是用内存映射加载二是服务启动时预加载三是用模型缓存。ONNX Runtime支持session_options.enable_mem_pattern False和session_options.enable_cpu_mem_arena False能减少内存占用但可能影响速度。更好的做法是服务启动时加载模型别在第一个请求时加载。app FastAPI() session None app.on_event(startup) async def load_model(): global session session ort.InferenceSession(model.onnx, providers[CUDAExecutionProvider])5.2 显存溢出怎么排查显存溢出OOM是GPU推理最常见的问题。原因通常是batch size太大、模型太大、或者显存碎片。排查步骤先用nvidia-smi看显存占用再用torch.cuda.memory_summary()看PyTorch显存分配。如果是ONNX Runtime用ort.InferenceSession的providers参数指定CUDAExecutionProvider并设置gpu_mem_limit。options ort.SessionOptions() options.graph_optimization_level ort.GraphOptimizationLevel.ORT_ENABLE_ALL session ort.InferenceSession( model.onnx, options, providers[(CUDAExecutionProvider, {gpu_mem_limit: 2 * 1024 * 1024 * 1024})] )限制显存后如果还是OOM就要减小batch size或者换更小的模型。动态批处理时设置最大batch size上限别让队列无限攒批。5.3 推理结果不一致怎么定位训练时精度正常推理时结果不对这种问题最让人头疼。常见原因有四个一是预处理不一致二是模型版本不对三是数值精度差异四是后处理逻辑不同。排查方法固定一个测试样本分别在训练代码和推理代码里跑一遍对比中间结果。用np.testing.assert_allclose检查数值差异。如果差异在1e-5以内是正常的浮点误差如果差异很大检查预处理和后处理。注意PyTorch默认用float32ONNX Runtime可能用float16或者int8。导出ONNX时指定opset_version13以上并在推理时显式设置输入类型为np.float32。5.4 服务并发上不去怎么优化并发上不去通常是三个瓶颈CPU预处理、GPU推理、网络IO。用py-spy做性能分析看时间花在哪。pip install py-spy py-spy top --pid uvicorn_pid如果CPU预处理占大头把tokenize逻辑用C或者Rust重写或者用tokenizers库的并行模式。如果GPU推理占大头开启动态批处理提高GPU利用率。如果网络IO占大头用HTTP/2或者gRPC替代HTTP/1.1。问题现象可能原因排查工具解决方案模型加载超过10秒模型太大、磁盘IO慢iostat、strace预加载、内存映射、模型量化显存OOMbatch太大、显存碎片nvidia-smi、memory_summary限制显存、减小batch、重启服务推理结果不一致预处理差异、版本不对逐层对比、assert_allclose统一预处理、锁定版本、显式类型并发低CPU瓶颈、GPU利用率低py-spy、nvidia-smi动态批处理、异步IO、模型加速P99延迟高批处理等待、GC停顿locust、gc日志调整等待时间、禁用GC、预热5.5 模型更新怎么做灰度模型更新不能直接全量替换要灰度。我的做法是新模型先加载到服务里但流量只切5%过去。观察24小时如果错误率和延迟没有明显变化再逐步加到50%、100%。如果出问题立即切回旧模型。实现上用两个InferenceSession根据请求头或者用户ID做路由。sessions { v1: ort.InferenceSession(model_v1.onnx), v2: ort.InferenceSession(model_v2.onnx) } app.post(/predict) async def predict(req: PredictRequest, model_version: str v1): session sessions.get(model_version, sessions[v1]) # ...灰度期间监控要对比两个版本的指标。如果新版本错误率超过旧版本2倍自动回滚。这个逻辑可以用Prometheus告警加脚本实现。5.6 日志太多怎么治理AI服务日志容易爆炸尤其是记录输入输出的时候。我的原则是生产环境只记录元数据不记录原始数据。请求ID、模型版本、耗时、状态码、输入长度、输出类别这些够了。原始输入输出只在调试时开而且要有采样率。import random if random.random() 0.01: # 1%采样 logger.info(finput_sample{req.inputs[:10]})日志用JSON格式方便ELK解析。别用print用logging。日志级别生产环境用INFO调试用DEBUG。uvicorn的access log可以关掉用中间件自己记录。6. 从零到一之后持续迭代的工程习惯跑通一个完整项目只是开始。AI工程能力的真正体现在于持续迭代。我自己的习惯是每周做一次模型性能回顾每月做一次数据质量检查每季度做一次架构评审。模型性能回顾看三个东西线上指标有没有下降、推理延迟有没有上升、GPU利用率有没有变化。数据质量检查看特征分布有没有漂移、缺失率有没有异常、标签分布有没有变化。架构评审看服务瓶颈在哪、有没有新的技术方案可以替换、成本能不能优化。还有一个容易被忽视的点文档。AI项目迭代快人员流动也快。没有文档新人接手要花几周才能搞明白。我要求每个模块必须有README每个接口必须有示例每个配置必须有注释。文档不用多漂亮但要能让人照着跑起来。最后分享一个小技巧把常用操作脚本化。训练、导出、部署、压测、回滚每个操作写成一个脚本放在scripts/下。别每次都手动敲命令容易出错还浪费时间。脚本里加参数校验和日志输出跑完自动发通知。#!/bin/bash # scripts/deploy.sh set -e MODEL_PATH$1 VERSION$2 if [ -z $MODEL_PATH ] || [ -z $VERSION ]; then echo Usage: $0 model_path version exit 1 fi echo Deploying model $VERSION from $MODEL_PATH cp $MODEL_PATH /opt/models/model_$VERSION.onnx curl -X POST http://localhost:8000/reload -d {\version\: \$VERSION\} echo Deploy complete这种脚本看起来简单但能省掉大量重复劳动。我见过太多团队部署靠手动scp回滚靠记忆出问题了手忙脚乱。工程能力就体现在这些细节里。