1. 从零手搓AI工程为什么我不建议你直接调包很多人一听到“AI工程”这四个字第一反应就是打开某个云平台拖几个组件调几个API然后跑通一个Demo就觉得自己已经掌握了。我刚开始也是这么想的直到有一次线上推理服务在凌晨两点崩了日志里全是显存溢出的报错而我对着那堆封装好的接口完全不知道从哪下手排查。那一刻我才意识到只会调包的人永远只能停留在“能用”的层面一旦出了问题连问题的边界在哪都摸不到。ai-engineering-from-scratch这个标题核心不在于“AI”而在于“from scratch”。它代表的是一种从底层理解并亲手构建AI工程链路的思路。这不是让你去手写CUDA核函数也不是让你从零实现一个Transformer而是说你要对数据怎么流入、模型怎么加载、推理怎么调度、服务怎么暴露、监控怎么做这些环节有足够的掌控力。适合谁来参考我认为有三类人一是刚入行做算法工程或MLOps的开发者想搞清楚一个AI服务从代码到上线到底经历了什么二是有一定Python基础但一直停留在Notebook阶段的同学想把自己的模型真正跑成一个服务三是被各种框架封装坑过、想回头补课的老手。这篇文章我会围绕一条完整的链路来展开从环境隔离与依赖管理到数据管道的构建再到模型推理的封装、服务的暴露、性能的压测最后是日志与监控的接入。每一步我都会解释为什么这么选、为什么不那么选以及我在实际操作中踩过的坑。你不需要有很深的数学背景但需要能看懂基本的Python代码和命令行操作。读完你至少能自己搭出一个可运行、可观测、可迭代的AI推理服务骨架而不是一个只能在本地跑通的玩具。2. 环境隔离与依赖管理别让版本冲突毁掉你的周末2.1 为什么虚拟环境不是可选项而是必选项我见过太多人直接在系统Python里pip install然后某天发现两个项目依赖的同一个库版本不兼容一个要numpy 1.24另一个要numpy 1.26结果就是两个项目轮流崩。AI工程涉及的东西尤其杂深度学习框架、CUDA运行时、各种图像处理库、Web框架它们之间的版本约束像一张蜘蛛网。虚拟环境的核心价值不是“干净”而是“可复现”。你今天跑通的代码三个月后换台机器还能跑通靠的就是环境隔离加依赖锁定。我的习惯是用conda管理大环境用pip管理包。原因很简单CUDA相关的运行时用conda装省心很多而纯Python包用pip更灵活。具体操作上我会先创建一个指定Python版本的环境conda create -n ai-eng python3.10 -y conda activate ai-eng选3.10而不是3.12是因为很多深度学习框架对最新Python版本的支持总是滞后半年左右3.10是目前兼容性最稳的版本之一。这一步没什么技术含量但选错版本后面可能连框架都装不上。2.2 依赖锁定的正确姿势装完包之后很多人会pip freeze requirements.txt然后就不管了。这个做法有个隐患freeze会把所有间接依赖都写进去包括那些跟你的系统或CUDA版本强相关的包换台机器可能直接装不上。我的做法是维护两个文件一个是requirements.in只写我直接依赖的顶层包比如torch、fastapi、uvicorn另一个是用pip-compile生成的requirements.txt里面是锁定了精确版本的全量依赖。pip install pip-tools pip-compile requirements.in --output-file requirements.txt pip-sync requirements.txtpip-sync的好处是它会把你环境里多余的包删掉保证环境和锁定文件完全一致。这个习惯在多人协作时尤其重要否则你永远不知道同事的“在我机器上能跑”到底差在哪。注意如果你的项目要用GPUtorch的安装命令一定要去官网查对应CUDA版本的命令不要直接pip install torch否则很可能装成CPU版本跑起来慢得让你怀疑人生。2.3 容器化从“能跑”到“到处能跑”虚拟环境解决了Python层面的隔离但解决不了系统库和CUDA驱动的差异。当你需要把服务部署到另一台机器时Docker几乎是绕不开的。我一般会写一个多阶段构建的Dockerfile第一阶段装依赖并编译第二阶段只拷贝运行时需要的东西这样镜像体积能小很多。FROM nvidia/cuda:12.1-runtime-ubuntu22.04 AS base RUN apt-get update apt-get install -y python3.10 python3-pip WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]这里选runtime而不是devel镜像是因为部署阶段不需要编译工具链runtime镜像能省下好几个G。基础镜像的CUDA版本必须和宿主机驱动兼容这个兼容关系表在NVIDIA官方文档里有选之前一定对一下否则容器起来就是CUDA driver version is insufficient。3. 数据管道模型还没跑数据先把你绊倒3.1 推理场景下的数据管道和训练有什么不同训练时的数据管道追求的是吞吐量和随机性推理时的数据管道追求的是低延迟和确定性。这个区别决定了你在推理场景下不能照搬训练那套DataLoader。训练时你可以开一堆worker预取数据推理时每个请求都是独立的你更需要的是一个轻量、快速、无状态的预处理函数。我通常会把预处理逻辑写成一个纯函数输入是原始请求数据输出是模型能吃的张量。这个函数里只做必要的操作解码、缩放、归一化、转张量。不要在这里做任何IO操作比如读文件、查数据库那些应该在上游完成。import numpy as np import torch def preprocess(image_bytes, target_size(224, 224)): # 假设输入是编码后的图像字节 import cv2 arr np.frombuffer(image_bytes, dtypenp.uint8) img cv2.imdecode(arr, cv2.IMREAD_COLOR) img cv2.resize(img, target_size) img cv2.cvtColor(img, cv2.COLOR_BGR2RGB) tensor torch.from_numpy(img).permute(2, 0, 1).float() / 255.0 mean torch.tensor([0.485, 0.456, 0.406]).view(3, 1, 1) std torch.tensor([0.229, 0.224, 0.225]).view(3, 1, 1) return (tensor - mean) / std这段代码里有个细节归一化的均值和标准差必须和训练时用的完全一致差一点都会导致精度下降。我见过有人推理时用了ImageNet的均值但训练时用的是自己算的结果模型表现莫名其妙地差查了两天才发现是这里的问题。3.2 批处理与动态形状的处理推理服务面临的一个现实问题是请求是一个一个来的但GPU喜欢批量处理。如果你每个请求都单独跑一次模型GPU利用率会低得可怜。解决办法是做一个微批处理队列请求进来先放进队列攒够一定数量或者等一小段时间再一起送进模型。import asyncio from collections import deque class BatchQueue: def __init__(self, max_batch8, max_wait0.01): self.queue deque() self.max_batch max_batch self.max_wait max_wait self.lock asyncio.Lock() async def add(self, item): async with self.lock: self.queue.append(item) if len(self.queue) self.max_batch: return self._flush() await asyncio.sleep(self.max_wait) async with self.lock: if item in self.queue: return self._flush() return None这个逻辑看起来简单但实际写的时候要注意并发安全否则会出现同一个请求被处理两次或者丢失的情况。max_wait这个参数需要根据你的延迟要求来调设得太大会增加尾延迟设得太小又攒不够批次。我一般从10毫秒开始试根据压测结果再调整。3.3 数据校验别让脏数据进到模型里线上服务最怕的就是异常输入。用户传了一张损坏的图片、一个空字符串、一个超大文件如果你的预处理函数没有防御轻则报错返回500重则把服务搞崩。我的做法是在预处理之前加一层校验检查数据类型、检查尺寸范围、检查解码是否成功。def validate_input(image_bytes, max_size_mb10): if not isinstance(image_bytes, bytes): raise ValueError(input must be bytes) if len(image_bytes) max_size_mb * 1024 * 1024: raise ValueError(input too large) if len(image_bytes) 0: raise ValueError(empty input) return True校验失败时返回明确的错误码和错误信息而不是让异常直接抛到框架层。这样前端能知道是用户输入的问题而不是服务挂了。这个习惯能帮你省下大量排查时间因为线上大部分“服务异常”其实都是输入异常。4. 模型加载与推理封装把模型当成一个需要照顾的组件4.1 模型加载时机与显存管理模型什么时候加载这个问题看起来简单但选错了会影响服务的启动速度和资源占用。我见过有人在每个请求里都加载一次模型结果QPS低到个位数。正确的做法是在服务启动时加载一次然后常驻显存。但这里有个坑如果你用uvicorn的多worker模式每个worker都会加载一份模型显存直接翻倍。所以推理服务通常建议单worker加异步或者用专门的模型服务框架来管理。import torch from transformers import AutoModelForImageClassification class ModelWrapper: def __init__(self, model_path, devicecuda): self.device torch.device(device if torch.cuda.is_available() else cpu) self.model AutoModelForImageClassification.from_pretrained(model_path) self.model.to(self.device) self.model.eval() torch.set_grad_enabled(False) def predict(self, batch_tensor): batch_tensor batch_tensor.to(self.device) with torch.inference_mode(): outputs self.model(batch_tensor) return outputs.logits.cpu()torch.inference_mode()比torch.no_grad()更彻底它会关闭自动求导相关的所有机制推理速度能快几个百分点。另外model.eval()一定要调否则BatchNorm和Dropout层的行为会和训练时一样结果完全不可控。4.2 推理封装的边界在哪里封装模型的时候要明确哪些逻辑放在封装层哪些放在外面。我的原则是封装层只负责“把张量变成张量”也就是输入张量到输出张量的映射。后处理、业务逻辑、数据库操作都放在外面。这样封装层可以复用换一个模型只需要改封装层业务代码不动。class InferenceEngine: def __init__(self, model_wrapper, postprocess_fn): self.model model_wrapper self.postprocess postprocess_fn def run(self, batch_tensor): logits self.model.predict(batch_tensor) return self.postprocess(logits)这个分层看起来多了一层但实际维护起来会轻松很多。我试过把后处理也塞进模型封装里结果后来换模型的时候后处理逻辑和模型逻辑缠在一起改一处崩三处。4.3 显存碎片与长稳运行服务跑一段时间后显存占用越来越高最后OOM这是推理服务最常见的问题之一。原因通常是显存碎片PyTorch的缓存分配器会缓存已分配的显存如果请求的形状变化频繁缓存块大小不一致就会产生碎片。解决办法有两个一是尽量固定输入形状比如统一resize到224x224二是定期调用torch.cuda.empty_cache()但这个操作会同步GPU频繁调用会影响性能我一般只在显存占用超过阈值时触发。def maybe_clear_cache(threshold_gb0.9): allocated torch.cuda.memory_allocated() / 1024**3 reserved torch.cuda.memory_reserved() / 1024**3 if reserved threshold_gb * torch.cuda.get_device_properties(0).total_memory / 1024**3: torch.cuda.empty_cache()这个阈值要根据你的显卡总显存来定我一般设在总显存的85%到90%之间。太低会频繁触发影响性能太高又起不到预防作用。5. 服务暴露与性能压测从本地跑通到扛住流量5.1 Web框架选型FastAPI还是Flask推理服务的Web层我几乎无脑选FastAPI。原因有三个原生异步支持、自动生成OpenAPI文档、基于Pydantic的请求校验。Flask虽然生态更成熟但同步模型在高并发下需要开很多线程而线程切换的开销在推理场景下不可忽视。FastAPI的异步能让你在等待GPU计算的时候处理其他请求吞吐量提升很明显。from fastapi import FastAPI, UploadFile, HTTPException from pydantic import BaseModel app FastAPI() engine None app.on_event(startup) async def load_model(): global engine engine InferenceEngine(ModelWrapper(model_path), postprocess_fn) app.post(/predict) async def predict(file: UploadFile): data await file.read() try: validate_input(data) except ValueError as e: raise HTTPException(status_code400, detailstr(e)) tensor preprocess(data).unsqueeze(0) result engine.run(tensor) return {result: result.tolist()}on_event(startup)里加载模型保证服务启动时只加载一次。请求体用UploadFile接收文件比用base64编码再解码效率高因为省去了编码解码的开销。5.2 压测别用“感觉”判断性能很多人上线前不压测上线后凭感觉说“好像有点慢”。压测的目的是拿到具体数字P50延迟、P99延迟、QPS、错误率。我常用的工具是locust因为它能用Python写压测脚本和我的技术栈一致。from locust import HttpUser, task, between class InferenceUser(HttpUser): wait_time between(0.01, 0.05) task def predict(self): with open(test.jpg, rb) as f: self.client.post(/predict, files{file: f})压测的时候要关注P99而不是平均值因为用户体验是由最慢的那部分请求决定的。如果P99延迟是平均延迟的5倍以上说明有长尾问题可能是批处理等待、显存回收、或者某个请求触发了慢路径。5.3 并发模型的选择同步、异步还是多进程FastAPI默认是异步的但如果你的推理代码里有阻塞操作比如同步的GPU调用它会阻塞事件循环。解决办法是用run_in_executor把阻塞操作放到线程池里执行。import asyncio from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers4) app.post(/predict) async def predict(file: UploadFile): data await file.read() tensor preprocess(data).unsqueeze(0) loop asyncio.get_event_loop() result await loop.run_in_executor(executor, engine.run, tensor) return {result: result.tolist()}线程池的大小需要根据GPU的能力来定。线程太多会导致GPU上下文切换频繁反而降低吞吐线程太少又无法充分利用GPU。我一般从2开始试逐步增加到QPS不再上升为止。6. 日志、监控与故障排查上线只是开始6.1 结构化日志让日志能被机器读懂print和logging.info(done)在排查问题时基本没用。我要求日志必须是结构化的JSON包含时间戳、请求ID、阶段、耗时、状态。这样可以用日志系统做聚合和告警。import logging import json import time logger logging.getLogger(inference) def log_request(request_id, stage, duration_ms, status, extraNone): record { ts: time.time(), request_id: request_id, stage: stage, duration_ms: duration_ms, status: status, } if extra: record.update(extra) logger.info(json.dumps(record))请求ID要贯穿整个链路从请求进来到返回每个阶段都带上同一个ID。这样出问题的时候你可以用ID把所有相关日志串起来快速定位是哪个环节慢了或者错了。6.2 关键监控指标别只看CPU和内存推理服务的监控除了CPU、内存、GPU利用率这些基础指标还要关注几个业务指标请求队列长度、批处理大小分布、预处理耗时、推理耗时、后处理耗时。这些指标能告诉你瓶颈在哪。比如队列长度持续增长说明处理速度跟不上请求速度批处理大小一直很小说明max_wait设得太短或者请求量不够。我一般用Prometheus的Python客户端暴露指标然后在Grafana里做面板。from prometheus_client import Histogram, Counter INFERENCE_LATENCY Histogram(inference_latency_ms, inference latency, buckets[10, 50, 100, 200, 500, 1000]) REQUEST_COUNT Counter(inference_requests_total, total requests, [status]) def predict_with_metrics(tensor): start time.time() try: result engine.run(tensor) REQUEST_COUNT.labels(statussuccess).inc() return result except Exception: REQUEST_COUNT.labels(statuserror).inc() raise finally: INFERENCE_LATENCY.observe((time.time() - start) * 1000)Histogram的buckets要根据你的实际延迟分布来设设得太粗看不出分位数设得太细又浪费存储。我一般先跑一轮压测拿到延迟分布后再定buckets。6.3 常见故障的排查链路线上出问题时最忌讳的是瞎猜。我总结了一个排查顺序先看错误率再看延迟再看资源。错误率突增先查日志里的异常类型是输入问题还是模型问题延迟突增但错误率正常查队列长度和批处理大小看是不是请求量上来了资源指标异常查GPU显存和利用率看是不是有内存泄漏或者碎片。有一次我遇到服务每隔几小时就OOM一次查了半天代码没发现问题最后用torch.cuda.memory_summary()打印显存快照发现是某个异常输入触发了形状变化导致缓存块大小不一致碎片逐渐累积。后来加了输入形状的强制校验问题就消失了。这个经历告诉我显存问题不一定是代码逻辑问题也可能是输入分布问题。7. 迭代与扩展这套骨架还能怎么用这套从零搭建的骨架最大的价值不是它现在能跑什么模型而是它的可扩展性。换模型只需要改ModelWrapper里的加载逻辑和preprocess函数服务层和监控层完全不用动。加一个新的推理接口只需要加一个路由和对应的预处理函数。这种分层设计让迭代成本变得很低。如果你想把服务做得更完善有几个方向可以继续深入。一是模型版本管理支持灰度发布和回滚这需要在ModelWrapper外面加一层版本路由。二是请求优先级给不同来源的请求打标签高优先级的走单独的队列。三是自动扩缩容根据队列长度动态调整worker数量。这些都是在现有骨架上加东西而不是推倒重来。我个人在实际操作中的体会是从零搭建的过程虽然比调包慢但你对每个环节的掌控力是完全不同的。出了问题你知道去哪找性能不够你知道往哪优化要加功能你知道在哪加。这种掌控力才是AI工程师和调包侠之间的真正差距。