1. 从零手搓AI工程：为什么我不建议你直接调包

很多人一听到“AI工程”这四个字，第一反应就是打开某个云平台，拖几个组件，调几个API，然后跑通了事。我刚开始接触这个领域的时候也是这么想的，直到有一次线上推理服务在高峰期直接雪崩，排查了整整两天才发现问题出在一个我从来没正眼看过的预处理环节上。那次事故之后，我开始系统性地把AI工程链路从头到尾自己实现了一遍，也就是今天想跟你聊的“ai-engineering-from-scratch”这件事。

所谓从零构建AI工程，不是让你去手写一个Transformer的注意力机制，也不是让你从汇编开始优化矩阵乘法。它的核心含义是：把AI系统从数据进入、特征处理、模型推理、结果后处理到服务暴露的完整链路，用你自己能完全掌控的方式搭建起来。这件事适合谁？适合那些已经会用现成框架跑模型，但一遇到性能瓶颈、线上故障、数据漂移就束手无策的工程师；也适合想真正理解AI系统全貌、不想永远停留在“调包侠”阶段的技术人。

我自己的体会是，你调包能跑通一个demo，和你从零搭一套能扛住真实流量的AI工程系统，中间隔着的不是几个API的距离，而是对整条链路上每个环节的深刻理解。这篇文章我会把我在这个过程中踩过的坑、总结的方法、以及可以直接抄作业的实现方案，毫无保留地分享出来。全文会比较长，但如果你能跟着走一遍，你对AI工程的理解会上一个台阶。

2. 整体架构设计与技术选型思路

2.1 为什么选择“分层解耦”而不是“端到端一把梭”

我见过很多团队的做法是：一个Python脚本从读数据开始，中间调模型，最后直接返回结果，所有逻辑揉在一个文件里。这种写法在原型阶段没问题，但一旦要上生产，问题就全暴露出来了。数据格式变了要改代码，模型换了要改代码，并发上来了要改代码，最后这个文件变成了一坨谁都不敢动的“屎山”。

我的方案是分层解耦，把整条链路拆成五个独立的层：数据接入层、预处理层、推理层、后处理层、服务层。每一层之间通过明确定义的接口通信，层与层之间可以独立替换、独立扩展、独立测试。

这么设计的好处是什么？举个例子，某天你发现预处理成了瓶颈，你可以单独把预处理层从Python换成C++实现，其他层完全不用动。再比如，你想同时支持两个不同版本的模型做A/B测试，只需要在推理层做路由，上层完全无感知。这种灵活性在真实业务里太重要了，因为需求变化的速度永远比你想象得快。

2.2 技术栈选型：不追新，只选对的

在技术选型上，我的原则是稳定优先、生态成熟、社区活跃。具体来说：

这里重点说一下为什么推理层我选了ONNX Runtime而不是直接上PyTorch Serving。原因很简单：ONNX Runtime在CPU上的推理性能通常比原生PyTorch好20%到40%，而且它不依赖完整的PyTorch环境，部署包体积小很多。当然，如果你用的是GPU集群，TensorRT可能是更好的选择，但那是另一个话题了。

注意：技术选型没有银弹，我列的这个组合适合中小规模、以CPU推理为主的场景。如果你的场景是超大规模GPU集群，选型策略需要重新评估。

2.3 数据流设计：每一步都要可观测

整条链路的数据流我是这样设计的：原始请求进来后，先经过数据接入层做格式校验和限流，然后进入预处理层做特征提取和归一化，接着推理层加载模型做前向计算，后处理层对输出做解码和业务规则过滤，最后服务层组装响应返回。

关键点在于：每一步的输入输出都要打日志、记指标。我在每个层的入口和出口都埋了监控点，记录数据形状、处理耗时、异常信息。这样做的好处是，一旦线上出问题，我能快速定位到是哪一层出了什么类型的错误，而不是像无头苍蝇一样到处猜。

这个设计思路说起来简单，但真正落地的时候有很多细节要注意。比如日志不能打太多，否则IO会成为瓶颈；指标要区分P50、P95、P99，只看平均值会掩盖长尾问题；异常处理要分级，有些错误可以重试，有些必须立即熔断。这些经验都是我在实际项目中一点点积累出来的。

3. 核心模块拆解与关键实现细节

3.1 预处理层：最容易被低估的性能杀手

很多人觉得预处理就是简单的数据清洗和格式转换，能有多难？我告诉你，在我经手的项目里，预处理层消耗的时间经常占到整个推理链路的60%以上。尤其是当输入是文本或图像的时候，分词、编码、缩放、归一化这些操作，每一个都可能成为瓶颈。

我的做法是把预处理拆成无状态操作和有状态操作两类。无状态操作比如字符串小写化、数值裁剪，可以并行执行；有状态操作比如依赖全局统计量的归一化，需要预先计算好参数并缓存。这样拆分之后，无状态部分可以用多线程或向量化加速，有状态部分只需要查表，整体耗时能降下来一大截。

具体实现上，我用了一个Pipeline模式：

class PreprocessPipeline: def __init__(self): self.steps = [] def add_step(self, step): self.steps.append(step) return self def run(self, data): for step in self.steps: data = step.process(data) return data

每个step是一个独立的处理单元，有统一的process接口。这样做的好处是，你可以像搭积木一样组合不同的预处理步骤，而且每个步骤可以单独做单元测试。

实操心得：预处理阶段一定要做输入校验。我踩过的坑是，线上突然来了一批格式异常的请求，预处理直接抛异常导致整个服务不可用。后来我在Pipeline最前面加了一个校验step，对不合规的输入直接返回错误码，不再往下传。

3.2 推理层：模型加载与批处理策略

推理层的核心问题就两个：怎么加载模型和怎么组织批处理。

模型加载方面，我的经验是启动时预加载，运行时零加载。什么意思？就是服务启动的时候就把所有需要的模型加载到内存里，运行过程中不再动态加载。这样做的好处是避免了首次请求的冷启动延迟，坏处是内存占用会高一些。如果你的模型特别大，可以考虑用内存映射的方式加载，或者做模型分片。

批处理策略是推理层性能的关键。我试过三种方案：

第一种是同步单条推理，来一个请求处理一个。实现最简单，但吞吐量极低，GPU利用率可能连10%都不到。

第二种是固定窗口批处理，攒够N条或者等M毫秒就触发一次推理。这个方案实现也不复杂，但需要调两个参数：批大小N和等待时间M。N太大延迟高，N太小吞吐上不去；M太长延迟高，M太短攒不够批。

第三种是动态批处理，根据当前队列长度和系统负载动态调整批大小。这个方案性能最好，但实现复杂度也最高。

我最终采用的是第二种方案的改进版：自适应窗口批处理。核心逻辑是维护一个请求队列，当队列长度达到阈值或者等待时间超过上限时触发推理。阈值不是固定的，而是根据最近的推理耗时动态调整。推理快的时候阈值调大，推理慢的时候阈值调小，这样能在延迟和吞吐之间找到一个动态平衡。

class AdaptiveBatcher: def __init__(self, max_batch_size=32, max_wait_ms=50): self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms self.queue = [] self.last_infer_time = 10 # 初始估计值，单位ms def should_trigger(self): if len(self.queue) >= self.max_batch_size: return True if self.queue and self._waited_too_long(): return True return False def _waited_too_long(self): # 根据最近推理耗时动态调整等待上限 dynamic_wait = min(self.max_wait_ms, self.last_infer_time * 2) return self._oldest_wait_time() > dynamic_wait

这段代码的核心思想是：如果最近推理很快，那就多等一会儿攒更大的批；如果最近推理很慢，那就少等一会儿赶紧处理，避免请求堆积。

3.3 后处理层：业务规则的灵活编排

后处理层是很多人会忽略的地方，但它恰恰是业务逻辑最集中的地方。模型输出的原始结果往往不能直接返回给用户，需要做解码、过滤、排序、格式化等一系列操作。

我的做法是引入一个轻量级规则引擎。每条规则是一个独立的函数，输入是上一步的输出，输出是处理后的结果。规则可以串行执行，也可以根据条件跳过。这样当业务规则变化时，只需要增删改规则，不需要动核心代码。

举个例子，假设你做一个文本分类服务，模型输出的是各类别的概率。后处理可能需要：过滤掉概率低于阈值的类别、按概率降序排列、只保留Top-K、把类别ID映射成可读标签、组装成JSON格式。这些操作每一个都是一条规则，可以独立测试和替换。

注意：后处理层的规则一定要做超时保护。我遇到过一条正则匹配规则因为输入文本过长导致耗时飙升，拖垮了整个服务。后来我给每条规则都加了执行时间上限，超时就直接跳过并记录告警。

3.4 服务层：并发模型与优雅降级

服务层直接面对用户请求，它的稳定性决定了整个系统的可用性。我用的是FastAPI + Uvicorn的组合，异步处理请求。但异步不是万能的，如果推理层是同步阻塞的，异步框架也救不了你。

我的方案是异步接入 + 同步推理 + 线程池隔离。服务层用异步方式接收请求，然后把推理任务提交到专门的线程池执行，避免阻塞事件循环。线程池的大小根据CPU核心数和推理任务的IO密集程度来定，一般是CPU核心数的2到4倍。

优雅降级是服务层必须考虑的问题。当系统负载过高时，不能直接拒绝所有请求，而是要有策略地降级。我的降级策略分三级：

• 一级降级：关闭非核心的后处理规则，只保留最基本的输出格式化
• 二级降级：跳过批处理，直接单条推理，降低延迟
• 三级降级：返回缓存结果或默认结果，保证服务不挂

这套降级策略在实际线上环境中救过我很多次。尤其是大促期间流量暴涨的时候，如果没有降级机制，服务早就被打挂了。

4. 完整实操流程：从零搭建一个可用的AI工程系统

4.1 环境准备与依赖安装

先把基础环境搭起来。我假设你用的是Linux系统，Python版本3.9以上。

# 创建虚拟环境 python -m venv ai-eng-env source ai-eng-env/bin/activate # 安装核心依赖 pip install fastapi uvicorn numpy onnxruntime pydantic # 安装辅助工具 pip install pytest locust # 测试和压测用

这里我特意没有装PyTorch或TensorFlow，因为推理阶段我们用ONNX Runtime就够了。模型转换是在开发阶段做的事情，生产环境不需要完整的训练框架。

实操心得：依赖版本一定要锁死。我吃过亏，线上环境因为某个依赖自动升级了小版本，导致行为不一致出了故障。建议用pip freeze生成requirements.txt，并且定期更新时要做完整的回归测试。

4.2 模型转换与优化

假设你已经有一个训练好的PyTorch模型，第一步是把它转成ONNX格式：

import torch import torch.onnx # 加载模型 model = YourModel() model.load_state_dict(torch.load("model.pth")) model.eval() # 构造示例输入 dummy_input = torch.randn(1, 3, 224, 224) # 导出ONNX torch.onnx.export( model, dummy_input, "model.onnx", input_names=["input"], output_names=["output"], dynamic_axes={"input": {0: "batch_size"}, "output": {0: "batch_size"}}, opset_version=13 )

关键参数说明：dynamic_axes指定了哪个维度是动态的，这里把batch维度设为动态，这样同一个模型可以处理不同批大小的输入。opset_version建议用13或更高，低版本可能不支持某些算子。

导出之后，可以用ONNX Runtime的工具做进一步优化：

import onnxruntime as ort from onnxruntime.transformers import optimizer # 图优化 optimized_model = optimizer.optimize_model("model.onnx") optimized_model.save_model_to_file("model_optimized.onnx")

图优化会自动做算子融合、常量折叠等操作，通常能带来10%到30%的性能提升。

4.3 推理服务核心代码实现

下面是推理服务的核心代码骨架：

import asyncio import numpy as np import onnxruntime as ort from concurrent.futures import ThreadPoolExecutor from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() # 全局初始化 session = ort.InferenceSession("model_optimized.onnx") executor = ThreadPoolExecutor(max_workers=4) class InferRequest(BaseModel): data: list class InferResponse(BaseModel): result: list latency_ms: float def _infer_sync(input_array): """同步推理函数，在线程池中执行""" inputs = {session.get_inputs()[0].name: input_array} outputs = session.run(None, inputs) return outputs[0] @app.post("/predict", response_model=InferResponse) async def predict(req: InferRequest): import time start = time.time() # 预处理 try: input_array = np.array(req.data, dtype=np.float32) if input_array.ndim == 1: input_array = input_array.reshape(1, -1) except Exception as e: raise HTTPException(status_code=400, detail=f"输入格式错误: {str(e)}") # 异步提交到线程池推理 loop = asyncio.get_event_loop() try: result = await loop.run_in_executor(executor, _infer_sync, input_array) except Exception as e: raise HTTPException(status_code=500, detail=f"推理失败: {str(e)}") # 后处理 result_list = result.tolist() latency = (time.time() - start) * 1000 return InferResponse(result=result_list, latency_ms=round(latency, 2))

这段代码虽然不长，但包含了几个关键设计：异步接口 + 线程池隔离、输入校验、异常分级处理、延迟统计。你可以直接拿这个骨架去填充自己的业务逻辑。

4.4 压测与性能调优

服务写完了，别急着上线，先压测。我用的是Locust：

from locust import HttpUser, task, between import random class InferUser(HttpUser): wait_time = between(0.01, 0.05) @task def predict(self): data = [random.random() for _ in range(128)] self.client.post("/predict", json={"data": data})

压测的时候重点关注三个指标：QPS（每秒查询数）、P99延迟、错误率。我的经验值是，如果P99延迟超过200ms，用户就能明显感觉到卡顿；如果错误率超过0.1%，就需要排查原因了。

调优的方向主要有几个：调整线程池大小、调整批处理参数、优化预处理逻辑、升级硬件。每次只调一个变量，观察指标变化，找到最优配置。

实操心得：压测环境一定要和线上环境尽量一致。我曾经在开发机上压测QPS能到500，上线后实际只有80，后来发现是开发机的CPU型号更新、主频更高。硬件差异对推理性能的影响非常大。

5. 常见问题与排查技巧实录

5.1 推理结果不一致：从浮点精度到算子实现

这个问题我遇到过不止一次：同一个模型，在开发环境和服务环境上推理结果有微小差异。排查下来原因通常有三个：

第一是浮点精度差异。不同硬件平台的浮点运算实现可能不同，尤其是涉及到exp、log等超越函数的时候。解决方案是统一用float32，并且在模型导出时固定算子实现。

第二是算子版本差异。ONNX Runtime不同版本对同一个算子的实现可能有变化。解决方案是锁定ONNX Runtime版本，升级前做完整的回归测试。

第三是预处理不一致。这个最隐蔽，比如开发环境用了PIL做图像缩放，服务环境用了OpenCV，两者的插值算法不同，导致输入有微小差异。解决方案是预处理逻辑统一用同一套代码。

5.2 内存泄漏：那些年我追过的幽灵

内存泄漏是AI服务最常见的慢性病。表现是服务运行一段时间后内存持续增长，最终OOM被杀。排查内存泄漏我一般用三步法：

第一步，确认是不是真泄漏。用tracemalloc或memory_profiler监控内存变化，如果内存持续增长且不回落，基本可以确定是泄漏。

第二步，定位泄漏点。在可疑的地方打快照，对比不同时间点的对象数量和大小。Python里最常见的是全局缓存没有清理、循环引用、C扩展没有释放内存。

第三步，修复并验证。修复后要跑长时间稳定性测试，至少观察24小时。

我踩过的一个坑是：ONNX Runtime的InferenceSession如果反复创建不释放，会泄漏内存。解决方案是全局只创建一个session，复用到底。

5.3 性能突然下降：排查思路速查表

这张表是我自己总结的，基本上覆盖了80%的线上问题。遇到性能问题的时候，先查表定位方向，再深入排查具体原因。

5.4 模型更新：如何做到不停机切换

模型更新是AI服务必须面对的问题。我的方案是双缓冲切换：同时加载新旧两个模型，新模型加载完成后，通过一个原子操作切换路由指针，旧模型等待所有进行中的请求处理完再释放。

class ModelManager: def __init__(self): self.current_model = None self.pending_model = None self.lock = threading.Lock() def load_new_model(self, path): new_session = ort.InferenceSession(path) with self.lock: self.pending_model = new_session def switch(self): with self.lock: if self.pending_model is not None: old = self.current_model self.current_model = self.pending_model self.pending_model = None return old return None

这个方案的关键是切换要快、要原子，不能有请求落在空档里。实际实现的时候还要考虑旧模型的优雅退出，等正在处理的请求完成后再释放资源。

注意：模型切换前一定要做验证。我遇到过新模型文件损坏导致切换后服务不可用的情况。后来我在加载新模型后会跑一组固定的测试用例，验证通过才允许切换。

6. 我在这条路上踩过的坑和总结的经验

从零搭建AI工程系统这件事，我前后做了差不多两年，踩过的坑不计其数。有几个经验我觉得特别值得分享。

第一个是不要过早优化。我刚开始的时候花了很多时间在推理性能优化上，结果后来发现真正的瓶颈在预处理。正确的做法是先跑通全链路，用 profiling 工具找到真正的瓶颈，再针对性地优化。

第二个是监控比功能更重要。一个没有监控的AI服务就像在黑暗中开车。我现在的习惯是，每加一个功能，同步加对应的监控指标。指标不用多，但一定要覆盖输入、输出、耗时、错误四个维度。

第三个是测试要覆盖边界情况。空输入、超长输入、非法字符、极端数值，这些边界情况在开发阶段就要测到。我吃过亏，线上因为一个空输入导致整个批次的推理全部失败。

第四个是文档和注释要写清楚。AI工程系统涉及的东西太多了，数据处理逻辑、模型版本、参数配置、降级策略，不写清楚的话，过两个月自己都看不懂。我现在要求自己每写一个模块，必须同步写清楚输入输出格式、依赖关系、异常处理策略。

最后再分享一个小技巧：用配置文件管理所有可变参数。批大小、超时时间、降级阈值、模型路径，这些全部放到配置文件里，不要硬编码在代码里。这样调参的时候不需要改代码重新部署，改配置重启就行。更进一步，可以用配置中心做热更新，连重启都省了。

这套系统后来支撑了我手上好几个项目的线上服务，最长的稳定运行了一年多没出过大故障。当然它也不是完美的，还有很多可以改进的地方，比如支持GPU推理、支持多模型编排、支持更复杂的降级策略。但作为一套从零构建的AI工程基础框架，它已经足够扎实了。如果你也在做类似的事情，希望这些经验能帮你少走一些弯路。