企业官网建设流程全解析

1. 从零搭建AI工程能力：为什么我劝你别再收藏夹吃灰了

"ai-engineering-from-scratch"这个标题，第一次看到的时候我愣了一下。不是因为它有多高深，恰恰相反——它太直白了，直白到像一句废话。AI工程，从零开始，谁不知道要从零开始？但真正让我停下来琢磨的，是"from scratch"这四个字背后藏着的那层意思：不是"从入门到精通"，不是"快速上手"，而是从一片空白的地基开始，把每一块砖都亲手砌上去。

我做了十多年一线开发，带过不少新人，也见过太多人在AI工程这条路上反复横跳。有人上来就调API，跑通了几个demo就觉得自己会了；有人一头扎进论文堆里，三个月后连一个能跑的服务都没部署出来；还有人收藏了上百个"AI工程学习路线"，硬盘里躺着几十个G的教程，实际动手写过的代码不超过五百行。这些坑我都踩过，或者说，我围观过别人踩。所以当我看到"ai-engineering-from-scratch"这个项目标题时，我脑子里第一反应是：终于有人愿意把"从零"这两个字当真了。

这个项目本质上是一个面向AI工程化落地的系统性实践框架，它要解决的问题不是"怎么训练一个大模型"——那是算法工程师的事——而是"怎么把一个AI能力变成稳定、可维护、可扩展的线上服务"。说得再直白一点：模型跑在notebook里是一回事，跑在千万级用户的请求链路里是另一回事。中间隔着的，就是AI工程要填的坑。适合谁来参考？如果你已经会写Python，能看懂基本的模型推理代码，但一提到"部署""监控""版本管理""灰度发布"就头皮发麻，那这个方向就是为你准备的。如果你还在纠结要不要学AI，那可以先划走，这篇内容对你来说太早了。

我接下来要拆的，是基于这个标题所指向的AI工程从零构建的完整路径。不是那种"三天速成"的爽文，而是一个从业者真正从零开始搭一套AI工程体系时，会遇到的选型逻辑、实操细节和那些文档里不会写的坑。我会尽量把每个"为什么"讲透，让你看完能直接抄作业，而不是又往收藏夹里塞一篇"以后再看"。

2. 整体设计思路：为什么"从零"不等于"从底层造轮子"

2.1 先搞清楚"零"的起点在哪里

很多人对"from scratch"有误解，以为要从零实现一个Transformer、从零写一个推理引擎、从零搭一个Kubernetes集群。我见过一个哥们儿，花了两个月时间用C++手写了一个简易的矩阵运算库，结果发现PyTorch一行代码就能搞定，而且性能是他的几百倍。这不是"从零"，这是"从负开始"。

AI工程里的"从零"，起点应该是"你有一个能跑的模型，但不知道怎么让它稳定服务"。这个起点很关键，因为它决定了你后面所有工作的重心。你的核心任务不是造模型，而是造一套让模型能持续产生价值的工程体系。这套体系包括但不限于：模型服务的封装、请求链路的编排、版本管理与回滚、性能监控与告警、数据回流与迭代。这些东西，才是AI工程师和算法工程师的分水岭。

所以整体设计的第一条原则就是：能用成熟工具解决的，绝不自己造。这不是偷懒，这是工程效率的基本素养。你花三个月造一个轮子，别人用现成方案三天就上线了，三个月后别人的系统已经迭代了五个版本，你的轮子还在修bug。AI工程这个领域，时间窗口比什么都重要。

2.2 分层架构：把复杂问题切成能一口吃掉的小块

我习惯把AI工程体系分成四层，从下往上依次是：基础设施层、模型服务层、应用编排层、业务接入层。这个分层不是教科书上的标准答案，是我自己在实际项目中反复调整后觉得最顺手的切法。

基础设施层管的是算力、存储、网络这些底层资源。这一层你基本不用从零写，但你要知道怎么用。比如GPU资源怎么调度、模型文件怎么存储和分发、服务之间的网络怎么隔离。我见过太多人在这一层翻车：模型文件放在本地磁盘上，服务扩容的时候新节点拉不到模型；GPU显存不够，多个模型抢资源导致服务雪崩。这些问题不是算法问题，是工程问题，但杀伤力一点不比算法问题小。

模型服务层是核心，管的是怎么把模型变成一个能接收请求、返回结果的HTTP或gRPC服务。这一层的关键词是封装和隔离。封装是指把模型的加载、推理、后处理逻辑打包成一个独立的服务单元；隔离是指不同模型、不同版本之间不能互相干扰。我试过在一个进程里加载多个模型，结果一个模型的内存泄漏把整个服务拖垮了。从那以后，我坚持一个模型一个服务实例，宁可多占点资源，也要保证故障隔离。

应用编排层管的是多个模型服务之间的协作。一个真实的AI应用往往不是单个模型能搞定的，比如一个智能客服系统，可能需要意图识别模型、实体抽取模型、情感分析模型、回复生成模型协同工作。这一层要解决的是：请求怎么在多个服务之间流转、超时怎么处理、某个服务挂了怎么降级。这一层是最能体现AI工程功力的地方，也是从零搭建时最容易忽略的地方。

业务接入层管的是怎么让业务方用起来。这一层要提供统一的API网关、鉴权、限流、计费、日志。很多技术出身的AI工程师觉得这一层"不够硬核"，不愿意花时间，结果业务方接入的时候发现连个像样的文档都没有，最后项目黄了。我的经验是：业务接入层的体验，直接决定了你的AI能力能不能真正被用起来。

2.3 技术选型的核心逻辑：别追新，追稳

AI工程领域的技术栈更新速度堪比时装周，今天流行这个框架，明天流行那个工具。我见过太多团队被新技术牵着鼻子走，半年换一套架构，最后什么都没沉淀下来。我的选型逻辑很简单：优先选社区活跃、文档齐全、有大规模生产验证的方案。

具体来说，模型服务化我首选FastAPI加Uvicorn，不是因为它性能最好，而是因为它足够简单、足够稳定、生态足够丰富。你可能会说Triton Inference Server性能更好，没错，但Triton的学习曲线和运维复杂度也更高。如果你的团队没有专门的推理优化工程师，FastAPI是更务实的选择。等你的QPS真的到了FastAPI扛不住的时候，再换Triton也不迟，而且那时候你已经有了足够的工程积累，换起来反而更快。

容器化我首选Docker加Docker Compose，而不是一上来就上Kubernetes。K8s很强，但它的复杂度对于从零开始的团队来说，往往是负担大于收益。我见过一个五人团队，花了两个月搭K8s集群，结果模型服务还没上线。Docker Compose足够支撑你从零到日请求百万级的场景，等真的到了瓶颈再迁移，迁移成本远小于你一开始就硬上K8s的维护成本。

监控我首选Prometheus加Grafana，日志我首选ELK或Loki。这些方案都是经过大规模验证的，社区里遇到问题一搜就有答案。不要自己造监控系统，那是一个无底洞。

3. 核心细节解析：从零搭建AI工程体系的五个关键环节

3.1 模型服务化：把notebook里的代码变成能扛住流量的服务

模型服务化是从零搭建AI工程体系的第一道坎。很多人觉得这事很简单：把notebook里的推理代码复制出来，包一个Flask接口，完事。我一开始也这么想，直到线上服务在压测的时候直接崩了。

问题出在几个地方。第一，模型加载时机。如果你在每次请求里都加载一次模型，那延迟会高到无法接受。正确的做法是在服务启动时加载模型，并常驻内存。但这里有个坑：如果你的服务是多worker的，每个worker都会加载一份模型，显存直接翻倍。我试过用Gunicorn起4个worker，结果24G显存的卡直接OOM。解决方案是用单worker多线程，或者用共享内存的方式加载模型。

第二，请求批处理。单个请求推理一次，GPU利用率极低。你需要把短时间内到达的多个请求合并成一个batch，一次性推理，然后拆分结果返回。这个逻辑听起来简单，但实现起来要考虑超时控制、batch大小动态调整、不同请求的输入长度差异等问题。我建议用现成的方案，比如NVIDIA的Triton或者Ray Serve，它们内置了动态批处理能力。如果你非要自己写，记住一个原则：batch等待时间不能超过业务可接受的额外延迟。比如你的业务要求P99延迟在200ms以内，那batch等待时间最多设20ms。

第三，健康检查与优雅退出。服务要能告诉负载均衡器"我还活着"，也要能在收到终止信号时把手头的请求处理完再退出。我见过一个服务，滚动更新的时候直接kill进程，导致正在处理的请求全部失败。这种问题在测试环境很难发现，一到生产环境就是事故。

下面是一个我常用的FastAPI模型服务模板，你可以直接参考：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import torch import asyncio from concurrent.futures import ThreadPoolExecutor import signal import sys app = FastAPI() model = None executor = ThreadPoolExecutor(max_workers=4) is_ready = False class PredictRequest(BaseModel): text: str max_length: int = 128 class PredictResponse(BaseModel): result: str latency_ms: float @app.on_event("startup") async def load_model(): global model, is_ready # 模型加载放在启动阶段，避免请求时加载 model = torch.load("/models/my_model.pt", map_location="cuda") model.eval() is_ready = True @app.get("/health") async def health(): if not is_ready: raise HTTPException(status_code=503, detail="Model not ready") return {"status": "ok"} @app.post("/predict", response_model=PredictResponse) async def predict(req: PredictRequest): if not is_ready: raise HTTPException(status_code=503, detail="Model not ready") loop = asyncio.get_event_loop() # 推理是CPU/GPU密集型操作，放到线程池避免阻塞事件循环 result = await loop.run_in_executor( executor, run_inference, req.text, req.max_length ) return result def run_inference(text, max_length): import time start = time.time() with torch.no_grad(): output = model(text, max_length=max_length) latency = (time.time() - start) * 1000 return PredictResponse(result=output, latency_ms=latency) def graceful_shutdown(signum, frame): global is_ready is_ready = False # 等待正在处理的请求完成 executor.shutdown(wait=True) sys.exit(0) signal.signal(signal.SIGTERM, graceful_shutdown)

这个模板里有几个细节值得说。is_ready标志位是为了让健康检查在模型加载完成前返回503，避免流量打到还没准备好的实例上。run_in_executor是为了不让推理阻塞FastAPI的事件循环，否则并发能力会大打折扣。graceful_shutdown是为了在收到终止信号时先停止接收新请求，再等待已有请求完成。

注意：模型加载时间可能很长，如果你的模型有几十个G，启动可能要几分钟。这时候要确保健康检查的初始延迟设置得足够长，否则负载均衡器会在服务还没准备好的时候就把它踢掉。

3.2 版本管理与灰度发布：别让新模型成为定时炸弹

模型版本管理是AI工程里最容易被低估的环节。我见过太多团队用文件名来区分版本，比如model_v1.pt、model_v2.pt，然后手动改配置切换。这种做法在模型少的时候还能凑合，一旦模型数量上来了，就是灾难。

正确的做法是把模型版本当成代码版本来管理。每个模型版本有唯一的标识符，有完整的元数据（训练数据、超参数、评估指标、上线时间），有可追溯的变更记录。我推荐用MLflow或者DVC来管理模型版本，它们能帮你把模型文件和元数据关联起来，还能和Git提交挂钩。

但版本管理只是第一步，更关键的是灰度发布。新模型上线，你不能直接把所有流量切过去，万一效果变差了呢？我一般的做法是：先切1%的流量到新模型，观察核心指标（准确率、延迟、错误率）24小时；如果没问题，切10%；再观察24小时；然后50%；最后100%。整个过程大概需要一周。听起来很慢，但比起新模型上线导致线上事故，这一周的时间成本几乎可以忽略不计。

灰度发布的实现方式有很多种。最简单的是在网关层做流量切分，根据请求的某个特征（比如用户ID的哈希值）决定走新模型还是旧模型。这种方式对模型服务本身没有侵入，但需要网关支持。另一种方式是在模型服务内部做，同一个服务加载新旧两个模型，根据请求头里的版本号决定用哪个。这种方式更灵活，但会占用双倍资源。

我倾向于第一种方式，因为模型服务应该保持纯粹，不应该承担流量调度的职责。下面是一个用Nginx做灰度发布的配置示例：

upstream model_v1 { server model-v1:8000; } upstream model_v2 { server model-v2:8000; } split_clients "${remote_addr}${http_user_agent}" $model_version { 1% "v2"; * "v1"; } server { listen 80; location /predict { if ($model_version = "v2") { proxy_pass http://model_v2; } if ($model_version = "v1") { proxy_pass http://model_v1; } } }

这个配置用split_clients模块根据客户端特征做流量切分，1%的流量走v2，99%走v1。调整百分比只需要改一个数字，非常方便。

实操心得：灰度发布期间一定要做好指标对比。我一般会在Grafana上放两个面板，一个显示新模型的指标，一个显示旧模型的指标，用同样的时间轴。这样一旦新模型有问题，一眼就能看出来。另外，灰度期间要确保日志里能区分请求走的是哪个版本，否则出了问题都不知道该查哪个模型的日志。

3.3 监控与告警：让问题在你发现之前就被发现

AI工程的监控和传统后端监控有一个很大的区别：除了系统指标，你还要监控模型指标。系统指标包括QPS、延迟、错误率、CPU/GPU利用率、内存占用；模型指标包括预测分布、置信度分布、输入数据分布。后者往往更重要，因为模型效果下降不一定体现在系统指标上。

我见过一个案例：一个推荐模型上线后，系统指标一切正常，延迟没变，错误率没变，但业务方反馈点击率下降了15%。查了半天才发现，模型对某一类用户的预测结果全部偏向同一个类别，导致推荐多样性大幅下降。这个问题在系统监控里完全看不出来，只有监控了预测分布才能发现。

所以我的监控体系里，模型指标和系统指标是同等重要的。具体来说，我会监控以下几类指标：

监控工具我用的是Prometheus加Grafana。Prometheus负责采集和存储指标，Grafana负责展示和告警。模型指标的采集需要在推理代码里埋点，比如每次推理后把预测结果的分布上报到Prometheus。下面是一个简单的埋点示例：

from prometheus_client import Counter, Histogram, Gauge import numpy as np # 定义指标 prediction_counter = Counter( 'model_predictions_total', 'Total predictions', ['predicted_class'] ) confidence_histogram = Histogram( 'model_confidence', 'Prediction confidence', buckets=[0.1, 0.3, 0.5, 0.7, 0.9, 0.95, 0.99] ) input_length_gauge = Gauge( 'model_input_length', 'Input text length' ) def run_inference_with_metrics(text): # 推理 result = model(text) predicted_class = result.argmax().item() confidence = result.softmax(dim=-1).max().item() # 上报指标 prediction_counter.labels(predicted_class=str(predicted_class)).inc() confidence_histogram.observe(confidence) input_length_gauge.set(len(text)) return result

注意：模型指标的告警阈值不能拍脑袋定，要用历史数据来算。我一般的做法是：上线后先观察一周，记录指标的均值和标准差，然后把告警阈值设为均值加减三倍标准差。这样既能发现异常，又不会因为正常波动频繁告警。

3.4 数据回流与持续迭代：让模型越用越聪明

AI工程和传统软件工程最大的区别在于：传统软件上线后行为是确定的，AI模型上线后行为会随着数据分布的变化而漂移。今天效果好的模型，三个月后可能就退化了。所以数据回流和持续迭代是AI工程体系里不可或缺的一环。

数据回流的核心是：把线上推理的输入数据和业务反馈的结果收集起来，用于后续的模型迭代。这里有几个关键点。第一，数据采集要合规。用户隐私数据不能随便存，该脱敏的脱敏，该加密的加密。第二，数据存储要结构化。不能把请求日志随便扔到一个文件里，要存到数据库或者数据湖里，方便后续查询和处理。第三，反馈闭环要建立。光收集数据没用，要有机制把数据转化成训练样本，定期触发模型迭代。

我一般的做法是：在推理服务里加一个异步的数据上报逻辑，把请求的输入、输出、时间戳、模型版本等信息发到Kafka，然后由一个消费程序写入数据仓库。业务侧的反馈（比如用户点击、评分）通过另一个通道写入，最后在数据仓库里做join，生成训练样本。

from kafka import KafkaProducer import json import hashlib producer = KafkaProducer( bootstrap_servers=['kafka:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8') ) def report_inference_data(text, result, model_version): # 脱敏处理 text_hash = hashlib.sha256(text.encode()).hexdigest() data = { 'text_hash': text_hash, 'text_length': len(text), 'predicted_class': str(result.argmax().item()), 'confidence': float(result.softmax(dim=-1).max().item()), 'model_version': model_version, 'timestamp': time.time() } producer.send('inference_logs', value=data)

这个逻辑是异步的，不会阻塞推理请求。Kafka的吞吐量足够大，不用担心成为瓶颈。

实操心得：数据回流最容易犯的错误是"只收集不处理"。我见过一个团队，收集了半年的推理数据，结果要用的时候发现数据格式乱七八糟，清洗就花了一个月。所以从一开始就要定好数据schema，并且写一个校验程序，确保写入的数据符合预期格式。另外，数据量大的时候要考虑冷热分离，最近一个月的数据放热存储方便查询，更早的数据归档到冷存储降低成本。

3.5 成本控制：AI工程不只是技术问题，更是经济问题

从零搭建AI工程体系，成本是一个绕不开的话题。GPU资源贵、存储贵、带宽贵，如果不加控制，账单能吓死人。我见过一个团队，模型服务用的是最高配的GPU实例，结果利用率只有5%，钱全浪费了。

成本控制的核心思路是按需分配、弹性伸缩。具体来说，有几个手段。第一，模型量化。把FP32的模型量化成FP16或者INT8，显存占用能减少一半到四分之三，推理速度还能提升。精度损失通常在可接受范围内，我实测下来，大部分模型量化后准确率下降不超过1%。第二，动态扩缩容。根据QPS自动调整实例数量，高峰期多开，低峰期少开。K8s的HPA或者云厂商的弹性伸缩服务都能做这个。第三，混合部署。把延迟不敏感的离线任务和延迟敏感的在线服务混部在同一张卡上，提高GPU利用率。

模型量化我用的是ONNX Runtime或者TensorRT，它们都支持训练后量化，不需要重新训练模型。下面是一个用ONNX Runtime做FP16量化的示例：

import onnx from onnxruntime.transformers import float16 # 加载原始FP32模型 model = onnx.load("model_fp32.onnx") # 转换为FP16 model_fp16 = float16.convert_float_to_float16(model) # 保存 onnx.save(model_fp16, "model_fp16.onnx")

量化后的模型推理速度通常能提升1.5到2倍，显存占用减少一半。如果你的模型是Transformer架构，还可以用ONNX Runtime的优化工具做算子融合和图层优化，进一步提升性能。

注意：量化不是万能的。有些模型对数值精度非常敏感，量化后效果会明显下降。所以量化后一定要做完整的评估，对比量化前后的指标。如果下降超过可接受范围，就要考虑混合精度量化，只量化对精度不敏感的部分。

4. 实操过程：从零搭建一个完整的AI工程Demo

4.1 环境准备与项目结构

说了这么多理论，接下来我带你走一遍完整的实操流程。我们以一个文本分类模型为例，从零搭建一个包含模型服务、监控、灰度发布、数据回流的完整AI工程Demo。

先看项目结构：

ai-engineering-demo/ ├── docker-compose.yml ├── model-service/ │ ├── Dockerfile │ ├── requirements.txt │ ├── app.py │ ├── model_loader.py │ └── metrics.py ├── gateway/ │ └── nginx.conf ├── monitoring/ │ ├── prometheus.yml │ └── grafana-dashboard.json ├──>import torch import torch.nn as nn from transformers import AutoTokenizer, AutoModelForSequenceClassification import logging logger = logging.getLogger(__name__) class ModelLoader: def __init__(self, model_path, device=None): self.model_path = model_path self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") self.model = None self.tokenizer = None def load(self): logger.info(f"Loading model from {self.model_path} on {self.device}") self.tokenizer = AutoTokenizer.from_pretrained(self.model_path) self.model = AutoModelForSequenceClassification.from_pretrained( self.model_path ) self.model.to(self.device) self.model.eval() logger.info("Model loaded successfully") return self def predict(self, text, max_length=128): inputs = self.tokenizer( text, return_tensors="pt", max_length=max_length, truncation=True, padding=True ) inputs = {k: v.to(self.device) for k, v in inputs.items()} with torch.no_grad(): outputs = self.model(**inputs) logits = outputs.logits probs = torch.softmax(logits, dim=-1) return { "predicted_class": int(probs.argmax().item()), "confidence": float(probs.max().item()), "probabilities": probs.cpu().numpy().tolist()[0] }

这个类封装了模型加载和推理逻辑。注意几个细节：model.eval()是必须的，否则Dropout和BatchNorm的行为会不一致；torch.no_grad()能减少显存占用；输入数据要移到和模型相同的设备上。

再看metrics.py：

from prometheus_client import Counter, Histogram, Gauge, generate_latest import time PREDICTION_COUNTER = Counter( 'model_predictions_total', 'Total number of predictions', ['predicted_class', 'model_version'] ) PREDICTION_LATENCY = Histogram( 'model_prediction_latency_seconds', 'Prediction latency in seconds', buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0] ) CONFIDENCE_HISTOGRAM = Histogram( 'model_confidence', 'Prediction confidence distribution', buckets=[0.1, 0.3, 0.5, 0.7, 0.9, 0.95, 0.99] ) MODEL_READY = Gauge( 'model_ready', 'Whether the model is ready to serve' ) def record_prediction(predicted_class, confidence, latency, model_version): PREDICTION_COUNTER.labels( predicted_class=str(predicted_class), model_version=model_version ).inc() PREDICTION_LATENCY.observe(latency) CONFIDENCE_HISTOGRAM.observe(confidence)

这些指标覆盖了系统性能和模型行为两个维度。PREDICTION_COUNTER按预测类别和模型版本分组，能看出模型是否偏向某个类别；CONFIDENCE_HISTOGRAM能看出模型的不确定性分布；PREDICTION_LATENCY能看出服务性能。

最后是app.py：

from fastapi import FastAPI, HTTPException, Request from fastapi.responses import Response from pydantic import BaseModel from prometheus_client import generate_latest, CONTENT_TYPE_LATEST import asyncio from concurrent.futures import ThreadPoolExecutor import time import signal import sys import os import logging from model_loader import ModelLoader from metrics import record_prediction, MODEL_READY logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="AI Model Service") executor = ThreadPoolExecutor(max_workers=4) model_loader = None is_ready = False MODEL_VERSION = os.getenv("MODEL_VERSION", "v1") class PredictRequest(BaseModel): text: str max_length: int = 128 class PredictResponse(BaseModel): predicted_class: int confidence: float probabilities: list model_version: str latency_ms: float @app.on_event("startup") async def startup(): global model_loader, is_ready model_path = os.getenv("MODEL_PATH", "/models/text-classifier") model_loader = ModelLoader(model_path).load() is_ready = True MODEL_READY.set(1) logger.info(f"Service ready with model version {MODEL_VERSION}") @app.get("/health") async def health(): if not is_ready: raise HTTPException(status_code=503, detail="Model not ready") return {"status": "ok", "model_version": MODEL_VERSION} @app.get("/metrics") async def metrics(): return Response( content=generate_latest(), media_type=CONTENT_TYPE_LATEST ) @app.post("/predict", response_model=PredictResponse) async def predict(req: PredictRequest): if not is_ready: raise HTTPException(status_code=503, detail="Model not ready") start = time.time() loop = asyncio.get_event_loop() try: result = await loop.run_in_executor( executor, model_loader.predict, req.text, req.max_length ) except Exception as e: logger.error(f"Prediction failed: {e}") raise HTTPException(status_code=500, detail="Prediction failed") latency = time.time() - start record_prediction( result["predicted_class"], result["confidence"], latency, MODEL_VERSION ) return PredictResponse( predicted_class=result["predicted_class"], confidence=result["confidence"], probabilities=result["probabilities"], model_version=MODEL_VERSION, latency_ms=latency * 1000 ) def graceful_shutdown(signum, frame): global is_ready logger.info("Shutting down gracefully...") is_ready = False MODEL_READY.set(0) executor.shutdown(wait=True) sys.exit(0) signal.signal(signal.SIGTERM, graceful_shutdown)

这个服务提供了三个接口：/health用于健康检查，/metrics用于Prometheus采集指标，/predict用于推理。MODEL_VERSION从环境变量读取，方便在部署时指定版本。

4.3 Docker Compose编排与网关配置

docker-compose.yml把各个组件串起来：

version: '3.8' services: model-v1: build: ./model-service environment: - MODEL_PATH=/models/text-classifier-v1 - MODEL_VERSION=v1 volumes: - ./models:/models deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] networks: - ai-net model-v2: build: ./model-service environment: - MODEL_PATH=/models/text-classifier-v2 - MODEL_VERSION=v2 volumes: - ./models:/models deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] networks: - ai-net gateway: image: nginx:alpine ports: - "80:80" volumes: - ./gateway/nginx.conf:/etc/nginx/nginx.conf:ro depends_on: - model-v1 - model-v2 networks: - ai-net prometheus: image: prom/prometheus:latest ports: - "9090:9090" volumes: - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro networks: - ai-net grafana: image: grafana/grafana:latest ports: - "3000:3000" environment: - GF_SECURITY_ADMIN_PASSWORD=admin volumes: - ./monitoring/grafana-dashboard.json:/etc/grafana/provisioning/dashboards/dashboard.json networks: - ai-net networks: ai-net: driver: bridge

这个编排文件起了五个服务：两个模型服务（v1和v2）、一个网关、一个Prometheus、一个Grafana。两个模型服务分别加载不同版本的模型，用于灰度发布。

gateway/nginx.conf的配置前面已经给过了，这里不再重复。核心就是用split_clients做流量切分。

monitoring/prometheus.yml配置采集目标：

global: scrape_interval: 15s scrape_configs: - job_name: 'model-v1' static_configs: - targets: ['model-v1:8000'] - job_name: 'model-v2' static_configs: - targets: ['model-v2:8000']

4.4 部署与验证

一切就绪后，执行部署脚本：

#!/bin/bash # scripts/deploy.sh set -e echo "Building model service image..." docker-compose build echo "Starting services..." docker-compose up -d echo "Waiting for services to be ready..." sleep 30 echo "Checking health..." curl -f http://localhost/health || exit 1 echo "Running smoke test..." curl -X POST http://localhost/predict \ -H "Content-Type: application/json" \ -d '{"text": "This is a test input for the model"}' \ | python -m json.tool echo "Deployment completed successfully"

部署完成后，你可以通过以下地址访问各个服务：

验证灰度发布是否生效：多次调用/predict接口，观察返回的model_version字段。大约1%的请求会返回v2，99%返回v1。如果比例不对，检查Nginx配置里的split_clients百分比。

验证监控是否生效：打开Prometheus的Graph页面，查询model_predictions_total，应该能看到按predicted_class和model_version分组的计数。查询model_prediction_latency_seconds_bucket，能看到延迟分布。

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

5.1 模型服务启动慢、健康检查失败

这是最常见的问题。模型加载需要时间，如果健康检查的初始延迟设置得太短，负载均衡器会在服务还没准备好的时候就判定为不健康，导致服务刚启动就被踢掉。

排查思路：先看服务日志，确认模型加载花了多长时间。然后在Docker Compose或者K8s的配置里，把健康检查的initialDelaySeconds设置为模型加载时间的1.5倍。比如模型加载需要60秒，就设90秒。

另一个可能的原因是模型文件太大，从磁盘加载慢。解决方案是把模型文件放到内存文件系统（tmpfs）里，或者用更快的存储。我试过把模型从机械硬盘移到SSD，加载时间从3分钟降到了30秒。

5.2 GPU显存不足导致服务崩溃

多worker加载模型、batch size太大、显存泄漏都可能导致OOM。排查方法是：在服务启动后，用nvidia-smi观察显存占用。如果启动后就占满了，说明模型加载有问题；如果运行一段时间后逐渐占满，说明有显存泄漏。

显存泄漏最常见的原因是推理时没有用torch.no_grad()，导致计算图被保留。另一个原因是把中间结果存到了全局变量里，没有及时释放。我一般的做法是：在推理函数里用with torch.no_grad():包裹所有计算，并且避免在全局作用域保存任何张量。

5.3 灰度发布流量比例不对

split_clients的百分比是基于请求特征的哈希值，理论上大样本下比例是准确的，但小样本下会有波动。如果你发现1%的配置实际切了5%的流量，先确认请求量是否足够大。如果请求量只有几百，波动是正常的。

另一个可能的原因是Nginx的split_clients模块没有正确加载。检查Nginx版本是否支持该模块，以及配置语法是否正确。可以用nginx -t测试配置。

5.4 监控指标缺失或不准

Prometheus采集不到指标，最常见的原因是网络不通或者采集路径不对。检查Prometheus的Targets页面，看目标是否处于UP状态。如果是DOWN，看错误信息。常见错误包括：服务没启动、端口不对、路径不对。

指标不准的另一个原因是埋点位置不对。比如你把延迟埋点放在了推理函数外面，那测出来的延迟就包含了网络传输和序列化的时间，比实际推理延迟大。我一般的做法是在推理函数内部埋点，只测模型前向传播的时间。

5.5 数据回流管道堵塞

Kafka消费者处理速度跟不上生产速度，导致消息积压。排查方法是看Kafka的Consumer Lag指标。如果Lag持续增长，说明消费能力不足。

解决方案有几个：增加消费者实例数、提高消费者的处理效率、批量消费。我一般的做法是批量消费，一次拉100条消息，批量写入数据仓库，这样能大幅提升吞吐量。但要注意，批量消费会增加延迟，如果业务对实时性要求高，就要权衡。

实操心得：数据回流管道一定要加监控和告警。我见过一个团队，Kafka积压了几千万条消息，直到磁盘满了才发现。从那以后，我坚持给所有数据管道加Lag监控，超过阈值就告警。

5.6 模型效果下降但系统指标正常

这是最棘手的问题，因为系统监控看不出来。排查思路是：对比新旧模型的预测分布，看是否有明显偏移。如果新模型对某一类输入的预测结果全部偏向同一个类别，说明模型可能过拟合了或者训练数据有偏。

另一个排查方向是看输入数据分布。如果线上输入的数据分布和训练数据分布差异很大，模型效果下降是必然的。这时候需要做数据回流，用线上数据重新训练模型。

我一般的做法是：每周跑一次模型效果评估，用最近的线上数据做测试集，对比模型的准确率、召回率、F1。如果指标下降超过5%，就触发模型迭代流程。

6. 我踩过的坑和给你的建议

从零搭建AI工程体系，我踩过的坑比写过的代码还多。挑几个最有代表性的说说。

第一个坑是过早优化。我一开始就想着要上K8s、上Service Mesh、上全套的CI/CD，结果花了两个月搭基础设施，模型服务一行代码没写。后来我学乖了，先用Docker Compose把核心链路跑通，等业务量上来了再逐步替换。能跑起来的简单方案，永远优于跑不起来的完美方案。

第二个坑是忽略业务反馈。我一度沉迷于优化模型的技术指标，准确率从92%提升到94%，结果业务方说"没感觉"。后来我才明白，业务方关心的是转化率、留存率这些业务指标，不是技术指标。技术指标提升1%可能对业务指标毫无影响，而业务指标的微小提升可能需要技术指标大幅改进。所以做AI工程，一定要盯着业务指标，别自嗨。

第三个坑是不做容量规划。有一次大促，流量涨了10倍，模型服务直接被打挂。事后复盘发现，我们从来没有做过容量测试，不知道系统能扛多少QPS。从那以后，我坚持每个季度做一次全链路压测，摸清系统的容量上限，并预留30%的buffer。

第四个坑是文档缺失。我接手过一个项目，前任工程师离职后，没人知道模型服务怎么部署、怎么回滚、怎么排查问题。我花了整整两周才把系统摸清楚。所以我现在要求团队：任何服务上线前，必须有一份完整的运维手册，包括部署步骤、回滚步骤、常见问题排查、联系人。这份文档可能永远用不上，但用上的时候能救命。

最后分享一个小技巧：给每个模型版本起一个有意义的名字，不要用v1、v2这种无意义的编号。比如2024-01-15-bert-base-lr2e5，包含了日期、模型架构、学习率。这样你看到版本号就能大致知道这个模型是什么时候训练的、用了什么配置。这个习惯能帮你在排查问题时快速定位到相关版本。

AI工程这个领域，技术更新快，但核心逻辑变化慢。把基础打牢，把工程习惯养好，比追新技术重要得多。我从零搭建过三套AI工程体系，每一套都比上一套更简单、更稳定。不是因为我技术退步了，而是因为我学会了做减法。