1. 为什么我要从零手搓一套AI工程流水线

第一次听到“ai-engineering-from-scratch”这个说法，是在一个做推荐系统的老哥群里。当时有人甩了个链接，说现在外面讲AI工程化的内容，要么是调包侠式的三行代码跑通一个demo，要么是云厂商的软文，真正把数据清洗、特征存储、模型训练、推理服务、监控告警这一整条链路拆开揉碎讲清楚的，少得可怜。我盯着屏幕想了半天，觉得这话没毛病。过去两年我参与过三个从零到一的AI项目，踩过的坑能写满一个笔记本，但市面上确实缺少一份“从裸机开始，把AI工程该有的东西一件件搭起来”的实操记录。

所以这个项目标题“ai-engineering-from-scratch”对我来说，不是一个课程名字，而是一种做事的方法论。它的核心意思是：不依赖任何现成的MLOps平台，不用那些一键部署的SaaS工具，从一台干净的Linux服务器开始，用最基础的开源组件，把数据管道、训练框架、模型仓库、推理API、监控面板全部手动串起来。这样做的好处非常直接——你会被迫理解每一个环节的输入输出、资源消耗和失败模式。坏处也很明显，前期搭建慢，文档要自己写，出了问题没人甩锅。

这篇文章适合谁看？如果你已经会用PyTorch或TensorFlow训练模型，但不知道训练完之后怎么让模型稳定地对外提供服务；如果你听过Feature Store、Model Registry、Drift Detection这些词，但没亲手搭过；如果你所在团队正准备把AI能力从Jupyter Notebook里搬出来，变成真正的后端服务，那这篇内容就是写给你的。我会按照实际搭建顺序，从硬件选型、系统配置、数据层、训练层、服务层到监控层，把每个环节的关键决策、参数计算和踩坑经验都摊开讲。全文基于我最近一次在四台二手服务器上复现整套流程的真实记录，所有命令和配置都经过验证，你可以直接抄作业。

2. 整体架构设计与技术选型逻辑

2.1 为什么不用Kubeflow和MLflow全家桶

很多人一上来就问，你搞AI工程化为什么不用Kubeflow？我的回答很实在：Kubeflow的抽象层太厚了。当你只有四台机器、两个GPU、一个兼职运维的时候，Kubeflow带来的复杂度远大于它解决的问题。我试过在测试环境部署Kubeflow Pipelines，光是Istio和Knative的配置就花了两天，最后发现一个简单的数据预处理任务，在Kubeflow里要写一堆YAML，而在裸机上就是一个Python脚本加cron。MLflow我也用过，它的Tracking和Model Registry确实好用，但当你需要自定义模型签名、处理非标准输入输出格式时，MLflow的约束就会变成障碍。

所以我的选型原则是：每一层只引入一个必要的组件，组件之间通过文件系统或HTTP API解耦。数据层用DVC做版本控制，训练层用PyTorch Lightning封装训练循环，模型仓库直接用文件系统加Git LFS，推理服务用FastAPI加ONNX Runtime，监控用Prometheus加Grafana。这些组件每一个都可以单独替换，不会牵一发动全身。下面这张表是我对比过的几套方案，你可以根据团队规模直接参考。

2.2 硬件配置与成本计算

我这次用的是一台二手Dell R730xd，配置如下：双路E5-2680 v4（共28核56线程），128GB DDR4 ECC内存，两块Tesla P40 24GB显卡，系统盘是两块480GB SSD做RAID1，数据盘是四块4TB SAS硬盘做RAID5。这套配置在二手市场大概一万二左右，加上显卡一共不到两万。为什么选P40？因为它的24GB显存对于7B参数以下的模型推理和微调足够用，而且价格只有RTX 3090的一半。当然P40的算力只有12 TFLOPS FP32，训练大模型会慢，但做工程化验证完全够。

这里有个关键计算：显存需求估算。假设你要部署一个7B参数的模型做推理，FP16精度下模型权重占14GB，加上KV Cache和中间激活值，大概需要18-20GB。P40的24GB刚好卡在线上。如果你要微调，用LoRA的话，显存需求可以降到16GB左右。但如果你要全量微调7B模型，至少需要4张A100 40GB，这不是个人能承受的。所以我的建议是：工程化验证阶段，用7B以下模型加LoRA微调，硬件成本控制在两万以内。等流程跑通了，再申请公司资源上大模型。

2.3 网络与存储规划

四台机器之间用千兆交换机连接，实测内网传输速度在110MB/s左右。这个速度对于传输模型文件（通常几百MB到几GB）来说可以接受，但如果你要频繁地在节点间同步数据集，千兆就会成为瓶颈。我的做法是：数据集只存一份，放在NFS服务器上，所有节点通过NFS挂载。NFS服务器就是那台R730xd，数据盘做RAID5后可用空间约11TB。训练时，数据加载器直接从NFS读取，虽然比本地SSD慢，但避免了数据同步的麻烦。实测下来，对于ImageNet级别的数据集，NFS读取速度能到80MB/s，配合PyTorch的DataLoader多进程预取，GPU利用率能保持在85%以上。

存储分层是这样的：系统盘SSD放操作系统和Python环境，数据盘RAID5放原始数据集和特征文件，另外挂载一块1TB NVMe SSD做训练时的临时缓存。为什么要单独加NVMe？因为RAID5的随机读写性能很差，而训练时DataLoader会频繁读取小文件，如果直接读RAID5，IOPS会成为瓶颈。我把当前训练任务的数据集预先拷贝到NVMe上，训练完再删掉，这样GPU利用率能从60%提升到90%以上。这个细节在大多数教程里都不会提，但实际影响非常大。

3. 数据管道搭建与特征工程实操

3.1 用DVC做数据版本控制

数据版本控制是AI工程化的第一步，也是最容易被忽略的一步。我见过太多团队用data_final_v2.csv这种方式管理数据，最后没人知道哪个文件对应哪次实验。DVC的原理很简单：它把大文件存在本地或远程存储，只在Git里保存一个.dvc文件记录哈希值。这样你切换Git分支时，dvc checkout就能把对应版本的数据拉出来。

安装和初始化命令如下：

pip install dvc dvc-s3 mkdir ai-engineering && cd ai-engineering git init dvc init dvc remote add -d myremote /mnt/nfs/dvc-storage

这里我把远程存储设在NFS上，因为团队其他成员也需要访问。如果你是一个人开发，直接设在本地也行。添加数据集的命令是：

dvc add data/raw/train.csv git add data/raw/train.csv.dvc data/raw/.gitignore git commit -m "add raw training data" dvc push

注意dvc add之后，原始文件会被移动到.dvc/cache目录，原地只留下一个.dvc文件。很多人第一次用会吓一跳，以为数据丢了。其实数据还在，只是被DVC接管了。如果你想恢复原始文件，运行dvc checkout即可。

注意：DVC的缓存目录默认在项目根目录的.dvc/cache下，如果数据集很大，这个目录会迅速膨胀。建议在dvc init之后立刻修改缓存位置到数据盘：dvc cache dir /mnt/data/dvc-cache。

3.2 特征存储的轻量级实现

Feature Store是AI工程化里的热词，但商业化的Feature Store（比如Feast、Tecton）对于小团队来说太重了。我的做法是用Parquet文件加SQLite元数据实现一个最小可用的特征存储。具体来说，每个特征组存为一个Parquet文件，文件名包含特征组名称和版本号，比如user_features_v1.parquet。元数据表记录特征组的名称、版本、创建时间、列名、数据类型和统计信息。

建表SQL如下：

CREATE TABLE feature_registry ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, version INTEGER NOT NULL, file_path TEXT NOT NULL, columns TEXT NOT NULL, dtypes TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, row_count INTEGER, UNIQUE(name, version) );

写入特征的Python代码：

import pandas as pd import sqlite3 import json def save_feature_group(df, name, version, base_path): file_path = f"{base_path}/{name}_v{version}.parquet" df.to_parquet(file_path, index=False) conn = sqlite3.connect(f"{base_path}/feature_registry.db") cursor = conn.cursor() cursor.execute(""" INSERT INTO feature_registry (name, version, file_path, columns, dtypes, row_count) VALUES (?, ?, ?, ?, ?, ?) """, ( name, version, file_path, json.dumps(list(df.columns)), json.dumps({col: str(dtype) for col, dtype in df.dtypes.items()}), len(df) )) conn.commit() conn.close()

读取特征时，先查SQLite拿到文件路径，再用pandas读取Parquet。这个方案的好处是零依赖、易调试，坏处是没有自动的线上/线下一致性保证。但对于小团队来说，一致性靠代码规范来保证比靠工具来保证更现实。我们约定：所有特征计算逻辑必须写在一个独立的Python模块里，训练和推理都调用同一个函数。

3.3 数据清洗的标准化流程

数据清洗是脏活累活，但必须标准化。我总结了一个五步流程：缺失值处理、异常值检测、类型转换、去重、采样。每一步都有对应的工具和参数。

缺失值处理：对于数值型特征，用中位数填充；对于类别型特征，用众数填充；对于缺失率超过80%的列，直接删除。这里有个经验：不要用均值填充，因为均值受异常值影响大。中位数更稳健。

异常值检测：用IQR方法，即计算第一四分位数Q1和第三四分位数Q3，定义正常范围为[Q1 - 1.5IQR, Q3 + 1.5IQR]。超出这个范围的值标记为异常，但不要直接删除，而是先分析原因。我遇到过很多次，所谓的“异常值”其实是真实的高价值样本，删掉会严重影响模型效果。

类型转换：确保所有数值列是float32或int64，所有类别列是category类型。这样做的好处是节省内存，同时避免后续训练时出现类型错误。对于类别列，还要检查类别数量，如果超过1000个，考虑用哈希编码或目标编码替代独热编码。

去重：用df.drop_duplicates(subset=key_columns)，key_columns是能唯一标识一条记录的列组合。注意，去重前要先做类型转换，否则1和1.0会被当成不同的值。

采样：如果数据量太大，用分层采样保持类别分布。sklearn.model_selection.train_test_split的stratify参数就是干这个的。

实操心得：数据清洗的每一步都要记录日志，包括处理前的行数、处理后的行数、删除的列名、填充的统计量。这些日志在排查模型效果下降时非常有用。我习惯把日志存成JSON文件，和数据集放在一起。

4. 模型训练与实验管理

4.1 PyTorch Lightning训练模板

PyTorch Lightning的好处是把训练循环、验证循环、优化器配置、学习率调度这些样板代码都封装好了，你只需要关注模型结构和数据加载。下面是我常用的训练模板，你可以直接复制到项目里。

import pytorch_lightning as pl import torch from torch import nn from torch.utils.data import DataLoader, Dataset class MyModel(pl.LightningModule): def __init__(self, input_dim, hidden_dim, output_dim, lr=1e-3): super().__init__() self.save_hyperparameters() self.layer1 = nn.Linear(input_dim, hidden_dim) self.layer2 = nn.Linear(hidden_dim, output_dim) self.relu = nn.ReLU() self.dropout = nn.Dropout(0.3) self.criterion = nn.CrossEntropyLoss() def forward(self, x): x = self.relu(self.layer1(x)) x = self.dropout(x) return self.layer2(x) def training_step(self, batch, batch_idx): x, y = batch logits = self(x) loss = self.criterion(logits, y) self.log('train_loss', loss, prog_bar=True) return loss def validation_step(self, batch, batch_idx): x, y = batch logits = self(x) loss = self.criterion(logits, y) acc = (logits.argmax(dim=1) == y).float().mean() self.log('val_loss', loss, prog_bar=True) self.log('val_acc', acc, prog_bar=True) def configure_optimizers(self): optimizer = torch.optim.AdamW(self.parameters(), lr=self.hparams.lr) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50) return [optimizer], [scheduler]

训练入口脚本：

from pytorch_lightning import Trainer from pytorch_lightning.callbacks import ModelCheckpoint, EarlyStopping from pytorch_lightning.loggers import TensorBoardLogger checkpoint_callback = ModelCheckpoint( dirpath='checkpoints/', filename='model-{epoch:02d}-{val_loss:.4f}', save_top_k=3, monitor='val_loss', mode='min' ) early_stop_callback = EarlyStopping( monitor='val_loss', patience=5, mode='min' ) logger = TensorBoardLogger('logs/', name='my_experiment') trainer = Trainer( max_epochs=100, gpus=1, callbacks=[checkpoint_callback, early_stop_callback], logger=logger, precision=16, accumulate_grad_batches=4 ) model = MyModel(input_dim=128, hidden_dim=256, output_dim=10) trainer.fit(model, train_dataloader, val_dataloader)

这里有几个关键参数需要解释。precision=16开启混合精度训练，能节省约40%显存，速度提升20%左右。accumulate_grad_batches=4表示梯度累积4步再更新一次参数，等效于把batch size扩大4倍。这两个参数配合使用，可以让你在单张24GB显卡上训练更大的模型。

4.2 超参数搜索的实用策略

超参数搜索不需要一上来就用Optuna或Ray Tune，手动网格搜索在小规模实验里效率更高。我的做法是：先粗后细，先关键后次要。第一轮只调学习率，范围从1e-5到1e-2，按对数均匀取5个值。第二轮固定最佳学习率，调batch size和dropout率。第三轮调网络层数和隐藏单元数。

这里有个计算：假设每个实验跑10个epoch，每个epoch耗时5分钟，那么一个实验就是50分钟。如果第一轮跑5个学习率，就是4个多小时。所以不要一次性跑太多实验，先用1%的数据子集快速筛选，再用全量数据验证。我通常用5%的数据跑第一轮，这样每个实验只要3分钟，5个实验15分钟就能出结果。

TensorBoard是查看实验结果的标配工具。启动命令：

tensorboard --logdir logs/ --port 6006 --host 0.0.0.0

然后在浏览器里打开http://服务器IP:6006，就能看到所有实验的loss曲线、准确率曲线和学习率变化。我习惯在TensorBoard里给每个实验加标签，比如lr=0.001_bs=64，这样对比起来一目了然。

4.3 模型版本管理与回滚

模型版本管理我用的是最笨但最可靠的方法：文件系统加Git LFS。每次训练完成后，把checkpoint文件重命名为{模型名}_{日期}_{git commit短哈希}_{验证集指标}.ckpt，然后提交到Git LFS。比如text_classifier_20240520_a3f2c1_val_acc_0.923.ckpt。这样从文件名就能看出模型版本、训练日期、代码版本和效果指标。

回滚时，只需要根据文件名找到对应的checkpoint，加载即可。加载代码：

import torch def load_model(checkpoint_path, model_class, **kwargs): model = model_class(**kwargs) checkpoint = torch.load(checkpoint_path, map_location='cpu') model.load_state_dict(checkpoint['state_dict']) model.eval() return model

注意：PyTorch Lightning保存的checkpoint包含state_dict、hyper_parameters、optimizer_states等信息。如果你只需要推理，加载state_dict就够了。但如果你要继续训练，需要保留完整的checkpoint。

我还会在模型仓库里放一个MODEL_CARD.md，记录这个模型的训练数据、超参数、评估指标、已知偏差和适用场景。这个习惯是从Google的Model Cards论文里学来的，在实际工作中非常有用，尤其是当团队人员流动时，新人能快速了解每个模型的来龙去脉。

5. 推理服务部署与性能优化

5.1 FastAPI加ONNX Runtime的推理服务

训练完模型，下一步是把它变成API。我选FastAPI是因为它异步性能好、自动生成文档、类型检查严格。ONNX Runtime是因为它比原生PyTorch推理快20%-30%，而且不依赖PyTorch环境，部署包更小。

先把PyTorch模型导出为ONNX格式：

import torch import torch.onnx model = MyModel(input_dim=128, hidden_dim=256, output_dim=10) model.load_state_dict(torch.load('checkpoints/best.ckpt')['state_dict']) model.eval() dummy_input = torch.randn(1, 128) 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维度是动态的。如果不设置，导出的模型只能处理固定batch size，线上服务会很不灵活。

FastAPI服务代码：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import onnxruntime as ort import numpy as np app = FastAPI(title="AI Inference Service") session = ort.InferenceSession("model.onnx", providers=["CUDAExecutionProvider"]) class PredictRequest(BaseModel): features: list[float] class PredictResponse(BaseModel): label: int confidence: float @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): if len(request.features) != 128: raise HTTPException(status_code=400, detail="Expected 128 features") input_array = np.array([request.features], dtype=np.float32) outputs = session.run(None, {"input": input_array}) logits = outputs[0][0] exp_logits = np.exp(logits - np.max(logits)) probs = exp_logits / exp_logits.sum() label = int(np.argmax(probs)) confidence = float(probs[label]) return PredictResponse(label=label, confidence=confidence) @app.get("/health") async def health(): return {"status": "ok"}

启动命令：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2

--workers 2表示启动两个工作进程，因为ONNX Runtime在推理时会释放GIL，多进程能更好地利用多核CPU。但注意，如果你的模型在GPU上推理，多个进程会竞争显存，这时候应该用单进程加异步批处理。

5.2 动态批处理与延迟优化

线上服务的请求是零散的，如果每个请求都单独推理，GPU利用率会很低。动态批处理的思想是：把短时间内到达的多个请求合并成一个batch，一起推理，然后拆分结果返回。这样能显著提升吞吐量。

实现动态批处理需要一个小型的请求队列。我用asyncio.Queue实现了一个简单的批处理调度器：

import asyncio import numpy as np class BatchScheduler: def __init__(self, session, max_batch_size=32, max_wait_ms=10): self.session = session self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms self.queue = asyncio.Queue() self.batch_task = None async def predict(self, features): future = asyncio.Future() await self.queue.put((features, future)) if self.batch_task is None or self.batch_task.done(): self.batch_task = asyncio.create_task(self._process_batch()) return await future async def _process_batch(self): await asyncio.sleep(self.max_wait_ms / 1000) batch = [] futures = [] while not self.queue.empty() and len(batch) < self.max_batch_size: features, future = self.queue.get_nowait() batch.append(features) futures.append(future) if not batch: return input_array = np.array(batch, dtype=np.float32) outputs = self.session.run(None, {"input": input_array}) logits_batch = outputs[0] for i, future in enumerate(futures): logits = logits_batch[i] exp_logits = np.exp(logits - np.max(logits)) probs = exp_logits / exp_logits.sum() label = int(np.argmax(probs)) confidence = float(probs[label]) future.set_result({"label": label, "confidence": confidence})

这个调度器的逻辑是：请求到达后先放入队列，然后等待最多10毫秒，或者等到队列里有32个请求，就触发一次批量推理。实测下来，在QPS为50的情况下，动态批处理能把GPU利用率从30%提升到75%，平均延迟从45毫秒降到18毫秒。

实操心得：max_wait_ms这个参数需要根据你的延迟要求来调。如果要求P99延迟低于50毫秒，就设5-10毫秒；如果追求最大吞吐量，可以设50毫秒。我一般从10毫秒开始调，观察延迟和吞吐的曲线，找到拐点。

5.3 服务健康检查与优雅关闭

线上服务必须要有健康检查接口，否则负载均衡器不知道你的服务是否还活着。/health接口返回200就表示健康，返回500就表示不健康。但简单的返回{"status": "ok"}还不够，最好加上模型加载状态和GPU显存使用情况。

import torch @app.get("/health") async def health(): gpu_available = torch.cuda.is_available() gpu_memory = torch.cuda.memory_allocated() / 1024**3 if gpu_available else 0 return { "status": "ok", "model_loaded": session is not None, "gpu_available": gpu_available, "gpu_memory_gb": round(gpu_memory, 2) }

优雅关闭是指服务收到SIGTERM信号后，先停止接收新请求，等正在处理的请求完成后再退出。FastAPI默认支持这个行为，但你需要确保推理任务不是无限循环的。我通常设置一个30秒的超时，超过就强制退出。

import signal import sys def graceful_shutdown(signum, frame): print("Received shutdown signal, waiting for ongoing requests...") sys.exit(0) signal.signal(signal.SIGTERM, graceful_shutdown)

6. 监控告警与线上问题排查

6.1 Prometheus指标暴露

监控是AI工程化的最后一环，也是最容易偷懒的一环。我的做法是用prometheus_client库在FastAPI服务里暴露指标，然后用Prometheus抓取，Grafana展示。

需要监控的指标分四类：请求量、延迟、错误率、资源使用。请求量用Counter，延迟用Histogram，错误率用Counter加标签，资源使用用Gauge。

from prometheus_client import Counter, Histogram, Gauge, generate_latest from fastapi import Response REQUEST_COUNT = Counter('inference_requests_total', 'Total inference requests', ['method', 'endpoint', 'status']) REQUEST_LATENCY = Histogram('inference_request_latency_seconds', 'Request latency', ['endpoint']) GPU_MEMORY = Gauge('gpu_memory_usage_gb', 'GPU memory usage in GB') MODEL_LOADED = Gauge('model_loaded', 'Whether model is loaded') @app.middleware("http") async def monitor_requests(request, call_next): start_time = time.time() response = await call_next(request) latency = time.time() - start_time REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path, status=response.status_code).inc() REQUEST_LATENCY.labels(endpoint=request.url.path).observe(latency) return response @app.get("/metrics") async def metrics(): if torch.cuda.is_available(): GPU_MEMORY.set(torch.cuda.memory_allocated() / 1024**3) MODEL_LOADED.set(1 if session else 0) return Response(generate_latest(), media_type="text/plain")

Prometheus配置：

scrape_configs: - job_name: 'inference-service' scrape_interval: 15s static_configs: - targets: ['localhost:8000']

Grafana面板我通常会放四个图：QPS曲线、P50/P95/P99延迟曲线、错误率曲线、GPU显存曲线。这四个图能覆盖90%的线上问题排查场景。

6.2 数据漂移检测的简易方案

数据漂移是指线上输入数据的分布和训练数据不一致，导致模型效果下降。检测漂移不需要复杂的统计检验，用群体稳定性指标（PSI）就够了。PSI的计算方法是：把训练数据的每个特征分成10个分箱，计算线上数据在每个分箱的占比，然后套公式。

import numpy as np def calculate_psi(expected, actual, bins=10): breakpoints = np.linspace(0, 100, bins + 1) expected_percents = np.percentile(expected, breakpoints) actual_percents = np.percentile(actual, breakpoints) expected_counts = np.histogram(expected, bins=expected_percents)[0] / len(expected) actual_counts = np.histogram(actual, bins=expected_percents)[0] / len(actual) expected_counts = np.where(expected_counts == 0, 0.0001, expected_counts) actual_counts = np.where(actual_counts == 0, 0.0001, actual_counts) psi_values = (actual_counts - expected_counts) * np.log(actual_counts / expected_counts) return np.sum(psi_values)

PSI小于0.1表示分布稳定，0.1到0.25表示轻微漂移，大于0.25表示显著漂移。我每天定时跑一次漂移检测，如果PSI超过0.25就发告警。告警渠道用邮件加企业微信机器人，消息里附上漂移最严重的特征名和PSI值。

注意：PSI对分箱数量敏感，bins=10是经验值。如果你的特征取值范围很大，可以先用np.log变换再计算PSI。另外，PSI只能检测单变量漂移，多变量漂移需要用KL散度或MMD，但那些计算复杂度高，小团队用PSI足够了。

6.3 常见线上问题速查表

下面这张表是我过去两年遇到过的典型线上问题，以及对应的排查思路和解决方法。你可以把它打印出来贴在工位上。

这张表里的每一条都是我实际踩过的坑。比如“服务频繁重启”那条，我遇到过是因为在FastAPI的全局作用域里加载了一个不断增长的缓存字典，每次请求都往里塞数据，最后内存耗尽被OOM Killer杀掉。解决方法很简单，把缓存改成LRU策略，限制最大条目数。

7. 我踩过的坑和最后分享几个技巧

第一个坑是NFS挂载导致的训练卡死。有一次训练到第3个epoch突然卡住，GPU利用率掉到0，日志也没有报错。排查了半天发现是NFS服务器重启了，而训练节点上的NFS挂载没有自动恢复。解决方法是在/etc/fstab里加上hard,intr选项，这样NFS不可用时会中断而不是无限等待。另外，训练脚本里要加超时机制，比如DataLoader的timeout参数设为30秒。

第二个坑是ONNX导出时的动态维度问题。我一开始没设dynamic_axes，导出的模型只能处理batch size=1的输入。线上服务用动态批处理时直接报错。后来加上dynamic_axes参数才解决。但注意，不是所有算子都支持动态维度，比如某些自定义的PyTorch算子导出后会变成静态的。导出后一定要用onnxruntime跑一遍不同batch size的输入验证。

第三个坑是Prometheus指标标签基数爆炸。我一开始把request_id作为标签加到了Counter里，结果每个请求都生成一个新的时间序列，Prometheus内存迅速涨到几十GB。后来把request_id去掉，只保留method、endpoint、status三个标签，内存就稳定了。记住一个原则：标签的取值组合不要超过1000个。

最后分享一个小技巧：用Makefile管理所有常用命令。AI工程化涉及的命令很多，训练、导出、部署、监控，每个环节都有好几条命令。我把它们都写进Makefile，这样新人进来只需要make train、make deploy就能跑起来，不用翻文档。

.PHONY: train export deploy monitor train: python train.py --config configs/train.yaml export: python export_onnx.py --checkpoint checkpoints/best.ckpt --output model.onnx deploy: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 monitor: tensorboard --logdir logs/ --port 6006 --host 0.0.0.0

这个Makefile我放在项目根目录，配合README里的说明，基本能做到“克隆下来就能跑”。对于小团队来说，这种朴素的工程化手段比任何高大上的平台都管用。