1. 从零搭建AI工程能力为什么“会用模型”和“会做工程”是两回事很多人第一次接触AI项目时都会经历一个相似的阶段在笔记本里跑通一个模型准确率看着还行于是觉得“AI不过如此”。可一旦要把这个东西放到真实业务里问题就全冒出来了——推理延迟忽高忽低、显存说爆就爆、模型版本一多就乱套、线上效果和离线评估对不上。这些问题的根源往往不是模型本身不够好而是AI工程能力没有跟上。ai-engineering-from-scratch这个标题核心讲的其实就是一件事把AI从“实验品”变成“可交付的产品”中间需要补哪些工程课。它适合两类人一类是算法出身、模型调得动但系统搭不起来的同学另一类是后端或运维出身、想切入AI方向但不知道从哪下手的同学。这两类人各有短板但目标是一致的——让AI系统稳定、可观测、可迭代。我自己带过几个从零起步的AI项目最深的体会是模型只是整个系统里很小的一块。一个能上线的AI服务背后至少有数据管道、特征处理、模型服务、监控告警、版本管理这几大块。任何一块塌了整体就不可用。所以这篇内容不会只讲怎么调模型而是把从零构建AI工程能力的完整链路拆开讲清楚每一步为什么这么做、坑在哪里、怎么验证。关键词里虽然没有给出具体的技术栈但“from scratch”本身就意味着我们要从最基础的工程决策讲起而不是假设你已经有一套成熟平台。下面我会按照一个真实项目的推进顺序来展开中间穿插大量实操细节和我自己踩过的坑。2. 环境与依赖管理别让“在我机器上能跑”成为口头禅2.1 为什么AI项目的环境问题比普通后端更棘手普通后端项目的依赖相对稳定一个requirements.txt或者go.mod基本能锁住。但AI项目不一样深度学习框架、CUDA驱动、算子库、编译工具链之间有一张复杂的版本兼容网。我见过太多次因为torch和cuda版本对不上导致整个团队卡半天的案例。更麻烦的是训练环境和推理环境往往还不一样——训练用A100推理用T4算子支持程度不同稍不注意就会出现“训练能跑、推理报错”。所以从零做AI工程第一件事不是写模型而是把环境当成代码来管理。我的做法是训练环境用容器镜像固化推理环境单独构建两者共享同一份依赖清单但允许差异化覆盖。具体来说基础镜像里装好CUDA、cuDNN、Python版本项目层面再用conda或uv锁定上层包版本。2.2 依赖锁定的实操方案与常见陷阱很多人习惯用pip freeze requirements.txt这在AI项目里其实不够。因为pip freeze会把间接依赖也写进去导致清单臃肿且难以维护。更推荐的方式是分层管理第一层系统级依赖CUDA、驱动用Dockerfile的FROM指定基础镜像。第二层框架级依赖torch、tensorflow在基础镜像里固定大版本。第三层项目级依赖transformers、datasets等用pyproject.toml或requirements.in管理直接依赖再用pip-compile生成锁定文件。这里有个坑我踩过transformers这个库更新极快不同小版本之间API可能不兼容。如果你在requirements.in里写transformers4.30某天CI重新构建时拉到了4.40可能某个from_pretrained的参数就变了。所以直接依赖也要锁到小版本比如transformers4.36.2。提示如果你的项目需要跨多台机器复现建议把pip-compile生成的锁定文件也纳入版本控制并且在CI里加一步“依赖一致性校验”确保本地和线上装的是同一套。2.3 一个可复用的Dockerfile骨架下面这个骨架是我在多个项目里沉淀下来的训练和推理共用通过构建参数区分ARG BASE_IMAGEnvidia/cuda:12.1.0-cudnn8-runtime-ubuntu22.04 FROM ${BASE_IMAGE} ARG PYTHON_VERSION3.10 RUN apt-get update apt-get install -y \ python${PYTHON_VERSION} python3-pip git curl \ rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.lock . RUN pip install --no-cache-dir -r requirements.lock COPY . . ENV PYTHONPATH/app关键点在于requirements.lock是编译后的锁定文件不是手写的。这样无论谁构建装出来的包版本都一致。另外PYTHONPATH设成/app避免运行时找不到模块——这个坑在本地开发时不容易发现因为本地通常是在项目根目录直接跑。3. 数据管道AI工程里最容易被低估的“脏活累活”3.1 数据质量决定模型上限但工程决定数据能不能用业内有一句话数据和特征决定了模型的上限模型只是逼近这个上限。但我想补一句数据管道工程决定了你能不能稳定地拿到高质量数据。很多团队模型调得很起劲但数据管道是几个脚本拼起来的没有调度、没有重试、没有数据校验结果就是训练时经常发现某天的数据缺了、格式变了、标签错了。从零构建数据管道我建议至少包含四个环节采集、清洗、校验、版本化。采集环节要处理的是数据源多样性——可能是数据库、对象存储、消息队列。清洗环节要处理缺失值、异常值、格式统一。校验环节是最容易被跳过的但恰恰最重要每次数据更新后自动检查行数波动、字段分布、标签比例一旦超出阈值就告警。3.2 用轻量级方案做数据校验不一定非要上Great Expectations这种重框架小项目用pandas加自定义断言就够了。比如import pandas as pd def validate(df: pd.DataFrame, expected_cols: list, label_col: str): assert set(expected_cols).issubset(df.columns), 字段缺失 assert df[label_col].isin([0, 1]).all(), 标签越界 assert df.isnull().mean().max() 0.05, 缺失率过高 assert 0.3 df[label_col].mean() 0.7, 标签比例异常这段代码看着简单但能拦住大部分低级错误。我自己的经验是校验规则要跟着业务走比如风控场景里正样本比例天然很低那阈值就不能照搬通用配置。规则本身也要版本化和数据一起管理。3.3 数据版本化别再用文件名区分了我见过太多团队用train_v2_final_真的final.csv这种方式管理数据版本。短期看没问题长期就是灾难。数据版本化要解决三个问题哪份数据训练了哪个模型、两份数据之间差了什么、能不能回滚。轻量级方案可以用DVC或者LakeFS如果不想引入新工具至少要做到每次数据生成时记录元数据时间、来源、行数、哈希存到一张表里训练时把数据哈希写进模型元数据。这样出问题时能快速定位。哈希计算用hashlib对排序后的内容做摘要即可不用逐行存。注意数据版本化不是把数据复制多份而是记录“指针”和“差异”。全量复制既浪费存储又容易混乱。4. 模型服务化从notebook到线上接口的鸿沟4.1 为什么notebook里的模型不能直接上线notebook里跑模型通常是全量数据加载、批量推理、没有并发。但线上服务要面对的是单条或小批量请求、高并发、低延迟、异常输入。这中间的鸿沟主要体现在三个方面。第一是预处理一致性。notebook里你可能手动做了归一化、分词、截断但线上请求进来时这些步骤必须自动且一致地执行。我见过因为线上分词器和训练时不一致导致效果掉十几个点的案例。解决办法是把预处理逻辑封装成独立模块训练和推理共用同一份代码。第二是批处理与延迟的权衡。线上请求是零散的但GPU批量推理效率高。所以需要做动态批处理dynamic batching把短时间内到达的请求攒成一批一起送进模型。这个攒批的时间窗口就是延迟和吞吐的权衡点通常设10到50毫秒。第三是资源隔离。模型服务往往和业务服务混部如果不做资源限制模型推理可能把CPU或内存吃满拖垮整个服务。用容器的话通过--cpus和--memory限制用K8s的话通过requests和limits控制。4.2 一个最小可用的模型服务实现用FastAPI加ONNX Runtime可以搭一个轻量且性能不错的服务。核心思路是把模型导出成ONNX推理时用ONNX Runtime避免依赖完整的训练框架。from fastapi import FastAPI import onnxruntime as ort import numpy as np app FastAPI() session ort.InferenceSession(model.onnx) app.post(/predict) def predict(payload: dict): x preprocess(payload) # 与训练一致的预处理 inputs {session.get_inputs()[0].name: x} logits session.run(None, inputs)[0] return {label: int(np.argmax(logits)), score: float(np.max(logits))}这里preprocess必须和训练时完全一致建议直接import训练代码里的函数而不是重写一遍。ONNX的好处是推理快、依赖少缺点是有些自定义算子导出不了需要提前验证。4.3 动态批处理的实现思路动态批处理的核心是一个队列加一个后台线程。请求进来先入队后台线程每隔固定时间或队列达到一定长度就取一批出来推理。用asyncio可以实现得比较优雅import asyncio queue asyncio.Queue() BATCH_SIZE 16 WAIT_MS 20 async def batch_worker(): while True: batch [] try: item await asyncio.wait_for(queue.get(), timeoutWAIT_MS / 1000) batch.append(item) except asyncio.TimeoutError: continue while len(batch) BATCH_SIZE and not queue.empty(): batch.append(queue.get_nowait()) # 批量推理并回填结果 results model_infer([b[0] for b in batch]) for (_, future), r in zip(batch, results): future.set_result(r)这段代码的关键是wait_for的超时控制如果20毫秒内没有新请求就把当前攒的批直接送出去避免小批量请求被无限等待。实际生产中还要考虑队列满时的拒绝策略以及推理异常时的错误传播。5. 监控与可观测性模型上线只是开始5.1 模型监控和普通服务监控的区别普通服务监控看的是QPS、延迟、错误率、资源使用率。模型服务这些也要看但还不够。模型特有的监控维度包括输入分布漂移、输出分布漂移、预测置信度分布、特征缺失率。因为这些指标的变化往往先于业务指标恶化。举个例子一个推荐模型上线后AUC没变但线上点击率掉了。排查发现是某个特征在最近一周的缺失率从1%涨到了15%模型虽然还能出结果但质量已经下降。如果只监控服务层指标这个问题要等业务方反馈才发现如果监控了特征缺失率当天就能告警。5.2 漂移检测的实用方法漂移检测不需要搞得很复杂。对于数值特征用PSIPopulation Stability Index或KS检验对于类别特征用卡方检验或简单的分布对比。PSI的计算公式是PSI sum((实际占比 - 预期占比) * ln(实际占比 / 预期占比))一般PSI小于0.1认为稳定0.1到0.25之间需要关注大于0.25就要告警。这个阈值不是绝对的要根据业务敏感度调整。我自己的做法是先跑一段时间收集基线再根据历史波动设定动态阈值而不是一上来就用固定值。5.3 日志与追踪的落地细节模型服务的日志要记录请求ID、输入摘要注意脱敏、输出结果、推理耗时、模型版本。请求ID用于串联整条链路方便排查。输入摘要不要记全量记哈希或关键字段即可避免日志膨胀和隐私问题。追踪方面如果团队已经有OpenTelemetry体系直接把模型推理作为一个span接入即可。没有的话至少要在日志里打上trace_id和上游服务对齐。我见过因为模型服务没打trace_id导致线上问题排查时无法定位是哪个请求出错的案例最后只能靠时间戳猜效率极低。6. 版本管理与迭代让每次变更都可追溯6.1 模型版本、数据版本、代码版本要绑定AI项目最怕的就是“三版本分离”模型是一个版本训练数据是另一个版本代码又是第三个版本三者之间没有关联。出了问题想复现根本不知道当时用的是什么。解决办法很简单每次训练产出的模型元数据里必须记录数据哈希和代码commit。具体做法是在训练脚本结束时把git rev-parse HEAD和数据哈希写进模型文件旁边的metadata.json。推理服务加载模型时也把这份元数据暴露到健康检查接口里。这样任何时候都能查到线上跑的是哪个组合。6.2 灰度发布与回滚策略模型更新不能全量直接上。标准做法是灰度先切5%流量到新模型观察核心指标延迟、错误率、业务指标没有异常再逐步放大到20%、50%、100%。灰度期间要能随时回滚所以旧模型必须保留。回滚的触发条件要提前定义好比如错误率超过基线2倍、延迟P99超过阈值、业务指标下降超过5%。这些条件写成自动化脚本避免人工判断延误。我自己的经验是回滚脚本要定期演练不然真出问题时才发现脚本跑不通那就尴尬了。6.3 模型迭代的节奏控制不是越频繁越好。模型迭代太频繁监控和归因会变得困难——指标波动了不知道是哪个版本引起的。我的建议是建立固定的迭代周期比如每周一次全量更新紧急修复走hotfix通道。每次更新前离线评估必须通过包括整体指标和分群指标。分群指标很重要因为整体AUC没降但某个重要人群的效果可能已经崩了。7. 一些从零起步时最容易忽略的工程细节7.1 配置文件管理别把参数写死在代码里从零做项目时很多人图省事把学习率、batch size、模型路径直接写在代码里。等到要调参或换环境时就得改代码重新提交。正确做法是把所有可变参数抽到配置文件用YAML或JSON管理代码只读配置。更进一步配置也要版本化和代码一起提交。# config/train.yaml model: name: bert-base max_length: 128 train: batch_size: 32 lr: 2e-5 epochs: 3 data: path: s3://bucket/dataset/v3读取时用omegaconf或pydantic做校验避免配置写错导致运行时才报错。7.2 异常处理模型推理的失败模式要区分对待模型推理可能因为多种原因失败输入格式错误、显存不足、模型文件损坏、依赖缺失。这些失败的处理方式不同。输入错误应该返回4xx让调用方修正显存不足应该触发降级或重试模型损坏属于严重故障要立即告警并回滚。我见过把所有异常都catch成500的写法结果调用方无法区分是自己参数错了还是服务挂了排查效率极低。建议定义清晰的错误码体系并在日志里记录足够的上下文。7.3 性能压测上线前必须做的一件事不压测就上线等于闭着眼睛开车。压测要关注几个指标吞吐量QPS、延迟分布P50/P95/P99、资源饱和度。压测工具可以用Locust或wrk关键是压测数据要贴近真实分布不能用全零或随机数据否则测出来的性能没有参考价值。压测时还要注意预热。很多模型第一次推理特别慢因为要加载权重、初始化算子。所以压测前先跑几百次预热请求让服务进入稳定状态再开始统计。线上服务也一样滚动重启后要有预热机制避免刚启动就被流量打挂。8. 我个人在从零构建AI工程能力时的一些体会回过头看从零搭建AI工程能力最难的不是某个具体技术点而是建立一套让事情可重复、可追溯、可回滚的机制。模型可以换、框架可以升级但这套机制是底座。我自己的习惯是每引入一个新组件先问三个问题——它出问题时我怎么发现、怎么定位、怎么恢复。这三个问题答不上来就不急着上。另外不要追求一步到位。我见过团队一开始就想搭一套大而全的MLOps平台结果半年过去还在搭平台业务需求一个没满足。更务实的做法是先用最简单的方式跑通闭环比如脚本加cron加手动回滚然后在痛点出现时逐步替换。工程能力是长出来的不是设计出来的。最后分享一个我常用的检查清单每次模型上线前过一遍环境锁定没有、数据校验通过没有、预处理一致没有、监控埋点全没有、回滚脚本测过没有、压测做过没有。这六条都过了基本不会出大问题。