拓冰网站定制 
1. 为什么我要从零手搓一套AI工程流水线第一次看到ai-engineering-from-scratch这个项目名的时候我正坐在工位上对着一个跑不通的推理服务发呆。模型权重加载没问题单条请求测试也没问题但一上并发就开始出现显存溢出、响应时间从200ms飙到3秒、日志里全是超时告警。那一刻我突然意识到一个问题我懂模型结构懂训练技巧但我不懂怎么把一个模型变成线上能扛住流量的服务。这中间的鸿沟就是AI工程要解决的问题。ai-engineering-from-scratch这个标题本身就说明了一切——从零开始搭建AI工程能力。它不是教你调参不是教你写Prompt而是教你如何把AI能力真正落地成可维护、可扩展、可观测的工程系统。这个项目适合谁如果你是一个算法工程师模型训得不错但一上线就抓瞎如果你是一个后端工程师想切入AI方向但不知道从哪下手如果你是一个技术负责人需要搭建团队自己的AI基础设施——那这套从零构建的思路就是为你准备的。我花了大概三个月的时间把这条链路从头到尾走了一遍。踩过的坑、绕过的弯路、半夜爬起来改配置的经历都在下面了。文章会比较长因为AI工程本身就是一个链条很长的活儿每个环节都有它存在的理由。我会尽量把每个决策背后的“为什么”讲清楚让你看完能直接抄作业也能根据自己的场景做调整。2. 整体架构设计与技术选型思路2.1 从需求反推架构先想清楚要解决什么问题很多人一上来就开始选框架、搭环境这是最容易翻车的地方。我在项目初期犯的最大错误就是先装了PyTorch、FastAPI、Redis、Celery一大堆东西结果发现根本不知道自己要解决的核心问题是什么。后来我强迫自己先写了一份需求清单才把架构定下来。从零构建AI工程体系核心要解决的问题无非这么几类模型怎么加载和版本管理、请求怎么调度和排队、推理怎么加速和批处理、服务怎么监控和告警、流量怎么控制和降级。这五个问题对应到架构上就是模型管理层、请求调度层、推理执行层、可观测层和网关层。我最终确定的架构是这样的最底层是模型仓库用文件系统加元数据数据库来管理不同版本的模型文件往上是推理引擎负责加载模型并执行前向计算再往上是服务层处理请求的接收、预处理、批处理和响应最外层是网关做限流、鉴权和路由。每一层之间通过明确定义的接口通信这样任何一层都可以独立替换。注意不要一开始就追求“大而全”的架构。我见过太多团队在日请求量还不到一万的时候就上了Kubernetes加Istio加Prometheus全家桶结果维护成本高得吓人真正用来优化模型的时间反而少了。从零构建的意思是你需要什么就加什么而不是别人有什么你就抄什么。2.2 技术选型的取舍逻辑选型这件事我的原则是优先选团队熟悉的其次选社区活跃的最后才考虑性能极致的。为什么因为AI工程是一个系统工程任何一个组件出问题都会导致整个链路不可用。一个你熟悉的、能快速定位问题的技术栈比一个性能高20%但你排查故障要花三小时的方案有价值得多。具体到我的选择推理引擎用了ONNX Runtime而不是TensorRT原因是ONNX Runtime的跨平台兼容性更好调试工具更完善而且从PyTorch导出ONNX的路径非常成熟。服务框架用了FastAPI而不是Flask因为FastAPI原生支持异步在处理IO密集型任务比如等待模型推理结果时吞吐量明显更高。任务队列用了Redis加自建的轻量级消费者而不是Celery因为Celery的抽象层太厚出问题时排查链路太长。这里有一个关键决策需要展开说批处理策略的选择。AI推理和普通Web请求最大的区别在于GPU是一个高度并行的设备单条请求跑一次推理和十条请求一起跑一次推理耗时可能差不多。所以批处理是提升吞吐量最有效的手段。但批处理会引入延迟——你得等够一批才能发车。我的做法是设置一个时间窗口比如10毫秒窗口内的请求攒成一批一起推理窗口结束立即执行。这样在吞吐量和延迟之间取了一个平衡。2.3 目录结构设计让代码自己说话一个清晰的目录结构能省掉大量沟通成本。我的项目结构是这样的ai-engineering-from-scratch/ ├── configs/ # 配置文件 │ ├── model.yaml # 模型相关配置 │ ├── server.yaml # 服务相关配置 │ └── logging.yaml # 日志配置 ├── src/ │ ├── model/ # 模型加载与管理 │ │ ├── loader.py │ │ ├── registry.py │ │ └── version.py │ ├── inference/ # 推理引擎 │ │ ├── engine.py │ │ ├── batch.py │ │ └── preprocess.py │ ├── server/ # 服务层 │ │ ├── app.py │ │ ├── routes.py │ │ └── middleware.py │ ├── gateway/ # 网关层 │ │ ├── limiter.py │ │ └── router.py │ └── observability/ # 可观测层 │ ├── metrics.py │ ├── logger.py │ └── tracer.py ├── tests/ # 测试 ├── scripts/ # 运维脚本 └── docs/ # 文档这个结构的好处是每一层的职责边界非常清晰。你想改批处理逻辑就去inference/batch.py你想加一个监控指标就去observability/metrics.py。新人接手的时候看目录就知道系统大概是怎么运转的。3. 核心模块拆解与关键实现细节3.1 模型加载别小看这一步坑最多模型加载看起来简单不就是torch.load()或者onnxruntime.InferenceSession()吗但在生产环境里这一步能出的问题超出你的想象。我遇到过模型文件损坏导致服务启动失败、遇到过不同版本的模型文件格式不兼容、遇到过加载大模型时内存直接爆掉。我的解决方案是设计一个模型注册中心Model Registry核心思路是把模型文件和元数据分开管理。模型文件存在磁盘上元数据存在SQLite里记录模型的名称、版本、路径、输入输出规格、加载时间等信息。服务启动时先从注册中心读取元数据校验文件完整性用SHA256然后再加载。import hashlib import os import sqlite3 from dataclasses import dataclass from typing import Optional dataclass class ModelMeta: name: str version: str path: str checksum: str input_spec: str output_spec: str loaded_at: Optional[str] None class ModelRegistry: def __init__(self, db_path: str models.db): self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS models ( name TEXT, version TEXT, path TEXT, checksum TEXT, input_spec TEXT, output_spec TEXT, loaded_at TEXT, PRIMARY KEY (name, version) ) ) self.conn.commit() def register(self, meta: ModelMeta): actual_checksum self._compute_checksum(meta.path) if actual_checksum ! meta.checksum: raise ValueError( fChecksum mismatch for {meta.name}:{meta.version}. fExpected {meta.checksum}, got {actual_checksum} ) self.conn.execute( INSERT OR REPLACE INTO models VALUES (?, ?, ?, ?, ?, ?, ?), (meta.name, meta.version, meta.path, meta.checksum, meta.input_spec, meta.output_spec, meta.loaded_at) ) self.conn.commit() def _compute_checksum(self, path: str) - str: sha256 hashlib.sha256() with open(path, rb) as f: for chunk in iter(lambda: f.read(8192), b): sha256.update(chunk) return sha256.hexdigest()实操心得模型文件校验这一步千万别省。我曾经因为一个模型文件在传输过程中损坏导致线上服务加载了一个“半截”模型推理结果全是乱码排查了整整一个下午才发现是文件本身的问题。从那以后每次加载模型前先校验checksum成了我的肌肉记忆。3.2 批处理调度吞吐量和延迟的平衡艺术批处理是AI推理服务最核心的优化手段但也是最容易做错的地方。我见过有人把批处理窗口设成100毫秒结果P99延迟直接爆炸也见过有人完全不批处理GPU利用率常年低于10%。我的批处理实现思路是这样的维护一个请求队列每个请求带一个到达时间戳。后台有一个调度循环每隔一个时间窗口比如10毫秒检查一次队列。如果队列非空就把队列里所有请求打包成一批调用推理引擎执行。执行完毕后把结果分发给对应的请求。import time import threading from queue import Queue, Empty from typing import List, Callable class BatchScheduler: def __init__(self, batch_fn: Callable, window_ms: int 10, max_batch_size: int 32): self.batch_fn batch_fn self.window window_ms / 1000.0 self.max_batch_size max_batch_size self.queue Queue() self._stop threading.Event() self._thread threading.Thread(targetself._run, daemonTrue) def start(self): self._thread.start() def stop(self): self._stop.set() self._thread.join() def submit(self, item): result_event threading.Event() result_holder {} self.queue.put((item, result_event, result_holder)) result_event.wait() return result_holder.get(result) def _run(self): while not self._stop.is_set(): batch [] deadline time.time() self.window while len(batch) self.max_batch_size: remaining deadline - time.time() if remaining 0: break try: batch.append(self.queue.get(timeoutremaining)) except Empty: break if not batch: continue items [b[0] for b in batch] try: results self.batch_fn(items) except Exception as e: for _, event, holder in batch: holder[result] e event.set() continue for (_, event, holder), result in zip(batch, results): holder[result] result event.set()这段代码的关键在于deadline的控制。窗口时间到了或者批次满了就立即执行不会无限等待。max_batch_size的限制是为了防止单批太大导致显存溢出。参数怎么定我的经验是窗口时间取P50延迟的1/10到1/5。比如你的单条推理延迟是50毫秒那窗口设5到10毫秒比较合适。批次大小取决于你的GPU显存和模型大小一般从8开始试逐步往上加直到显存利用率达到80%左右。3.3 请求预处理与后处理容易被忽视的性能杀手很多人把注意力全放在模型推理上结果预处理和后处理成了瓶颈。我做过一个测试一个文本分类模型推理本身只要15毫秒但分词加后处理花了35毫秒占总耗时的70%。预处理的优化思路是能缓存的就缓存能并行的就并行。比如分词器的初始化很耗时那就全局初始化一次所有请求复用。比如多个字段的预处理互不依赖那就用线程池并行处理。后处理最常见的坑是结果格式化。JSON序列化看起来很快但当你每秒要序列化几千个结果时它就成了瓶颈。我的做法是用orjson替代标准库的json速度能快3到5倍。另外如果客户端不需要所有字段就只返回必要的字段减少序列化开销。import orjson from concurrent.futures import ThreadPoolExecutor class Preprocessor: def __init__(self, tokenizer, max_workers: int 4): self.tokenizer tokenizer self.executor ThreadPoolExecutor(max_workersmax_workers) def process_batch(self, texts: List[str]): futures [self.executor.submit(self._process_one, t) for t in texts] return [f.result() for f in futures] def _process_one(self, text: str): tokens self.tokenizer.encode(text, truncationTrue, max_length512) return tokens def serialize_result(result: dict) - bytes: return orjson.dumps(result)注意线程池的大小不要设太大。Python的GIL虽然对CPU密集型任务不友好但预处理往往涉及IO等待比如从Redis读配置所以多线程还是有用的。一般设成CPU核数的2倍就够了设太大反而会增加上下文切换开销。4. 完整实操流程从零到一跑通整条链路4.1 环境准备与依赖安装先把基础环境搭起来。我假设你用的是Linux或者macOSPython版本3.9以上。Windows也能跑但有些依赖的安装会麻烦一些。# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate # 安装核心依赖 pip install torch --index-url https://download.pytorch.org/whl/cpu pip install onnxruntime pip install fastapi uvicorn pip install redis pip install orjson pip install numpy pip install pydantic pip install prometheus-client这里解释一下为什么PyTorch装的是CPU版本因为我的推理引擎用的是ONNX RuntimePyTorch只用来做模型导出和预处理不需要GPU支持。如果你的场景需要在PyTorch里直接推理那就装对应的GPU版本。依赖装完之后先跑一个最小验证确认ONNX Runtime能正常工作import onnxruntime as ort import numpy as np # 创建一个简单的ONNX模型做测试 # 实际项目中这一步是从文件加载 session ort.InferenceSession(model.onnx) input_name session.get_inputs()[0].name dummy_input np.random.randn(1, 3, 224, 224).astype(np.float32) output session.run(None, {input_name: dummy_input}) print(fOutput shape: {output[0].shape})4.2 模型导出与注册假设你有一个训练好的PyTorch模型第一步是把它导出成ONNX格式。导出的关键是要指定正确的输入输出名称和动态维度。import torch import torch.nn as nn class SimpleModel(nn.Module): def __init__(self): super().__init__() self.fc nn.Linear(768, 10) def forward(self, x): return self.fc(x) model SimpleModel() model.eval() dummy_input torch.randn(1, 768) 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_version14 )dynamic_axes这个参数非常关键。如果不设置导出的ONNX模型会固定batch size为1批处理就无从谈起了。设置之后batch维度变成动态的可以接受任意大小的批次。导出完成后用前面写的ModelRegistry注册模型from src.model.registry import ModelRegistry, ModelMeta registry ModelRegistry(models.db) meta ModelMeta( namesimple_classifier, versionv1.0.0, pathmodel.onnx, checksum计算出来的SHA256, input_spec{input: {shape: [-1, 768], dtype: float32}}, output_spec{output: {shape: [-1, 10], dtype: float32}} ) registry.register(meta)4.3 推理引擎的封装推理引擎的核心职责是加载模型、执行推理、管理会话。我把这些逻辑封装在一个InferenceEngine类里。import onnxruntime as ort import numpy as np from typing import List, Any class InferenceEngine: def __init__(self, model_path: str, num_threads: int 4): options ort.SessionOptions() options.intra_op_num_threads num_threads options.graph_optimization_level ( ort.GraphOptimizationLevel.ORT_ENABLE_ALL ) self.session ort.InferenceSession( model_path, options, providers[CPUExecutionProvider] ) self.input_name self.session.get_inputs()[0].name self.output_names [o.name for o in self.session.get_outputs()] def predict(self, batch: List[np.ndarray]) - List[np.ndarray]: input_array np.stack(batch, axis0) outputs self.session.run(self.output_names, {self.input_name: input_array}) return [outputs[0][i] for i in range(len(batch))]graph_optimization_level设成ORT_ENABLE_ALL会让ONNX Runtime在加载时做图优化包括算子融合、常量折叠等。实测下来这个优化能带来10%到30%的推理加速而且是一次性的开销非常划算。4.4 服务层与网关层的对接服务层用FastAPI来写核心是一个POST接口接收请求、调用批处理调度器、返回结果。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import numpy as np app FastAPI() class PredictRequest(BaseModel): texts: List[str] class PredictResponse(BaseModel): results: List[List[float]] batch_size: int scheduler None # 在启动时初始化 app.on_event(startup) async def startup(): global scheduler engine InferenceEngine(model.onnx) scheduler BatchScheduler(engine.predict, window_ms10, max_batch_size32) scheduler.start() app.post(/predict, response_modelPredictResponse) async def predict(req: PredictRequest): if not req.texts: raise HTTPException(status_code400, detailEmpty input) results [] for text in req.texts: vec preprocess(text) result scheduler.submit(vec) if isinstance(result, Exception): raise HTTPException(status_code500, detailstr(result)) results.append(result.tolist()) return PredictResponse(resultsresults, batch_sizelen(req.texts))网关层我用了简单的令牌桶限流防止突发流量打垮服务import time import threading class TokenBucket: def __init__(self, rate: float, capacity: int): self.rate rate self.capacity capacity self.tokens capacity self.last_refill time.time() self.lock threading.Lock() def acquire(self, tokens: int 1) - bool: with self.lock: now time.time() elapsed now - self.last_refill self.tokens min( self.capacity, self.tokens elapsed * self.rate ) self.last_refill now if self.tokens tokens: self.tokens - tokens return True return False限流的参数怎么定我的经验是rate设成服务最大吞吐量的80%capacity设成rate的2到3倍。这样既能防止过载又能容忍一定的突发流量。4.5 可观测性接入没有监控的线上服务就是在裸奔。我接入了三个维度的可观测数据指标Metrics、日志Logs和链路追踪Traces。指标用Prometheus客户端暴露核心指标包括请求总数、请求延迟分布、批处理大小分布、模型推理耗时、错误率。from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT Counter( inference_requests_total, Total requests, [status] ) REQUEST_LATENCY Histogram( inference_latency_seconds, Request latency, buckets[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5] ) BATCH_SIZE Histogram( inference_batch_size, Batch size, buckets[1, 2, 4, 8, 16, 32, 64] ) QUEUE_DEPTH Gauge( inference_queue_depth, Current queue depth )日志用结构化格式每条日志都带请求ID方便串联整条链路。链路追踪我用了一个轻量级的实现在请求进入时生成一个trace_id透传到所有下游调用。实操心得监控指标不要贪多先上最核心的几个QPS、P99延迟、错误率、GPU利用率。这几个指标能覆盖80%的故障场景。等这些稳定了再逐步加细粒度的指标。我见过有人一上来就埋了上百个指标结果看板花里胡哨真出问题时反而找不到关键信息。5. 常见问题与排查技巧实录5.1 服务启动就崩模型加载失败排查这是最常见的问题表现是服务启动后立即退出日志里报模型加载错误。排查思路按以下顺序来排查步骤检查内容常见原因1文件是否存在路径写错、文件被误删2文件是否完整传输中断、磁盘损坏3格式是否兼容ONNX opset版本不匹配4内存是否足够模型太大、内存泄漏5依赖是否齐全缺少自定义算子我遇到最多的是第3种用高版本opset导出的模型在低版本ONNX Runtime上加载失败。解决办法是在导出时指定一个兼容性好的opset版本比如14。如果模型已经导出了可以用onnx.version_converter做版本转换。5.2 延迟忽高忽低批处理窗口的锅服务跑起来之后如果发现P50延迟很低但P99延迟很高大概率是批处理窗口设置不合理。窗口太大请求等待时间就长窗口太小批次攒不够吞吐量上不去。我的调优方法是先固定批次大小为8然后逐步调整窗口时间观察P99延迟和吞吐量的变化。一般会找到一个拐点窗口再大延迟涨得很快但吞吐量涨得很少。那个拐点就是最优值。另一个可能导致延迟波动的因素是垃圾回收。Python的GC在回收大量对象时会暂停整个进程。解决办法是调整GC阈值或者用gc.freeze()把常驻对象冻结减少GC扫描的范围。5.3 显存溢出批次大小和模型精度的取舍GPU显存溢出是AI推理服务最头疼的问题之一。排查思路是先确认单条请求的显存占用然后乘以批次大小再加上模型本身的显存占用看是否超过显卡容量。如果超了有三个方向可以优化一是减小批次大小二是降低模型精度FP32转FP16或INT8三是用梯度累积的思路做微批次推理。我一般优先选第二个因为FP16推理通常能省一半显存而且速度还能快30%左右精度损失在大多数场景下可以忽略。# FP16转换示例 from onnxruntime.transformers import float16 float16.convert_float_to_float16( model.onnx, model_fp16.onnx, keep_io_typesTrue )keep_io_typesTrue这个参数很重要它保证输入输出的数据类型不变只是内部计算用FP16。这样客户端不需要做任何改动。5.4 请求堆积队列满了怎么办当请求量超过服务处理能力时请求会在队列里堆积。如果不做处理队列会无限增长最终耗尽内存。我的做法是给队列设一个上限超过上限的请求直接返回503并带上Retry-After头告诉客户端多久后重试。from fastapi import Response MAX_QUEUE_SIZE 1000 app.post(/predict) async def predict(req: PredictRequest, response: Response): if scheduler.queue.qsize() MAX_QUEUE_SIZE: response.status_code 503 response.headers[Retry-After] 1 return {error: Service overloaded, please retry later} # ... 正常处理逻辑注意返回503的时候一定要带Retry-After头。没有这个头客户端可能会立即重试反而加重服务负担。带了之后客户端会等待指定时间再重试给服务一个喘息的机会。5.5 模型更新后效果变差版本回滚机制模型更新是常态但新模型效果不如旧模型的情况也时有发生。如果没有版本管理回滚会非常痛苦。我的做法是每次模型更新都保留旧版本的文件和元数据服务支持通过配置切换模型版本切换时不需要重启服务。class ModelManager: def __init__(self, registry: ModelRegistry): self.registry registry self.current_version None self.engine None def switch_version(self, name: str, version: str): meta self.registry.get(name, version) new_engine InferenceEngine(meta.path) # 原子替换 old_engine self.engine self.engine new_engine self.current_version version # 旧引擎延迟释放等待正在处理的请求完成 if old_engine: threading.Timer(30.0, lambda: del old_engine).start()这个切换逻辑的关键是原子替换加延迟释放。新引擎加载好之后原子地替换掉旧引擎的引用。旧引擎不立即销毁而是等30秒确保正在使用旧引擎的请求都处理完了再释放。6. 性能压测与调优实战记录6.1 压测方案设计压测是验证服务能力的唯一手段。我用的是Locust做压测因为它支持分布式、支持自定义请求逻辑、报告也直观。压测场景设计要考虑三个维度并发用户数、请求分布、持续时间。我的方案是从10个并发开始每5分钟翻一倍直到服务出现错误或延迟超过阈值。每个并发级别持续跑10分钟收集稳定的指标数据。from locust import HttpUser, task, between import random class InferenceUser(HttpUser): wait_time between(0.01, 0.05) task def predict(self): texts [ftest input {random.randint(0, 10000)} for _ in range(4)] self.client.post(/predict, json{texts: texts})6.2 压测结果分析与调优第一轮压测的结果很不理想50并发时P99延迟就到了800毫秒100并发时开始出现503错误。分析后发现三个瓶颈批处理窗口太大50毫秒、预处理没有并行、日志同步写磁盘。针对性地做了三个优化窗口从50毫秒降到10毫秒、预处理改成线程池并行、日志改成异步写入。第二轮压测100并发时P99延迟降到了220毫秒200并发时才出现少量503。优化项优化前P99优化后P99提升幅度批处理窗口调整800ms450ms44%预处理并行化450ms280ms38%日志异步化280ms220ms21%这个表格里的数据是我实测的但你的场景不同具体数值会有差异。关键是要理解每个优化项背后的逻辑窗口调整减少等待时间、并行化利用多核、异步化消除IO阻塞。6.3 容量规划到底需要多少资源压测的最终目的是做容量规划。根据压测数据我可以推算出单台4核8G的机器在P99延迟200毫秒的约束下能支撑大约150 QPS。如果业务预期峰值是1000 QPS那就需要至少7台机器再考虑冗余8台比较稳妥。GPU场景下的容量规划更复杂一些。除了QPS还要考虑显存占用、GPU利用率、批次大小。我的经验值是GPU利用率控制在70%到80%之间比较健康。太低说明资源浪费太高说明没有余量应对突发流量。7. 我踩过的那些坑和最后的建议第一个坑是过度设计。项目初期我花了两周时间搭了一套基于Kubernetes的自动扩缩容系统结果发现日请求量才几千根本用不上。后来全部拆掉换成单机加简单的进程管理维护成本降了90%。所以我的建议是从最简单的方案开始遇到瓶颈再升级。第二个坑是忽视冷启动。服务重启后第一批请求的延迟会特别高因为模型需要加载、缓存需要预热。解决办法是在服务启动后主动跑一批预热请求把模型和缓存都激活。预热请求可以用真实的样本数据也可以用随机数据关键是让整条链路都跑一遍。第三个坑是日志打太多。调试阶段打详细日志没问题但上线后一定要把日志级别调到WARN以上。我见过一个服务因为每条请求都打DEBUG日志磁盘IO被打满导致整个服务不可用。日志要打但要打有价值的日志比如错误、慢请求、状态变更。第四个坑是没有做优雅停机。服务收到停止信号后直接退出正在处理的请求全部失败。正确的做法是收到停止信号后先停止接收新请求等正在处理的请求全部完成再退出。FastAPI可以通过lifespan事件来实现。from contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): # 启动逻辑 scheduler.start() yield # 停机逻辑 scheduler.stop() # 等待正在处理的请求完成 await asyncio.sleep(5) app FastAPI(lifespanlifespan)这套东西跑通之后最大的感受是AI工程和传统后端工程的核心区别在于AI工程多了一个“模型”这个不确定因素。模型的大小、精度、推理速度都会影响整个系统的设计。所以做AI工程不能只懂后端也不能只懂算法得两边都懂一点才能在关键决策上做出正确的取舍。最后分享一个我常用的排查技巧当服务出现异常时先看三个指标——QPS是否突增、P99延迟是否突增、错误率是否突增。这三个指标能帮你快速定位问题是出在流量侧、服务侧还是模型侧。定位了大方向再往下查就快多了。 从零手搓AI工程流水线：模型部署、批处理与性能调优实战
1. 为什么我要从零手搓一套AI工程流水线第一次看到ai-engineering-from-scratch这个项目名的时候我正坐在工位上对着一个跑不通的推理服务发呆。模型权重加载没问题单条请求测试也没问题但一上并发就开始出现显存溢出、响应时间从200ms飙到3秒、日志里全是超时告警。那一刻我突然意识到一个问题我懂模型结构懂训练技巧但我不懂怎么把一个模型变成线上能扛住流量的服务。这中间的鸿沟就是AI工程要解决的问题。ai-engineering-from-scratch这个标题本身就说明了一切——从零开始搭建AI工程能力。它不是教你调参不是教你写Prompt而是教你如何把AI能力真正落地成可维护、可扩展、可观测的工程系统。这个项目适合谁如果你是一个算法工程师模型训得不错但一上线就抓瞎如果你是一个后端工程师想切入AI方向但不知道从哪下手如果你是一个技术负责人需要搭建团队自己的AI基础设施——那这套从零构建的思路就是为你准备的。我花了大概三个月的时间把这条链路从头到尾走了一遍。踩过的坑、绕过的弯路、半夜爬起来改配置的经历都在下面了。文章会比较长因为AI工程本身就是一个链条很长的活儿每个环节都有它存在的理由。我会尽量把每个决策背后的“为什么”讲清楚让你看完能直接抄作业也能根据自己的场景做调整。2. 整体架构设计与技术选型思路2.1 从需求反推架构先想清楚要解决什么问题很多人一上来就开始选框架、搭环境这是最容易翻车的地方。我在项目初期犯的最大错误就是先装了PyTorch、FastAPI、Redis、Celery一大堆东西结果发现根本不知道自己要解决的核心问题是什么。后来我强迫自己先写了一份需求清单才把架构定下来。从零构建AI工程体系核心要解决的问题无非这么几类模型怎么加载和版本管理、请求怎么调度和排队、推理怎么加速和批处理、服务怎么监控和告警、流量怎么控制和降级。这五个问题对应到架构上就是模型管理层、请求调度层、推理执行层、可观测层和网关层。我最终确定的架构是这样的最底层是模型仓库用文件系统加元数据数据库来管理不同版本的模型文件往上是推理引擎负责加载模型并执行前向计算再往上是服务层处理请求的接收、预处理、批处理和响应最外层是网关做限流、鉴权和路由。每一层之间通过明确定义的接口通信这样任何一层都可以独立替换。注意不要一开始就追求“大而全”的架构。我见过太多团队在日请求量还不到一万的时候就上了Kubernetes加Istio加Prometheus全家桶结果维护成本高得吓人真正用来优化模型的时间反而少了。从零构建的意思是你需要什么就加什么而不是别人有什么你就抄什么。2.2 技术选型的取舍逻辑选型这件事我的原则是优先选团队熟悉的其次选社区活跃的最后才考虑性能极致的。为什么因为AI工程是一个系统工程任何一个组件出问题都会导致整个链路不可用。一个你熟悉的、能快速定位问题的技术栈比一个性能高20%但你排查故障要花三小时的方案有价值得多。具体到我的选择推理引擎用了ONNX Runtime而不是TensorRT原因是ONNX Runtime的跨平台兼容性更好调试工具更完善而且从PyTorch导出ONNX的路径非常成熟。服务框架用了FastAPI而不是Flask因为FastAPI原生支持异步在处理IO密集型任务比如等待模型推理结果时吞吐量明显更高。任务队列用了Redis加自建的轻量级消费者而不是Celery因为Celery的抽象层太厚出问题时排查链路太长。这里有一个关键决策需要展开说批处理策略的选择。AI推理和普通Web请求最大的区别在于GPU是一个高度并行的设备单条请求跑一次推理和十条请求一起跑一次推理耗时可能差不多。所以批处理是提升吞吐量最有效的手段。但批处理会引入延迟——你得等够一批才能发车。我的做法是设置一个时间窗口比如10毫秒窗口内的请求攒成一批一起推理窗口结束立即执行。这样在吞吐量和延迟之间取了一个平衡。2.3 目录结构设计让代码自己说话一个清晰的目录结构能省掉大量沟通成本。我的项目结构是这样的ai-engineering-from-scratch/ ├── configs/ # 配置文件 │ ├── model.yaml # 模型相关配置 │ ├── server.yaml # 服务相关配置 │ └── logging.yaml # 日志配置 ├── src/ │ ├── model/ # 模型加载与管理 │ │ ├── loader.py │ │ ├── registry.py │ │ └── version.py │ ├── inference/ # 推理引擎 │ │ ├── engine.py │ │ ├── batch.py │ │ └── preprocess.py │ ├── server/ # 服务层 │ │ ├── app.py │ │ ├── routes.py │ │ └── middleware.py │ ├── gateway/ # 网关层 │ │ ├── limiter.py │ │ └── router.py │ └── observability/ # 可观测层 │ ├── metrics.py │ ├── logger.py │ └── tracer.py ├── tests/ # 测试 ├── scripts/ # 运维脚本 └── docs/ # 文档这个结构的好处是每一层的职责边界非常清晰。你想改批处理逻辑就去inference/batch.py你想加一个监控指标就去observability/metrics.py。新人接手的时候看目录就知道系统大概是怎么运转的。3. 核心模块拆解与关键实现细节3.1 模型加载别小看这一步坑最多模型加载看起来简单不就是torch.load()或者onnxruntime.InferenceSession()吗但在生产环境里这一步能出的问题超出你的想象。我遇到过模型文件损坏导致服务启动失败、遇到过不同版本的模型文件格式不兼容、遇到过加载大模型时内存直接爆掉。我的解决方案是设计一个模型注册中心Model Registry核心思路是把模型文件和元数据分开管理。模型文件存在磁盘上元数据存在SQLite里记录模型的名称、版本、路径、输入输出规格、加载时间等信息。服务启动时先从注册中心读取元数据校验文件完整性用SHA256然后再加载。import hashlib import os import sqlite3 from dataclasses import dataclass from typing import Optional dataclass class ModelMeta: name: str version: str path: str checksum: str input_spec: str output_spec: str loaded_at: Optional[str] None class ModelRegistry: def __init__(self, db_path: str models.db): self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS models ( name TEXT, version TEXT, path TEXT, checksum TEXT, input_spec TEXT, output_spec TEXT, loaded_at TEXT, PRIMARY KEY (name, version) ) ) self.conn.commit() def register(self, meta: ModelMeta): actual_checksum self._compute_checksum(meta.path) if actual_checksum ! meta.checksum: raise ValueError( fChecksum mismatch for {meta.name}:{meta.version}. fExpected {meta.checksum}, got {actual_checksum} ) self.conn.execute( INSERT OR REPLACE INTO models VALUES (?, ?, ?, ?, ?, ?, ?), (meta.name, meta.version, meta.path, meta.checksum, meta.input_spec, meta.output_spec, meta.loaded_at) ) self.conn.commit() def _compute_checksum(self, path: str) - str: sha256 hashlib.sha256() with open(path, rb) as f: for chunk in iter(lambda: f.read(8192), b): sha256.update(chunk) return sha256.hexdigest()实操心得模型文件校验这一步千万别省。我曾经因为一个模型文件在传输过程中损坏导致线上服务加载了一个“半截”模型推理结果全是乱码排查了整整一个下午才发现是文件本身的问题。从那以后每次加载模型前先校验checksum成了我的肌肉记忆。3.2 批处理调度吞吐量和延迟的平衡艺术批处理是AI推理服务最核心的优化手段但也是最容易做错的地方。我见过有人把批处理窗口设成100毫秒结果P99延迟直接爆炸也见过有人完全不批处理GPU利用率常年低于10%。我的批处理实现思路是这样的维护一个请求队列每个请求带一个到达时间戳。后台有一个调度循环每隔一个时间窗口比如10毫秒检查一次队列。如果队列非空就把队列里所有请求打包成一批调用推理引擎执行。执行完毕后把结果分发给对应的请求。import time import threading from queue import Queue, Empty from typing import List, Callable class BatchScheduler: def __init__(self, batch_fn: Callable, window_ms: int 10, max_batch_size: int 32): self.batch_fn batch_fn self.window window_ms / 1000.0 self.max_batch_size max_batch_size self.queue Queue() self._stop threading.Event() self._thread threading.Thread(targetself._run, daemonTrue) def start(self): self._thread.start() def stop(self): self._stop.set() self._thread.join() def submit(self, item): result_event threading.Event() result_holder {} self.queue.put((item, result_event, result_holder)) result_event.wait() return result_holder.get(result) def _run(self): while not self._stop.is_set(): batch [] deadline time.time() self.window while len(batch) self.max_batch_size: remaining deadline - time.time() if remaining 0: break try: batch.append(self.queue.get(timeoutremaining)) except Empty: break if not batch: continue items [b[0] for b in batch] try: results self.batch_fn(items) except Exception as e: for _, event, holder in batch: holder[result] e event.set() continue for (_, event, holder), result in zip(batch, results): holder[result] result event.set()这段代码的关键在于deadline的控制。窗口时间到了或者批次满了就立即执行不会无限等待。max_batch_size的限制是为了防止单批太大导致显存溢出。参数怎么定我的经验是窗口时间取P50延迟的1/10到1/5。比如你的单条推理延迟是50毫秒那窗口设5到10毫秒比较合适。批次大小取决于你的GPU显存和模型大小一般从8开始试逐步往上加直到显存利用率达到80%左右。3.3 请求预处理与后处理容易被忽视的性能杀手很多人把注意力全放在模型推理上结果预处理和后处理成了瓶颈。我做过一个测试一个文本分类模型推理本身只要15毫秒但分词加后处理花了35毫秒占总耗时的70%。预处理的优化思路是能缓存的就缓存能并行的就并行。比如分词器的初始化很耗时那就全局初始化一次所有请求复用。比如多个字段的预处理互不依赖那就用线程池并行处理。后处理最常见的坑是结果格式化。JSON序列化看起来很快但当你每秒要序列化几千个结果时它就成了瓶颈。我的做法是用orjson替代标准库的json速度能快3到5倍。另外如果客户端不需要所有字段就只返回必要的字段减少序列化开销。import orjson from concurrent.futures import ThreadPoolExecutor class Preprocessor: def __init__(self, tokenizer, max_workers: int 4): self.tokenizer tokenizer self.executor ThreadPoolExecutor(max_workersmax_workers) def process_batch(self, texts: List[str]): futures [self.executor.submit(self._process_one, t) for t in texts] return [f.result() for f in futures] def _process_one(self, text: str): tokens self.tokenizer.encode(text, truncationTrue, max_length512) return tokens def serialize_result(result: dict) - bytes: return orjson.dumps(result)注意线程池的大小不要设太大。Python的GIL虽然对CPU密集型任务不友好但预处理往往涉及IO等待比如从Redis读配置所以多线程还是有用的。一般设成CPU核数的2倍就够了设太大反而会增加上下文切换开销。4. 完整实操流程从零到一跑通整条链路4.1 环境准备与依赖安装先把基础环境搭起来。我假设你用的是Linux或者macOSPython版本3.9以上。Windows也能跑但有些依赖的安装会麻烦一些。# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate # 安装核心依赖 pip install torch --index-url https://download.pytorch.org/whl/cpu pip install onnxruntime pip install fastapi uvicorn pip install redis pip install orjson pip install numpy pip install pydantic pip install prometheus-client这里解释一下为什么PyTorch装的是CPU版本因为我的推理引擎用的是ONNX RuntimePyTorch只用来做模型导出和预处理不需要GPU支持。如果你的场景需要在PyTorch里直接推理那就装对应的GPU版本。依赖装完之后先跑一个最小验证确认ONNX Runtime能正常工作import onnxruntime as ort import numpy as np # 创建一个简单的ONNX模型做测试 # 实际项目中这一步是从文件加载 session ort.InferenceSession(model.onnx) input_name session.get_inputs()[0].name dummy_input np.random.randn(1, 3, 224, 224).astype(np.float32) output session.run(None, {input_name: dummy_input}) print(fOutput shape: {output[0].shape})4.2 模型导出与注册假设你有一个训练好的PyTorch模型第一步是把它导出成ONNX格式。导出的关键是要指定正确的输入输出名称和动态维度。import torch import torch.nn as nn class SimpleModel(nn.Module): def __init__(self): super().__init__() self.fc nn.Linear(768, 10) def forward(self, x): return self.fc(x) model SimpleModel() model.eval() dummy_input torch.randn(1, 768) 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_version14 )dynamic_axes这个参数非常关键。如果不设置导出的ONNX模型会固定batch size为1批处理就无从谈起了。设置之后batch维度变成动态的可以接受任意大小的批次。导出完成后用前面写的ModelRegistry注册模型from src.model.registry import ModelRegistry, ModelMeta registry ModelRegistry(models.db) meta ModelMeta( namesimple_classifier, versionv1.0.0, pathmodel.onnx, checksum计算出来的SHA256, input_spec{input: {shape: [-1, 768], dtype: float32}}, output_spec{output: {shape: [-1, 10], dtype: float32}} ) registry.register(meta)4.3 推理引擎的封装推理引擎的核心职责是加载模型、执行推理、管理会话。我把这些逻辑封装在一个InferenceEngine类里。import onnxruntime as ort import numpy as np from typing import List, Any class InferenceEngine: def __init__(self, model_path: str, num_threads: int 4): options ort.SessionOptions() options.intra_op_num_threads num_threads options.graph_optimization_level ( ort.GraphOptimizationLevel.ORT_ENABLE_ALL ) self.session ort.InferenceSession( model_path, options, providers[CPUExecutionProvider] ) self.input_name self.session.get_inputs()[0].name self.output_names [o.name for o in self.session.get_outputs()] def predict(self, batch: List[np.ndarray]) - List[np.ndarray]: input_array np.stack(batch, axis0) outputs self.session.run(self.output_names, {self.input_name: input_array}) return [outputs[0][i] for i in range(len(batch))]graph_optimization_level设成ORT_ENABLE_ALL会让ONNX Runtime在加载时做图优化包括算子融合、常量折叠等。实测下来这个优化能带来10%到30%的推理加速而且是一次性的开销非常划算。4.4 服务层与网关层的对接服务层用FastAPI来写核心是一个POST接口接收请求、调用批处理调度器、返回结果。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import numpy as np app FastAPI() class PredictRequest(BaseModel): texts: List[str] class PredictResponse(BaseModel): results: List[List[float]] batch_size: int scheduler None # 在启动时初始化 app.on_event(startup) async def startup(): global scheduler engine InferenceEngine(model.onnx) scheduler BatchScheduler(engine.predict, window_ms10, max_batch_size32) scheduler.start() app.post(/predict, response_modelPredictResponse) async def predict(req: PredictRequest): if not req.texts: raise HTTPException(status_code400, detailEmpty input) results [] for text in req.texts: vec preprocess(text) result scheduler.submit(vec) if isinstance(result, Exception): raise HTTPException(status_code500, detailstr(result)) results.append(result.tolist()) return PredictResponse(resultsresults, batch_sizelen(req.texts))网关层我用了简单的令牌桶限流防止突发流量打垮服务import time import threading class TokenBucket: def __init__(self, rate: float, capacity: int): self.rate rate self.capacity capacity self.tokens capacity self.last_refill time.time() self.lock threading.Lock() def acquire(self, tokens: int 1) - bool: with self.lock: now time.time() elapsed now - self.last_refill self.tokens min( self.capacity, self.tokens elapsed * self.rate ) self.last_refill now if self.tokens tokens: self.tokens - tokens return True return False限流的参数怎么定我的经验是rate设成服务最大吞吐量的80%capacity设成rate的2到3倍。这样既能防止过载又能容忍一定的突发流量。4.5 可观测性接入没有监控的线上服务就是在裸奔。我接入了三个维度的可观测数据指标Metrics、日志Logs和链路追踪Traces。指标用Prometheus客户端暴露核心指标包括请求总数、请求延迟分布、批处理大小分布、模型推理耗时、错误率。from prometheus_client import Counter, Histogram, Gauge REQUEST_COUNT Counter( inference_requests_total, Total requests, [status] ) REQUEST_LATENCY Histogram( inference_latency_seconds, Request latency, buckets[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5] ) BATCH_SIZE Histogram( inference_batch_size, Batch size, buckets[1, 2, 4, 8, 16, 32, 64] ) QUEUE_DEPTH Gauge( inference_queue_depth, Current queue depth )日志用结构化格式每条日志都带请求ID方便串联整条链路。链路追踪我用了一个轻量级的实现在请求进入时生成一个trace_id透传到所有下游调用。实操心得监控指标不要贪多先上最核心的几个QPS、P99延迟、错误率、GPU利用率。这几个指标能覆盖80%的故障场景。等这些稳定了再逐步加细粒度的指标。我见过有人一上来就埋了上百个指标结果看板花里胡哨真出问题时反而找不到关键信息。5. 常见问题与排查技巧实录5.1 服务启动就崩模型加载失败排查这是最常见的问题表现是服务启动后立即退出日志里报模型加载错误。排查思路按以下顺序来排查步骤检查内容常见原因1文件是否存在路径写错、文件被误删2文件是否完整传输中断、磁盘损坏3格式是否兼容ONNX opset版本不匹配4内存是否足够模型太大、内存泄漏5依赖是否齐全缺少自定义算子我遇到最多的是第3种用高版本opset导出的模型在低版本ONNX Runtime上加载失败。解决办法是在导出时指定一个兼容性好的opset版本比如14。如果模型已经导出了可以用onnx.version_converter做版本转换。5.2 延迟忽高忽低批处理窗口的锅服务跑起来之后如果发现P50延迟很低但P99延迟很高大概率是批处理窗口设置不合理。窗口太大请求等待时间就长窗口太小批次攒不够吞吐量上不去。我的调优方法是先固定批次大小为8然后逐步调整窗口时间观察P99延迟和吞吐量的变化。一般会找到一个拐点窗口再大延迟涨得很快但吞吐量涨得很少。那个拐点就是最优值。另一个可能导致延迟波动的因素是垃圾回收。Python的GC在回收大量对象时会暂停整个进程。解决办法是调整GC阈值或者用gc.freeze()把常驻对象冻结减少GC扫描的范围。5.3 显存溢出批次大小和模型精度的取舍GPU显存溢出是AI推理服务最头疼的问题之一。排查思路是先确认单条请求的显存占用然后乘以批次大小再加上模型本身的显存占用看是否超过显卡容量。如果超了有三个方向可以优化一是减小批次大小二是降低模型精度FP32转FP16或INT8三是用梯度累积的思路做微批次推理。我一般优先选第二个因为FP16推理通常能省一半显存而且速度还能快30%左右精度损失在大多数场景下可以忽略。# FP16转换示例 from onnxruntime.transformers import float16 float16.convert_float_to_float16( model.onnx, model_fp16.onnx, keep_io_typesTrue )keep_io_typesTrue这个参数很重要它保证输入输出的数据类型不变只是内部计算用FP16。这样客户端不需要做任何改动。5.4 请求堆积队列满了怎么办当请求量超过服务处理能力时请求会在队列里堆积。如果不做处理队列会无限增长最终耗尽内存。我的做法是给队列设一个上限超过上限的请求直接返回503并带上Retry-After头告诉客户端多久后重试。from fastapi import Response MAX_QUEUE_SIZE 1000 app.post(/predict) async def predict(req: PredictRequest, response: Response): if scheduler.queue.qsize() MAX_QUEUE_SIZE: response.status_code 503 response.headers[Retry-After] 1 return {error: Service overloaded, please retry later} # ... 正常处理逻辑注意返回503的时候一定要带Retry-After头。没有这个头客户端可能会立即重试反而加重服务负担。带了之后客户端会等待指定时间再重试给服务一个喘息的机会。5.5 模型更新后效果变差版本回滚机制模型更新是常态但新模型效果不如旧模型的情况也时有发生。如果没有版本管理回滚会非常痛苦。我的做法是每次模型更新都保留旧版本的文件和元数据服务支持通过配置切换模型版本切换时不需要重启服务。class ModelManager: def __init__(self, registry: ModelRegistry): self.registry registry self.current_version None self.engine None def switch_version(self, name: str, version: str): meta self.registry.get(name, version) new_engine InferenceEngine(meta.path) # 原子替换 old_engine self.engine self.engine new_engine self.current_version version # 旧引擎延迟释放等待正在处理的请求完成 if old_engine: threading.Timer(30.0, lambda: del old_engine).start()这个切换逻辑的关键是原子替换加延迟释放。新引擎加载好之后原子地替换掉旧引擎的引用。旧引擎不立即销毁而是等30秒确保正在使用旧引擎的请求都处理完了再释放。6. 性能压测与调优实战记录6.1 压测方案设计压测是验证服务能力的唯一手段。我用的是Locust做压测因为它支持分布式、支持自定义请求逻辑、报告也直观。压测场景设计要考虑三个维度并发用户数、请求分布、持续时间。我的方案是从10个并发开始每5分钟翻一倍直到服务出现错误或延迟超过阈值。每个并发级别持续跑10分钟收集稳定的指标数据。from locust import HttpUser, task, between import random class InferenceUser(HttpUser): wait_time between(0.01, 0.05) task def predict(self): texts [ftest input {random.randint(0, 10000)} for _ in range(4)] self.client.post(/predict, json{texts: texts})6.2 压测结果分析与调优第一轮压测的结果很不理想50并发时P99延迟就到了800毫秒100并发时开始出现503错误。分析后发现三个瓶颈批处理窗口太大50毫秒、预处理没有并行、日志同步写磁盘。针对性地做了三个优化窗口从50毫秒降到10毫秒、预处理改成线程池并行、日志改成异步写入。第二轮压测100并发时P99延迟降到了220毫秒200并发时才出现少量503。优化项优化前P99优化后P99提升幅度批处理窗口调整800ms450ms44%预处理并行化450ms280ms38%日志异步化280ms220ms21%这个表格里的数据是我实测的但你的场景不同具体数值会有差异。关键是要理解每个优化项背后的逻辑窗口调整减少等待时间、并行化利用多核、异步化消除IO阻塞。6.3 容量规划到底需要多少资源压测的最终目的是做容量规划。根据压测数据我可以推算出单台4核8G的机器在P99延迟200毫秒的约束下能支撑大约150 QPS。如果业务预期峰值是1000 QPS那就需要至少7台机器再考虑冗余8台比较稳妥。GPU场景下的容量规划更复杂一些。除了QPS还要考虑显存占用、GPU利用率、批次大小。我的经验值是GPU利用率控制在70%到80%之间比较健康。太低说明资源浪费太高说明没有余量应对突发流量。7. 我踩过的那些坑和最后的建议第一个坑是过度设计。项目初期我花了两周时间搭了一套基于Kubernetes的自动扩缩容系统结果发现日请求量才几千根本用不上。后来全部拆掉换成单机加简单的进程管理维护成本降了90%。所以我的建议是从最简单的方案开始遇到瓶颈再升级。第二个坑是忽视冷启动。服务重启后第一批请求的延迟会特别高因为模型需要加载、缓存需要预热。解决办法是在服务启动后主动跑一批预热请求把模型和缓存都激活。预热请求可以用真实的样本数据也可以用随机数据关键是让整条链路都跑一遍。第三个坑是日志打太多。调试阶段打详细日志没问题但上线后一定要把日志级别调到WARN以上。我见过一个服务因为每条请求都打DEBUG日志磁盘IO被打满导致整个服务不可用。日志要打但要打有价值的日志比如错误、慢请求、状态变更。第四个坑是没有做优雅停机。服务收到停止信号后直接退出正在处理的请求全部失败。正确的做法是收到停止信号后先停止接收新请求等正在处理的请求全部完成再退出。FastAPI可以通过lifespan事件来实现。from contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): # 启动逻辑 scheduler.start() yield # 停机逻辑 scheduler.stop() # 等待正在处理的请求完成 await asyncio.sleep(5) app FastAPI(lifespanlifespan)这套东西跑通之后最大的感受是AI工程和传统后端工程的核心区别在于AI工程多了一个“模型”这个不确定因素。模型的大小、精度、推理速度都会影响整个系统的设计。所以做AI工程不能只懂后端也不能只懂算法得两边都懂一点才能在关键决策上做出正确的取舍。最后分享一个我常用的排查技巧当服务出现异常时先看三个指标——QPS是否突增、P99延迟是否突增、错误率是否突增。这三个指标能帮你快速定位问题是出在流量侧、服务侧还是模型侧。定位了大方向再往下查就快多了。