1. 从零搭建AI工程能力为什么大多数人卡在第一步就放弃了如果你最近在技术社区里频繁看到“ai-engineering-from-scratch”这个说法不用怀疑它不是什么新出的框架或者工具库而是一种越来越被认可的学习路径——从最底层开始把AI工程化所需要的各项能力一块一块搭起来。我身边有不少朋友有的是后端转AI有的是刚入行的算法工程师还有纯粹对AI应用感兴趣的产品经理他们都在问同一个问题到底怎么才能系统地掌握AI工程而不是东一榔头西一棒子地学这个问题的答案其实就藏在“from scratch”这四个字里。市面上大部分AI课程和教程要么是教你调包——import这个、pip install那个跑通一个demo就结束了要么是纯理论——推导公式、讲反向传播学完之后连一个能上线的推理服务都搭不出来。而真正在企业里做AI工程的人都知道从模型训练到服务上线中间隔着一整套工程化的东西数据处理管道、特征存储、模型版本管理、推理优化、监控告警、灰度发布……这些东西没有任何一个课程会完整地教给你因为它们太琐碎、太依赖具体场景了。所以这篇内容我想从一个从业者的角度把“从零搭建AI工程能力”这件事拆开来讲。不是给你一个学习路线图那种泛泛的东西而是把每个阶段真正需要掌握的核心能力、容易踩的坑、以及我自己的实操经验尽可能详细地分享出来。无论你是刚入行的工程师还是想从其他方向转到AI工程或者只是想把AI能力集成到自己项目里的开发者都能从中找到可以直接用的东西。2. 先搞清楚AI工程到底在工程什么2.1 AI工程师和算法工程师、数据工程师的边界在哪里很多人一开始就搞混了这三个角色的职责。我刚开始接触这个领域的时候也迷糊后来在几个不同规模的团队里待过之后才慢慢理清楚。算法工程师的核心产出是模型——他们关心的是模型结构、损失函数、训练策略、效果指标。数据工程师的核心产出是数据管道——他们关心的是数据怎么采集、怎么清洗、怎么存储、怎么保证时效性和准确性。而AI工程师的核心产出是“能跑起来的AI系统”——你需要把算法工程师产出的模型和数据工程师产出的数据变成一个稳定、高效、可维护的服务。这意味着AI工程师需要懂模型但不需要懂到能改模型结构的程度需要懂数据但不需要懂到能设计分布式存储系统的程度。你需要的是“连接”的能力——知道模型需要什么样的输入格式知道数据管道能提供什么样的输出然后在中间做适配、做优化、做容错。举个例子算法工程师给你一个PyTorch的模型文件数据工程师给你一个Kafka的实时数据流你要做的事情是把数据流里的原始数据做预处理转成模型需要的张量格式调用模型做推理然后把结果写回到下游系统。这中间涉及到序列化、批处理、超时控制、错误重试、性能监控等等一系列工程问题。2.2 一个完整的AI系统包含哪些模块我画过很多次架构图后来发现不管什么场景一个完整的AI系统基本上都包含这几个模块数据接入层、预处理层、模型推理层、后处理层、服务接口层、监控层。数据接入层负责从各种来源获取数据——可能是数据库、消息队列、文件系统、或者HTTP接口。预处理层做的是把原始数据转成模型能吃的格式包括分词、归一化、特征编码等等。模型推理层就是加载模型、执行前向计算。后处理层把模型的原始输出转成业务需要的结果比如把logits转成类别标签、把检测框做NMS、把生成的文本做后处理。服务接口层对外提供API可能是RESTful、gRPC、或者消息队列的消费接口。监控层则贯穿整个链路收集延迟、吞吐量、错误率、资源利用率等指标。这个划分看起来很简单但实际做的时候每个模块都有大量的细节。比如预处理层如果模型需要固定长度的输入你就要做padding和truncation如果模型需要特定的归一化参数你就要保证训练和推理时用的是同一套参数如果输入数据有缺失值你还要决定是丢弃、填充还是用默认值。这些决策看起来很小但直接影响最终的推理效果和系统稳定性。2.3 为什么“能跑通demo”和“能上线服务”之间隔着巨大的鸿沟我见过太多这样的情况在Jupyter Notebook里跑得好好的模型一上线就各种问题。延迟从几十毫秒变成几秒吞吐量上不去内存泄漏GPU利用率忽高忽低。这些问题的根源在于demo环境和服务环境有本质区别。demo里你处理的是单条数据服务里你要处理的是并发请求demo里你可以慢慢等模型加载服务里你需要在毫秒级完成推理demo里出错了你重启一下就行服务里你需要保证7x24小时可用。具体来说从demo到服务你需要解决这些问题第一批处理。单条推理的效率极低你需要把多个请求攒成一批一起推理但攒批会引入延迟需要在吞吐量和延迟之间做权衡。第二并发控制。多个请求同时进来你需要用线程池或者异步IO来处理但Python的GIL会让多线程在CPU密集型任务上表现很差所以通常需要用多进程或者把推理放到单独的C服务里。第三资源管理。GPU内存是有限的模型加载、中间张量、批处理缓存都会占用显存你需要精确控制每个部分的内存使用。第四错误处理。输入数据可能格式不对、可能超出模型的处理范围、可能包含恶意内容你需要做输入校验和异常捕获。第五版本管理。模型会更新你需要支持多版本共存、灰度发布、快速回滚。3. 从零搭建的第一个阶段把单机推理服务跑起来3.1 选型为什么我建议从FastAPI加ONNX Runtime开始当你第一次搭建推理服务的时候最容易犯的错误是直接上Kubernetes加Triton Inference Server。这套组合确实强大但对于刚开始的人来说复杂度太高了。你需要先理解容器编排、服务发现、负载均衡、GPU调度这些东西才能把服务跑起来。而实际上在你还没有搞清楚模型推理本身有哪些坑之前这些基础设施层面的东西只会分散你的注意力。我的建议是先用FastAPI加ONNX Runtime把单机服务跑通。FastAPI是一个Python的Web框架轻量、异步、自带文档非常适合做推理服务的接口层。ONNX Runtime是一个跨平台的推理引擎支持CPU和GPU性能不错而且不依赖PyTorch或TensorFlow的运行时环境。这个组合的好处是依赖少、启动快、调试方便。你可以在一台开发机上就把整个链路跑通然后再考虑怎么扩展到多机多卡。选ONNX Runtime而不是直接加载PyTorch模型还有一个重要的原因ONNX是一个开放的模型交换格式你可以把PyTorch、TensorFlow、甚至传统机器学习框架训练出来的模型都转成ONNX格式然后用同一个运行时来推理。这在多团队协作的场景下特别有用因为算法团队可能用不同的框架但工程团队只需要维护一套推理环境。3.2 模型转换从PyTorch到ONNX的实操步骤和常见坑把PyTorch模型转成ONNX看起来就是调用一个torch.onnx.export函数的事情但实际操作中会遇到各种问题。我整理了一个标准的转换流程以及每个步骤需要注意的地方。首先你需要确保模型处于eval模式。这个很简单调用model.eval()就行但很多人会忘记。如果模型还在train模式dropout和batch normalization的行为会不一样导出的ONNX模型推理结果会和预期有偏差。其次你需要准备一个示例输入。这个输入的shape要和实际推理时的shape一致数据类型也要一致。如果模型支持动态shape你需要在export的时候指定dynamic_axes参数。比如对于一个文本分类模型输入是[batch_size, sequence_length]你可以把batch_size和sequence_length都设为动态的。然后调用torch.onnx.export指定opset_version。我一般用opset_version13或更高因为高版本的opset支持更多的算子而且对动态shape的支持更好。导出之后一定要用onnxruntime加载一下跑几条测试数据和PyTorch的输出做对比。如果误差在1e-4以内基本就没问题。常见的坑包括第一模型里有自定义算子ONNX不支持这时候你需要自己写自定义算子或者换一种实现方式。第二模型里有控制流比如if-else或者循环低版本的opset对控制流的支持不好需要升级opset或者改写模型。第三输入输出的名字和顺序不对导致推理时拿不到正确的结果。第四动态shape设置不对导致推理时shape不匹配报错。3.3 服务封装请求解析、批处理、超时控制的代码骨架服务封装的核心是把HTTP请求转成模型输入把模型输出转成HTTP响应同时处理好并发和超时。下面是一个简化的代码骨架用FastAPI和ONNX Runtime实现。import asyncio import numpy as np import onnxruntime as ort from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import time app FastAPI() # 加载ONNX模型 session ort.InferenceSession(model.onnx, providers[CUDAExecutionProvider, CPUExecutionProvider]) class PredictRequest(BaseModel): texts: List[str] class PredictResponse(BaseModel): labels: List[int] scores: List[float] # 预处理函数 def preprocess(texts: List[str]) - np.ndarray: # 这里做tokenization、padding等操作 # 返回shape为[batch_size, sequence_length]的numpy数组 pass # 后处理函数 def postprocess(outputs: np.ndarray) - tuple: # 这里做argmax、softmax等操作 pass app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): start_time time.time() # 输入校验 if not request.texts: raise HTTPException(status_code400, detailtexts cannot be empty) if len(request.texts) 128: raise HTTPException(status_code400, detailbatch size too large) try: # 预处理 inputs preprocess(request.texts) # 推理 input_name session.get_inputs()[0].name outputs session.run(None, {input_name: inputs}) # 后处理 labels, scores postprocess(outputs[0]) # 超时检查 elapsed time.time() - start_time if elapsed 5.0: raise HTTPException(status_code504, detailinference timeout) return PredictResponse(labelslabels, scoresscores) except Exception as e: raise HTTPException(status_code500, detailstr(e))这个骨架里有几个关键点需要展开说。第一批处理。上面的代码是每个请求单独推理效率很低。实际生产中你需要一个批处理调度器把多个请求攒成一批一起推理。可以用asyncio的Queue来实现请求进来后放到队列里后台有一个worker不断从队列里取请求攒够一批或者等待超时后就执行推理。第二超时控制。推理时间可能因为输入长度、批大小、GPU状态而波动你需要设置一个合理的超时时间超时后返回错误而不是让请求一直挂着。第三错误处理。输入可能包含模型无法处理的字符、可能超出最大长度限制、可能格式不对这些都需要在预处理阶段就拦截掉返回明确的错误信息。3.4 实测性能单条推理和批处理的差距有多大我在一台配有T4 GPU的机器上做过测试模型是一个BERT-base的文本分类模型序列长度128。单条推理的延迟大约是15毫秒吞吐量大约是60 QPS。改成批处理之后批大小设为32单批的推理延迟大约是45毫秒但吞吐量提升到了大约700 QPS。也就是说批处理把吞吐量提升了10倍以上代价是单条请求的延迟从15毫秒增加到了45毫秒因为要等攒批。这个权衡在实际场景中怎么选取决于你的业务需求。如果是离线任务对延迟不敏感那就把批大小设大一点最大化吞吐量。如果是在线服务对延迟敏感那就把批大小设小一点或者设置一个最大等待时间比如10毫秒内攒到多少算多少。我一般会设置一个动态批处理策略最小批大小8最大批大小64最大等待时间20毫秒。这样在低负载时延迟低在高负载时吞吐量高。4. 第二个阶段让服务变得可靠和可观测4.1 监控指标除了延迟和QPS你还应该关注什么延迟和QPS是最基本的指标但只靠这两个指标你无法定位问题。我一般会监控这几类指标第一资源指标包括CPU利用率、GPU利用率、GPU显存使用量、内存使用量、网络IO。这些指标能告诉你系统是不是到了瓶颈。第二业务指标包括请求量、成功率、错误率、错误类型分布。这些指标能告诉你系统的健康状态。第三模型指标包括输入长度分布、输出置信度分布、预测类别分布。这些指标能告诉你模型的行为是不是正常。特别要强调的是模型指标。我遇到过一次线上事故模型的预测结果突然全部偏向某一个类别。查了半天才发现是上游数据管道出了问题输入的文本里混入了大量空字符串。如果当时监控了输入长度分布就能第一时间发现异常。所以不要只监控系统层面的指标模型层面的指标同样重要。4.2 日志设计怎么打日志才能快速定位问题日志是排查问题的第一手资料但很多人的日志打得乱七八糟出了问题根本查不到有用信息。我总结了一个日志设计的原则每条日志都要包含请求ID、时间戳、阶段标识、关键参数、耗时。请求ID用于串联一个请求的所有日志时间戳用于分析时序阶段标识用于区分是预处理、推理还是后处理关键参数用于复现问题耗时用于定位性能瓶颈。具体来说一个请求进来的时候生成一个唯一的请求ID然后在每个阶段结束时打一条日志。比如[req-12345] [2024-01-15 10:30:00.123] [preprocess] input_length128, batch_size16, elapsed2ms [req-12345] [2024-01-15 10:30:00.168] [inference] batch_size16, elapsed45ms [req-12345] [2024-01-15 10:30:00.170] [postprocess] num_classes5, elapsed2ms [req-12345] [2024-01-15 10:30:00.170] [total] elapsed47ms, statussuccess这样的日志出了问题一眼就能看出是哪个阶段慢了、哪个参数不对。另外对于错误日志一定要把异常堆栈和输入数据的关键信息打出来但要注意脱敏不要把用户隐私数据写到日志里。4.3 容错机制模型加载失败、输入异常、GPU OOM怎么处理生产环境里什么奇怪的事情都可能发生。模型文件可能损坏、GPU可能突然OOM、输入数据可能包含恶意内容。你需要为每一种异常情况设计好容错机制。模型加载失败是最严重的情况因为这意味着服务完全不可用。我的做法是在服务启动时加载模型如果加载失败服务直接启动失败让容器编排系统重新调度。同时保留上一个版本的模型文件如果新版本加载失败自动回滚到旧版本。输入异常是最常见的情况。比如输入文本超长、包含非法字符、编码格式不对。这些应该在预处理阶段就拦截掉返回明确的错误码和错误信息。我一般会定义一个错误码规范比如4001表示输入为空4002表示输入超长4003表示编码错误等等。这样上游系统可以根据错误码做相应的处理。GPU OOM是比较棘手的情况。当显存不够时推理会直接失败。我的做法是第一精确计算模型和中间张量的显存占用设置一个安全的批大小上限。第二在推理前检查显存使用量如果超过阈值就拒绝新请求或者降低批大小。第三捕获OOM异常清理缓存返回错误而不是让服务崩溃。4.4 压测实战用Locust模拟真实流量找到系统的瓶颈点压测是验证系统可靠性的必要手段。我一般用Locust来做压测因为它支持用Python写测试脚本可以灵活地模拟各种流量模式。下面是一个简单的Locust脚本示例from locust import HttpUser, task, between import random class PredictUser(HttpUser): wait_time between(0.01, 0.1) task def predict(self): # 模拟不同长度的输入 length random.choice([16, 32, 64, 128]) text .join([word] * length) self.client.post(/predict, json{texts: [text]})压测的时候我一般会分几个阶段第一阶段低并发验证功能正确性。第二阶段逐步增加并发观察延迟和吞吐量的变化。第三阶段找到拐点也就是延迟开始急剧上升、吞吐量不再增长的并发数。这个拐点就是系统的瓶颈点。然后分析瓶颈在哪里是CPU、GPU、内存、还是网络。如果是GPU利用率满了那就需要优化模型或者增加GPU。如果是CPU利用率满了那就需要优化预处理或者增加CPU。如果是内存不够那就需要减少批大小或者优化内存使用。5. 第三个阶段从单机到多机服务化架构的演进5.1 什么时候需要从单机扩展到多机单机服务能支撑的QPS是有限的。以我之前的测试为例一台T4 GPU的机器BERT-base模型批处理优化后大概能跑到700 QPS。如果你的业务需要更高的QPS或者需要更高的可用性单机挂了服务就挂了那就需要扩展到多机。但扩展之前先问自己几个问题第一能不能通过优化模型来提升单机性能比如量化、剪枝、蒸馏这些都能显著降低模型的计算量。第二能不能通过增加批大小来提升吞吐量如果延迟允许把批大小从32增加到128吞吐量可能翻倍。第三能不能通过异步处理来削峰填谷如果流量有波峰波谷可以用消息队列把请求缓存起来后台慢慢处理。如果这些优化都做了还是不够那就需要考虑多机部署。5.2 负载均衡策略轮询、最少连接、一致性哈希怎么选多机部署的第一个问题是负载均衡。常见的策略有轮询、最少连接、一致性哈希。轮询最简单每个请求依次分发给每台机器适合机器配置相同、请求处理时间相近的场景。最少连接会把请求发给当前连接数最少的机器适合请求处理时间差异较大的场景。一致性哈希会根据请求的某个特征比如用户ID做哈希把同一个用户的请求总是发给同一台机器适合需要会话保持的场景。对于推理服务我一般用最少连接策略因为推理时间会随着输入长度和批大小波动最少连接能更好地平衡负载。如果服务是无状态的轮询也可以。一致性哈希在推理服务里用得比较少除非你有特殊的缓存需求。5.3 模型版本管理与灰度发布怎么做到不停机更新模型更新是常态但更新过程中不能中断服务。我的做法是用Kubernetes的Deployment来管理推理服务每个模型版本对应一个Deployment。更新的时候先创建一个新版本的Deployment等新版本的Pod全部就绪后通过Service的selector把流量切到新版本。如果新版本有问题把selector切回旧版本就行整个过程秒级完成用户无感知。灰度发布是在此基础上更进一步先把一小部分流量切到新版本观察一段时间如果没有问题再逐步扩大流量比例。Kubernetes本身不直接支持按比例切流量但可以通过Istio或者Linkerd这样的服务网格来实现。如果不想引入服务网格也可以用两个Service一个指向旧版本一个指向新版本然后在入口层比如Nginx做流量分配。5.4 成本控制GPU利用率上不去钱都白花了GPU很贵如果利用率上不去就是在烧钱。我见过很多团队GPU利用率长期在20%以下原因通常是批大小太小、请求不均匀、模型太大。提升GPU利用率的方法有几个第一增大批大小直到GPU利用率达到80%以上。第二用多模型共享GPU比如用Triton Inference Server的ensemble功能把多个小模型放在同一张GPU上。第三用GPU共享技术比如NVIDIA的MPSMulti-Process Service让多个进程共享同一张GPU。第四用竞价实例或者抢占式实例来降低成本但要做好被回收的准备。6. 那些只有踩过坑才知道的实操经验6.1 预处理和后处理的不一致是最大的隐形杀手我遇到过好几次线上效果和离线评估不一致的情况最后查出来都是预处理或后处理的问题。最常见的是训练时用的分词器和推理时用的分词器版本不一样导致token ID对不上。或者训练时做了某种归一化推理时忘了做。或者训练时输出的类别索引和推理时解析的类别索引顺序不一样。这类问题的可怕之处在于它不会报错服务正常运行但结果就是不对。你可能要花好几天才能定位到。我的经验是把预处理和后处理的代码封装成独立的模块训练和推理共用同一份代码。如果做不到共用那就写一个一致性测试用同一批数据分别跑训练管道和推理管道对比中间结果和最终结果。6.2 别小看序列化和反序列化的开销在服务化架构里数据需要在不同模块之间传递序列化和反序列化的开销经常被忽略。我做过一个测试用JSON序列化一个包含1000个浮点数的数组耗时大约是1毫秒。用Protobuf序列化同样的数据耗时大约是0.1毫秒。如果QPS是1000那JSON的序列化开销就是1秒的CPU时间而Protobuf只有0.1秒。在高并发场景下这个差距会被放大。所以如果对性能有要求尽量用二进制格式而不是文本格式。常见的二进制格式有Protobuf、MessagePack、FlatBuffers。Protobuf最通用MessagePack更紧凑FlatBuffers读取速度最快。选哪个取决于你的具体需求。6.3 模型文件的管理比你想的复杂模型文件通常很大几百MB到几个GB。如果每次服务启动都从远程存储下载模型启动时间会很长。如果模型文件更新了怎么保证所有实例都用到新版本我的做法是把模型文件打包到容器镜像里。这样启动时不需要下载而且版本和镜像绑定不会出现版本混乱。缺点是镜像会很大构建和推送的时间会比较长。折中方案是把模型文件放在一个共享存储上容器启动时挂载但需要一个版本管理机制来保证一致性。6.4 日志和监控的存储成本可能超过计算成本这个坑我是真金白银踩过的。有一次服务上线后日志量暴涨一个月下来日志存储的费用比GPU的费用还高。后来分析发现大量的日志是重复的、无用的。比如每个请求都打一条完整的输入输出但99%的请求都是正常的这些日志根本没人看。我的做法是第一分级日志。正常请求只打摘要信息错误请求才打详细信息。第二采样日志。对于高QPS的服务只记录1%的请求的详细日志。第三设置日志保留期限比如只保留最近7天的日志更早的日志归档到冷存储。第四用结构化日志方便后续做聚合分析而不是把日志当文本存。7. 持续演进AI工程能力不是学完就会的7.1 从“能跑”到“跑得好”需要持续迭代搭建一个能跑的AI服务可能只需要几天但让它跑得好、跑得稳、跑得省需要持续迭代几个月甚至几年。我负责过的一个服务上线第一版的QPS是200延迟是100毫秒。经过半年的优化QPS提升到了2000延迟降到了30毫秒。这中间的优化包括模型量化、批处理调优、预处理用C重写、缓存热点请求、GPU显存优化等等。每一项优化看起来都不大但累积起来效果显著。所以不要指望一次就把系统做到完美。先让它跑起来然后根据监控数据找到瓶颈一个一个优化。每次优化后做压测验证效果然后继续找下一个瓶颈。7.2 关注新技术但不要盲目追新AI工程领域的新技术层出不穷新的推理引擎、新的服务框架、新的硬件加速器。保持关注是必要的但不要盲目追新。我见过一些团队看到新的推理引擎出来就换结果换完之后性能没提升多少反而引入了一堆新问题。我的原则是只有当现有方案遇到无法解决的瓶颈时才考虑换方案。换之前先做小规模的对比测试用数据说话。7.3 建立自己的工具箱和最佳实践库最后一点也是我觉得最重要的一点建立自己的工具箱和最佳实践库。每次解决一个问题就把解决方案整理成可复用的代码片段或者文档。时间长了你就会有一套自己的工具遇到类似问题的时候可以直接拿来用而不是从头再来。比如我现在有一套标准的Dockerfile模板、一套标准的监控配置、一套标准的压测脚本。这些东西看起来不起眼但能节省大量的时间。我在实际操作中的体会是AI工程能力的提升20%靠学习新知识80%靠踩坑和总结。你踩的坑越多总结得越认真能力提升得越快。所以不要怕出问题出了问题认真排查、认真记录下次遇到类似问题就能快速解决。这个从零搭建的过程本身就是最好的学习方式。