1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。不是因为陌生恰恰相反是因为它戳中了我这几年带团队、做项目最痛的一个点太多人把“AI工程”等同于“会调几个API、会跑几个开源模型”结果一遇到线上问题就抓瞎一遇到性能瓶颈就只会加机器一遇到数据漂移就完全不知道从哪下手。我自己是从传统后端转过来的最早做推荐系统那会儿还没有“AI工程”这个明确的岗位划分大家都是算法工程师顺手把工程也干了。后来模型越来越大、链路越来越长、上线频率越来越高才慢慢意识到AI工程不是算法工程的附属品它是一套独立的、有自己核心方法论的工程体系。这个项目标题里的“from scratch”我理解有两层意思一是从零开始搭建一套完整的AI工程能力二是不依赖现成的高级封装把底层的东西自己走一遍搞清楚每个环节到底在干什么。这篇文章适合谁看如果你是刚入行一两年、平时主要在用现成框架跑实验的算法同学想搞清楚模型上线之后到底发生了什么如果你是后端或数据工程师想转AI方向但不知道从哪切入或者你是技术负责人团队要开始正经做AI产品了需要一套可落地的工程规范——那这篇内容应该能帮你省下不少踩坑的时间。我会按照一个真实项目的推进节奏从整体设计思路讲到核心细节、实操过程、问题排查中间穿插我自己踩过的坑和总结出来的经验。2. 整体设计与思路拆解AI工程到底在工程什么2.1 先搞清楚AI工程和算法研究的边界很多人刚接触这个领域的时候会把AI工程和算法研究混在一起。我刚开始也这样觉得不就是把模型训出来然后部署上去吗后来发现完全不是一回事。算法研究的核心目标是在给定数据上把指标刷上去关注的是模型结构、损失函数、训练策略而AI工程的核心目标是让模型在真实业务场景下稳定、高效、可维护地产生价值关注的是数据管道、特征一致性、服务架构、监控告警、迭代效率。举个具体的例子。算法同学在实验室里训了一个模型AUC比基线高了两个点很开心。但工程同学要问的问题是这个模型推理一次要多久显存占用多少能不能在现有机器上部署训练时的特征和线上推理时的特征怎么保证一致模型更新了怎么灰度出问题了怎么回滚这些问题的答案决定了这个模型能不能真正上线而不是停留在notebook里。所以“from scratch”搭建AI工程体系第一步不是选框架、不是买机器而是先把边界划清楚。我的经验是一个完整的AI工程体系至少包含五个核心模块数据管道、特征平台、训练流水线、模型服务、监控与反馈闭环。这五个模块环环相扣任何一个环节薄弱整个系统的上限就被卡住了。2.2 为什么我选择“先跑通再优化”的架构策略确定了要做什么之后接下来就是怎么做。这里有一个很关键的决策是一开始就设计一个“完美”的架构还是先用最简单的方式跑通全流程再逐步优化我试过前者也试过后者。前者的结果是花了两个月设计了一套自认为很优雅的架构结果第一版模型上线的时候发现数据管道的吞吐量根本撑不住特征平台的设计和实际业务特征对不上最后推倒重来。后者的结果是第一周就用最土的办法把整个链路跑通了虽然每个环节都很粗糙但至少知道瓶颈在哪、哪里需要重点优化。所以我现在坚定地推荐**“先跑通再优化”的策略。具体来说第一版可以这样设计数据管道用最简单的定时脚本从数据库拉数据特征直接在训练脚本里现算模型服务用Flask起一个HTTP接口监控就打印日志。这套东西肯定撑不住生产流量但它的价值在于让你快速验证整个链路的可行性**发现真正的瓶颈在哪里。注意这里说的“简单”是指实现方式简单不是指可以跳过某些环节。五个核心模块一个都不能少只是每个模块先用最轻量的方式实现。2.3 技术选型的核心考量可替换性比性能更重要在“from scratch”的阶段技术选型有一个很容易踩的坑过度关注性能指标忽略了可替换性。比如选特征存储的时候看到某个方案QPS特别高就选了结果发现它的数据模型和业务特征结构不匹配改起来特别痛苦。我的经验是在早期阶段可替换性比性能更重要。因为你对业务的理解还在快速变化今天觉得合适的方案下个月可能就不合适了。如果每个组件都是紧耦合的换一个就要动全身那迭代速度根本起不来。具体怎么做核心原则是接口标准化、实现可插拔。比如数据管道不管底层用的是Kafka还是RabbitMQ上层消费的逻辑应该通过统一的接口来读数据特征平台不管底层用的是Redis还是本地文件上层取特征的API应该保持一致。这样后面要换实现的时候只需要改适配层不用动业务逻辑。3. 核心细节解析与实操要点五个模块的关键设计3.1 数据管道别让脏数据毁掉整个系统数据管道是AI工程的地基但也是最容易被忽视的环节。我见过太多团队模型效果不好就怪算法结果排查半天发现是数据管道里混进了脏数据。数据管道的核心设计要点有三个幂等性、可回溯、Schema约束。幂等性是指同一条数据重复处理多次结果应该是一样的。这个听起来简单但实际做的时候很容易出问题。比如你用“插入”而不是“更新插入”来写数据重复消费就会产生重复记录。我的做法是每条数据都带一个唯一ID写入的时候用upsert语义保证重复消费不会产生副作用。可回溯是指任何一条数据你都能查到它是从哪来的、经过了哪些处理。这个在排查问题的时候特别重要。我一般会在数据管道里加一个“血缘追踪”的字段记录数据的来源、处理时间、处理版本。出问题的时候顺着这个字段就能定位到是哪一步出了错。Schema约束是指数据的结构必须符合预定义的模式。这个在早期容易被忽略觉得数据格式差不多就行。但等到字段类型不一致、字段缺失、字段名拼写错误这些问题积累起来排查成本会指数级上升。我的做法是在数据管道的入口和出口都做Schema校验不符合的直接拒绝并告警。# 一个简单的Schema校验示例 from pydantic import BaseModel, ValidationError from typing import Optional class UserFeature(BaseModel): user_id: int age: Optional[int] None gender: Optional[str] None last_active_days: int def validate_record(record: dict) - bool: try: UserFeature(**record) return True except ValidationError as e: print(fSchema校验失败: {e}) return False实操心得Schema校验的严格程度要循序渐进。一开始可以只校验关键字段等数据质量稳定了再逐步加严。一上来就卡太死会导致大量数据被拒绝影响业务。3.2 特征平台训练和推理的一致性怎么保证特征平台是AI工程里最容易出问题的环节没有之一。核心问题就一个训练时用的特征和推理时用的特征怎么保证是一致的这个问题听起来简单但实际做的时候坑特别多。比如训练的时候用SQL从数据仓库拉特征推理的时候用Redis查特征两边的计算逻辑稍微有一点不一致模型效果就会大打折扣。更隐蔽的是时间窗口的问题训练的时候用的是“过去7天的平均值”推理的时候如果也用“过去7天的平均值”但时间窗口的起止点没对齐结果就是错的。我的解决方案是特征定义统一化、计算逻辑统一化、存储统一化。特征定义统一化是指所有特征都在一个地方定义训练和推理都引用同一个定义。我一般会用一份YAML或JSON文件来描述特征包括特征名、数据类型、计算逻辑、时间窗口等。计算逻辑统一化是指特征的计算逻辑只实现一次训练和推理都调用同一个实现。这个在Python里可以用装饰器或者函数注册的方式来做。存储统一化是指训练和推理用同一份特征数据。这个在早期可以用离线特征存储加在线缓存的方式来做后面再逐步过渡到专门的特征存储系统。# 特征定义示例 feature_config { user_avg_order_amount_7d: { type: float, window: 7d, computation: mean(order_amount), source: orders }, user_order_count_30d: { type: int, window: 30d, computation: count(order_id), source: orders } }注意特征的时间窗口一定要明确起止点的计算方式。是“自然日”还是“滚动窗口”是“包含当天”还是“不包含当天”这些细节不明确训练和推理的结果就会不一致。3.3 训练流水线可复现比跑得快更重要训练流水线的核心要求是可复现。什么意思就是给定同样的数据、同样的代码、同样的配置任何时候都能跑出同样的模型。这个在实验阶段可能觉得无所谓但到了生产环境可复现性是排查问题的前提。可复现性的三个关键点代码版本化、数据版本化、配置版本化。代码版本化用Git就能解决但要注意的是不仅要记录代码的版本还要记录依赖的版本。我一般会用requirements.txt或conda env export把环境也固定下来。数据版本化是指每次训练用的数据都要有一个唯一的版本标识。这个可以用数据快照的方式来做也可以用数据版本管理工具。关键是任何时候都能根据版本标识找到当时用的数据。配置版本化是指训练用的超参数、特征配置、模型结构配置都要记录下来。我一般会把所有配置写在一个YAML文件里和代码一起提交到Git。# 训练配置示例 model: name: wide_and_deep embedding_dim: 16 hidden_units: [128, 64] training: batch_size: 1024 learning_rate: 0.001 epochs: 10 data: train_path: s3://bucket/data/train_v20240101.parquet valid_path: s3://bucket/data/valid_v20240101.parquet feature_config: configs/features_v3.yaml实操心得训练流水线一定要支持“断点续训”和“失败重试”。训练大模型动辄几个小时甚至几天中间因为机器故障或者网络问题中断了如果要从头开始成本太高了。3.4 模型服务延迟和吞吐的平衡艺术模型服务是AI工程里最接近传统后端工程的环节但也有一些特有的挑战。核心挑战是延迟和吞吐的平衡。延迟是指单次推理的响应时间吞吐是指单位时间内能处理的请求数。这两个指标通常是矛盾的提高吞吐往往会增加延迟降低延迟往往会牺牲吞吐。怎么平衡取决于具体的业务场景。如果是实时推荐、搜索排序这类场景延迟是硬指标一般要求P99在100ms以内。这时候可能需要用模型量化、剪枝、蒸馏等手段来压缩模型或者用GPU加速、批处理优化等手段来提升推理速度。如果是离线批量打分、数据分析这类场景吞吐是更重要的指标。这时候可以用更大的批处理、更多的并发来提升吞吐延迟稍微高一点没关系。我的经验是先明确业务的延迟和吞吐要求再倒推技术方案。不要一上来就追求极致的延迟或吞吐那样很容易过度设计。# 一个简单的模型服务示例 from fastapi import FastAPI import torch import numpy as np app FastAPI() model torch.load(model.pt) model.eval() app.post(/predict) async def predict(features: dict): input_tensor torch.tensor([list(features.values())], dtypetorch.float32) with torch.no_grad(): output model(input_tensor) return {prediction: output.item()}注意模型服务一定要做版本管理和灰度发布。新模型上线之前先切一小部分流量过去观察一段时间确认没问题再全量。出问题了要能快速回滚到上一个版本。3.5 监控与反馈闭环模型上线只是开始很多人以为模型上线就完事了其实上线只是开始。模型上线之后数据分布会变、业务逻辑会变、用户行为会变模型效果会逐渐衰减。如果没有监控和反馈闭环模型效果衰减了都不知道。监控的核心指标有三类系统指标、模型指标、业务指标。系统指标包括QPS、延迟、错误率、资源利用率等这些和传统后端服务的监控类似。模型指标包括预测分布、特征分布、模型置信度等。预测分布突然偏移可能意味着数据分布变了特征分布突然偏移可能意味着上游数据出了问题。业务指标包括点击率、转化率、GMV等。这些是最终衡量模型价值的指标但反馈周期比较长需要耐心观察。反馈闭环是指把线上的反馈数据收集起来用于模型的迭代优化。这个闭环跑得越快模型的迭代速度就越快竞争力就越强。4. 实操过程与核心环节实现从零到一的完整记录4.1 环境准备与基础依赖安装假设我们现在要搭建一套完整的AI工程体系从最基础的环境开始。我一般会用Docker来管理环境这样能保证开发、测试、生产环境的一致性。# 基础镜像 FROM python:3.10-slim # 安装系统依赖 RUN apt-get update apt-get install -y \ build-essential \ libgomp1 \ rm -rf /var/lib/apt/lists/* # 安装Python依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 设置工作目录 WORKDIR /app COPY . . CMD [python, main.py]requirements.txt里包含核心依赖numpy1.24.0 pandas2.0.0 scikit-learn1.3.0 torch2.1.0 fastapi0.104.0 uvicorn0.24.0 pydantic2.4.0 redis5.0.0 sqlalchemy2.0.0实操心得依赖版本一定要固定不要用或latest。我踩过好几次坑本地跑得好好的到了服务器上因为某个依赖版本不一致直接报错。4.2 数据管道的搭建与调试数据管道的第一步是数据接入。假设我们的数据源是MySQL需要定时把增量数据同步到数据仓库。import pandas as pd from sqlalchemy import create_engine from datetime import datetime, timedelta def extract_incremental_data(last_sync_time: datetime): engine create_engine(mysql://user:passhost:3306/db) query f SELECT * FROM orders WHERE updated_at {last_sync_time} df pd.read_sql(query, engine) return df def transform_data(df: pd.DataFrame) - pd.DataFrame: # 数据清洗 df df.dropna(subset[user_id, order_amount]) df df[df[order_amount] 0] # 类型转换 df[user_id] df[user_id].astype(int) df[order_amount] df[order_amount].astype(float) return df def load_data(df: pd.DataFrame, target_path: str): df.to_parquet(target_path, indexFalse)调试的时候我一般会先跑一小批数据确认每个环节的输出符合预期再跑全量。全量跑的时候要盯着日志看有没有异常。注意增量同步一定要处理好时区问题。我遇到过好几次数据库存的是UTC时间脚本按本地时间算增量结果要么漏数据要么重复。4.3 特征计算与存储的实现特征计算的核心是保证训练和推理的一致性。我的做法是把特征计算逻辑封装成独立的函数训练和推理都调用同一个函数。class FeatureCalculator: def __init__(self, config: dict): self.config config def compute_user_features(self, user_id: int, as_of_date: datetime) - dict: features {} for feature_name, feature_def in self.config.items(): window feature_def[window] start_date as_of_date - parse_duration(window) # 从数据源拉取窗口内的数据 data self._fetch_data(user_id, start_date, as_of_date) # 根据计算逻辑计算特征值 if feature_def[computation] mean(order_amount): features[feature_name] data[order_amount].mean() elif feature_def[computation] count(order_id): features[feature_name] len(data) return features def _fetch_data(self, user_id: int, start_date: datetime, end_date: datetime): # 实际实现中从数据仓库或缓存拉取 pass存储方面离线特征存Parquet文件在线特征存Redis。训练的时候从Parquet读推理的时候从Redis读但计算逻辑是同一套。实操心得特征计算一定要做单元测试。我一般会构造一批已知输入和预期输出的测试用例每次改特征逻辑都跑一遍确保没有引入bug。4.4 模型训练与评估的完整流程模型训练我一般用PyTorch因为灵活度高方便做各种实验。训练流程包括数据加载、模型定义、训练循环、评估、保存。import torch import torch.nn as nn from torch.utils.data import DataLoader class WideAndDeep(nn.Module): def __init__(self, num_features, embedding_dim16): super().__init__() self.linear nn.Linear(num_features, 1) self.embedding nn.Embedding(num_features, embedding_dim) self.deep nn.Sequential( nn.Linear(num_features * embedding_dim, 128), nn.ReLU(), nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, 1) ) def forward(self, x): linear_out self.linear(x) embed_out self.embedding(x).flatten(start_dim1) deep_out self.deep(embed_out) return torch.sigmoid(linear_out deep_out) def train_model(model, train_loader, valid_loader, epochs10, lr0.001): optimizer torch.optim.Adam(model.parameters(), lrlr) criterion nn.BCELoss() for epoch in range(epochs): model.train() for batch_x, batch_y in train_loader: optimizer.zero_grad() output model(batch_x) loss criterion(output, batch_y) loss.backward() optimizer.step() # 验证 model.eval() valid_loss 0 with torch.no_grad(): for batch_x, batch_y in valid_loader: output model(batch_x) valid_loss criterion(output, batch_y).item() print(fEpoch {epoch1}, Valid Loss: {valid_loss / len(valid_loader):.4f}) return model评估的时候除了看AUC、准确率这些常规指标我还会看特征重要性和预测分布。特征重要性突然变化可能意味着数据管道出了问题预测分布突然偏移可能意味着模型需要重新训练了。4.5 模型部署与线上服务的配置模型部署我一般用FastAPI加Uvicorn轻量且性能足够。部署的时候要注意几个点模型加载一次不要每次请求都加载做好输入校验防止异常输入导致服务崩溃加好日志方便排查问题。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch import logging app FastAPI() logger logging.getLogger(__name__) class PredictRequest(BaseModel): user_id: int features: list[float] class PredictResponse(BaseModel): prediction: float model_version: str # 启动时加载模型 model None model_version v1.0.0 app.on_event(startup) async def load_model(): global model model torch.load(fmodels/model_{model_version}.pt) model.eval() logger.info(f模型 {model_version} 加载完成) app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): try: input_tensor torch.tensor([request.features], dtypetorch.float32) with torch.no_grad(): output model(input_tensor) return PredictResponse( predictionoutput.item(), model_versionmodel_version ) except Exception as e: logger.error(f预测失败: {e}) raise HTTPException(status_code500, detail预测服务异常)注意模型服务一定要加超时控制和限流。我遇到过因为某个请求特征特别多导致推理时间过长把整个服务拖垮的情况。5. 常见问题与排查技巧实录5.1 训练和推理结果不一致的排查思路这是AI工程里最经典的问题没有之一。排查思路我一般按这个顺序来第一步检查特征计算逻辑是否一致。把训练时用的特征值和推理时用的特征值打出来对比看有没有差异。如果有差异定位到具体的特征检查计算逻辑。第二步检查特征的时间窗口是否对齐。训练时用的“过去7天”和推理时用的“过去7天”起止点是不是一样的。这个很容易出问题尤其是跨天、跨时区的时候。第三步检查模型加载是否正确。有时候模型保存和加载的方式不一致会导致权重对不上。我一般会在保存模型的时候同时保存一份权重哈希加载的时候校验一下。第四步检查输入数据的预处理是否一致。比如归一化的参数、缺失值的填充方式训练和推理必须完全一致。排查步骤检查内容常见问题1特征值对比计算逻辑不一致、数据源不一致2时间窗口对齐时区问题、起止点定义不一致3模型加载权重哈希不匹配、版本不对4数据预处理归一化参数不一致、缺失值处理不一致5.2 模型效果突然下降的应急处理模型效果突然下降一般有三种原因数据问题、模型问题、业务问题。数据问题最常见比如上游数据管道断了、数据分布变了、特征计算逻辑改了。排查方法是看监控面板对比出问题前后的数据分布和特征分布。模型问题包括模型文件损坏、模型版本切换出错、推理服务异常等。排查方法是看模型服务的日志和指标。业务问题包括用户行为变化、竞争对手动作、季节性因素等。这类问题比较难排查需要结合业务数据一起分析。应急处理的原则是先止损再排查。如果确认是模型问题先回滚到上一个稳定版本如果是数据问题先切到备用数据源或者暂停相关特征如果是业务问题先观察一段时间确认不是短期波动。实操心得我一般会保留最近三个版本的模型出问题了可以快速回滚。回滚的流程要提前演练不要等到出问题了才现学。5.3 特征平台性能瓶颈的优化经验特征平台的性能瓶颈一般出现在两个地方特征读取和特征计算。特征读取的瓶颈通常是网络延迟或存储系统的QPS限制。优化方法包括加本地缓存、批量读取、预取等。我一般会在服务启动的时候把热点特征预加载到本地内存请求来了直接读内存延迟能降一个数量级。特征计算的瓶颈通常是计算逻辑太复杂或者数据量太大。优化方法包括预计算、增量计算、并行计算等。我一般会把能预计算的特征提前算好存起来推理的时候直接查不用现算。# 特征预加载示例 class FeatureCache: def __init__(self, redis_client, local_cache_size10000): self.redis redis_client self.local_cache {} self.local_cache_size local_cache_size def get_features(self, user_id: int) - dict: # 先查本地缓存 if user_id in self.local_cache: return self.local_cache[user_id] # 本地缓存没有查Redis features self.redis.hgetall(fuser_features:{user_id}) # 写入本地缓存 if len(self.local_cache) self.local_cache_size: self.local_cache.pop(next(iter(self.local_cache))) self.local_cache[user_id] features return features5.4 线上服务延迟毛刺的定位方法延迟毛刺是指P99延迟突然飙升但平均延迟看起来正常。这个问题特别难排查因为毛刺是偶发的很难复现。我的排查方法分三步第一步看毛刺的时间分布。是集中在某个时间段还是随机分布集中在某个时间段可能是定时任务导致的资源竞争随机分布可能是某个特定请求触发的。第二步看毛刺的请求特征。是不是某个特定用户、某个特定特征组合导致的我一般会在日志里记录每个请求的特征哈希出问题的时候可以快速定位。第三步看系统资源指标。CPU、内存、网络、磁盘IO哪个指标在毛刺时间点有异常我遇到过因为磁盘IO抖动导致模型加载变慢进而引发延迟毛刺的情况。注意延迟毛刺的排查一定要有完整的链路追踪。从请求进来到特征读取、模型推理、结果返回每个环节的耗时都要记录下来。不然只能看到总延迟不知道慢在哪。6. 一些踩坑之后的个人体会做AI工程这几年最大的体会是工程能力决定AI产品的下限算法能力决定AI产品的上限。很多团队算法很强但工程能力跟不上模型效果再好也落不了地。反过来工程能力强的团队即使算法不是最先进的也能通过快速迭代把效果一点点磨上去。另一个体会是不要追求一步到位。我见过太多团队一开始就想搭一套“完美”的AI工程体系结果半年过去了还在设计阶段。正确的做法是先跑通最小闭环然后根据实际遇到的问题逐步优化。每个优化都要有明确的业务价值不要为了技术而技术。最后分享一个小技巧把每次线上问题都当成一次学习机会。我有个习惯每次排查完线上问题都会写一份复盘文档记录问题的现象、排查过程、根因、解决方案、后续改进措施。这些文档积累下来就是团队最宝贵的知识库。下次遇到类似问题直接翻文档就能解决不用从头排查。这个内容后续还可以这样扩展如果你已经跑通了基本的AI工程链路下一步可以研究自动化机器学习和持续训练让模型能够自动迭代减少人工干预。再往后可以研究多模型融合和在线学习进一步提升模型效果和响应速度。不过这些都是后话先把基础打牢再说。