1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地降低了一个刚入门的开发者花一个下午就能用现成的框架跑通一个对话机器人。但我带过不少新人也面试过不少号称“做过AI项目”的候选人发现一个很普遍的现象模型能跑起来但一旦线上出问题或者需要做性能优化、成本控制、效果调优很多人就完全不知道从哪里下手了。这就是我特别想聊“ai-engineering-from-scratch”这个方向的原因——不是让你去造轮子而是让你真正理解轮子是怎么转的。所谓AI工程能力从零构建核心不是从零训练一个大模型那既不现实也没必要。它指的是从最基础的数学直觉、数据处理、模型推理、服务封装到部署上线、监控迭代这一整条链路上你都能自己动手搭一遍、拆一遍、改一遍。这个过程能帮你建立起对AI系统的“手感”就像学编程不能只学调库得自己写过链表和排序一样。这套能力适合谁适合已经会用Python、了解基本机器学习概念但总觉得自己的AI项目“浮在表面”的开发者也适合那些想从传统后端转向AI工程方向的工程师。你不需要有很深的算法背景但需要有耐心把每个环节都亲手过一遍。我自己的体会是真正拉开AI工程师差距的从来不是谁会用更多框架而是谁能在模型效果不达预期时快速定位是数据问题、特征问题、推理配置问题还是服务架构问题。这种定位能力只有你从零把整条链路搭过才能长在身上。接下来我会按照我实际带人做项目的顺序把这条链路拆开讲清楚每个环节都告诉你为什么这么做、怎么做、容易踩什么坑。2. 整体思路与方案选型先跑通最小闭环再逐层加码2.1 为什么我坚持“最小闭环优先”而不是“功能优先”很多人做AI项目一上来就想把功能做全又要支持多轮对话又要接知识库又要做意图识别还要加个前端界面。结果往往是每个模块都半生不熟出了问题互相甩锅最后项目烂尾。我从零构建AI工程能力时始终坚持一个原则先跑通一个最小闭环哪怕它只做一件极其简单的事。什么是最小闭环比如你要做一个文本分类服务最小闭环就是读入一条文本经过一个最简单的模型哪怕就是个逻辑回归或者一个预训练模型加一层分类头输出一个类别然后把这个过程封装成一个HTTP接口。就这么简单。这个闭环里包含了数据读取、模型推理、服务封装三个核心环节每个环节你都能看到输入和输出。跑通之后你再往里面加东西加预处理、加缓存、加批量推理、加监控。每加一层你都知道它为什么被加进来解决了什么问题。这个思路的好处是你始终有一个能工作的系统作为参照。当新加的模块导致效果变差或性能下降时你可以快速回退到上一个可用状态对比排查。我见过太多人一上来就搭了一个复杂架构结果连数据从哪进、从哪出都说不清楚调试全靠猜。2.2 技术栈选型为什么我推荐从Python原生生态起步在工具选型上我的建议可能和很多教程不一样不要一上来就用那些高度封装的AI应用框架。那些框架确实能让你几分钟跑通一个Demo但它们把太多细节藏起来了你学不到东西。我推荐从Python原生生态起步具体来说数据处理用NumPy和Pandas就够了不需要上Spark。除非你的数据量真的到了TB级别否则单机Pandas处理几百万条数据完全没问题。用原生库的好处是你对每一步操作都心里有数知道内存里发生了什么。模型推理直接用PyTorch或TensorFlow的原生API加载模型、做前向推理。不要用那些自动帮你做批处理、自动管理显存的封装。你自己写一遍推理循环才能理解batch size、序列长度、显存占用之间的关系。服务封装用FastAPI或者Flask这种轻量级框架自己写路由、自己处理请求解析和响应序列化。不要用那些自动生成API的框架因为你不知道它背后帮你做了多少事。部署先用最朴素的进程管理方式比如用systemd或者supervisor跑一个Python进程。等你理解了服务的基本生命周期再去考虑Docker和Kubernetes。这套选型的逻辑是每一层都让你接触到最本质的东西。等你把这些都摸透了再去用那些高级框架你就能一眼看出它们帮你省了什么、代价是什么。这时候你用框架是你在驾驭框架而不是框架在驾驭你。2.3 环境隔离与依赖管理别让环境问题浪费你半天时间从零构建AI工程能力环境管理是第一个拦路虎。我强烈建议从第一天就用虚拟环境Python自带的venv就够用不需要上Conda。为什么因为venv更轻量和pip配合更直接而且你以后部署到服务器上时venv的迁移成本更低。具体操作上我会给每个项目建一个独立的目录然后在目录下执行python -m venv .venv激活后所有依赖都装在这个环境里。依赖记录用pip freeze requirements.txt但要注意这个命令会把所有间接依赖都写进去有时候会导致版本冲突。更好的做法是手动维护一个requirements.in只写你直接依赖的包然后用pip-compile生成锁定版本的requirements.txt。这个习惯能帮你在不同机器上复现环境时省下大量时间。还有一个坑是Python版本。我建议统一用3.10或3.11这两个版本在AI生态里兼容性最好。3.12虽然新但有些库的wheel还没跟上编译安装会让你痛不欲生。CUDA版本也要注意PyTorch的版本和CUDA版本有严格的对应关系装错了就是各种莫名其妙的报错。我的经验是先去PyTorch官网查好对应关系再用pip安装不要用conda自动解析那个解析过程有时候会把你带沟里。3. 核心细节解析与实操要点数据、模型、服务三件套3.1 数据处理从原始文本到模型输入的完整链路数据处理是AI工程里最脏最累但最重要的一环。我见过太多项目模型结构调来调去最后发现是数据清洗没做好。从零构建时我建议你把数据处理拆成四个明确的步骤每一步都写成独立的函数方便单独测试和替换。第一步是原始数据加载。不管你的数据是CSV、JSON还是数据库里的先写一个函数把它读成Python的列表或字典。这一步不要做任何转换就读进来。目的是让你确认数据能正常读取字段名和内容符合预期。第二步是清洗与过滤。这一步要处理缺失值、异常字符、重复样本。比如文本分类任务里空文本、纯符号文本、超长文本都要处理。我的经验是先统计一下文本长度的分布看看有没有异常长的样本。如果有要么截断要么单独处理。截断长度怎么定一般取覆盖95%样本的长度作为max_length这样既能保留大部分信息又不会让padding太多浪费计算。第三步是分词与编码。如果你用的是预训练模型就用它对应的tokenizer。这里有个细节tokenizer的padding和truncation参数一定要显式设置不要依赖默认值。paddingmax_length和paddingTrue动态padding在效果上可能有细微差别动态padding通常更快但需要你手动处理batch内的对齐。我一般训练时用动态padding推理时用固定长度这样能兼顾效率和稳定性。第四步是特征组装。把input_ids、attention_mask、token_type_ids如果需要组装成模型需要的格式。这一步要注意数据类型PyTorch的模型通常要求LongTensor如果你从NumPy转过来忘了转类型就会报类型错误。我习惯在组装完后再检查一遍shape和dtype这个习惯帮我省了很多调试时间。注意数据处理函数一定要写单元测试。拿几条典型样本手动算一遍预期输出然后和函数输出对比。这个习惯能帮你提前发现90%的数据bug。3.2 模型推理自己写一遍前向循环理解每个参数的含义模型推理看起来简单不就是model(input)吗但真正从零写一遍你会发现有很多细节需要理解。我建议你从单条推理开始然后扩展到批量推理最后再考虑性能优化。单条推理的代码大概长这样import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification tokenizer AutoTokenizer.from_pretrained(bert-base-chinese) model AutoModelForSequenceClassification.from_pretrained(bert-base-chinese, num_labels2) model.eval() text 这个产品非常好用 inputs tokenizer(text, return_tensorspt, truncationTrue, max_length128) with torch.no_grad(): outputs model(**inputs) logits outputs.logits pred torch.argmax(logits, dim-1) print(pred.item())这段代码里每个参数都值得你停下来想一想。return_tensorspt是告诉tokenizer返回PyTorch张量如果你忘了写返回的就是Python列表模型没法直接用。truncationTrue是超长截断max_length128是截断长度。torch.no_grad()是关闭梯度计算推理时一定要加否则显存占用会大很多。model.eval()是切换到评估模式影响Dropout和BatchNorm的行为推理时必须加。批量推理时你要自己控制batch size。batch size太大会爆显存太小则GPU利用率低。怎么找合适的batch size我的经验是从8开始试逐步翻倍直到显存占用达到80%左右。同时要观察推理时间如果batch size翻倍后时间没有明显增加说明GPU还没吃满可以继续加。这个过程需要你实际跑几次记录不同batch size下的显存和耗时画个曲线就清楚了。还有一个容易忽略的点是推理结果的解析。模型输出的是logits你需要自己做softmax得到概率再取argmax得到类别。如果是多标签分类还要自己设阈值。这些步骤看起来简单但阈值怎么定、概率怎么校准都是需要你根据业务场景去调的。我一般会先跑一批验证集看看概率分布再决定阈值。3.3 服务封装从脚本到API的跨越把推理脚本变成一个HTTP服务是AI工程能力的关键一步。我推荐用FastAPI因为它自带异步支持和自动文档而且性能比Flask好。下面是一个最简服务的样子from fastapi import FastAPI from pydantic import BaseModel import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification app FastAPI() tokenizer AutoTokenizer.from_pretrained(bert-base-chinese) model AutoModelForSequenceClassification.from_pretrained(bert-base-chinese, num_labels2) model.eval() class Request(BaseModel): text: str class Response(BaseModel): label: int score: float app.post(/predict, response_modelResponse) def predict(req: Request): inputs tokenizer(req.text, return_tensorspt, truncationTrue, max_length128) with torch.no_grad(): outputs model(**inputs) probs torch.softmax(outputs.logits, dim-1) pred torch.argmax(probs, dim-1) return Response(labelpred.item(), scoreprobs[0][pred].item())这个服务跑起来后你可以用curl或者Postman测试。但要注意几个问题第一模型加载要放在服务启动时不要放在每次请求里否则每次请求都重新加载模型慢得没法用。第二要考虑并发请求FastAPI默认是同步的如果推理是CPU密集型多个请求会排队。你可以用async def配合线程池或者直接用多进程部署。第三要加异常处理比如输入文本为空、超长、包含特殊字符时服务不能崩。我一般会在服务里加一个健康检查接口/health返回模型是否加载成功、当前显存占用等信息。这个接口在部署和监控时非常有用。另外日志一定要打全每个请求的输入长度、推理耗时、输出结果都记下来出问题时才有据可查。4. 实操过程与核心环节实现从零搭一个文本分类服务4.1 项目结构设计与初始化我习惯把项目结构设计得清晰一点方便后续扩展。一个典型的从零构建的AI工程项目结构是这样的ai-service/ ├── .venv/ ├── data/ │ ├── raw/ │ └── processed/ ├── src/ │ ├── data_loader.py │ ├── preprocess.py │ ├── model.py │ ├── inference.py │ └── service.py ├── tests/ │ ├── test_preprocess.py │ └── test_inference.py ├── requirements.in ├── requirements.txt └── README.md这个结构里data_loader.py负责读原始数据preprocess.py负责清洗和编码model.py负责加载模型inference.py负责推理逻辑service.py负责HTTP服务。每个文件职责单一方便单独测试。tests/目录放单元测试我至少会写预处理和推理的测试。requirements.in只写直接依赖比如torch、transformers、fastapi、uvicorn然后用pip-compile生成requirements.txt。初始化步骤很简单先建目录再建虚拟环境再装依赖。但有个细节要注意transformers库的版本要和torch匹配我一般会先装torch再装transformers让pip自动解析兼容版本。如果手动指定版本很容易出现版本冲突。4.2 数据准备与预处理实操假设我们要做一个情感分类服务数据是CSV格式两列text和label。第一步是加载数据import pandas as pd def load_data(path): df pd.read_csv(path) assert text in df.columns and label in df.columns df df.dropna(subset[text, label]) df df[df[text].str.strip() ! ] return df这段代码做了三件事读CSV、检查列名、去掉空文本和空标签。assert语句是防御性编程如果数据格式不对直接报错不要让它悄悄跑下去。第二步是分析文本长度分布def analyze_length(df): lengths df[text].str.len() print(lengths.describe()) print(f95分位长度: {lengths.quantile(0.95)}) print(f99分位长度: {lengths.quantile(0.99)})跑一下这个函数你就能知道大部分文本有多长。如果95分位是100那max_length设128就够如果99分位是500那可能要设512。这个决策直接影响推理速度和显存占用一定要基于数据来定不要拍脑袋。第三步是编码。我一般会写一个encode_batch函数接收文本列表返回模型输入def encode_batch(tokenizer, texts, max_length128): inputs tokenizer( texts, paddingTrue, truncationTrue, max_lengthmax_length, return_tensorspt ) return inputs注意这里用了paddingTrue这是动态padding会按batch内最长文本对齐。训练时用这个能加速推理时如果batch size是1动态padding和固定padding没区别。但如果你要做批量推理动态padding会让每个batch的shape不一样需要你手动处理。4.3 模型加载与推理性能优化模型加载我建议单独写一个类把tokenizer和model封装在一起class TextClassifier: def __init__(self, model_name, num_labels2, deviceNone): self.tokenizer AutoTokenizer.from_pretrained(model_name) self.model AutoModelForSequenceClassification.from_pretrained( model_name, num_labelsnum_labels ) self.device device or (cuda if torch.cuda.is_available() else cpu) self.model.to(self.device) self.model.eval() def predict(self, texts, max_length128): inputs self.tokenizer( texts, paddingTrue, truncationTrue, max_lengthmax_length, return_tensorspt ) inputs {k: v.to(self.device) for k, v in inputs.items()} with torch.no_grad(): outputs self.model(**inputs) probs torch.softmax(outputs.logits, dim-1) preds torch.argmax(probs, dim-1) return preds.cpu().tolist(), probs.cpu().tolist()这个类里device的自动选择很关键。有GPU就用GPU没有就用CPU。inputs要搬到和模型同一个设备上否则会报设备不匹配的错误。probs.cpu()是把结果搬回CPU因为后续处理通常不需要GPU。性能优化方面我试过几个手段。第一是半精度推理把模型转成torch.float16显存占用能降一半速度也能提升但精度可能有微小损失。对于分类任务这个损失通常可以接受。第二是ONNX Runtime把PyTorch模型导出成ONNX格式用ONNX Runtime推理CPU上能快2到3倍。第三是批处理把多个请求攒成一个batch一起推理能显著提升GPU利用率。但批处理会引入延迟需要根据业务场景权衡。提示半精度推理在有些模型上会导致数值不稳定建议先在验证集上对比一下精度确认没问题再上线。4.4 服务部署与监控实操服务写好后部署方式我推荐先用uvicorn直接跑uvicorn src.service:app --host 0.0.0.0 --port 8000 --workers 2--workers 2表示起两个工作进程能处理并发请求。但要注意每个worker都会加载一份模型显存占用会翻倍。如果显存不够就只起一个worker用异步处理并发。监控方面我至少会记录这几个指标请求量、平均延迟、P95延迟、错误率、显存占用。这些指标可以用Prometheus采集用Grafana展示。如果不想搞这么复杂至少要在日志里打出来方便出问题时排查。还有一个实用技巧是请求队列。当请求量超过服务处理能力时不要直接拒绝而是把请求放进队列慢慢处理。这样可以避免服务崩溃但会增加延迟。队列长度要设上限超过上限再拒绝防止内存爆掉。5. 常见问题与排查技巧实录5.1 模型推理报错速查表从零构建AI工程能力的过程中你会遇到各种各样的报错。我把最常见的几类整理成表格方便你快速定位报错信息可能原因排查方法RuntimeError: Expected all tensors to be on the same device模型和输入不在同一设备检查model.device和inputs的设备用.to(device)统一CUDA out of memory显存不足减小batch size或改用半精度或清理缓存torch.cuda.empty_cache()Token indices sequence length is longer than...输入超过模型最大长度设置truncationTrue和max_lengthValueError: too many values to unpack返回值数量不对检查模型输出结构用outputs.logits而不是直接解包ConnectionErrorwhen loading model网络问题或模型名错误检查模型名拼写确认网络能访问模型仓库这张表里的问题我几乎每个都踩过。特别是设备不匹配那个刚开始用GPU时经常忘后来我养成了一个习惯在推理函数开头就打印一下模型和输入的设备确认一致再往下跑。5.2 效果不达预期的排查思路模型跑起来了但效果不好这是更让人头疼的问题。我的排查顺序是这样的第一看数据。把预测错误的样本拿出来一条条看。是标注错了还是文本本身有歧义还是模型确实没学到我遇到过好几次最后发现是训练数据里有大量重复样本导致模型过拟合。也有一次是标签映射搞反了正负样本弄颠倒这种低级错误一定要先排除。第二看输入。把模型实际收到的input_ids打印出来看看tokenizer有没有把文本切错。中文分词有时候会把关键词切碎导致模型学不到。如果发现这个问题可以考虑换tokenizer或者在预处理时加一些规则。第三看模型。如果数据和输入都没问题那就是模型本身的问题。可以试试换一个预训练模型或者调一下学习率、batch size。我一般会先跑一个baseline记录准确率和F1然后每次只改一个变量对比效果。这样能清楚知道哪个改动有效。第四看评估。有时候不是模型不好是评估指标选错了。比如类别不平衡时准确率会虚高这时候要看F1或者AUC。我习惯同时看多个指标避免被单一指标误导。5.3 服务上线的避坑经验服务上线后有几个坑我踩过这里分享给你第一个坑是冷启动。服务刚启动时模型还没加载完请求就进来了会报错。解决办法是在服务启动时先加载模型加载完再对外提供服务。FastAPI可以用startup事件来做这件事。第二个坑是内存泄漏。如果每次请求都创建新的tokenizer或model内存会一直涨。一定要把模型加载放在全局所有请求共用。另外PyTorch的缓存也要注意长时间运行后torch.cuda.empty_cache()可以清理一下。第三个坑是日志太多。每个请求都打详细日志磁盘很快满。我一般会分级打日志正常请求只打摘要错误请求打详细堆栈。日志要定期轮转别让一个文件无限增长。第四个坑是版本不一致。开发环境跑得好好的上线就报错多半是依赖版本不一致。解决办法是用pip-compile锁定版本部署时严格按requirements.txt安装。Docker镜像能进一步保证环境一致但会增加构建时间看你的场景权衡。6. 从零构建之后能力扩展与持续迭代6.1 从单模型到多模型编排当你把单模型服务跑通后下一步自然是多模型编排。比如一个对话系统可能需要意图识别模型、实体抽取模型、情感分析模型协同工作。这时候你要考虑的是模型之间怎么通信是串行还是并行串行简单但慢并行快但复杂。我的建议是先用串行跑通再逐步把独立的模型改成并行。多模型编排的另一个问题是资源管理。多个模型同时加载显存可能不够。解决办法是模型共享底层编码器或者用模型量化压缩。如果实在不够就做模型路由根据请求类型只加载需要的模型。6.2 从手动调参到自动化评估手动调参效率太低我建议尽早建立自动化评估流程。具体来说就是准备一个固定的验证集每次改动后自动跑一遍评估输出准确率、F1、延迟等指标。这个流程可以用脚本实现也可以用MLflow这类工具管理。关键是让评估标准化避免每次凭感觉判断好坏。自动化评估还有一个好处是能发现回归。有时候你改了一个地方以为只影响某个功能结果评估发现其他指标也掉了。这种问题手动测试很难发现自动化评估能帮你兜住。6.3 从单机部署到弹性伸缩单机部署能撑住的流量有限当请求量上来后你需要考虑弹性伸缩。最朴素的方式是加机器前面挂个负载均衡。再进一步是用容器编排根据CPU或GPU利用率自动扩缩容。但弹性伸缩有个前提你的服务必须是无状态的模型加载要快否则扩容时冷启动太慢用户会感知到。我自己的经验是先把单机服务做到极致把延迟和吞吐优化到瓶颈再考虑分布式。很多时候单机优化能解决80%的问题分布式反而引入更多复杂性。等你真的需要分布式了再上也不迟。最后分享一个小技巧不管你的服务多简单一定要写一个/metrics接口暴露关键指标。这个接口在排查问题时能救命而且成本极低。我从第一个AI服务开始就加了这个接口后来每次出问题都是靠它快速定位的。