1. 从零搭建AI工程体系为什么我劝你别急着调包很多人第一次接触AI工程脑子里想的都是“找个开源模型跑起来再说”。我刚开始也这样clone一个仓库pip install一堆依赖跑通一个demo就觉得自己会了。结果真到了要上线一个能扛住真实流量的AI服务时才发现自己连最基本的工程化思维都没有——模型加载慢、显存泄漏、推理延迟忽高忽低、日志一团糟、版本管理全靠手动改文件名。ai-engineering-from-scratch这个标题核心不在“AI”而在“from scratch”。它讲的不是怎么调一个API而是从零开始把AI能力变成一套可维护、可扩展、可观测的工程系统。这件事的难度远比训练一个模型大得多。训练模型是实验室里的事工程化是生产环境里的事两者之间的鸿沟我见过太多团队摔进去。这篇文章适合谁看如果你已经会写Python用过PyTorch或TensorFlow但对“怎么把模型变成服务”这件事还停留在Flask包一层就完事的阶段那这篇内容就是给你写的。如果你已经做过一些AI项目但每次上线都像打仗版本混乱、性能不稳、排查问题靠猜那这篇也能帮你梳理出一套从零搭建的工程框架。我不讲虚的全是实操中踩出来的经验。2. 整体设计思路先想清楚“工程”到底解决什么问题2.1 从脚本到系统中间差的是工程约束我见过太多AI项目的起点是一个Jupyter Notebook终点也是一个Jupyter Notebook。中间所谓的“工程化”就是把Notebook里的代码复制到一个.py文件里然后写个Flask路由。这种做法在demo阶段没问题但一旦要面对真实用户问题就全暴露了。AI工程化的本质是给模型推理过程加上一系列约束输入输出要可验证、资源要可管理、行为要可观测、变更要可追溯。这四个约束对应到具体的技术选型上就是接口层、资源层、监控层和版本层。ai-engineering-from-scratch这个项目标题下的核心工作就是把这四层从零搭起来而不是依赖某个大而全的框架一键生成。为什么不用现成的MLOps平台因为从零搭建的过程才是真正理解每一层在干什么的过程。你用MLflow、用BentoML、用Triton当然可以但如果你不知道它们内部怎么处理模型加载、怎么管理显存、怎么做批处理出了问题你连日志都看不懂。我的建议是先手动搭一遍最小可用系统再根据实际瓶颈引入工具。2.2 技术选型的核心逻辑控制变量逐步替换从零搭建AI工程体系最容易犯的错误是一上来就追求“生产级”。我试过在一个日请求量不到一千的项目里上Kubernetes加Istio结果运维复杂度直接把开发效率拖垮了。后来我总结出一个原则每个阶段只解决当前最大的瓶颈其他部分用最简单的方式实现。具体来说我的选型逻辑是这样的阶段核心瓶颈选型策略典型工具单机验证模型能否跑通最简依赖直接调用Python PyTorch服务化并发请求处理轻量Web框架 进程管理FastAPI Gunicorn性能优化推理延迟与吞吐批处理 异步 缓存ONNX Runtime Redis可观测问题定位困难结构化日志 指标暴露Prometheus Grafana版本管理模型与代码不一致制品库 配置分离DVC 环境变量这个表格不是让你照搬而是让你理解每一步的“为什么”。比如为什么服务化阶段选FastAPI而不是Flask因为FastAPI原生支持异步和Pydantic校验这两点在做AI服务时太重要了。输入校验能帮你挡掉大量脏数据导致的推理异常异步能让你在等待模型推理时处理其他请求。2.3 目录结构设计让工程边界一目了然从零搭建的另一个关键点是目录结构。我见过太多项目把所有代码堆在一个文件夹里模型文件、配置文件、业务逻辑、工具函数混在一起。这种结构在初期看似方便但一旦要改一个参数你根本不知道会影响哪些地方。我推荐的结构是这样的ai-service/ ├── configs/ # 配置文件按环境分离 │ ├── base.yaml │ ├── dev.yaml │ └── prod.yaml ├── src/ │ ├── api/ # 接口层路由和请求响应模型 │ ├── core/ # 核心配置加载、日志初始化 │ ├── models/ # 模型定义与加载逻辑 │ ├── services/ # 业务逻辑编排模型调用 │ └── utils/ # 通用工具如预处理、后处理 ├── tests/ # 测试用例 ├── scripts/ # 运维脚本如模型导出、数据迁移 ├── artifacts/ # 模型文件、词表等制品 └── pyproject.toml # 依赖管理这个结构的关键在于services层。很多人会把模型调用直接写在API路由里导致路由函数又长又难测。把业务逻辑抽到services层API层只负责参数校验和响应封装这样你可以在不启动Web服务的情况下测试核心逻辑。我实测下来这种分层能让单元测试的覆盖率提升至少30%因为模型推理部分可以被mock掉。3. 核心细节解析模型加载、推理优化与资源管理3.1 模型加载别小看那几秒钟模型加载是AI服务启动时最耗时的环节。一个BERT-base模型加载到CPU上大概需要2-3秒加载到GPU上加上CUDA初始化可能要5-10秒。如果你用的是更大的模型比如LLaMA系列加载时间可能超过30秒。这期间如果服务已经开始接收请求用户会直接超时。我的做法是在服务启动时同步加载模型加载完成后再注册健康检查端点。具体来说用FastAPI的lifespan机制from contextlib import asynccontextmanager from fastapi import FastAPI ml_models {} asynccontextmanager async def lifespan(app: FastAPI): # 启动时加载模型 ml_models[text_classifier] load_model(artifacts/classifier.onnx) yield # 关闭时释放资源 ml_models.clear() app FastAPI(lifespanlifespan)这里有个细节load_model函数内部要做异常处理。如果模型文件损坏或版本不匹配服务应该启动失败并给出明确错误而不是启动成功但推理时报错。我踩过的坑是模型文件路径写错服务正常启动但第一个请求进来才报FileNotFoundError排查了半天。另一个关键点是模型加载的幂等性。如果你用Gunicorn启动多个worker每个worker都会加载一份模型。对于大模型这会导致显存爆炸。解决方案有两种一是用--preload参数让Gunicorn在主进程加载模型然后fork给worker二是用共享内存或模型服务化的方式把模型推理独立成一个进程worker通过IPC调用。前者简单但只适合CPU推理后者复杂但更适合GPU场景。3.2 推理优化批处理与异步的取舍推理性能优化的核心就两个字批处理。GPU的并行计算能力意味着一次处理16个样本和一次处理1个样本的耗时差不多。但批处理引入了一个新问题延迟。如果你等够16个请求再一起推理第一个请求可能要等几百毫秒。我的经验是根据业务场景选择策略。如果是离线批量处理直接攒够一个batch再推理吞吐量最大化。如果是在线服务用动态批处理dynamic batching设置一个最大等待时间比如10毫秒超时或者攒够batch size就触发推理。实现动态批处理不需要自己写调度器可以用onnxruntime的I/O Binding配合异步推理或者用Triton Inference Server的dynamic batching功能。但如果你想从零理解原理可以自己用asyncio.Queue实现一个简易版本import asyncio class BatchProcessor: def __init__(self, model, max_batch_size16, max_wait0.01): self.model model self.max_batch_size max_batch_size self.max_wait max_wait self.queue asyncio.Queue() self.batch_task None async def process(self, input_data): future asyncio.Future() await self.queue.put((input_data, future)) if self.batch_task is None or self.batch_task.done(): self.batch_task asyncio.create_task(self._batch_loop()) return await future async def _batch_loop(self): batch [] futures [] try: while len(batch) self.max_batch_size: item await asyncio.wait_for( self.queue.get(), timeoutself.max_wait ) batch.append(item[0]) futures.append(item[1]) except asyncio.TimeoutError: pass if batch: results self.model(batch) for future, result in zip(futures, results): future.set_result(result)这段代码的关键在于asyncio.wait_for的超时控制。max_wait设置太小批处理效果差设置太大延迟高。我实测下来对于文本分类任务10毫秒是一个比较平衡的值。对于生成式任务因为单次推理时间长可以适当增大到50毫秒。3.3 资源管理显存泄漏的排查与预防GPU显存泄漏是AI服务最头疼的问题之一。表现是服务运行一段时间后显存占用越来越高最终OOM崩溃。原因通常有几个PyTorch的缓存没有释放、中间变量被意外持有、CUDA上下文没有正确清理。预防显存泄漏我总结了几条硬规则推理代码必须用torch.no_grad()包裹禁止在推理时构建计算图。如果用了torch.cuda.empty_cache()要知道它只是释放缓存不解决根本问题。对于动态形状的输入尽量固定输入尺寸或者用torch.jit.trace导出静态图。定期用nvidia-smi或pynvml监控显存设置告警阈值。排查显存泄漏的工具我推荐torch.cuda.memory_summary()它能打印出显存分配的详细情况。另外tracemalloc可以追踪Python层面的内存分配虽然不能直接看显存但能帮你找到持有大对象的代码位置。4. 实操过程从零搭建一个可用的AI服务4.1 环境准备与依赖锁定第一步永远是环境。我强烈建议用pyproject.toml加uv或poetry来管理依赖不要用requirements.txt裸奔。原因很简单AI项目的依赖冲突太常见了torch、transformers、onnxruntime之间的版本兼容性是个雷区。# 用uv创建虚拟环境并安装依赖 uv venv .venv source .venv/bin/activate uv pip install fastapi uvicorn onnxruntime pydantic pyyaml锁定版本后把uv.lock或poetry.lock提交到版本控制。这样别人clone你的项目能复现出一模一样的环境。我踩过的坑是本地用torch 2.0服务器上是torch 1.13模型加载直接报错排查了一下午。4.2 配置加载与日志初始化配置和日志是工程化的地基。配置要支持多环境日志要结构化。我用pydantic-settings来管理配置from pydantic_settings import BaseSettings class Settings(BaseSettings): model_path: str artifacts/model.onnx max_batch_size: int 16 max_wait_ms: int 10 log_level: str INFO class Config: env_file .env env_prefix AI_SERVICE_日志用structlog或标准库的logging加JSON formatter。关键是要把请求ID、模型版本、推理耗时这些字段打进去。这样出问题时你能快速过滤出某个请求的完整链路。import logging import json class JsonFormatter(logging.Formatter): def format(self, record): log_data { timestamp: self.formatTime(record), level: record.levelname, message: record.getMessage(), request_id: getattr(record, request_id, None), model_version: getattr(record, model_version, None), } return json.dumps(log_data)4.3 模型导出与推理封装训练好的模型不能直接用于生产需要先导出成推理友好的格式。PyTorch模型导出ONNX是常见做法import torch model MyModel() model.load_state_dict(torch.load(checkpoint.pt)) model.eval() dummy_input torch.randn(1, 3, 224, 224) torch.onnx.export( model, dummy_input, artifacts/model.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}, output: {0: batch_size}}, opset_version14, )导出时要注意dynamic_axes的设置。如果你不确定输入形状是否固定就加上动态轴。但动态轴会影响推理性能因为ONNX Runtime无法做形状相关的优化。我的建议是如果业务场景中输入长度变化不大直接固定形状性能提升很明显。推理封装层要处理几件事输入预处理、模型调用、输出后处理。这三步要分开写方便单独测试。预处理里最常见的坑是归一化参数不一致训练时用ImageNet的均值和方差推理时忘了加结果准确率暴跌。4.4 接口层实现与请求校验FastAPI的Pydantic模型是请求校验的利器。定义好输入输出的schemaFastAPI会自动生成文档并做类型检查from pydantic import BaseModel, Field from typing import List class PredictRequest(BaseModel): text: str Field(..., min_length1, max_length512) top_k: int Field(default3, ge1, le10) class PredictResponse(BaseModel): labels: List[str] scores: List[float] model_version: str latency_ms: float这里Field的约束很重要。min_length1能挡掉空字符串max_length512能防止超长输入导致推理时间失控。我见过一个服务因为没限制输入长度被一个用户提交了10万字的文本直接把GPU显存打满。接口实现里要记录推理耗时并把它返回给客户端。这样客户端能感知服务性能你也能在监控里看到P99延迟。import time app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): start time.perf_counter() result await service.predict(request.text, request.top_k) latency (time.perf_counter() - start) * 1000 logger.info(prediction, extra{latency_ms: latency}) return PredictResponse( labelsresult.labels, scoresresult.scores, model_versionsettings.model_version, latency_mslatency, )4.5 容器化与部署脚本容器化不是为了时髦是为了环境一致性。Dockerfile要分层构建把依赖安装和代码复制分开利用缓存加速构建FROM python:3.11-slim WORKDIR /app # 先复制依赖文件安装依赖 COPY pyproject.toml uv.lock ./ RUN pip install uv uv sync --frozen # 再复制代码和模型 COPY src/ ./src/ COPY artifacts/ ./artifacts/ COPY configs/ ./configs/ EXPOSE 8000 CMD [uvicorn, src.api.main:app, --host, 0.0.0.0, --port, 8000]注意模型文件不要打进镜像。模型动辄几百MB打进镜像会让镜像体积爆炸推送和拉取都很慢。正确做法是把模型放在对象存储或共享卷里容器启动时挂载或下载。如果模型不大小于100MB打进镜像也可以接受但要用.dockerignore排除掉训练用的checkpoint。5. 常见问题与排查技巧实录5.1 推理结果不稳定从随机性到确定性问题现象同一个输入多次请求返回的结果不一样。原因通常是模型没有切换到eval模式Dropout和BatchNorm还在训练状态。解决方案很简单加载模型后立刻调用model.eval()。但如果你用的是ONNX导出时就要确保模型是eval状态否则导出的图里会包含Dropout节点。另一个原因是预处理中的随机增强没有关闭。训练时用随机裁剪、随机翻转做数据增强推理时必须关掉。我踩过的坑是在预处理函数里写了个if random.random() 0.5的分支训练和推理共用同一个函数结果推理时也随机翻转准确率直接掉了一半。5.2 服务启动慢模型加载与依赖导入的优化问题现象服务启动要等30秒以上健康检查一直失败。排查思路先看是模型加载慢还是依赖导入慢。在main.py开头加时间戳打印出每个阶段的耗时。依赖导入慢的常见原因是transformers或torch的懒加载。这些库在import时会做很多初始化工作。解决方案是把模型加载放在lifespan里但把import torch放在文件顶部让Python在启动时并行处理。另外用python -X importtime可以分析每个模块的导入耗时。模型加载慢的优化手段用ONNX格式替代PyTorch原生格式ONNX Runtime的加载速度通常快2-3倍。如果模型特别大可以考虑用内存映射文件memory-mapped file加载减少启动时的IO等待。5.3 显存不足批处理大小与模型精度的权衡问题现象服务运行一段时间后OOM。排查步骤先用nvidia-smi看显存占用曲线如果呈锯齿状上升说明有泄漏如果一直很高说明批处理大小设置过大。显存不足的解决方案按优先级排序减小max_batch_size这是最直接有效的。用FP16代替FP32推理显存占用减半精度损失通常很小。用ONNX Runtime的arena配置限制显存池大小。如果模型支持用动态量化dynamic quantization把权重转成INT8。我实测下来FP16推理在大多数CV和NLP任务上精度损失不到0.5%但显存占用和推理速度都有明显改善。不过要注意有些操作在FP16下会溢出比如softmax的指数运算需要用torch.float32做中间计算。5.4 常见问题速查表问题现象可能原因排查方法解决方案服务启动失败模型文件缺失或损坏检查artifacts/目录和文件MD5重新导出模型校验文件完整性推理结果全相同模型未切换到eval模式打印模型training属性加载后调用model.eval()延迟忽高忽低批处理等待时间不稳定查看日志中的latency_ms分布调整max_wait_ms增加预热请求显存持续增长计算图未释放或缓存未清理用torch.cuda.memory_summary()确保torch.no_grad()定期empty_cache()请求超时输入过长或模型过大记录输入长度和推理耗时限制输入长度优化模型或升级硬件日志缺失请求ID日志上下文未传递检查中间件是否注入request_id用contextvars传递请求上下文5.5 独家避坑技巧预热与降级预热是AI服务上线前必须做的。模型第一次推理往往比后续慢很多因为CUDA内核需要编译、内存池需要分配。我的做法是在lifespan里加载模型后立刻用几条典型输入跑一遍推理把预热做掉。这样服务真正接收流量时延迟是稳定的。降级策略是另一个保命手段。当GPU显存不足或推理超时时服务应该能自动降级到CPU推理或者返回一个默认结果并记录告警。实现方式可以用try/except包裹推理调用捕获RuntimeError后切换到备用模型。备用模型可以是小模型或者是一个基于规则的简单分类器。我踩过最惨的坑是一个服务没有降级逻辑GPU OOM后整个服务崩溃所有请求都失败。后来加了降级虽然CPU推理慢但至少服务可用用户不会看到500错误。6. 版本管理与持续迭代让每次变更都可追溯6.1 模型版本与代码版本的解耦AI工程和传统软件工程最大的区别在于模型是数据产物代码是逻辑产物两者的版本节奏不一致。你可能一周改三次代码但模型一个月才更新一次。如果把模型文件直接提交到Git仓库会变得巨大且难以管理。我的做法是用DVCData Version Control管理模型文件Git只存一个.dvc指针文件。这样模型版本和代码版本可以独立演进但又能通过dvc checkout精确复现某个组合。具体操作dvc init dvc add artifacts/model.onnx git add artifacts/model.onnx.dvc .gitignore git commit -m add model v1.0当模型更新时重新dvc add并提交新的.dvc文件。部署时CI/CD流水线根据Git commit找到对应的.dvc文件再dvc pull拉取模型。这样你永远知道线上跑的是哪个模型版本。6.2 配置分离与环境变量注入配置不能硬编码在代码里。我见过一个项目把模型路径写死在service.py里换环境要改代码重新构建镜像。正确做法是用环境变量注入配置代码只读环境变量。import os MODEL_PATH os.environ.get(MODEL_PATH, artifacts/model.onnx) MAX_BATCH_SIZE int(os.environ.get(MAX_BATCH_SIZE, 16))在Kubernetes里这些环境变量通过ConfigMap注入。在Docker Compose里通过environment字段注入。这样同一个镜像可以在不同环境运行只需要改环境变量。6.3 灰度发布与A/B测试模型更新不能一刀切。新模型上线前应该先让一小部分流量走新模型对比新旧模型的指标。实现方式可以用请求头里的model_version字段做路由app.post(/predict) async def predict(request: PredictRequest, x_model_version: str Header(None)): if x_model_version v2: result await service_v2.predict(request.text) else: result await service_v1.predict(request.text) return result灰度期间要重点监控新模型的延迟、错误率和业务指标。如果新模型延迟明显升高或者业务指标下降立刻回滚。回滚只需要把流量切回旧版本不需要重新部署。7. 监控与可观测性让问题自己暴露出来7.1 指标暴露Prometheus与自定义指标AI服务的监控不能只看CPU和内存。你需要暴露推理相关的指标请求量、延迟分布、批处理大小、模型加载时间、显存占用。用prometheus_client库可以轻松暴露这些指标from prometheus_client import Histogram, Counter, Gauge INFERENCE_LATENCY Histogram( inference_latency_ms, Inference latency in milliseconds, buckets[10, 50, 100, 200, 500, 1000, 2000], ) REQUEST_COUNT Counter( inference_requests_total, Total inference requests, [model_version, status], ) GPU_MEMORY Gauge( gpu_memory_used_mb, GPU memory used in MB, )Histogram的buckets设置很关键。如果你的P99延迟在200毫秒左右buckets要覆盖到500毫秒以上否则P99会落在Inf桶里算出来不准。我一般会先跑一段时间看实际延迟分布再调整buckets。7.2 日志聚合与链路追踪结构化日志要集中收集方便搜索和告警。我用Loki加Grafana做日志聚合成本低查询快。关键是在日志里带上request_id这样你能把一个请求的所有日志串起来。链路追踪用OpenTelemetry它能自动埋点FastAPI和ONNX Runtime的调用。配置好后你可以在Jaeger里看到一个请求从进入API到模型推理再到返回的完整链路每个阶段的耗时一目了然。这对于定位“到底是网络慢还是模型慢”这类问题特别有用。7.3 告警规则什么情况下该叫醒你告警不能太多否则会麻木。我设置的告警规则只有三条P99延迟超过1秒持续5分钟。错误率超过1%持续5分钟。GPU显存占用超过90%持续10分钟。前两条是业务告警说明服务出了问题。第三条是资源告警说明需要扩容或优化。其他指标比如请求量下降可以做成看板但不告警因为可能是正常的业务波动。告警通道用Alertmanager路由到不同的接收端。P0告警打电话P1告警发消息P2告警发邮件。AI服务的告警我一般定在P1因为模型推理出问题通常不会导致整个系统不可用但需要尽快处理。8. 从零搭建的扩展方向下一步可以做什么这套从零搭建的AI工程体系最小可用版本大概需要两三天就能跑起来。但它的扩展空间很大你可以根据业务需求逐步增强。第一个扩展方向是模型服务化。把模型推理独立成一个微服务用gRPC通信这样API层和推理层可以独立扩缩容。推理层可以用Triton Inference Server它支持多模型、多框架、动态批处理性能比手写的Python服务好很多。第二个扩展方向是特征存储。如果你的模型需要实时特征比如用户画像、上下文信息就需要一个低延迟的特征存储。Redis或Aerospike是常见选择。关键是要保证训练和推理的特征计算逻辑一致否则会出现训练-推理偏差。第三个扩展方向是自动化流水线。用GitHub Actions或GitLab CI把模型导出、测试、构建镜像、部署串成一条流水线。每次代码提交自动跑单元测试每次模型更新自动触发集成测试。这样能把人为失误降到最低。我个人在实际操作中的体会是从零搭建的价值不在于最终的系统有多复杂而在于你亲手处理过每一个环节的细节。模型加载慢、显存泄漏、批处理延迟、版本混乱这些问题你只有自己踩过一遍才能真正理解那些成熟框架为什么要那样设计。等你把最小系统跑通了再去用MLflow或BentoML你会发现它们解决的每一个问题你都似曾相识。