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_version=13或更高，因为高版本的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_model=PredictResponse) async def predict(request: PredictRequest): start_time = time.time() # 输入校验 if not request.texts: raise HTTPException(status_code=400, detail="texts cannot be empty") if len(request.texts) > 128: raise HTTPException(status_code=400, detail="batch 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_code=504, detail="inference timeout") return PredictResponse(labels=labels, scores=scores) except Exception as e: raise HTTPException(status_code=500, detail=str(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_length=128, batch_size=16, elapsed=2ms [req-12345] [2024-01-15 10:30:00.168] [inference] batch_size=16, elapsed=45ms [req-12345] [2024-01-15 10:30:00.170] [postprocess] num_classes=5, elapsed=2ms [req-12345] [2024-01-15 10:30:00.170] [total] elapsed=47ms, status=success

这样的日志，出了问题一眼就能看出是哪个阶段慢了、哪个参数不对。另外，对于错误日志，一定要把异常堆栈和输入数据的关键信息打出来，但要注意脱敏，不要把用户隐私数据写到日志里。

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的MPS（Multi-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%靠踩坑和总结。你踩的坑越多，总结得越认真，能力提升得越快。所以，不要怕出问题，出了问题认真排查、认真记录，下次遇到类似问题就能快速解决。这个从零搭建的过程，本身就是最好的学习方式。