1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台调一个现成的大模型接口写几行胶水代码然后对外宣称自己做了个AI应用。我承认这条路确实能在半天内跑通一个Demo但如果你真的想在这个领域站稳脚跟靠这种“调包式开发”是走不远的。ai-engineering-from-scratch这个标题背后的核心诉求其实不是让你去重复造轮子而是让你具备一种“拆解轮子”的能力——知道一个AI系统从数据进来、到特征处理、到模型推理、再到结果输出中间到底发生了什么。我见过太多简历上写着“精通AI应用开发”的人结果连一个简单的向量相似度计算都要去查文档连模型输出的logits怎么转成概率都说不清楚。这不是他们的错因为现在的工具链太方便了方便到把所有的中间过程都封装成了黑盒。但问题在于一旦线上出了bad case一旦推理延迟飙升一旦显存溢出你连从哪里下手排查都不知道。所以这篇内容的目标很明确给那些想真正理解AI工程底层逻辑的人提供一条从零开始、手搓核心组件的学习路径。它适合有一定Python基础、但没深入过AI系统内部实现的开发者也适合那些调了很久API、想回头补课“为什么”的工程师。我不会教你从零训练一个千亿参数的大模型那不现实也没必要。我要讲的是AI工程中最核心、最常被忽视、但又最值得亲手实现的那几个模块数据管道、特征工程、模型推理引擎、以及服务化部署。每一个模块我都会给出可运行的代码骨架解释为什么这么设计以及在实际操作中我会怎么取舍。你不需要GPU集群一台普通的开发机甚至一个Colab环境就能跟着走完整个流程。2. 数据管道AI工程里最脏最累但最不能省的一步2.1 为什么数据管道的质量直接决定模型上限在任何AI系统里数据管道都是那个“做好了没人夸做砸了全盘皆输”的角色。我见过一个团队花了三个月调模型结构最后发现是训练数据里的标签噪声太大导致模型一直在拟合错误的目标。也见过一个线上服务因为数据预处理阶段的一个空值处理逻辑写错了导致推理结果整体偏移。这些问题的根源都不在模型本身而在数据管道。从零构建数据管道第一步不是写代码而是定义数据的“契约”。什么叫数据契约就是明确每一条数据在进入管道时应该长什么样字段名、类型、取值范围、是否允许为空、默认值是什么。这个契约一旦定下来后续所有的清洗、转换、增强操作都必须围绕它来执行。我习惯用一个简单的YAML文件来管理这个契约然后在管道入口处做一次强校验。这样做的好处是当上游数据源发生变更时你能在第一时间发现而不是等到模型效果下降才去回溯。# data_contract.yaml fields: user_id: type: int required: true item_text: type: str required: true max_length: 512 label: type: int required: false allowed_values: [0, 1]有了契约之后管道的设计就变成了一个“流水线”问题。我通常会把管道拆成三个独立的阶段提取Extract、转换Transform、加载Load。提取阶段负责从各种数据源数据库、日志文件、消息队列把原始数据拉过来这个阶段的核心是“幂等性”——同样的输入必须产生同样的输出不能因为重跑一次就多出一批重复数据。转换阶段是真正做脏活的地方包括去重、填充缺失值、类型转换、文本归一化等等。加载阶段则是把处理好的数据写入下游的存储或直接喂给训练/推理流程。2.2 手写一个可复用的数据清洗器很多人喜欢用pandas的dropna()和fillna()一把梭这在探索性分析阶段没问题但在生产级的AI工程里是远远不够的。你需要一个可配置、可扩展、可追踪的清洗器。我下面给出一个简化版的实现核心思路是把每个清洗步骤定义成一个独立的函数然后通过配置文件来组合这些步骤。import re from typing import Any, Dict, List, Callable class DataCleaner: def __init__(self, steps: List[Dict[str, Any]]): self.steps steps self.registry: Dict[str, Callable] { strip_whitespace: self._strip_whitespace, remove_html: self._remove_html, truncate: self._truncate, fill_default: self._fill_default, } def _strip_whitespace(self, value: str, **kwargs) - str: return value.strip() def _remove_html(self, value: str, **kwargs) - str: return re.sub(r[^], , value) def _truncate(self, value: str, max_length: int 512, **kwargs) - str: return value[:max_length] def _fill_default(self, value: Any, default: Any , **kwargs) - Any: return value if value is not None else default def clean(self, record: Dict[str, Any]) - Dict[str, Any]: for step in self.steps: field step[field] op step[op] params step.get(params, {}) if field in record: record[field] self.registry[op](record[field], **params) return record这个清洗器的设计要点在于每个操作都是纯函数不依赖外部状态操作通过注册表管理方便扩展清洗步骤通过配置驱动方便调整。在实际使用中我会把清洗前后的数据都做一次采样落盘方便出问题时对比排查。这个习惯帮我省过很多次命——有一次线上模型突然开始输出乱码最后查出来是上游某个字段的编码从UTF-8变成了GBK而清洗器里的strip_whitespace恰好把这个异常掩盖了导致问题拖了三天才定位到。注意数据清洗中最危险的操作不是“做错了”而是“悄悄做错了”。任何清洗步骤都应该有日志记录至少记录下被修改的字段和修改前后的值。不要为了性能省掉这一步。2.3 数据版本管理与可复现性AI工程和传统软件工程最大的区别之一就是AI系统对数据的依赖极强。代码版本管理有Git但数据版本管理往往被忽视。我强烈建议在项目初期就引入数据版本的概念。最简单的做法是每次数据管道跑完计算一个数据指纹比如所有记录的哈希值的汇总然后把这个指纹和管道配置一起存下来。这样当模型效果出现波动时你可以快速判断是数据变了还是代码变了。更进一步的做法是使用DVCData Version Control这样的工具把大数据文件用指针的方式管理起来。但即使不用工具手动维护一个data_manifest.json也能解决80%的问题。这个文件里记录每次数据生成的元信息时间戳、数据量、字段统计、管道配置哈希。我自己的项目里这个文件是自动生成的每次数据更新都会追加一条记录。看起来很简单但当你需要回溯“上周三的模型到底是用哪批数据训练的”时这个文件就是救命稻草。3. 特征工程从原始文本到模型能吃的数字3.1 文本特征提取的几种手搓方案特征工程是AI工程里最考验功力的环节。同样的数据不同的人做出来的特征模型效果可能差出十几个百分点。对于文本数据来说最基础的特征提取方式就是词袋模型Bag of Words和TF-IDF。虽然现在大家都用预训练模型的embedding但理解这些基础方法对于排查问题非常有帮助。比如当你的embedding模型表现异常时用TF-IDF跑一个baseline往往能快速判断是数据问题还是模型问题。手搓一个TF-IDF计算器其实不难核心就是统计词频和逆文档频率。下面是一个简化版的实现重点在于理解计算过程而不是追求性能。import math from collections import Counter from typing import List, Dict class TFIDFVectorizer: def __init__(self): self.vocab: Dict[str, int] {} self.idf: Dict[str, float] {} def fit(self, corpus: List[str]): doc_count len(corpus) df Counter() for doc in corpus: tokens set(doc.lower().split()) for token in tokens: df[token] 1 for token, count in df.items(): self.idf[token] math.log((doc_count 1) / (count 1)) 1 self.vocab {token: idx for idx, token in enumerate(sorted(self.idf.keys()))} def transform(self, doc: str) - List[float]: tokens doc.lower().split() tf Counter(tokens) total len(tokens) if tokens else 1 vector [0.0] * len(self.vocab) for token, count in tf.items(): if token in self.vocab: vector[self.vocab[token]] (count / total) * self.idf[token] return vector这个实现里有一个细节值得注意IDF的计算我用了平滑处理分子分母都加1这是为了避免某个词在所有文档中都没出现时导致除零错误。在实际工程中这种边界条件的处理往往比算法本身更重要。我见过一个线上服务因为IDF计算时没有做平滑导致某个新词出现时整个向量变成NaN进而让模型输出全乱。3.2 特征归一化与分桶的实操细节数值型特征的处理同样有很多坑。最常见的操作是归一化Normalization和分桶Binning。归一化把不同量纲的特征拉到同一个尺度分桶则把连续值离散化有时候能带来更稳定的模型表现。但这两个操作都有各自的适用场景和注意事项。归一化最常用的方法是Min-Max缩放和Z-Score标准化。Min-Max把值映射到[0,1]区间适合分布比较均匀的特征Z-Score把值转换成均值为0、方差为1的分布适合有异常值的特征。我个人的经验是如果特征的最大最小值在训练集和测试集之间差异很大优先用Z-Score如果特征有明确的物理边界比如年龄、评分优先用Min-Max。分桶的坑更多。等宽分桶和等频分桶是两种基本策略但实际使用中往往需要根据业务含义来定制分桶边界。比如用户年龄等宽分桶可能把0-10岁和10-20岁放在一个桶里这显然不合理。我通常的做法是先画一下特征的分布直方图然后根据业务经验手动调整分桶边界。这个过程没有标准答案但有一个原则每个桶里的样本量不要相差太悬殊否则模型会偏向样本多的桶。提示特征归一化的参数均值、方差、最大最小值必须从训练集计算然后应用到验证集和测试集。千万不要在全部数据上计算归一化参数那会造成数据泄露让模型在离线评估时表现虚高。3.3 特征存储与在线离线一致性当AI系统从离线实验走向在线服务时特征工程会面临一个经典难题离线训练用的特征和在线推理用的特征不一致。这种不一致可能来自计算逻辑的差异、数据源的延迟、或者时间窗口的错位。解决这个问题的标准做法是引入特征存储Feature Store但即使不引入完整的特征存储系统也可以通过一些工程手段来保证一致性。我的做法是把特征计算逻辑封装成独立的模块离线和在线共用同一份代码。离线时这个模块从历史数据中批量计算特征在线时这个模块从实时数据流中计算特征。关键在于模块的输入输出接口必须严格定义不能因为运行环境不同就改变行为。另外对于依赖时间窗口的特征比如“过去7天的点击次数”离线计算和在线计算的时间边界必须对齐。我通常会在代码里显式地传入一个reference_time参数离线时用数据的时间戳在线时用当前时间确保两边的时间语义一致。4. 模型推理引擎不调API自己跑一次前向传播4.1 从PyTorch模型到ONNX的导出与验证训练好的模型要上线第一步是把它从训练框架里“拿出来”。PyTorch的模型默认是动态图直接用于推理会有性能开销而且依赖训练时的环境。我通常会把模型导出成ONNX格式这样既能跨框架运行又能利用ONNX Runtime的图优化能力。导出过程本身不难但验证导出后的模型和原模型行为一致才是真正花时间的地方。import torch import torch.onnx import onnxruntime as ort import numpy as np def export_and_verify(model, dummy_input, onnx_path): model.eval() torch.onnx.export( model, dummy_input, onnx_path, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}}, opset_version13, ) # 验证 with torch.no_grad(): torch_out model(dummy_input).numpy() sess ort.InferenceSession(onnx_path) onnx_out sess.run(None, {input: dummy_input.numpy()})[0] diff np.abs(torch_out - onnx_out).max() print(fMax difference: {diff}) assert diff 1e-4, ONNX model output differs from PyTorch model这段代码里dynamic_axes的设置很关键。如果不设置导出的ONNX模型会固定batch size线上服务就没法处理变长请求了。opset_version的选择也要注意不同版本的ONNX Runtime支持的opset不同选太高可能导致兼容性问题。我一般用13比较稳。验证环节的diff阈值我设的是1e-4对于大多数模型来说这个精度足够了。如果diff超过这个值通常是因为模型里有某些操作在ONNX里的实现和PyTorch有细微差异比如某些激活函数或者归一化层。4.2 手写一个简单的推理服务模型导出之后下一步是把它包装成一个服务。很多人直接用Flask写一个接口把模型加载到内存里然后每次请求调一次model.predict()。这种做法在低并发场景下没问题但在生产环境里会遇到性能瓶颈。我下面给出一个基于FastAPI的推理服务骨架重点在于批处理Batching和异步处理的设计。import asyncio from fastapi import FastAPI from pydantic import BaseModel import numpy as np import onnxruntime as ort app FastAPI() sess ort.InferenceSession(model.onnx) class Request(BaseModel): text: str class BatchProcessor: def __init__(self, max_batch_size32, max_wait_time0.01): self.max_batch_size max_batch_size self.max_wait_time max_wait_time self.queue asyncio.Queue() self.task None async def add_request(self, request): future asyncio.Future() await self.queue.put((request, future)) if self.task is None or self.task.done(): self.task asyncio.create_task(self._process()) return await future async def _process(self): batch [] futures [] try: while len(batch) self.max_batch_size: request, future await asyncio.wait_for( self.queue.get(), timeoutself.max_wait_time ) batch.append(request) futures.append(future) except asyncio.TimeoutError: pass if batch: inputs np.array([r.text for r in batch]) outputs sess.run(None, {input: inputs})[0] for future, output in zip(futures, outputs): future.set_result(output.tolist()) processor BatchProcessor() app.post(/predict) async def predict(request: Request): result await processor.add_request(request) return {result: result}这个服务的核心思想是“攒批”把短时间内到达的多个请求合并成一个批次一次性送给模型推理。这样做的好处是充分利用GPU的并行能力显著提高吞吐量。max_wait_time的设置需要根据实际延迟要求来调整设得太大会增加延迟设得太小则攒不到足够的请求。我一般从10毫秒开始调根据线上监控的P99延迟来微调。4.3 推理性能的监控与调优推理服务上线之后监控是必不可少的。我关注的指标主要有四个QPS每秒查询数、P99延迟、GPU利用率和显存占用。这四个指标之间往往需要权衡。比如提高批处理大小能提升QPS和GPU利用率但会增加P99延迟。降低批处理大小则相反。调优的过程就是找到业务能接受的最大延迟下的最优批处理配置。除了批处理还有一些其他的优化手段。比如对于文本模型可以在预处理阶段做动态padding避免所有样本都padding到最大长度。再比如如果模型有多个输出头但线上只需要其中一个可以在导出ONNX时就把不需要的输出裁掉减少计算量。这些优化单独看可能只提升几个百分点但叠加起来效果就很可观了。我在一个项目里通过动态padding和输出裁剪把推理延迟从120毫秒降到了75毫秒用户体验提升非常明显。注意推理服务的性能调优一定要在真实流量下进行。离线压测工具如wrk、locust虽然能模拟并发但请求的分布和真实场景往往差异很大。我习惯在服务上线初期保留一个“影子模式”把真实请求复制一份到新版本服务上对比两者的延迟和输出确认无误后再切换流量。5. 服务化部署让模型真正跑在线上5.1 容器化与依赖管理模型服务要部署到线上容器化是标准做法。但AI项目的容器化有一个特殊难点依赖太重。PyTorch、CUDA、各种科学计算库加起来动辄几个GB。如果直接用官方的基础镜像构建出来的镜像可能超过10GB拉取和启动都很慢。我的做法是分阶段构建第一阶段用完整的基础镜像安装依赖并导出ONNX模型第二阶段用精简的运行时镜像只安装ONNX Runtime和必要的服务框架。这样最终镜像可以控制在1GB以内。# 第一阶段构建和导出 FROM pytorch/pytorch:2.0.1-cuda11.7-cudnn8-runtime AS builder WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . RUN python export_onnx.py # 第二阶段运行时 FROM python:3.10-slim WORKDIR /app COPY --frombuilder /app/model.onnx . COPY --frombuilder /app/requirements-runtime.txt . RUN pip install -r requirements-runtime.txt COPY app.py . CMD [uvicorn, app:app, --host, 0.0.0.0, --port, 8000]这个Dockerfile的关键在于requirements-runtime.txt只包含运行时必需的包比如onnxruntime、fastapi、uvicorn而不包含torch、transformers这些训练时才用到的库。这样镜像体积能大幅缩小启动速度也更快。5.2 灰度发布与回滚策略模型上线最怕的是什么是新模型效果不如旧模型但又没法快速回滚。所以灰度发布是必须的。我的做法是新模型上线时先切5%的流量过去观察一段时间至少一个业务周期确认核心指标准确率、延迟、错误率没有下降再逐步扩大流量比例。如果发现异常立即把流量切回旧模型。实现灰度发布最简单的方式是在服务入口处加一个路由层根据请求的某些特征比如用户ID的哈希值来决定走新模型还是旧模型。这个路由逻辑可以放在网关层也可以放在服务内部。我倾向于放在服务内部因为这样更容易做A/B测试对比。下面是一个简单的路由实现import hashlib class ModelRouter: def __init__(self, old_model, new_model, new_model_ratio0.05): self.old_model old_model self.new_model new_model self.new_model_ratio new_model_ratio def route(self, request_id: str): hash_val int(hashlib.md5(request_id.encode()).hexdigest(), 16) if (hash_val % 100) (self.new_model_ratio * 100): return self.new_model return self.old_model这个路由器的核心是“一致性哈希”同一个请求ID总是路由到同一个模型这样用户体验是稳定的不会出现同一个用户一会儿用新模型一会儿用旧模型的情况。new_model_ratio从0.05开始逐步调到1.0整个过程可能需要几天甚至几周取决于业务对稳定性的要求。5.3 线上问题的排查链路即使做了充分的测试和灰度线上还是可能出问题。我总结了一套排查链路按顺序执行能解决大部分常见问题。第一步看服务是否存活检查进程状态、端口监听、健康检查接口。第二步看请求是否正常检查QPS曲线、错误率、超时率。第三步看模型是否正常检查输入数据的分布是否和训练时一致检查模型输出的分布是否发生偏移。第四步看资源是否正常检查CPU、内存、GPU、显存的占用情况。这个链路里第三步是最容易被忽视的。很多人只关注服务层面的指标却忘了模型本身也是一个需要监控的对象。我习惯在推理服务里加一个“数据漂移检测”模块定期统计输入特征的均值和方差和训练时的基准做对比。如果偏差超过阈值就触发告警。这个机制帮我提前发现过好几次上游数据源变更导致的问题避免了更严重的线上事故。提示线上排查时日志是第一手资料。但AI服务的日志往往很大全量记录不现实。我的做法是正常请求只记录摘要信息请求ID、延迟、模型版本异常请求记录完整输入输出。这样既能控制日志量又能在出问题时拿到足够的信息。6. 从手搓到生产我踩过的那些坑6.1 数据管道里的“幽灵空值”有一次我们的推荐模型线上效果突然下降了5个百分点。排查了一圈模型没变、代码没变、服务没变最后发现是上游数据源里某个字段的空值表示方式从null变成了空字符串。我们的数据清洗器里有一个fill_default步骤只处理了None没处理空字符串导致这个字段的特征全部变成了默认值。这个问题之所以难查是因为它不报错、不崩溃只是悄悄地让模型效果变差。从那以后我在数据管道的入口处加了一个“数据质量检查”环节对每个字段做统计空值率、唯一值数量、最大最小长度、类型分布。这些统计和上一次运行的结果做对比如果偏差超过阈值就告警。这个检查看起来很简单但它是数据管道的第一道防线能拦住大部分“幽灵问题”。6.2 ONNX导出的“算子不支持”陷阱ONNX虽然好用但并不是所有PyTorch算子都能完美导出。我遇到过一次模型里用了一个自定义的激活函数导出ONNX时没有报错但推理结果和PyTorch差了十万八千里。后来发现是ONNX Runtime把这个自定义算子当成了一个近似实现精度损失很大。解决办法是把自定义算子拆解成ONNX支持的基础算子组合或者用ONNX的自定义算子接口注册。这个坑给我的教训是导出ONNX之后一定要做数值验证而且验证数据要覆盖各种边界情况。我现在的验证流程是用训练集、验证集、测试集各采样1000条数据对比PyTorch和ONNX的输出计算最大绝对误差和平均绝对误差。只有两个误差都在可接受范围内才允许上线。6.3 批处理带来的“尾延迟”问题批处理能提升吞吐量但会引入尾延迟。我遇到过一个情况平均延迟只有50毫秒但P99延迟高达2秒。排查后发现是批处理器的max_wait_time设置成了100毫秒当请求量稀疏时每个请求都要等满100毫秒才被处理。更糟糕的是当请求量突然增大时队列积压导致后续请求的等待时间越来越长。解决办法是动态调整max_wait_time请求量大的时候减小等待时间请求量小的时候增大等待时间。具体实现可以用一个滑动窗口统计最近的请求速率然后根据速率来调整。这个优化把P99延迟从2秒降到了200毫秒以内效果非常明显。6.4 灰度发布时的“模型版本混乱”灰度发布期间新旧模型同时在线如果日志里不记录模型版本排查问题时会非常痛苦。我遇到过一次用户反馈结果异常但查日志发现同一个请求ID在不同时间被路由到了不同模型导致无法复现。后来我们在每个响应里都加了一个model_version字段并且在日志里强制记录。这样任何一次请求都能追溯到具体是哪个模型处理的。这个经验告诉我灰度发布不仅仅是流量切换更是一套完整的可观测性体系。没有完善的日志和监控灰度发布就是盲人摸象。7. 这套手搓方案适合谁以及后续怎么扩展如果你跟着上面的内容走了一遍你应该已经拥有了一个可以运行的AI工程最小系统数据能进来、特征能算出来、模型能跑起来、服务能对外提供。这套系统当然不完美它没有分布式训练、没有自动扩缩容、没有复杂的特征存储但它的每一个组件你都能看懂、能修改、能排查。这就是“from scratch”的意义——不是让你抛弃现有工具而是让你在工具出问题时有能力深入到内部去解决问题。这套方案最适合那些正在从“调包侠”向“AI工程师”转型的开发者。如果你已经能熟练调用各种API但总觉得心里没底不知道背后发生了什么那亲手实现一遍这些核心组件会让你的认知提升一个档次。后续的扩展方向也很明确数据管道可以接入流式处理框架特征工程可以引入特征存储推理引擎可以支持多模型编排服务化可以接入Kubernetes做自动扩缩容。但无论怎么扩展底层的那些原理和坑都是绕不过去的。我个人在实际操作中的体会是手搓一遍最大的收获不是代码本身而是那种“我知道这里为什么这么设计”的底气。当线上出问题时你不再是一脸茫然地重启服务而是能顺着数据流、特征流、模型流一路排查下去快速定位到根因。这种能力是任何现成工具都给不了你的。