1. 从零手搓AI工程为什么我不建议你直接调包1.1 一个让我彻底改变主意的真实场景去年帮一个朋友排查线上推理服务的问题现象很典型模型在测试集上指标漂亮得不行一上生产环境延迟直接飙到800msGPU利用率却只有30%出头。他第一反应是“模型太大换个小模型”第二反应是“加机器”。我让他把推理链路拆开看结果发现瓶颈根本不在模型本身——数据预处理阶段有个Python循环在逐条处理文本单条耗时就有十几毫秒batch size设成1GPU全程在等CPU喂数据。这个问题用任何“调包式”方案都解决不了因为框架帮你屏蔽了底层细节你根本不知道数据是怎么流的。后来我花了两个晚上用最原始的方式重写了一遍推理管线手动做tokenize的批处理、手动管理KV cache、手动控制数据预取。延迟降到120msGPU利用率拉到85%以上。这件事让我彻底想明白一个道理AI工程不是调包工程。你可以用PyTorch、用Transformers、用vLLM但你必须知道这些工具在背后做了什么否则出了问题你连排查的方向都没有。“ai-engineering-from-scratch”这个项目标题之所以值得认真对待就是因为它代表了一种被很多人忽略的能力——从零开始理解并构建AI系统的每一个环节。1.2 这个项目到底在做什么适合谁看“ai-engineering-from-scratch”从字面拆解就是“从零开始的AI工程”。它不是教你从零训练一个大模型——那是研究机构干的事——而是教你从零搭建一套完整的AI工程体系包括数据处理管线、模型推理服务、特征存储、实验追踪、部署监控这些真正在生产环境里要命的东西。适合看这个内容的人有三类。第一类是有一定Python基础、用过PyTorch或TensorFlow、但从来没自己搭过完整AI系统的开发者。第二类是在小团队里“一个人当三个人用”的全栈工程师老板让你搞个AI功能上线你总不能只丢一个Jupyter Notebook过去。第三类是已经在大厂做AI相关工作的工程师日常被各种内部平台包裹着想搞清楚这些平台底层到底怎么实现的。不适合谁看如果你只是想快速调个API做个demo那确实没必要从零折腾。但如果你想真正理解AI系统为什么这么设计、出了问题怎么定位、性能瓶颈在哪里那从零构建是唯一的路。1.3 从零构建的核心思路分层解耦我见过太多人一上来就想搞个大一统的架构结果代码写到三千行自己都看不懂了。“ai-engineering-from-scratch”的核心思路应该是分层解耦把整个AI工程拆成几个相对独立的层每层只做一件事层与层之间通过明确的接口通信。具体来说我习惯分成五层数据层负责原始数据的读取、清洗、转换特征层负责特征的计算、存储、版本管理模型层负责模型的加载、推理、批处理服务层负责请求路由、限流、熔断监控层负责指标采集、日志聚合、告警。每一层都可以独立替换和升级比如你从PyTorch换到ONNX Runtime只需要改模型层的实现其他层完全不受影响。这种分层的好处是调试极其方便。线上出了问题你只需要沿着调用链一层层往下查看是哪一层的输入输出不符合预期。而不是像调包方案那样一个报错信息丢出来你连它是在哪个环节抛的都不知道。提示分层不是目的解耦才是。如果你的两层之间需要频繁共享状态那说明分层没分对应该考虑合并或者重新划分边界。2. 核心模块拆解每个环节到底在解决什么问题2.1 数据管线别让预处理成为性能黑洞数据管线是AI工程里最容易被低估的环节。很多人觉得“不就是读数据、洗数据吗”但实际生产环境里数据管线往往是整个系统最慢、最不稳定、最难调试的部分。从零构建数据管线核心要解决三个问题吞吐量、一致性、可恢复性。吞吐量好理解你的模型推理可能只要10ms但数据预处理如果花了100ms那整体延迟就被拖垮了。一致性指的是训练和推理时的数据处理逻辑必须完全一致否则会出现“训练时指标很好、上线后效果暴跌”的经典问题。可恢复性是指管线跑了一半挂了能不能从断点继续而不是从头再来。我通常的做法是用生成器模式构建数据管线。每个处理步骤是一个独立的生成器函数通过管道串联起来。这样做的好处是内存占用低——数据是流式处理的不需要一次性全部加载到内存——而且每个步骤可以独立测试和替换。def read_data(source): for record in source: yield record def clean_text(records): for record in records: record[text] record[text].strip().lower() yield record def tokenize(records, tokenizer): for record in records: record[tokens] tokenizer.encode(record[text]) yield record # 串联成管线 pipeline tokenize(clean_text(read_data(source)), tokenizer)这个模式看起来简单但实际用起来非常灵活。你可以在任何一步插入缓存、插入日志、插入异常处理。而且因为每个步骤都是纯函数式的生成器测试起来极其方便——给一个输入检查输出是否符合预期就行。注意生成器模式虽然优雅但在多进程场景下需要额外处理。如果你用PyTorch的DataLoader它内部已经做了多进程封装你只需要把生成器包装成Dataset即可。但如果你自己写多进程管线要小心生成器不能被pickle的问题通常需要改成可序列化的类。2.2 特征存储训练和推理的“同一份真相”特征存储是AI工程里最容易被跳过、但后期最让人头疼的环节。我见过太多团队在训练时用一套代码算特征在推理时用另一套代码算特征结果两边算出来的值对不上模型效果直接崩掉。从零构建特征存储核心目标是训练和推理使用完全相同的特征计算逻辑。实现方式有很多种最简单的是把特征计算逻辑封装成一个独立的模块训练和推理都调用这个模块。但这样做有个问题训练时通常是批量计算推理时是单条计算如果特征计算逻辑里有依赖全局统计量的操作比如归一化批量计算和单条计算的结果可能不一致。更稳妥的做法是引入特征版本管理。每次特征计算逻辑变更就生成一个新的版本号训练和推理都明确指定使用哪个版本。这样即使逻辑变了旧版本的模型仍然能正常工作。class FeatureStore: def __init__(self): self._features {} self._versions {} def register(self, name, version, compute_fn): key f{name}:{version} self._features[key] compute_fn self._versions[name] version def compute(self, name, version, raw_data): key f{name}:{version} if key not in self._features: raise ValueError(fFeature {key} not found) return self._features[key](raw_data)这个简易实现虽然比不上 Feast、Tecton 这些专业特征平台但核心思想是一样的特征计算逻辑必须版本化、可追溯、训练推理共用。等你真正踩过“训练推理特征不一致”的坑就会明白这个设计有多重要。2.3 模型推理批处理、缓存与并发控制模型推理是AI工程里最“重”的环节也是优化空间最大的环节。从零构建推理服务核心要解决三个问题批处理、缓存、并发控制。批处理是指把多个请求合并成一个批次送给模型充分利用GPU的并行计算能力。但批处理会引入延迟——你得等够一定数量的请求才能组成一个批次。这里有个经典的权衡批次越大吞吐越高但延迟也越大。实际生产中通常设置一个最大等待时间比如10ms超时或者凑够批次就立即推理。缓存是指对相同的请求直接返回之前的结果避免重复计算。缓存的难点在于如何判断两个请求是否“相同”。对于文本生成任务如果输入文本完全一样且解码参数temperature、top_p等也完全一样那结果可以缓存。但如果temperature大于0每次生成的结果可能不同缓存就不适用了。并发控制是指限制同时处理的请求数量防止GPU内存溢出。通常用信号量或者队列来实现。我习惯用asyncio的Semaphore配合异步推理接口可以做到既不过载又不浪费资源。import asyncio class InferenceServer: def __init__(self, model, max_concurrent4, max_batch_size8, max_wait_ms10): self.model model self.semaphore asyncio.Semaphore(max_concurrent) self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self._queue [] async def predict(self, input_data): async with self.semaphore: future asyncio.Future() self._queue.append((input_data, future)) if len(self._queue) self.max_batch_size: await self._flush() else: asyncio.get_event_loop().call_later( self.max_wait_ms / 1000, lambda: asyncio.ensure_future(self._flush()) ) return await future async def _flush(self): if not self._queue: return batch self._queue[:self.max_batch_size] self._queue self._queue[self.max_batch_size:] inputs [item[0] for item in batch] results self.model.batch_predict(inputs) for (_, future), result in zip(batch, results): if not future.done(): future.set_result(result)这段代码展示了一个最简化的动态批处理实现。实际生产中还需要考虑超时处理、错误传播、优雅关闭等细节但核心逻辑就是这样。提示批处理的大小不是越大越好。我实测下来对于大多数7B到13B的模型batch size在8到16之间性价比最高。再往上吞吐提升不明显但延迟和显存占用会显著增加。2.4 服务层限流、熔断与优雅降级服务层是AI系统对外的门面也是保证系统稳定性的最后一道防线。从零构建服务层核心要实现三个机制限流、熔断、优雅降级。限流是指限制单位时间内的请求数量防止突发流量打垮后端。最简单的限流是固定窗口计数器但它在窗口边界处会有突刺问题。更平滑的是令牌桶算法以恒定速率往桶里放令牌请求来了就从桶里取令牌取不到就拒绝。熔断是指当后端错误率超过阈值时自动切断请求避免雪崩。熔断器通常有三种状态关闭正常放行、打开直接拒绝、半开放少量请求试探。当错误率超过阈值时从关闭切到打开经过一段时间后切到半开如果试探请求成功则回到关闭否则继续打开。优雅降级是指当系统压力过大时主动关闭一些非核心功能保证核心功能可用。比如当GPU排队时间过长时可以关闭流式输出、降低最大生成长度、甚至返回缓存结果。import time from enum import Enum class CircuitState(Enum): CLOSED closed OPEN open HALF_OPEN half_open class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.state CircuitState.CLOSED self.failure_count 0 self.last_failure_time 0 def call(self, fn, *args, **kwargs): if self.state CircuitState.OPEN: if time.time() - self.last_failure_time self.recovery_timeout: self.state CircuitState.HALF_OPEN else: raise Exception(Circuit is open) try: result fn(*args, **kwargs) if self.state CircuitState.HALF_OPEN: self.state CircuitState.CLOSED self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state CircuitState.OPEN raise e这个熔断器实现虽然简单但已经覆盖了核心逻辑。实际生产中你可能需要更细粒度的控制比如按错误类型区分、按接口维度隔离等。3. 完整实操从零搭建一个可用的推理服务3.1 环境准备与依赖选择动手之前先把环境理清楚。我的建议是能用标准库就用标准库必须用第三方库时选最轻量的。这不是洁癖而是从零构建的核心精神——你引入的每一个依赖都是你未来要理解和维护的负担。Python版本选3.10或3.11这两个版本在性能和稳定性上平衡得最好。3.12虽然更新但有些AI相关的库还没完全适配。虚拟环境用venv就够了没必要上conda除非你需要管理CUDA版本。核心依赖我通常只装这几个numpy做数值计算torch或onnxruntime做模型推理fastapi和uvicorn做HTTP服务pydantic做数据校验。其他的能省则省。像transformers这种库如果你只是做推理完全可以用onnxruntime直接加载ONNX模型省掉一大坨依赖。python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install numpy torch fastapi uvicorn pydantic装完之后检查一下版本确保没有冲突。我遇到过好几次torch和numpy版本不兼容导致段错误的情况排查起来非常痛苦。注意如果你用GPU推理torch的安装命令要去官网查对应的CUDA版本。别直接pip install torch那样装的是CPU版本跑起来慢得让你怀疑人生。3.2 模型加载与推理封装模型加载这块我建议把模型封装成一个类对外只暴露predict和batch_predict两个方法。这样做的好处是模型的具体实现PyTorch还是ONNX、CPU还是GPU对外部完全透明以后想换实现只需要改这个类。import torch import numpy as np class ModelWrapper: def __init__(self, model_path, devicecpu): self.device torch.device(device) self.model torch.jit.load(model_path, map_locationself.device) self.model.eval() torch.no_grad() def predict(self, input_data): tensor torch.tensor(input_data, dtypetorch.float32).to(self.device) if tensor.dim() 1: tensor tensor.unsqueeze(0) output self.model(tensor) return output.cpu().numpy() torch.no_grad() def batch_predict(self, inputs): tensors [torch.tensor(inp, dtypetorch.float32) for inp in inputs] batch torch.stack(tensors).to(self.device) outputs self.model(batch) return outputs.cpu().numpy()这里用torch.jit.load加载TorchScript模型而不是直接加载PyTorch的state_dict。原因是TorchScript模型是序列化的加载速度快而且不依赖原始模型定义代码。如果你用的是ONNX把torch.jit.load换成onnxruntime.InferenceSession就行接口逻辑完全一样。torch.no_grad()装饰器很重要它告诉PyTorch不要构建计算图能省不少内存。model.eval()也很关键它会把Dropout和BatchNorm切到推理模式否则结果会不稳定。3.3 HTTP服务与请求处理服务层用FastAPI因为它原生支持异步而且自动生成API文档调试起来很方便。核心是要处理好请求的解析、校验、路由和响应。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import asyncio app FastAPI() model None inference_server None class PredictRequest(BaseModel): inputs: List[List[float]] max_batch_size: int 8 class PredictResponse(BaseModel): outputs: List[List[float]] latency_ms: float app.on_event(startup) async def startup(): global model, inference_server model ModelWrapper(model.pt, devicecuda) inference_server InferenceServer(model) app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): import time start time.time() try: results await asyncio.gather(*[ inference_server.predict(inp) for inp in request.inputs ]) except Exception as e: raise HTTPException(status_code500, detailstr(e)) latency (time.time() - start) * 1000 return PredictResponse( outputs[r.tolist() for r in results], latency_mslatency )这段代码里有个细节值得说asyncio.gather并发处理多个输入。因为推理是IO密集型和计算密集型混合的操作用异步并发可以充分利用等待GPU计算的时间来处理其他请求。当然前提是你的推理服务内部做了批处理否则并发请求只会让GPU排队更长。启动命令很简单uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1注意--workers参数。对于GPU推理服务通常设成1因为多个worker会争抢GPU资源反而降低效率。如果你有多个GPU可以每个GPU起一个worker然后用Nginx做负载均衡。3.4 监控指标与日志采集服务跑起来只是第一步能观测才是关键。我习惯在服务里内置几个核心指标请求总数、请求延迟分布、错误率、GPU利用率、显存占用。这些指标用Prometheus的格式暴露出来配合Grafana做可视化。from prometheus_client import Counter, Histogram, Gauge, generate_latest from fastapi import Response REQUEST_COUNT Counter(inference_requests_total, Total requests, [status]) REQUEST_LATENCY Histogram(inference_latency_ms, Request latency in ms) GPU_MEMORY Gauge(gpu_memory_used_mb, GPU memory used in MB) app.get(/metrics) async def metrics(): if torch.cuda.is_available(): GPU_MEMORY.set(torch.cuda.memory_allocated() / 1024 / 1024) return Response(generate_latest(), media_typetext/plain)日志这块我建议用结构化日志每条日志是一个JSON对象包含时间戳、请求ID、输入摘要、输出摘要、延迟、错误信息。这样后期做日志分析的时候可以直接用jq或者ELK不用写正则去解析。import logging import json class JsonFormatter(logging.Formatter): def format(self, record): log_obj { timestamp: self.formatTime(record), level: record.levelname, message: record.getMessage(), } if hasattr(record, request_id): log_obj[request_id] record.request_id return json.dumps(log_obj) logger logging.getLogger(inference) handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO)提示日志里不要记录完整的输入输出尤其是涉及用户数据的时候。记录摘要或者哈希值就够了既方便排查问题又避免隐私泄露。4. 踩坑实录那些让我熬夜到凌晨三点的典型问题4.1 内存泄漏为什么服务跑一天就崩这是最经典的问题。服务刚启动时内存占用2GB跑一天后变成16GB然后OOM崩溃。排查内存泄漏最有效的工具是tracemalloc和objgraph但更关键的是知道常见泄漏点在哪里。第一个常见泄漏点是全局缓存没有淘汰策略。我见过有人用字典做缓存只往里塞从不清理跑久了内存自然爆炸。解决办法是用functools.lru_cache或者自己实现一个带TTL的缓存。第二个泄漏点是PyTorch的CUDA缓存。PyTorch会缓存GPU内存以加速后续分配但如果你的输入尺寸变化很大缓存会越积越多。解决办法是定期调用torch.cuda.empty_cache()或者在服务启动时设置PYTORCH_CUDA_ALLOC_CONF环境变量来调整分配策略。第三个泄漏点是异步任务没有正确取消。如果你用asyncio.create_task创建了任务但没有保存引用任务可能会在后台一直运行持有资源不释放。解决办法是用asyncio.TaskGroup或者手动管理任务生命周期。import gc import torch def periodic_cleanup(): gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() # 每处理1000个请求清理一次 request_counter 0 def maybe_cleanup(): global request_counter request_counter 1 if request_counter % 1000 0: periodic_cleanup()4.2 批处理导致的延迟毛刺动态批处理虽然能提升吞吐但会引入延迟毛刺。我遇到过一种情况P99延迟突然从50ms飙到500ms但平均延迟没怎么变。排查后发现是批处理逻辑的问题——当请求量突然增大时批次被填满的速度变快但_flush方法是异步调用的多个flush任务并发执行导致GPU上同时跑了多个批次互相争抢资源。解决办法是给flush操作加锁保证同一时间只有一个批次在GPU上执行。同时设置一个最大队列长度超过就拒绝新请求而不是无限排队。class InferenceServer: def __init__(self, model, max_queue_size100): self._flush_lock asyncio.Lock() self.max_queue_size max_queue_size # ... 其他初始化 async def predict(self, input_data): if len(self._queue) self.max_queue_size: raise Exception(Queue is full) # ... 其余逻辑 async def _flush(self): async with self._flush_lock: if not self._queue: return # ... 批处理逻辑4.3 模型版本更新导致的兼容性问题模型更新是另一个容易出问题的环节。你训练了一个新模型指标比旧模型好兴冲冲地部署上去结果服务直接挂了。原因可能是新模型的输入输出格式变了、依赖的库版本变了、甚至只是权重文件的保存方式变了。我的经验是模型更新必须走灰度流程。新模型先部署到一个独立的实例上用少量真实流量测试确认没问题后再逐步扩大流量比例。同时要保留旧模型的权重文件万一新模型出问题可以快速回滚。class ModelRouter: def __init__(self, primary_model, canary_modelNone, canary_ratio0.0): self.primary primary_model self.canary canary_model self.canary_ratio canary_ratio def get_model(self): import random if self.canary and random.random() self.canary_ratio: return self.canary return self.primary这个路由逻辑很简单但配合监控指标就能实现安全的灰度发布。你可以观察canary模型的错误率和延迟如果明显差于primary就把canary_ratio调回0。4.4 常见问题速查表问题现象可能原因排查方法解决方案服务启动后第一次请求特别慢模型懒加载、CUDA初始化看启动日志时间戳启动时预热一次推理GPU利用率低但延迟高数据预处理瓶颈、批处理未生效用py-spy看CPU热点优化预处理、检查批处理逻辑内存持续增长不释放缓存无淘汰、CUDA缓存累积tracemalloc、nvidia-smi加TTL缓存、定期empty_cacheP99延迟远高于P50批处理毛刺、GC停顿分位数监控、GC日志加flush锁、调整GC阈值模型输出不稳定Dropout未关闭、随机种子未固定对比多次推理结果model.eval()、固定随机种子并发请求下结果错乱共享状态未隔离、线程不安全压测复现、代码审查加锁、用不可变数据结构注意这张表里的“解决方案”都是治标真正治本的方法是理解每个问题背后的原理。比如“GPU利用率低”可能是数据管线的问题也可能是批处理的问题还可能是模型本身计算量太小。只有理解了整个链路才能快速定位到根因。5. 从能跑到好用性能优化的几个关键抓手5.1 数据预取的流水线设计数据预取是提升GPU利用率最有效的手段之一。核心思想是在GPU计算当前批次的同时CPU已经在准备下一个批次的数据。这样GPU永远不用等数据利用率自然就上去了。实现方式是用一个后台线程或协程专门做数据准备准备好的数据放入一个有限大小的队列。GPU计算线程从队列取数据取不到就等待。队列的大小需要权衡太小了起不到缓冲作用太大了内存占用高且延迟大。我通常设成2到4。import threading import queue class DataPrefetcher: def __init__(self, data_source, batch_size8, queue_size4): self.queue queue.Queue(maxsizequeue_size) self.data_source data_source self.batch_size batch_size self._stop threading.Event() self._thread threading.Thread(targetself._worker, daemonTrue) self._thread.start() def _worker(self): batch [] for item in self.data_source: if self._stop.is_set(): break batch.append(item) if len(batch) self.batch_size: self.queue.put(batch) batch [] if batch: self.queue.put(batch) def __iter__(self): while True: try: batch self.queue.get(timeout1) yield batch except queue.Empty: if self._stop.is_set(): break def stop(self): self._stop.set() self._thread.join(timeout5)这个预取器的逻辑很直白后台线程不断从数据源读数据、组批次、放入队列前台线程从队列取批次做推理。队列满的时候后台线程会阻塞防止内存无限增长。5.2 量化与推理加速的取舍量化是另一个常用的加速手段把FP32的权重转成INT8或FP16能显著减少显存占用和计算量。但量化会带来精度损失需要根据具体任务权衡。我的一般原则是先试FP16不够再试INT8。FP16通常能带来1.5到2倍的加速精度损失几乎可以忽略。INT8能带来3到4倍加速但精度损失可能需要仔细评估。对于分类任务INT8通常没问题对于生成任务INT8可能会导致输出质量明显下降。PyTorch的量化工具链比较成熟动态量化一行代码就能搞定import torch.quantization # 动态量化适用于LSTM、Linear等 quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 ) # 静态量化需要校准数据 model.qconfig torch.quantization.get_default_qconfig(fbgemm) torch.quantization.prepare(model, inplaceTrue) # 用校准数据跑一遍 for batch in calibration_data: model(batch) torch.quantization.convert(model, inplaceTrue)提示量化后的模型一定要做精度对比测试。我见过有人量化完直接上线结果模型输出全是乱码回滚都来不及。测试方法很简单用同一批测试数据对比量化前后的输出差异如果差异超过阈值就不能用。5.3 缓存策略什么该缓存什么不该缓存缓存用得好能大幅降低延迟和计算量用不好会导致结果不一致甚至内存爆炸。我的经验是只缓存确定性计算的结果。确定性计算是指同样的输入永远产生同样的输出。比如特征提取、embedding计算、分类模型的推理结果这些都是确定性的可以缓存。但生成模型的输出尤其是temperature大于0时是不确定的缓存了反而会导致用户每次看到一样的结果体验很差。缓存的key设计也很关键。对于文本输入直接用原文做key可能太长通常用哈希值。但哈希有碰撞风险所以最好用SHA256这种碰撞概率极低的算法。对于数值输入要注意浮点数的精度问题通常需要先量化再哈希。import hashlib import json def make_cache_key(input_data, model_version): if isinstance(input_data, str): content input_data else: content json.dumps(input_data, sort_keysTrue) raw f{model_version}:{content} return hashlib.sha256(raw.encode()).hexdigest()缓存淘汰策略我通常用LRU最近最少使用配合TTL生存时间。LRU保证热数据留在缓存里TTL保证冷数据不会永远占着内存。Python的cachetools库提供了现成的实现没必要自己造轮子。5.4 压测与容量规划服务上线前一定要做压测搞清楚系统的容量边界在哪里。压测工具我用locust因为它用Python写测试脚本和我们的技术栈一致学习成本低。压测的核心是找到三个关键指标最大QPS、P99延迟拐点、错误率拐点。最大QPS是指系统能稳定处理的最大请求速率。P99延迟拐点是指延迟开始急剧上升的QPS值。错误率拐点是指错误率开始明显上升的QPS值。这三个值通常不一样容量规划要以最小的那个为准再留30%的余量。from locust import HttpUser, task, between class InferenceUser(HttpUser): wait_time between(0.01, 0.05) task def predict(self): self.client.post(/predict, json{ inputs: [[0.1] * 128] })压测的时候要注意几点一是压测环境要和生产环境尽量一致否则数据没有参考价值二是要逐步增加压力不要一上来就满负荷否则测不出拐点三是要监控服务端的资源使用情况CPU、内存、GPU、网络都要看瓶颈往往不在你以为的地方。6. 工程化收尾让项目从“能跑”到“能维护”6.1 配置管理与环境隔离从零构建的项目最容易犯的错误是把配置硬编码在代码里。数据库地址、模型路径、批处理大小、超时时间这些都应该抽出来放到配置文件或环境变量里。我习惯用YAML做配置文件因为它比JSON可读性好比INI表达能力强。配置加载用pydantic的BaseSettings它能自动从环境变量读取配置还能做类型校验。from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): model_path: str Field(..., descriptionPath to model file) device: str Field(cpu, descriptionDevice: cpu or cuda) max_batch_size: int Field(8, ge1, le64) max_wait_ms: int Field(10, ge1, le1000) max_queue_size: int Field(100, ge10) class Config: env_prefix AI_ENG_ env_file .env settings Settings()这样配置既可以通过环境变量覆盖比如AI_ENG_MAX_BATCH_SIZE16也可以写在.env文件里。不同环境开发、测试、生产用不同的.env文件代码完全不用改。6.2 测试策略单元测试、集成测试与端到端测试从零构建的项目如果没有测试后期维护会非常痛苦。我的测试策略是三层单元测试覆盖核心逻辑集成测试覆盖模块间交互端到端测试覆盖完整链路。单元测试用pytest重点测数据管线的每个步骤、特征计算逻辑、批处理逻辑。这些测试跑得很快每次改代码都应该跑一遍。集成测试测模型加载、推理服务、缓存、熔断这些模块的交互。这些测试需要加载真实模型跑得慢一些通常只在合并代码前跑。端到端测试模拟真实请求从HTTP接口打到模型推理再返回结果。这个最慢但最能发现问题。我通常用httpx的测试客户端来做。import pytest from fastapi.testclient import TestClient from main import app pytest.fixture def client(): with TestClient(app) as c: yield c def test_predict_endpoint(client): response client.post(/predict, json{ inputs: [[0.1] * 128] }) assert response.status_code 200 data response.json() assert outputs in data assert len(data[outputs]) 1 assert data[latency_ms] 0注意端到端测试里不要用真实的大模型用一个小的随机模型就行。测试的目的是验证链路通不通不是验证模型效果好不好。用大模型会让测试跑得极慢最后没人愿意跑。6.3 部署与回滚的标准化流程部署不是把代码传到服务器上跑起来就完事了。标准化的部署流程应该包括构建镜像、推送镜像、更新配置、滚动重启、健康检查、流量切换。每一步都要可回滚。我用Docker做镜像Dockerfile尽量精简只装必要的依赖。镜像打标签用Git commit hash这样能精确追溯到代码版本。FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]部署的时候用docker-compose或者Kubernetes关键是配置健康检查接口。健康检查接口要检查模型是否加载成功、GPU是否可用、依赖服务是否可达。只有健康检查通过才把流量切过来。app.get(/health) async def health(): checks { model_loaded: model is not None, gpu_available: torch.cuda.is_available() if settings.device cuda else True, } if not all(checks.values()): raise HTTPException(status_code503, detailchecks) return {status: healthy, checks: checks}回滚就更简单了把镜像标签切回上一个版本重新部署。前提是你保留了旧版本的镜像而且数据库 schema 没有不兼容的变更。所以数据库变更一定要向前兼容至少保留两个版本的兼容性。6.4 文档与知识沉淀最后说一个最容易被忽略但长期价值最高的环节文档。从零构建的项目如果不写文档三个月后你自己都忘了某个模块为什么这么设计。文档不需要多正式但至少要包含架构图用文字描述也行、每个模块的职责和接口、关键设计决策的理由、常见问题的排查方法。我习惯在代码仓库里放一个DESIGN.md每次做重要变更时更新它。# 推理服务设计文档 ## 模块划分 - data/: 数据管线负责读取、清洗、tokenize - features/: 特征存储负责特征计算和版本管理 - model/: 模型封装负责加载和推理 - server/: HTTP服务负责路由、限流、熔断 - monitor/: 监控负责指标采集和日志 ## 关键决策 - 为什么用生成器而不是列表内存占用低支持流式处理 - 为什么用动态批处理而不是固定批处理请求量波动大固定批处理浪费资源 - 为什么用SHA256做缓存key碰撞概率极低避免缓存错乱 ## 已知问题 - 批处理在请求量突增时会有延迟毛刺正在优化 - INT8量化在生成任务上效果下降明显暂不启用这份文档不需要写得多漂亮关键是持续更新。每次踩坑之后把问题和解决方案记进去下次遇到类似问题就能快速定位。我个人在实际操作中的体会是从零构建AI工程系统最大的价值不在于省了多少钱或者性能提升了多少而在于你对整个系统的掌控力。你知道每个环节在做什么、为什么这么做、出了问题该去哪里找。这种掌控力是调包永远给不了的。当然从零构建不意味着拒绝一切现成工具而是说你要理解工具背后的原理在需要的时候能够自己实现一个简化版。等你真正手搓过一遍之后再用那些成熟的框架你会发现自己的理解完全不一样了。 从零手搓AI工程：拒绝调包，构建高性能推理服务
1. 从零手搓AI工程为什么我不建议你直接调包1.1 一个让我彻底改变主意的真实场景去年帮一个朋友排查线上推理服务的问题现象很典型模型在测试集上指标漂亮得不行一上生产环境延迟直接飙到800msGPU利用率却只有30%出头。他第一反应是“模型太大换个小模型”第二反应是“加机器”。我让他把推理链路拆开看结果发现瓶颈根本不在模型本身——数据预处理阶段有个Python循环在逐条处理文本单条耗时就有十几毫秒batch size设成1GPU全程在等CPU喂数据。这个问题用任何“调包式”方案都解决不了因为框架帮你屏蔽了底层细节你根本不知道数据是怎么流的。后来我花了两个晚上用最原始的方式重写了一遍推理管线手动做tokenize的批处理、手动管理KV cache、手动控制数据预取。延迟降到120msGPU利用率拉到85%以上。这件事让我彻底想明白一个道理AI工程不是调包工程。你可以用PyTorch、用Transformers、用vLLM但你必须知道这些工具在背后做了什么否则出了问题你连排查的方向都没有。“ai-engineering-from-scratch”这个项目标题之所以值得认真对待就是因为它代表了一种被很多人忽略的能力——从零开始理解并构建AI系统的每一个环节。1.2 这个项目到底在做什么适合谁看“ai-engineering-from-scratch”从字面拆解就是“从零开始的AI工程”。它不是教你从零训练一个大模型——那是研究机构干的事——而是教你从零搭建一套完整的AI工程体系包括数据处理管线、模型推理服务、特征存储、实验追踪、部署监控这些真正在生产环境里要命的东西。适合看这个内容的人有三类。第一类是有一定Python基础、用过PyTorch或TensorFlow、但从来没自己搭过完整AI系统的开发者。第二类是在小团队里“一个人当三个人用”的全栈工程师老板让你搞个AI功能上线你总不能只丢一个Jupyter Notebook过去。第三类是已经在大厂做AI相关工作的工程师日常被各种内部平台包裹着想搞清楚这些平台底层到底怎么实现的。不适合谁看如果你只是想快速调个API做个demo那确实没必要从零折腾。但如果你想真正理解AI系统为什么这么设计、出了问题怎么定位、性能瓶颈在哪里那从零构建是唯一的路。1.3 从零构建的核心思路分层解耦我见过太多人一上来就想搞个大一统的架构结果代码写到三千行自己都看不懂了。“ai-engineering-from-scratch”的核心思路应该是分层解耦把整个AI工程拆成几个相对独立的层每层只做一件事层与层之间通过明确的接口通信。具体来说我习惯分成五层数据层负责原始数据的读取、清洗、转换特征层负责特征的计算、存储、版本管理模型层负责模型的加载、推理、批处理服务层负责请求路由、限流、熔断监控层负责指标采集、日志聚合、告警。每一层都可以独立替换和升级比如你从PyTorch换到ONNX Runtime只需要改模型层的实现其他层完全不受影响。这种分层的好处是调试极其方便。线上出了问题你只需要沿着调用链一层层往下查看是哪一层的输入输出不符合预期。而不是像调包方案那样一个报错信息丢出来你连它是在哪个环节抛的都不知道。提示分层不是目的解耦才是。如果你的两层之间需要频繁共享状态那说明分层没分对应该考虑合并或者重新划分边界。2. 核心模块拆解每个环节到底在解决什么问题2.1 数据管线别让预处理成为性能黑洞数据管线是AI工程里最容易被低估的环节。很多人觉得“不就是读数据、洗数据吗”但实际生产环境里数据管线往往是整个系统最慢、最不稳定、最难调试的部分。从零构建数据管线核心要解决三个问题吞吐量、一致性、可恢复性。吞吐量好理解你的模型推理可能只要10ms但数据预处理如果花了100ms那整体延迟就被拖垮了。一致性指的是训练和推理时的数据处理逻辑必须完全一致否则会出现“训练时指标很好、上线后效果暴跌”的经典问题。可恢复性是指管线跑了一半挂了能不能从断点继续而不是从头再来。我通常的做法是用生成器模式构建数据管线。每个处理步骤是一个独立的生成器函数通过管道串联起来。这样做的好处是内存占用低——数据是流式处理的不需要一次性全部加载到内存——而且每个步骤可以独立测试和替换。def read_data(source): for record in source: yield record def clean_text(records): for record in records: record[text] record[text].strip().lower() yield record def tokenize(records, tokenizer): for record in records: record[tokens] tokenizer.encode(record[text]) yield record # 串联成管线 pipeline tokenize(clean_text(read_data(source)), tokenizer)这个模式看起来简单但实际用起来非常灵活。你可以在任何一步插入缓存、插入日志、插入异常处理。而且因为每个步骤都是纯函数式的生成器测试起来极其方便——给一个输入检查输出是否符合预期就行。注意生成器模式虽然优雅但在多进程场景下需要额外处理。如果你用PyTorch的DataLoader它内部已经做了多进程封装你只需要把生成器包装成Dataset即可。但如果你自己写多进程管线要小心生成器不能被pickle的问题通常需要改成可序列化的类。2.2 特征存储训练和推理的“同一份真相”特征存储是AI工程里最容易被跳过、但后期最让人头疼的环节。我见过太多团队在训练时用一套代码算特征在推理时用另一套代码算特征结果两边算出来的值对不上模型效果直接崩掉。从零构建特征存储核心目标是训练和推理使用完全相同的特征计算逻辑。实现方式有很多种最简单的是把特征计算逻辑封装成一个独立的模块训练和推理都调用这个模块。但这样做有个问题训练时通常是批量计算推理时是单条计算如果特征计算逻辑里有依赖全局统计量的操作比如归一化批量计算和单条计算的结果可能不一致。更稳妥的做法是引入特征版本管理。每次特征计算逻辑变更就生成一个新的版本号训练和推理都明确指定使用哪个版本。这样即使逻辑变了旧版本的模型仍然能正常工作。class FeatureStore: def __init__(self): self._features {} self._versions {} def register(self, name, version, compute_fn): key f{name}:{version} self._features[key] compute_fn self._versions[name] version def compute(self, name, version, raw_data): key f{name}:{version} if key not in self._features: raise ValueError(fFeature {key} not found) return self._features[key](raw_data)这个简易实现虽然比不上 Feast、Tecton 这些专业特征平台但核心思想是一样的特征计算逻辑必须版本化、可追溯、训练推理共用。等你真正踩过“训练推理特征不一致”的坑就会明白这个设计有多重要。2.3 模型推理批处理、缓存与并发控制模型推理是AI工程里最“重”的环节也是优化空间最大的环节。从零构建推理服务核心要解决三个问题批处理、缓存、并发控制。批处理是指把多个请求合并成一个批次送给模型充分利用GPU的并行计算能力。但批处理会引入延迟——你得等够一定数量的请求才能组成一个批次。这里有个经典的权衡批次越大吞吐越高但延迟也越大。实际生产中通常设置一个最大等待时间比如10ms超时或者凑够批次就立即推理。缓存是指对相同的请求直接返回之前的结果避免重复计算。缓存的难点在于如何判断两个请求是否“相同”。对于文本生成任务如果输入文本完全一样且解码参数temperature、top_p等也完全一样那结果可以缓存。但如果temperature大于0每次生成的结果可能不同缓存就不适用了。并发控制是指限制同时处理的请求数量防止GPU内存溢出。通常用信号量或者队列来实现。我习惯用asyncio的Semaphore配合异步推理接口可以做到既不过载又不浪费资源。import asyncio class InferenceServer: def __init__(self, model, max_concurrent4, max_batch_size8, max_wait_ms10): self.model model self.semaphore asyncio.Semaphore(max_concurrent) self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self._queue [] async def predict(self, input_data): async with self.semaphore: future asyncio.Future() self._queue.append((input_data, future)) if len(self._queue) self.max_batch_size: await self._flush() else: asyncio.get_event_loop().call_later( self.max_wait_ms / 1000, lambda: asyncio.ensure_future(self._flush()) ) return await future async def _flush(self): if not self._queue: return batch self._queue[:self.max_batch_size] self._queue self._queue[self.max_batch_size:] inputs [item[0] for item in batch] results self.model.batch_predict(inputs) for (_, future), result in zip(batch, results): if not future.done(): future.set_result(result)这段代码展示了一个最简化的动态批处理实现。实际生产中还需要考虑超时处理、错误传播、优雅关闭等细节但核心逻辑就是这样。提示批处理的大小不是越大越好。我实测下来对于大多数7B到13B的模型batch size在8到16之间性价比最高。再往上吞吐提升不明显但延迟和显存占用会显著增加。2.4 服务层限流、熔断与优雅降级服务层是AI系统对外的门面也是保证系统稳定性的最后一道防线。从零构建服务层核心要实现三个机制限流、熔断、优雅降级。限流是指限制单位时间内的请求数量防止突发流量打垮后端。最简单的限流是固定窗口计数器但它在窗口边界处会有突刺问题。更平滑的是令牌桶算法以恒定速率往桶里放令牌请求来了就从桶里取令牌取不到就拒绝。熔断是指当后端错误率超过阈值时自动切断请求避免雪崩。熔断器通常有三种状态关闭正常放行、打开直接拒绝、半开放少量请求试探。当错误率超过阈值时从关闭切到打开经过一段时间后切到半开如果试探请求成功则回到关闭否则继续打开。优雅降级是指当系统压力过大时主动关闭一些非核心功能保证核心功能可用。比如当GPU排队时间过长时可以关闭流式输出、降低最大生成长度、甚至返回缓存结果。import time from enum import Enum class CircuitState(Enum): CLOSED closed OPEN open HALF_OPEN half_open class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.state CircuitState.CLOSED self.failure_count 0 self.last_failure_time 0 def call(self, fn, *args, **kwargs): if self.state CircuitState.OPEN: if time.time() - self.last_failure_time self.recovery_timeout: self.state CircuitState.HALF_OPEN else: raise Exception(Circuit is open) try: result fn(*args, **kwargs) if self.state CircuitState.HALF_OPEN: self.state CircuitState.CLOSED self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state CircuitState.OPEN raise e这个熔断器实现虽然简单但已经覆盖了核心逻辑。实际生产中你可能需要更细粒度的控制比如按错误类型区分、按接口维度隔离等。3. 完整实操从零搭建一个可用的推理服务3.1 环境准备与依赖选择动手之前先把环境理清楚。我的建议是能用标准库就用标准库必须用第三方库时选最轻量的。这不是洁癖而是从零构建的核心精神——你引入的每一个依赖都是你未来要理解和维护的负担。Python版本选3.10或3.11这两个版本在性能和稳定性上平衡得最好。3.12虽然更新但有些AI相关的库还没完全适配。虚拟环境用venv就够了没必要上conda除非你需要管理CUDA版本。核心依赖我通常只装这几个numpy做数值计算torch或onnxruntime做模型推理fastapi和uvicorn做HTTP服务pydantic做数据校验。其他的能省则省。像transformers这种库如果你只是做推理完全可以用onnxruntime直接加载ONNX模型省掉一大坨依赖。python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install numpy torch fastapi uvicorn pydantic装完之后检查一下版本确保没有冲突。我遇到过好几次torch和numpy版本不兼容导致段错误的情况排查起来非常痛苦。注意如果你用GPU推理torch的安装命令要去官网查对应的CUDA版本。别直接pip install torch那样装的是CPU版本跑起来慢得让你怀疑人生。3.2 模型加载与推理封装模型加载这块我建议把模型封装成一个类对外只暴露predict和batch_predict两个方法。这样做的好处是模型的具体实现PyTorch还是ONNX、CPU还是GPU对外部完全透明以后想换实现只需要改这个类。import torch import numpy as np class ModelWrapper: def __init__(self, model_path, devicecpu): self.device torch.device(device) self.model torch.jit.load(model_path, map_locationself.device) self.model.eval() torch.no_grad() def predict(self, input_data): tensor torch.tensor(input_data, dtypetorch.float32).to(self.device) if tensor.dim() 1: tensor tensor.unsqueeze(0) output self.model(tensor) return output.cpu().numpy() torch.no_grad() def batch_predict(self, inputs): tensors [torch.tensor(inp, dtypetorch.float32) for inp in inputs] batch torch.stack(tensors).to(self.device) outputs self.model(batch) return outputs.cpu().numpy()这里用torch.jit.load加载TorchScript模型而不是直接加载PyTorch的state_dict。原因是TorchScript模型是序列化的加载速度快而且不依赖原始模型定义代码。如果你用的是ONNX把torch.jit.load换成onnxruntime.InferenceSession就行接口逻辑完全一样。torch.no_grad()装饰器很重要它告诉PyTorch不要构建计算图能省不少内存。model.eval()也很关键它会把Dropout和BatchNorm切到推理模式否则结果会不稳定。3.3 HTTP服务与请求处理服务层用FastAPI因为它原生支持异步而且自动生成API文档调试起来很方便。核心是要处理好请求的解析、校验、路由和响应。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import asyncio app FastAPI() model None inference_server None class PredictRequest(BaseModel): inputs: List[List[float]] max_batch_size: int 8 class PredictResponse(BaseModel): outputs: List[List[float]] latency_ms: float app.on_event(startup) async def startup(): global model, inference_server model ModelWrapper(model.pt, devicecuda) inference_server InferenceServer(model) app.post(/predict, response_modelPredictResponse) async def predict(request: PredictRequest): import time start time.time() try: results await asyncio.gather(*[ inference_server.predict(inp) for inp in request.inputs ]) except Exception as e: raise HTTPException(status_code500, detailstr(e)) latency (time.time() - start) * 1000 return PredictResponse( outputs[r.tolist() for r in results], latency_mslatency )这段代码里有个细节值得说asyncio.gather并发处理多个输入。因为推理是IO密集型和计算密集型混合的操作用异步并发可以充分利用等待GPU计算的时间来处理其他请求。当然前提是你的推理服务内部做了批处理否则并发请求只会让GPU排队更长。启动命令很简单uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1注意--workers参数。对于GPU推理服务通常设成1因为多个worker会争抢GPU资源反而降低效率。如果你有多个GPU可以每个GPU起一个worker然后用Nginx做负载均衡。3.4 监控指标与日志采集服务跑起来只是第一步能观测才是关键。我习惯在服务里内置几个核心指标请求总数、请求延迟分布、错误率、GPU利用率、显存占用。这些指标用Prometheus的格式暴露出来配合Grafana做可视化。from prometheus_client import Counter, Histogram, Gauge, generate_latest from fastapi import Response REQUEST_COUNT Counter(inference_requests_total, Total requests, [status]) REQUEST_LATENCY Histogram(inference_latency_ms, Request latency in ms) GPU_MEMORY Gauge(gpu_memory_used_mb, GPU memory used in MB) app.get(/metrics) async def metrics(): if torch.cuda.is_available(): GPU_MEMORY.set(torch.cuda.memory_allocated() / 1024 / 1024) return Response(generate_latest(), media_typetext/plain)日志这块我建议用结构化日志每条日志是一个JSON对象包含时间戳、请求ID、输入摘要、输出摘要、延迟、错误信息。这样后期做日志分析的时候可以直接用jq或者ELK不用写正则去解析。import logging import json class JsonFormatter(logging.Formatter): def format(self, record): log_obj { timestamp: self.formatTime(record), level: record.levelname, message: record.getMessage(), } if hasattr(record, request_id): log_obj[request_id] record.request_id return json.dumps(log_obj) logger logging.getLogger(inference) handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO)提示日志里不要记录完整的输入输出尤其是涉及用户数据的时候。记录摘要或者哈希值就够了既方便排查问题又避免隐私泄露。4. 踩坑实录那些让我熬夜到凌晨三点的典型问题4.1 内存泄漏为什么服务跑一天就崩这是最经典的问题。服务刚启动时内存占用2GB跑一天后变成16GB然后OOM崩溃。排查内存泄漏最有效的工具是tracemalloc和objgraph但更关键的是知道常见泄漏点在哪里。第一个常见泄漏点是全局缓存没有淘汰策略。我见过有人用字典做缓存只往里塞从不清理跑久了内存自然爆炸。解决办法是用functools.lru_cache或者自己实现一个带TTL的缓存。第二个泄漏点是PyTorch的CUDA缓存。PyTorch会缓存GPU内存以加速后续分配但如果你的输入尺寸变化很大缓存会越积越多。解决办法是定期调用torch.cuda.empty_cache()或者在服务启动时设置PYTORCH_CUDA_ALLOC_CONF环境变量来调整分配策略。第三个泄漏点是异步任务没有正确取消。如果你用asyncio.create_task创建了任务但没有保存引用任务可能会在后台一直运行持有资源不释放。解决办法是用asyncio.TaskGroup或者手动管理任务生命周期。import gc import torch def periodic_cleanup(): gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() # 每处理1000个请求清理一次 request_counter 0 def maybe_cleanup(): global request_counter request_counter 1 if request_counter % 1000 0: periodic_cleanup()4.2 批处理导致的延迟毛刺动态批处理虽然能提升吞吐但会引入延迟毛刺。我遇到过一种情况P99延迟突然从50ms飙到500ms但平均延迟没怎么变。排查后发现是批处理逻辑的问题——当请求量突然增大时批次被填满的速度变快但_flush方法是异步调用的多个flush任务并发执行导致GPU上同时跑了多个批次互相争抢资源。解决办法是给flush操作加锁保证同一时间只有一个批次在GPU上执行。同时设置一个最大队列长度超过就拒绝新请求而不是无限排队。class InferenceServer: def __init__(self, model, max_queue_size100): self._flush_lock asyncio.Lock() self.max_queue_size max_queue_size # ... 其他初始化 async def predict(self, input_data): if len(self._queue) self.max_queue_size: raise Exception(Queue is full) # ... 其余逻辑 async def _flush(self): async with self._flush_lock: if not self._queue: return # ... 批处理逻辑4.3 模型版本更新导致的兼容性问题模型更新是另一个容易出问题的环节。你训练了一个新模型指标比旧模型好兴冲冲地部署上去结果服务直接挂了。原因可能是新模型的输入输出格式变了、依赖的库版本变了、甚至只是权重文件的保存方式变了。我的经验是模型更新必须走灰度流程。新模型先部署到一个独立的实例上用少量真实流量测试确认没问题后再逐步扩大流量比例。同时要保留旧模型的权重文件万一新模型出问题可以快速回滚。class ModelRouter: def __init__(self, primary_model, canary_modelNone, canary_ratio0.0): self.primary primary_model self.canary canary_model self.canary_ratio canary_ratio def get_model(self): import random if self.canary and random.random() self.canary_ratio: return self.canary return self.primary这个路由逻辑很简单但配合监控指标就能实现安全的灰度发布。你可以观察canary模型的错误率和延迟如果明显差于primary就把canary_ratio调回0。4.4 常见问题速查表问题现象可能原因排查方法解决方案服务启动后第一次请求特别慢模型懒加载、CUDA初始化看启动日志时间戳启动时预热一次推理GPU利用率低但延迟高数据预处理瓶颈、批处理未生效用py-spy看CPU热点优化预处理、检查批处理逻辑内存持续增长不释放缓存无淘汰、CUDA缓存累积tracemalloc、nvidia-smi加TTL缓存、定期empty_cacheP99延迟远高于P50批处理毛刺、GC停顿分位数监控、GC日志加flush锁、调整GC阈值模型输出不稳定Dropout未关闭、随机种子未固定对比多次推理结果model.eval()、固定随机种子并发请求下结果错乱共享状态未隔离、线程不安全压测复现、代码审查加锁、用不可变数据结构注意这张表里的“解决方案”都是治标真正治本的方法是理解每个问题背后的原理。比如“GPU利用率低”可能是数据管线的问题也可能是批处理的问题还可能是模型本身计算量太小。只有理解了整个链路才能快速定位到根因。5. 从能跑到好用性能优化的几个关键抓手5.1 数据预取的流水线设计数据预取是提升GPU利用率最有效的手段之一。核心思想是在GPU计算当前批次的同时CPU已经在准备下一个批次的数据。这样GPU永远不用等数据利用率自然就上去了。实现方式是用一个后台线程或协程专门做数据准备准备好的数据放入一个有限大小的队列。GPU计算线程从队列取数据取不到就等待。队列的大小需要权衡太小了起不到缓冲作用太大了内存占用高且延迟大。我通常设成2到4。import threading import queue class DataPrefetcher: def __init__(self, data_source, batch_size8, queue_size4): self.queue queue.Queue(maxsizequeue_size) self.data_source data_source self.batch_size batch_size self._stop threading.Event() self._thread threading.Thread(targetself._worker, daemonTrue) self._thread.start() def _worker(self): batch [] for item in self.data_source: if self._stop.is_set(): break batch.append(item) if len(batch) self.batch_size: self.queue.put(batch) batch [] if batch: self.queue.put(batch) def __iter__(self): while True: try: batch self.queue.get(timeout1) yield batch except queue.Empty: if self._stop.is_set(): break def stop(self): self._stop.set() self._thread.join(timeout5)这个预取器的逻辑很直白后台线程不断从数据源读数据、组批次、放入队列前台线程从队列取批次做推理。队列满的时候后台线程会阻塞防止内存无限增长。5.2 量化与推理加速的取舍量化是另一个常用的加速手段把FP32的权重转成INT8或FP16能显著减少显存占用和计算量。但量化会带来精度损失需要根据具体任务权衡。我的一般原则是先试FP16不够再试INT8。FP16通常能带来1.5到2倍的加速精度损失几乎可以忽略。INT8能带来3到4倍加速但精度损失可能需要仔细评估。对于分类任务INT8通常没问题对于生成任务INT8可能会导致输出质量明显下降。PyTorch的量化工具链比较成熟动态量化一行代码就能搞定import torch.quantization # 动态量化适用于LSTM、Linear等 quantized_model torch.quantization.quantize_dynamic( model, {torch.nn.Linear}, dtypetorch.qint8 ) # 静态量化需要校准数据 model.qconfig torch.quantization.get_default_qconfig(fbgemm) torch.quantization.prepare(model, inplaceTrue) # 用校准数据跑一遍 for batch in calibration_data: model(batch) torch.quantization.convert(model, inplaceTrue)提示量化后的模型一定要做精度对比测试。我见过有人量化完直接上线结果模型输出全是乱码回滚都来不及。测试方法很简单用同一批测试数据对比量化前后的输出差异如果差异超过阈值就不能用。5.3 缓存策略什么该缓存什么不该缓存缓存用得好能大幅降低延迟和计算量用不好会导致结果不一致甚至内存爆炸。我的经验是只缓存确定性计算的结果。确定性计算是指同样的输入永远产生同样的输出。比如特征提取、embedding计算、分类模型的推理结果这些都是确定性的可以缓存。但生成模型的输出尤其是temperature大于0时是不确定的缓存了反而会导致用户每次看到一样的结果体验很差。缓存的key设计也很关键。对于文本输入直接用原文做key可能太长通常用哈希值。但哈希有碰撞风险所以最好用SHA256这种碰撞概率极低的算法。对于数值输入要注意浮点数的精度问题通常需要先量化再哈希。import hashlib import json def make_cache_key(input_data, model_version): if isinstance(input_data, str): content input_data else: content json.dumps(input_data, sort_keysTrue) raw f{model_version}:{content} return hashlib.sha256(raw.encode()).hexdigest()缓存淘汰策略我通常用LRU最近最少使用配合TTL生存时间。LRU保证热数据留在缓存里TTL保证冷数据不会永远占着内存。Python的cachetools库提供了现成的实现没必要自己造轮子。5.4 压测与容量规划服务上线前一定要做压测搞清楚系统的容量边界在哪里。压测工具我用locust因为它用Python写测试脚本和我们的技术栈一致学习成本低。压测的核心是找到三个关键指标最大QPS、P99延迟拐点、错误率拐点。最大QPS是指系统能稳定处理的最大请求速率。P99延迟拐点是指延迟开始急剧上升的QPS值。错误率拐点是指错误率开始明显上升的QPS值。这三个值通常不一样容量规划要以最小的那个为准再留30%的余量。from locust import HttpUser, task, between class InferenceUser(HttpUser): wait_time between(0.01, 0.05) task def predict(self): self.client.post(/predict, json{ inputs: [[0.1] * 128] })压测的时候要注意几点一是压测环境要和生产环境尽量一致否则数据没有参考价值二是要逐步增加压力不要一上来就满负荷否则测不出拐点三是要监控服务端的资源使用情况CPU、内存、GPU、网络都要看瓶颈往往不在你以为的地方。6. 工程化收尾让项目从“能跑”到“能维护”6.1 配置管理与环境隔离从零构建的项目最容易犯的错误是把配置硬编码在代码里。数据库地址、模型路径、批处理大小、超时时间这些都应该抽出来放到配置文件或环境变量里。我习惯用YAML做配置文件因为它比JSON可读性好比INI表达能力强。配置加载用pydantic的BaseSettings它能自动从环境变量读取配置还能做类型校验。from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): model_path: str Field(..., descriptionPath to model file) device: str Field(cpu, descriptionDevice: cpu or cuda) max_batch_size: int Field(8, ge1, le64) max_wait_ms: int Field(10, ge1, le1000) max_queue_size: int Field(100, ge10) class Config: env_prefix AI_ENG_ env_file .env settings Settings()这样配置既可以通过环境变量覆盖比如AI_ENG_MAX_BATCH_SIZE16也可以写在.env文件里。不同环境开发、测试、生产用不同的.env文件代码完全不用改。6.2 测试策略单元测试、集成测试与端到端测试从零构建的项目如果没有测试后期维护会非常痛苦。我的测试策略是三层单元测试覆盖核心逻辑集成测试覆盖模块间交互端到端测试覆盖完整链路。单元测试用pytest重点测数据管线的每个步骤、特征计算逻辑、批处理逻辑。这些测试跑得很快每次改代码都应该跑一遍。集成测试测模型加载、推理服务、缓存、熔断这些模块的交互。这些测试需要加载真实模型跑得慢一些通常只在合并代码前跑。端到端测试模拟真实请求从HTTP接口打到模型推理再返回结果。这个最慢但最能发现问题。我通常用httpx的测试客户端来做。import pytest from fastapi.testclient import TestClient from main import app pytest.fixture def client(): with TestClient(app) as c: yield c def test_predict_endpoint(client): response client.post(/predict, json{ inputs: [[0.1] * 128] }) assert response.status_code 200 data response.json() assert outputs in data assert len(data[outputs]) 1 assert data[latency_ms] 0注意端到端测试里不要用真实的大模型用一个小的随机模型就行。测试的目的是验证链路通不通不是验证模型效果好不好。用大模型会让测试跑得极慢最后没人愿意跑。6.3 部署与回滚的标准化流程部署不是把代码传到服务器上跑起来就完事了。标准化的部署流程应该包括构建镜像、推送镜像、更新配置、滚动重启、健康检查、流量切换。每一步都要可回滚。我用Docker做镜像Dockerfile尽量精简只装必要的依赖。镜像打标签用Git commit hash这样能精确追溯到代码版本。FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]部署的时候用docker-compose或者Kubernetes关键是配置健康检查接口。健康检查接口要检查模型是否加载成功、GPU是否可用、依赖服务是否可达。只有健康检查通过才把流量切过来。app.get(/health) async def health(): checks { model_loaded: model is not None, gpu_available: torch.cuda.is_available() if settings.device cuda else True, } if not all(checks.values()): raise HTTPException(status_code503, detailchecks) return {status: healthy, checks: checks}回滚就更简单了把镜像标签切回上一个版本重新部署。前提是你保留了旧版本的镜像而且数据库 schema 没有不兼容的变更。所以数据库变更一定要向前兼容至少保留两个版本的兼容性。6.4 文档与知识沉淀最后说一个最容易被忽略但长期价值最高的环节文档。从零构建的项目如果不写文档三个月后你自己都忘了某个模块为什么这么设计。文档不需要多正式但至少要包含架构图用文字描述也行、每个模块的职责和接口、关键设计决策的理由、常见问题的排查方法。我习惯在代码仓库里放一个DESIGN.md每次做重要变更时更新它。# 推理服务设计文档 ## 模块划分 - data/: 数据管线负责读取、清洗、tokenize - features/: 特征存储负责特征计算和版本管理 - model/: 模型封装负责加载和推理 - server/: HTTP服务负责路由、限流、熔断 - monitor/: 监控负责指标采集和日志 ## 关键决策 - 为什么用生成器而不是列表内存占用低支持流式处理 - 为什么用动态批处理而不是固定批处理请求量波动大固定批处理浪费资源 - 为什么用SHA256做缓存key碰撞概率极低避免缓存错乱 ## 已知问题 - 批处理在请求量突增时会有延迟毛刺正在优化 - INT8量化在生成任务上效果下降明显暂不启用这份文档不需要写得多漂亮关键是持续更新。每次踩坑之后把问题和解决方案记进去下次遇到类似问题就能快速定位。我个人在实际操作中的体会是从零构建AI工程系统最大的价值不在于省了多少钱或者性能提升了多少而在于你对整个系统的掌控力。你知道每个环节在做什么、为什么这么做、出了问题该去哪里找。这种掌控力是调包永远给不了的。当然从零构建不意味着拒绝一切现成工具而是说你要理解工具背后的原理在需要的时候能够自己实现一个简化版。等你真正手搓过一遍之后再用那些成熟的框架你会发现自己的理解完全不一样了。