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 技术栈选型不追新只选对的在技术选型上我的原则是稳定优先、生态成熟、社区活跃。具体来说层级选型理由数据接入Python FastAPI异步性能足够开发效率高生态好预处理NumPy 自研Pipeline避免过度依赖框架可控性强推理ONNX Runtime跨平台性能优秀支持多种硬件后端后处理纯Python 规则引擎业务逻辑多变需要快速迭代服务层Uvicorn Nginx成熟稳定运维成本低这里重点说一下为什么推理层我选了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_size32, max_wait_ms50): 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_version13 )关键参数说明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_workers4) 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_modelInferResponse) async def predict(req: InferRequest): import time start time.time() # 预处理 try: input_array np.array(req.data, dtypenp.float32) if input_array.ndim 1: input_array input_array.reshape(1, -1) except Exception as e: raise HTTPException(status_code400, detailf输入格式错误: {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_code500, detailf推理失败: {str(e)}) # 后处理 result_list result.tolist() latency (time.time() - start) * 1000 return InferResponse(resultresult_list, latency_msround(latency, 2))这段代码虽然不长但包含了几个关键设计异步接口 线程池隔离、输入校验、异常分级处理、延迟统计。你可以直接拿这个骨架去填充自己的业务逻辑。4.4 压测与性能调优服务写完了别急着上线先压测。我用的是Locustfrom 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 性能突然下降排查思路速查表现象可能原因排查方法解决方案QPS骤降下游依赖超时检查各层耗时指标加超时熔断P99延迟飙升批处理攒批过大查看批大小分布调小最大批大小错误率上升输入数据异常采样错误请求加强输入校验内存持续增长内存泄漏内存快照对比修复泄漏点CPU利用率低IO阻塞检查线程状态异步化IO操作GPU利用率低批大小太小查看批大小分布增大批大小这张表是我自己总结的基本上覆盖了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工程基础框架它已经足够扎实了。如果你也在做类似的事情希望这些经验能帮你少走一些弯路。