企业官网建设流程全解析

1. 从零搭建AI工程体系，为什么我劝你别一上来就搞模型

“ai-engineering-from-scratch”这个标题，第一次看到的时候我以为是又一个教你调包的教程。点进去翻了翻，发现它想做的事情比调包大得多——它试图回答一个很实际的问题：如果不用那些高度封装的框架，从最底层开始搭一套能跑起来的AI工程体系，到底需要哪些东西？

这个问题在2024年之后变得特别有意义。因为大部分人的AI开发路径是这样的：装个PyTorch，pip install transformers，加载预训练权重，微调一下，部署到某个云服务上。这条路能跑通，但一旦出了问题——推理延迟突然飙升、显存莫名其妙爆掉、模型输出不稳定——很多人就懵了，因为他们不知道底层发生了什么。

“ai-engineering-from-scratch”的核心价值就在这里：它把AI工程拆解成了一条从数据到部署的完整链路，每一环都要求你理解原理，而不是只会调API。适合谁看？我觉得有三类人：一是刚入行做AI应用开发、想搞清楚“黑盒”里到底有什么的工程师；二是有后端或数据工程背景、想转AI工程方向的开发者；三是带团队的技术负责人，需要判断哪些环节可以自研、哪些可以依赖现成方案。

这篇文章我会按照这个项目的思路，把AI工程从零搭建的关键环节拆开讲。不是泛泛而谈“AI很重要”，而是具体到每一步怎么做、为什么这么做、踩过哪些坑。全文大概会覆盖环境搭建、数据处理、模型训练与推理、服务化部署、监控与迭代这几个核心模块，每个模块都会给出可操作的方案和参数选择的依据。

2. 整体设计思路：为什么“从零”不等于“重复造轮子”

2.1 先搞清楚“从零”的边界在哪里

很多人看到“from scratch”就以为要自己写矩阵乘法、手推反向传播。不是这样的。AI工程里的“从零”指的是：你清楚每一层在做什么，并且有能力在需要的时候替换掉任何一层。比如你知道PyTorch的DataLoader底层是怎么做批处理的，所以当数据加载成为瓶颈时，你能自己写一个更高效的迭代器；你知道Transformer的注意力计算复杂度是O(n²)，所以当序列变长时你知道该换什么方案。

这个项目在设计上遵循了一个原则：能用标准库就不用第三方库，能用轻量库就不用重框架。比如数据处理用NumPy和Pandas而不是Spark，模型训练用PyTorch但不用Lightning，服务化用FastAPI而不是Triton。这样做的好处是依赖少、启动快、调试方便，缺点是当规模上去之后需要自己补很多工程化的东西。

我自己的经验是，在单机单卡、数据量在百万级以下的场景里，这套“轻量从零”的方案比堆框架效率高得多。因为框架带来的抽象层在排查问题时全是障碍。你调一个DataLoader的num_workers参数，结果发现是底层共享内存的问题，这时候如果你不懂操作系统层面的东西，就只能靠试。

2.2 技术选型的三个核心考量

选型这件事，我一般看三个维度：可控性、性能、开发效率。这三个不可能同时最大化，必须做取舍。

可控性方面，PyTorch比TensorFlow好，因为它的动态图机制让调试更直观。你可以在forward函数里直接print中间张量的值，TensorFlow 1.x的静态图时代这是做不到的。性能方面，如果你要做推理优化，ONNX Runtime或者TensorRT比原生PyTorch快，但转换过程会丢失一些灵活性。开发效率方面，HuggingFace的transformers库确实省事，但它的抽象层太厚，出了问题很难定位。

“ai-engineering-from-scratch”的选择是：训练阶段用PyTorch原生，推理阶段用ONNX Runtime，服务化用FastAPI，监控用Prometheus + Grafana。这套组合的好处是每一层都有清晰的边界，你可以单独替换任何一层而不影响其他部分。

注意：不要为了“从零”而拒绝所有现成工具。用requests库发HTTP请求不丢人，自己写socket才是浪费时间。关键是你要知道requests底层用的是urllib3，连接池是怎么管理的。

2.3 目录结构设计：让项目自己说话

一个AI工程项目最容易烂掉的地方就是目录结构。我见过太多项目把所有代码堆在根目录下，train.py、model.py、utils.py、test.py混在一起，三个月后自己都找不到东西。

这个项目的目录结构是这样的：

ai-engineering-from-scratch/ ├── configs/ # 配置文件，YAML格式 ├── data/ # 数据相关 │ ├── raw/ # 原始数据，只读 │ ├── processed/ # 处理后数据 │ └── loaders/ # 数据加载代码 ├── models/ # 模型定义 ├── training/ # 训练脚本 ├── inference/ # 推理相关 ├── serving/ # 服务化代码 ├── monitoring/ # 监控配置 ├── tests/ # 测试 └── scripts/ # 运维脚本

这个结构的关键点是数据和代码分离、训练和推理分离、配置和逻辑分离。数据目录下的raw永远不修改，processed可以随时重新生成。训练和推理共享模型定义但有不同的入口。配置文件用YAML而不是Python字典，因为YAML可以被非技术人员修改。

我踩过的一个坑是：早期把预处理逻辑写在了训练脚本里，结果推理的时候发现预处理方式不一致，导致线上效果差了一大截。后来把预处理抽成独立的模块，训练和推理都调用同一个函数，问题才解决。所以数据预处理一定要独立成模块，这是血泪教训。

3. 核心细节解析：数据、模型、推理的三层拆解

3.1 数据层：从原始数据到训练样本的完整链路

数据层是整个AI工程里最脏最累但最重要的一环。模型再好，数据有问题，结果一定不行。这个项目把数据层拆成了四个步骤：采集、清洗、标注、格式化。

采集阶段，如果是结构化数据，直接用SQL从数据库拉；如果是非结构化数据（文本、图片），需要写爬虫或者对接API。这里要注意的是数据版本管理。我推荐用DVC（Data Version Control）来管理数据版本，因为它和Git集成得好，大文件存在远程存储上，本地只保留元数据。

清洗阶段，核心是处理缺失值、异常值、重复值。文本数据还要做去重、去噪、分词。这里有一个经验：不要过度清洗。我见过有人把文本里的所有标点都去掉，结果模型学出来的东西连句子边界都分不清。清洗的原则是保留对任务有用的信息，去掉纯噪声。

标注阶段，如果是监督学习，标注质量直接决定模型上限。我的建议是：先标一批小样本，训练一个基线模型，用模型来辅助标注。这叫主动学习（Active Learning），可以大幅减少标注成本。具体做法是：模型对未标注样本给出预测，人工只修正那些模型不确定的样本。

格式化阶段，就是把数据转成模型能吃的格式。对于PyTorch来说，你需要实现一个Dataset类和一个DataLoader。Dataset负责单样本的读取和预处理，DataLoader负责批处理、打乱、多进程加载。

class CustomDataset(Dataset): def __init__(self, data_path, tokenizer, max_len=512): self.data = load_data(data_path) self.tokenizer = tokenizer self.max_len = max_len def __len__(self): return len(self.data) def __getitem__(self, idx): item = self.data[idx] encoding = self.tokenizer( item['text'], max_length=self.max_len, padding='max_length', truncation=True, return_tensors='pt' ) return { 'input_ids': encoding['input_ids'].squeeze(), 'attention_mask': encoding['attention_mask'].squeeze(), 'label': torch.tensor(item['label']) }

这个Dataset类看起来简单，但有几个细节要注意。padding='max_length'会导致大量无效计算，因为短句子也被填充到了最大长度。更好的做法是用动态padding，在DataLoader的collate_fn里根据当前batch的最大长度来padding。这样能显著提升训练速度，尤其是在句子长度分布不均匀的时候。

提示：num_workers的设置不是越大越好。一般来说，num_workers等于CPU核心数就够了。设置太大反而会因为进程间通信开销导致速度下降。另外在Windows上num_workers>0可能会有问题，建议设为0。

3.2 模型层：训练一个能用的模型需要关注什么

模型层是大多数人最熟悉的部分，但也是最容易出问题的地方。这个项目在模型层的设计上强调三点：可复现、可调试、可扩展。

可复现意味着你要固定随机种子。PyTorch的随机性来自多个地方：权重初始化、数据打乱、Dropout。你需要设置torch.manual_seed、np.random.seed、random.seed，还要设置cudnn的deterministic模式。

def set_seed(seed=42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False

注意cudnn.benchmark设为False会降低一点速度，但能保证结果可复现。如果你更看重速度，可以设为True，但每次运行结果会有微小差异。

可调试意味着你要能方便地查看中间结果。我的做法是在模型的关键位置加hook，记录每一层的输出统计量（均值、方差、最大值、最小值）。这样当loss不下降时，你能快速定位是哪一层出了问题。

def register_hooks(model): stats = {} def hook_fn(name): def fn(module, input, output): if isinstance(output, torch.Tensor): stats[name] = { 'mean': output.mean().item(), 'std': output.std().item(), 'max': output.max().item(), 'min': output.min().item() } return fn for name, module in model.named_modules(): if isinstance(module, (nn.Linear, nn.Conv2d, nn.LayerNorm)): module.register_forward_hook(hook_fn(name)) return stats

可扩展意味着模型定义要模块化。不要把整个模型写在一个forward函数里，而是拆成多个子模块。这样你想换掉某一层的时候，只需要改一个地方。

训练循环的设计也有讲究。这个项目用的是标准的训练-验证-保存最佳模型的流程，但加了几个实用的功能：梯度裁剪、学习率预热、早停。

梯度裁剪是防止梯度爆炸的，特别是RNN和Transformer模型。一般设max_norm=1.0或5.0。学习率预热是在训练初期用较小的学习率，逐渐增加到设定值，避免初期震荡。早停是当验证集loss连续N个epoch不下降时停止训练，防止过拟合。

# 梯度裁剪 torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 学习率预热 from transformers import get_linear_schedule_with_warmup scheduler = get_linear_schedule_with_warmup( optimizer, num_warmup_steps=1000, num_training_steps=total_steps ) # 早停 best_val_loss = float('inf') patience = 3 counter = 0 for epoch in range(epochs): val_loss = validate(model, val_loader) if val_loss < best_val_loss: best_val_loss = val_loss torch.save(model.state_dict(), 'best_model.pt') counter = 0 else: counter += 1 if counter >= patience: print(f"Early stopping at epoch {epoch}") break

注意：早停的patience不要设太小。我见过有人设patience=1，结果模型在第二个epoch刚好遇到一个难学的batch，loss上升了一点就停了，其实再训练几个epoch效果会更好。一般设3-5比较合理。

3.3 推理层：从模型文件到可调用的服务

推理层是把训练好的模型变成别人能用的服务。这里有几个关键决策：用什么格式保存模型、用什么引擎推理、怎么处理并发请求。

模型保存格式，PyTorch原生的是.pt或.pth，但生产环境我更推荐ONNX。因为ONNX是跨平台的，可以在没有PyTorch的环境里运行，而且ONNX Runtime的推理速度通常比PyTorch快20%-50%。

导出ONNX的代码：

dummy_input = torch.randint(0, 10000, (1, 512)).to(device) torch.onnx.export( model, dummy_input, "model.onnx", input_names=['input_ids', 'attention_mask'], output_names=['logits'], dynamic_axes={ 'input_ids': {0: 'batch_size', 1: 'sequence_length'}, 'attention_mask': {0: 'batch_size', 1: 'sequence_length'}, 'logits': {0: 'batch_size'} }, opset_version=14 )

dynamic_axes很重要，它让模型能处理不同长度的输入。如果不设置，导出的模型只能处理固定长度的输入，那就很鸡肋了。

推理引擎，ONNX Runtime是首选。它的Python API很简单：

import onnxruntime as ort session = ort.InferenceSession( "model.onnx", providers=['CUDAExecutionProvider', 'CPUExecutionProvider'] ) def predict(input_ids, attention_mask): outputs = session.run( ['logits'], { 'input_ids': input_ids.numpy(), 'attention_mask': attention_mask.numpy() } ) return outputs[0]

providers的顺序很重要，CUDA在前表示优先用GPU，GPU不可用时自动回退到CPU。

并发处理，FastAPI是当前最流行的选择。它的异步特性适合IO密集型的场景，但AI推理是计算密集型的，所以要用线程池或者进程池来跑推理。

from fastapi import FastAPI from concurrent.futures import ThreadPoolExecutor import asyncio app = FastAPI() executor = ThreadPoolExecutor(max_workers=4) @app.post("/predict") async def predict_endpoint(request: PredictRequest): loop = asyncio.get_event_loop() result = await loop.run_in_executor( executor, run_inference, request.text ) return {"result": result}

max_workers的设置要看GPU显存和模型大小。一般来说，如果模型占2GB显存，GPU有16GB，那最多同时跑4-6个推理实例。设太多会导致显存溢出。

提示：FastAPI的默认超时时间可能不够，特别是模型第一次加载的时候。建议在启动时做一次warmup，用假数据跑几次推理，让CUDA kernel编译完成。

4. 实操过程：从零搭建一个文本分类服务的完整记录

4.1 环境准备与依赖安装

我用的环境是Ubuntu 22.04，Python 3.10，CUDA 11.8。Python版本不要用太新的，3.10或3.11比较稳，3.12有些库还没适配。

创建虚拟环境：

python -m venv venv source venv/bin/activate

安装核心依赖：

pip install torch==2.1.0 torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 pip install transformers==4.35.0 pip install onnx onnxruntime-gpu==1.16.0 pip install fastapi uvicorn pip install pandas numpy scikit-learn pip install prometheus-client

这里有几个版本要注意。onnxruntime-gpu的版本要和CUDA版本匹配，1.16.0对应CUDA 11.8。transformers的版本和torch版本也有对应关系，4.35.0配torch 2.1.0是验证过的组合。

安装完之后验证一下：

import torch print(torch.__version__) print(torch.cuda.is_available()) print(torch.cuda.get_device_name(0))

如果cuda.is_available()返回False，检查一下CUDA驱动和CUDA Toolkit的版本是否匹配。用nvidia-smi看驱动支持的CUDA版本，用nvcc --version看Toolkit版本，两者要兼容。

4.2 数据准备与预处理脚本

我用的数据集是一个中文情感分类数据集，大概5万条样本，正面负面各一半。原始数据是CSV格式，两列：text和label。

预处理脚本做了这几件事：去重、去空、长度过滤、划分训练验证测试集。

import pandas as pd from sklearn.model_selection import train_test_split def preprocess(input_path, output_dir): df = pd.read_csv(input_path) # 去重 df = df.drop_duplicates(subset=['text']) # 去空 df = df.dropna(subset=['text', 'label']) df = df[df['text'].str.strip() != ''] # 长度过滤，去掉太短和太长的 df['length'] = df['text'].str.len() df = df[(df['length'] >= 5) & (df['length'] <= 500)] # 划分数据集 train_df, temp_df = train_test_split( df, test_size=0.2, random_state=42, stratify=df['label'] ) val_df, test_df = train_test_split( temp_df, test_size=0.5, random_state=42, stratify=temp_df['label'] ) train_df.to_csv(f'{output_dir}/train.csv', index=False) val_df.to_csv(f'{output_dir}/val.csv', index=False) test_df.to_csv(f'{output_dir}/test.csv', index=False) print(f"Train: {len(train_df)}, Val: {len(val_df)}, Test: {len(test_df)}")

stratify参数很重要，它保证划分后各类别的比例和原始数据一致。如果不加，可能训练集里正面样本多、验证集里负面样本多，导致评估结果不可靠。

长度过滤的阈值5和500是根据数据分布定的。我先画了长度分布直方图，发现大部分样本在10-200字之间，超过500字的只有不到1%。这些超长样本可能是噪声，去掉对模型影响不大。

4.3 模型训练与调参记录

我用的基座模型是BERT-base-chinese，这是一个12层、768隐藏维度、110M参数的模型。对于5万条样本的二分类任务，这个规模足够了。更大的模型容易过拟合，更小的模型效果可能不够。

训练参数：

训练脚本的核心逻辑：

from transformers import BertTokenizer, BertForSequenceClassification from torch.utils.data import DataLoader from transformers import AdamW, get_linear_schedule_with_warmup tokenizer = BertTokenizer.from_pretrained('bert-base-chinese') model = BertForSequenceClassification.from_pretrained( 'bert-base-chinese', num_labels=2 ).to(device) train_dataset = CustomDataset('data/processed/train.csv', tokenizer, max_len=128) train_loader = DataLoader( train_dataset, batch_size=32, shuffle=True, num_workers=4 ) optimizer = AdamW(model.parameters(), lr=2e-5, weight_decay=0.01) total_steps = len(train_loader) * 5 scheduler = get_linear_schedule_with_warmup( optimizer, num_warmup_steps=500, num_training_steps=total_steps ) for epoch in range(5): model.train() total_loss = 0 for batch in train_loader: input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['label'].to(device) outputs = model( input_ids=input_ids, attention_mask=attention_mask, labels=labels ) loss = outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() scheduler.step() optimizer.zero_grad() total_loss += loss.item() avg_loss = total_loss / len(train_loader) print(f"Epoch {epoch+1}, Loss: {avg_loss:.4f}") # 验证 val_acc = evaluate(model, val_loader) print(f"Val Accuracy: {val_acc:.4f}")

实际跑下来，第一个epoch loss从0.69降到0.35，验证准确率0.89。第二个epoch loss降到0.22，准确率0.92。第三个epoch准确率0.93。第四个epoch准确率0.93，loss 0.18。第五个epoch准确率0.93，loss 0.16。早停在第5个epoch后触发，因为验证准确率不再提升。

最终测试集准确率0.92，F1分数0.91。这个结果对于5万条样本的二分类任务来说算正常水平。如果想到0.95以上，需要更多数据或者数据增强。

注意：BERT微调的学习率很敏感。我试过1e-5，收敛太慢；试过5e-5，loss震荡严重。2e-5是经过多次实验比较稳的值。如果你的数据集更小，学习率要更小，比如1e-5。

4.4 导出ONNX并搭建推理服务

训练完之后，把最好的模型导出为ONNX：

model.load_state_dict(torch.load('best_model.pt')) model.eval() dummy_input = torch.randint(0, 10000, (1, 128)).to(device) dummy_mask = torch.ones(1, 128).to(device) torch.onnx.export( model, (dummy_input, dummy_mask), "model.onnx", input_names=['input_ids', 'attention_mask'], output_names=['logits'], dynamic_axes={ 'input_ids': {0: 'batch_size', 1: 'sequence_length'}, 'attention_mask': {0: 'batch_size', 1: 'sequence_length'}, 'logits': {0: 'batch_size'} }, opset_version=14 )

导出之后验证一下ONNX模型的输出和PyTorch是否一致：

import onnxruntime as ort import numpy as np session = ort.InferenceSession("model.onnx", providers=['CUDAExecutionProvider']) # PyTorch输出 with torch.no_grad(): pt_output = model(dummy_input, dummy_mask).logits.cpu().numpy() # ONNX输出 onnx_output = session.run( ['logits'], { 'input_ids': dummy_input.cpu().numpy(), 'attention_mask': dummy_mask.cpu().numpy() } )[0] print(np.allclose(pt_output, onnx_output, atol=1e-4))

atol=1e-4是容忍度，因为浮点运算顺序不同会有微小差异。如果差异很大，说明导出有问题。

推理服务用FastAPI：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import onnxruntime as ort from transformers import BertTokenizer import numpy as np app = FastAPI() tokenizer = BertTokenizer.from_pretrained('bert-base-chinese') session = ort.InferenceSession( "model.onnx", providers=['CUDAExecutionProvider', 'CPUExecutionProvider'] ) class PredictRequest(BaseModel): text: str class PredictResponse(BaseModel): label: int confidence: float @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): if not request.text.strip(): raise HTTPException(status_code=400, detail="Text cannot be empty") encoding = tokenizer( request.text, max_length=128, padding='max_length', truncation=True, return_tensors='np' ) outputs = session.run( ['logits'], { 'input_ids': encoding['input_ids'].astype(np.int64), 'attention_mask': encoding['attention_mask'].astype(np.int64) } ) logits = outputs[0][0] probs = np.exp(logits) / np.exp(logits).sum() label = int(np.argmax(probs)) confidence = float(probs[label]) return PredictResponse(label=label, confidence=confidence)

启动服务：

uvicorn serving.app:app --host 0.0.0.0 --port 8000 --workers 1

workers设为1，因为ONNX Runtime本身会管理线程。如果设多个workers，每个worker都会加载一份模型，显存会翻倍。

4.5 监控与性能指标采集

服务上线之后，你需要知道它跑得怎么样。Prometheus + Grafana是标准方案。

在FastAPI里加Prometheus中间件：

from prometheus_client import Counter, Histogram, generate_latest from fastapi import Response import time REQUEST_COUNT = Counter( 'predict_requests_total', 'Total predict requests', ['method', 'endpoint', 'status'] ) REQUEST_LATENCY = Histogram( 'predict_request_latency_seconds', 'Request latency', ['endpoint'] ) @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(): return Response(generate_latest(), media_type="text/plain")

需要监控的核心指标：

延迟P99比平均延迟更重要，因为用户感知到的是最慢的那次请求。如果P99超过500ms，说明有长尾请求需要优化。

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

5.1 训练阶段的典型问题

问题一：loss不下降或者震荡严重。

这是最常见的问题。排查顺序是：先看数据有没有问题（标签是否正确、有没有脏数据），再看学习率是不是太大，再看模型初始化是不是有问题。

我遇到过一次loss一直震荡，最后发现是数据里有重复样本，同一个batch里出现了多次，导致梯度方向来回拉扯。去重之后问题解决。

问题二：显存溢出（OOM）。

OOM的原因有很多：batch_size太大、模型太大、中间激活值太多。解决方法按优先级：减小batch_size、用梯度累积、用混合精度训练、用梯度检查点。

梯度累积的代码：

accumulation_steps = 4 for i, batch in enumerate(train_loader): outputs = model(**batch) loss = outputs.loss / accumulation_steps loss.backward() if (i + 1) % accumulation_steps == 0: optimizer.step() scheduler.step() optimizer.zero_grad()

这样等效于batch_size扩大了4倍，但显存占用不变。

问题三：验证集准确率远低于训练集准确率。

这是过拟合的典型表现。解决方法：增加数据、加Dropout、加权重衰减、早停。如果这些都不行，说明模型太大了，换个小模型。

5.2 推理阶段的典型问题

问题一：ONNX模型输出和PyTorch不一致。

最常见的原因是opset_version不匹配。BERT模型建议用opset 14或更高。另外检查一下tokenizer的padding方式是否一致，PyTorch用max_length，ONNX也要用max_length。

问题二：推理延迟高。

先看是不是第一次推理慢（CUDA kernel编译），如果是，做warmup。如果每次都慢，看GPU利用率，如果利用率低，说明是CPU瓶颈（数据预处理、tokenization），可以把tokenization也放到GPU上或者用更快的tokenizer。

问题三：并发请求时显存溢出。

ONNX Runtime默认会为每个推理请求分配显存。如果并发高，显存会爆。解决方法是设置显存限制：

options = ort.SessionOptions() options.intra_op_num_threads = 1 options.inter_op_num_threads = 1 session = ort.InferenceSession( "model.onnx", sess_options=options, providers=[('CUDAExecutionProvider', { 'gpu_mem_limit': 4 * 1024 * 1024 * 1024, # 4GB 'arena_extend_strategy': 'kSameAsRequested' })] )

gpu_mem_limit限制每个session的显存使用量，arena_extend_strategy控制显存分配策略。

5.3 服务化阶段的典型问题

问题一：服务启动慢。

FastAPI启动时加载模型和tokenizer需要时间。如果模型大，可能要几十秒。解决方法是用lazy loading，第一次请求时才加载模型。但这样第一次请求会很慢。更好的方法是启动时加载，但用健康检查接口让负载均衡器知道服务还没准备好。

问题二：内存泄漏。

Python的垃圾回收机制在处理大对象时可能不及时。如果发现内存持续增长，检查是不是有全局变量在累积。比如把每次请求的结果都存在一个列表里，时间长了就爆了。

问题三：日志太多或太少。

日志太多会拖慢服务，太少出问题没法排查。我的做法是：正常请求只记录关键信息（请求ID、耗时、结果标签），错误请求记录完整堆栈。用结构化日志（JSON格式），方便后续分析。

import logging import json logger = logging.getLogger(__name__) def log_request(request_id, text, latency, result): logger.info(json.dumps({ 'request_id': request_id, 'text_length': len(text), 'latency_ms': latency * 1000, 'label': result.label, 'confidence': result.confidence }))

5.4 常见问题速查表

提示：排查问题时，二分法是最有效的。先确认是数据问题还是模型问题，再确认是训练问题还是推理问题。不要一上来就改模型结构，大部分问题出在数据和配置上。

6. 迭代与扩展：从能跑到好用还有多远

6.1 模型迭代的节奏控制

一个AI服务上线只是开始，后续的迭代才是重头戏。我的经验是：上线后第一周每天看一次指标，第二周每三天看一次，之后每周看一次。重点看的是线上数据和训练数据的分布差异。

如果发现线上请求的文本长度分布和训练数据不一样，比如线上有很多超长文本，那就需要补充这类数据重新训练。如果发现某些类别的准确率明显偏低，那就针对性地标注更多这类样本。

模型更新的频率不要太高。每次更新都有风险，可能引入新的问题。我一般是一个月更新一次，除非有紧急问题。更新前一定要做A/B测试，用10%的流量跑新模型，对比一周的数据再决定是否全量。

6.2 性能优化的几个方向

当QPS上去之后，性能优化就变得重要了。几个有效的方向：

量化。把FP32模型转成INT8，推理速度能提升2-3倍，精度损失通常在1%以内。ONNX Runtime支持动态量化：

from onnxruntime.quantization import quantize_dynamic, QuantType quantize_dynamic( "model.onnx", "model_quantized.onnx", weight_type=QuantType.QUInt8 )

批处理。如果请求量大，可以把多个请求攒成一个batch一起推理。这样能充分利用GPU的并行能力。但会增加延迟，需要权衡。

缓存。如果有很多重复请求，可以加一层缓存。用请求文本的hash作为key，结果作为value。缓存命中率通常能到20%-30%。

模型蒸馏。用大模型教小模型，小模型推理快但效果接近大模型。适合对延迟敏感的场景。

6.3 从单模型到多模型的扩展

当业务变复杂时，一个模型可能不够用。比如既要分类又要抽取实体，就需要多个模型。这时候服务架构要调整。

我的做法是用一个路由层，根据请求类型分发到不同的模型服务。每个模型服务独立部署，独立扩缩容。路由层用简单的规则或者一个小模型来做意图识别。

MODEL_REGISTRY = { 'classification': 'http://classifier:8000/predict', 'ner': 'http://ner:8001/predict', 'sentiment': 'http://sentiment:8002/predict' } @app.post("/route") async def route_request(request: RouteRequest): intent = detect_intent(request.text) target_url = MODEL_REGISTRY.get(intent) if not target_url: raise HTTPException(status_code=400, detail="Unknown intent") async with httpx.AsyncClient() as client: response = await client.post(target_url, json={'text': request.text}) return response.json()

这样每个模型可以独立迭代，互不影响。缺点是增加了网络开销和运维复杂度。

6.4 我踩过的最大的一个坑

最后分享一个我踩过的最大的坑。早期做AI服务的时候，我把模型文件直接打进了Docker镜像。每次更新模型都要重新构建镜像、重新部署，整个过程要半小时。而且镜像越来越大，从几百MB涨到了几个GB。

后来改成了模型文件放在共享存储上，服务启动时从存储加载。更新模型只需要替换存储上的文件，然后重启服务，几分钟搞定。镜像也小了很多，只有几百MB。

这个改动的关键是模型和代码分离。代码的更新频率低，模型的更新频率高，两者应该独立管理。模型文件用版本号命名，服务启动时指定版本号，回滚的时候改个版本号就行。

# 启动时指定模型版本 MODEL_VERSION=v2.1 uvicorn serving.app:app --host 0.0.0.0 --port 8000

服务里读取环境变量加载对应版本的模型：

import os model_version = os.getenv('MODEL_VERSION', 'v1.0') model_path = f"/mnt/models/{model_version}/model.onnx" session = ort.InferenceSession(model_path, providers=['CUDAExecutionProvider'])

这个方案简单但有效，推荐给所有做AI服务的人。不要小看部署流程的优化，它直接影响你迭代的速度。迭代速度上去了，模型效果才能快速提升。