1. 为什么我要从零手搓一套AI工程流水线第一次看到ai-engineering-from-scratch这个项目名的时候我正被一堆调包式AI开发折磨得够呛。那会儿团队里新来的几个小伙伴问他们模型怎么部署的回答是就调了个API问推理延迟怎么优化回答是换了个更贵的实例。这种状态在业务量小的时候没问题一旦请求量上来、成本报表一摊开问题就全暴露了。所以我特别理解这个标题背后的诉求——它想解决的不是怎么用AI而是怎么把AI当成一个正经工程来做从最底层的数据管道、模型加载、推理服务、监控告警一路搭起来而不是站在一堆黑盒SDK上拼乐高。ai-engineering-from-scratch这个项目说白了就是一套从零构建AI工程能力的实践路线。它适合三类人一是刚转行做AI应用、只会调API但不懂底层原理的开发者二是想把自己项目从能跑升级到能扛的后端工程师三是带团队的技术负责人需要一套可复用的工程规范来约束大家的交付质量。它不教你训练一个千亿参数大模型那是另一回事它教你的是当你手上有一个已经能用的模型之后怎么把它包装成一个稳定、可观测、可扩展、成本可控的线上服务。这个定位非常务实也是我在实际工作中反复踩坑之后最想分享的部分。我打算按我自己搭这套东西的顺序来写先讲整体设计思路和选型逻辑再拆核心模块的细节然后是完整的实操落地过程最后把我踩过的坑和排查经验整理出来。全程不堆术语能上代码就上代码能算参数就算参数你看完应该能直接照着搭一套属于自己的版本。2. 整体设计与技术选型思路拆解2.1 先想清楚从零到底指哪一层很多人对from scratch有误解以为是要自己写矩阵乘法、自己实现反向传播。真没必要那是造轮子不是做工程。我理解的从零是指不依赖任何一站式AI平台自己把数据、模型、服务、监控这几块拼起来。模型本身可以用开源的推理框架可以用现成的但把它们串成一条能上线的流水线这个串的过程必须你自己掌控。为什么强调这个因为一站式平台最大的问题是看不见。你不知道它内部怎么批处理、怎么调度、怎么计费出了问题只能提工单。而自己搭的流水线每一层你都能打日志、能改参数、能替换组件。我做过一个对比同样一个7B模型用平台托管和自建服务在QPS 50的场景下自建的成本能压到托管的三分之一左右延迟还更可控。这个差距在业务量小的时候不明显量一大就是真金白银。所以这套项目的核心思路是分层解耦每层可替换。数据层、模型层、服务层、观测层各自独立层与层之间用清晰的接口通信。这样你换一个推理框架不用动数据管道换一个向量库不用改服务代码。这种设计在初期会显得麻烦但到了后期迭代的时候你会感谢自己当初没图省事。2.2 技术栈选型的几个关键取舍选型这块我踩过不少坑说几个核心决策点。推理框架选什么。市面上主流的几个我都试过。如果你的模型是Transformer架构、追求吞吐vLLM是目前比较稳的选择它的PagedAttention对显存利用很友好如果追求极致的低延迟、模型结构比较特殊可以考虑TensorRT-LLM但编译过程比较折腾如果只是想快速跑起来、模型不大HuggingFace的transformers加个简单的批处理也能用。我的建议是先用transformers把流程跑通再根据压测结果决定要不要换vLLM。别一上来就上最复杂的调试成本太高。服务框架选什么。FastAPI是我用得最多的异步支持好、生态成熟、写起来快。如果你追求极致性能可以考虑用Rust写的推理服务框架但开发效率会下降。对于大多数业务场景FastAPI加uvicorn加gunicorn的组合足够用了。这里有个细节uvicorn的worker数量不是越多越好因为每个worker都会加载一份模型显存会成倍增长。通常建议worker数等于GPU数或者用单worker加异步批处理。向量库选什么。如果你的场景需要RAG向量库是绕不开的。Milvus功能全但部署重Qdrant轻量且API友好pgvector适合已经有Postgres的场景。我个人的偏好是小规模用pgvector中大规模用Qdrant。Milvus适合那种数据量上亿、需要分布式部署的场景普通业务用不上。监控用什么。Prometheus加Grafana是标配这个没什么好纠结的。关键是埋点要埋对地方后面我会详细讲。下面这张表是我整理的核心组件选型对照你可以根据自己的场景调整层级组件轻量方案生产方案选择依据推理模型加载transformersvLLM吞吐需求服务API框架FastAPIFastAPI 异步批处理并发量存储向量库pgvectorQdrant/Milvus数据规模缓存结果缓存内存字典Redis命中率监控指标采集日志文件Prometheus可观测性部署容器DockerDocker K8s扩缩容需求2.3 目录结构怎么组织才不乱项目结构这件事看起来是小事实际上决定了你后期维护的痛苦程度。我见过太多项目所有代码堆在一个main.py里改一个功能要翻半天。这套项目我建议按职责分层大致长这样ai-engineering-from-scratch/ ├── configs/ # 配置文件按环境分 │ ├── dev.yaml │ └── prod.yaml ├── data/ # 数据处理管道 │ ├── loaders/ # 数据加载 │ ├── processors/ # 清洗、分块 │ └── pipelines/ # 管道编排 ├── models/ # 模型相关 │ ├── loader.py # 模型加载封装 │ └── inference.py # 推理逻辑 ├── services/ # 服务层 │ ├── api.py # API入口 │ ├── batch.py # 批处理调度 │ └── cache.py # 缓存 ├── observability/ # 观测层 │ ├── metrics.py # 指标埋点 │ └── logging.py # 日志配置 ├── tests/ # 测试 └── scripts/ # 运维脚本这个结构的好处是每个目录的职责单一新人进来能快速定位。比如要改数据清洗逻辑直接去data/processors/要加一个监控指标去observability/metrics.py。不用在几千行代码里大海捞针。提示配置文件一定要和代码分离并且按环境区分。我见过把数据库密码硬编码在代码里的项目换环境的时候改得满头大汗。用pydantic-settings或者dynaconf这类库能省很多事。3. 核心模块的细节解析与实操要点3.1 数据管道别小看清洗和分块数据管道是整条流水线的入口也是最容易被忽视的环节。很多人觉得数据嘛读进来就行结果线上效果一塌糊涂回头查发现是数据里混了一堆乱码和重复内容。清洗这一步要做的事去除HTML标签、统一编码、过滤超短文本、去重。去重尤其重要我做过一个实验同一批数据不去重和去重之后检索的准确率差了将近15个百分点。去重可以用SimHash或者MinHash对于文本量不大的场景简单的MD5哈希加集合判断就够了。分块chunking是RAG场景的核心。分块策略直接决定了检索质量。常见的做法有固定长度分块、按句子分块、按语义分块。我的经验是固定长度加重叠是最稳的baseline。比如每块512个token相邻块重叠50个token。重叠的作用是防止关键信息被切断在边界上。语义分块听起来高级但实现复杂、效果不稳定除非你有明确的调优需求否则不建议一上来就用。分块大小怎么定这要看你的嵌入模型和下游任务。一般来说嵌入模型有最大输入长度限制比如512或8192。分块不能超过这个限制。另外块太小会导致检索到的上下文不完整块太大会引入噪声。我通常从256到512这个区间开始试根据实际检索效果调整。def chunk_text(text, chunk_size512, overlap50): tokens tokenize(text) chunks [] start 0 while start len(tokens): end start chunk_size chunk tokens[start:end] chunks.append(detokenize(chunk)) start end - overlap return chunks这段代码看起来简单但有几个坑tokenize和detokenize必须用和嵌入模型一致的tokenizer否则长度对不上重叠部分不能大于块大小否则会死循环边界处理要小心最后一块可能不足chunk_size。3.2 模型加载显存管理和冷启动优化模型加载这块核心矛盾是显存占用和加载速度。一个7B的模型FP16精度下大概占14GB显存加上KV Cache和中间激活实际占用会更高。如果你的GPU显存不够就得考虑量化。量化方案我试过几种INT8量化能把显存压到一半左右精度损失通常在1%以内INT4量化能压到四分之一但精度损失明显一些适合对精度要求不极端的场景。GPTQ和AWQ是目前比较成熟的量化方法加载速度快效果也稳定。冷启动是另一个大问题。模型加载本身就要几十秒如果每次扩容都重新加载用户体验会很差。解决办法有两个一是预热服务启动后先跑几条请求把模型热起来二是常驻保持一定数量的实例不销毁用的时候直接接流量。K8s的HPA配合就绪探针能缓解这个问题但探针的配置要调好否则会出现还没加载完就被打流量的情况。class ModelLoader: def __init__(self, model_path, quantizeNone): self.model_path model_path self.quantize quantize self.model None self.tokenizer None def load(self): self.tokenizer AutoTokenizer.from_pretrained(self.model_path) if self.quantize int8: self.model AutoModelForCausalLM.from_pretrained( self.model_path, load_in_8bitTrue, device_mapauto ) else: self.model AutoModelForCausalLM.from_pretrained( self.model_path, torch_dtypetorch.float16, device_mapauto ) self.model.eval() return selfdevice_mapauto这个参数很关键它会让transformers自动把模型层分配到可用的GPU上。如果你有多张卡这个参数能帮你省不少手动分配的功夫。但要注意多卡推理的通信开销不小如果模型能塞进单卡就别用多卡。3.3 推理服务批处理是吞吐的命门推理服务的性能很大程度上取决于**批处理batching**做得好不好。单条请求一条条处理GPU利用率极低因为每次计算都要等数据搬运。批处理就是把多个请求攒在一起一次性送进模型这样GPU的计算单元才能跑满。批处理有两种模式静态批处理和连续批处理。静态批处理是攒够N条或者等M毫秒就发一批实现简单但会有等待延迟。连续批处理continuous batching是vLLM这类框架的核心能力它能在一条请求生成结束的瞬间就插入新的请求GPU几乎不停歇。如果你的QPS比较高强烈建议用支持连续批处理的框架。批大小怎么定这要看显存和延迟要求。批越大吞吐越高但单条延迟也会增加因为要等同一批里最慢的那条。我通常的做法是先测出显存能承受的最大批大小然后取它的70%作为线上配置留点余量防止OOM。class BatchScheduler: def __init__(self, max_batch_size8, max_wait_ms50): self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms self.queue [] async def add_request(self, request): self.queue.append(request) if len(self.queue) self.max_batch_size: return await self.flush() await asyncio.sleep(self.max_wait_ms / 1000) return await self.flush() async def flush(self): batch self.queue[:self.max_batch_size] self.queue self.queue[self.max_batch_size:] return await self.run_inference(batch)这段代码是个简化版实际生产里要考虑超时、异常、优先级等。但核心逻辑就是攒批超时触发。max_wait_ms这个参数很关键设太小批不起来设太大延迟高。50ms是个比较平衡的起点。3.4 缓存层省下的都是利润缓存是性价比最高的优化手段没有之一。很多请求其实是重复的或者高度相似的。把结果缓存起来命中一次就省一次推理省下的GPU时间就是利润。缓存分两级精确缓存和语义缓存。精确缓存就是请求内容完全一致时直接返回用Redis的字符串键值对就能实现。语义缓存是把请求向量化在向量库里找相似度超过阈值的缓存结果。语义缓存能覆盖换个说法问同一个问题的场景但阈值要调好太高命中率低太低会返回不相关的结果。class SemanticCache: def __init__(self, vector_store, threshold0.95): self.vector_store vector_store self.threshold threshold def get(self, query_embedding): results self.vector_store.search(query_embedding, top_k1) if results and results[0].score self.threshold: return results[0].payload[response] return None def set(self, query_embedding, response): self.vector_store.upsert(query_embedding, {response: response})阈值0.95是我实测下来比较稳的值。低于这个值容易把不相关的缓存返回给用户体验很差。另外缓存要有过期策略尤其是知识类问答数据更新了缓存还返回旧答案就尴尬了。TTL设个几小时到一天比较合理。注意语义缓存的向量化本身也要消耗算力如果嵌入模型很大缓存带来的收益可能被抵消。建议用小的嵌入模型做缓存大的嵌入模型做检索。4. 完整实操过程与核心环节落地4.1 环境准备与依赖安装先把环境搭起来。我假设你用的是Linux服务器有NVIDIA GPU驱动和CUDA已经装好。Python版本建议3.10以上太老的版本有些库不支持。# 创建虚拟环境 python -m venv venv source venv/bin/activate # 安装核心依赖 pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118 pip install transformers accelerate fastapi uvicorn pip install redis qdrant-client prometheus-client pip install pydantic-settings pyyaml这里有个细节torch的版本要和CUDA版本匹配。上面用的是cu118对应CUDA 11.8。如果你的CUDA是12.1就把index-url换成对应的。装错了会报CUDA error排查起来很烦。依赖装完之后先跑个简单的脚本验证GPU能用import torch print(torch.cuda.is_available()) print(torch.cuda.get_device_name(0)) print(f显存: {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f} GB)如果输出是True和你的显卡型号说明环境没问题。显存信息很重要后面配批大小要用到。4.2 配置文件的设计与加载配置文件我用YAML按环境分。核心配置项包括模型路径、批处理参数、缓存配置、监控端口等。# configs/prod.yaml model: path: /models/qwen-7b quantize: int8 max_batch_size: 8 max_wait_ms: 50 cache: redis_url: redis://localhost:6379/0 ttl_seconds: 3600 semantic_threshold: 0.95 server: host: 0.0.0.0 port: 8000 workers: 1 monitoring: metrics_port: 9090 log_level: INFO加载配置用pydantic-settings能自动做类型校验配置写错了启动就报错比运行时才发现问题好得多。from pydantic_settings import BaseSettings from pydantic import Field class ModelConfig(BaseSettings): path: str quantize: str None max_batch_size: int 8 max_wait_ms: int 50 class AppConfig(BaseSettings): model: ModelConfig cache: dict server: dict monitoring: dict classmethod def load(cls, envprod): import yaml with open(fconfigs/{env}.yaml) as f: data yaml.safe_load(f) return cls(**data)4.3 服务入口与请求生命周期服务入口用FastAPI一个/generate接口处理生成请求一个/health接口给探针用。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import time app FastAPI() config AppConfig.load() loader ModelLoader(config.model.path, config.model.quantize).load() scheduler BatchScheduler(config.model.max_batch_size, config.model.max_wait_ms) cache SemanticCache(...) class GenerateRequest(BaseModel): prompt: str max_tokens: int 256 temperature: float 0.7 app.post(/generate) async def generate(req: GenerateRequest): start time.time() # 1. 查缓存 query_embedding embed(req.prompt) cached cache.get(query_embedding) if cached: metrics.cache_hit.inc() return {text: cached, cached: True} # 2. 走推理 result await scheduler.add_request(req) # 3. 写缓存 cache.set(query_embedding, result) # 4. 埋点 latency time.time() - start metrics.request_latency.observe(latency) metrics.request_count.inc() return {text: result, cached: False}这个请求生命周期里埋点的位置很关键。延迟要分两段记缓存查询的延迟和推理的延迟分开记才能知道瓶颈在哪。如果缓存查询占了大部分时间说明向量库该优化了如果推理占大头说明该调批处理参数了。4.4 监控指标的设计与埋点监控指标我分四类流量、延迟、错误、资源。流量看QPS延迟看P50/P95/P99错误看错误率资源看GPU利用率和显存占用。from prometheus_client import Counter, Histogram, Gauge request_count Counter(ai_requests_total, Total requests, [status]) request_latency Histogram(ai_request_latency_seconds, Request latency, buckets[0.1, 0.5, 1, 2, 5, 10]) cache_hit Counter(ai_cache_hits_total, Cache hits) gpu_memory Gauge(ai_gpu_memory_bytes, GPU memory usage) batch_size Histogram(ai_batch_size, Batch size distribution, buckets[1, 2, 4, 8, 16])request_latency的buckets要按你的实际延迟分布来设。如果P99是3秒buckets里就要有3附近的边界否则算出来的分位数不准。我见过有人用默认buckets结果P99算出来是10秒实际只有2秒就是因为buckets没覆盖到。GPU显存和利用率可以用pynvml采集起个后台线程定时更新Gauge。import pynvml import threading def collect_gpu_metrics(interval5): pynvml.nvmlInit() handle pynvml.nvmlDeviceGetHandleByIndex(0) while True: info pynvml.nvmlDeviceGetMemoryInfo(handle) gpu_memory.set(info.used) time.sleep(interval) threading.Thread(targetcollect_gpu_metrics, daemonTrue).start()4.5 压测与参数调优服务搭好之后别急着上线先压测。我用的是locust写个简单的压测脚本模拟不同并发下的表现。from locust import HttpUser, task, between class AIUser(HttpUser): wait_time between(0.1, 0.5) task def generate(self): self.client.post(/generate, json{ prompt: 介绍一下机器学习的基本概念, max_tokens: 128 })压测的时候重点看几个数QPS、P99延迟、GPU利用率、显存峰值。如果GPU利用率上不去说明批处理没生效检查max_wait_ms是不是设太小了如果显存峰值接近上限说明批大小该调小如果P99延迟远高于P50说明有长尾请求可能是某些输入特别长导致的要考虑对输入长度做限制。我实测下来的一组参考数据7B模型INT8量化单卡A10批大小8max_wait_ms50QPS能到15左右P99延迟在2秒以内。这个数据供你参考实际会因模型和硬件而异。5. 常见问题与排查技巧实录5.1 显存溢出OOM的排查路径OOM是最高频的问题。排查思路是先定位是加载时OOM还是推理时OOM。加载时OOM通常是模型太大或者量化没生效。检查load_in_8bit参数有没有传对检查device_map是不是把模型分到了多卡但通信失败。如果模型确实太大考虑用更激进的量化或者换更小的模型。推理时OOM通常是批太大或者输入太长。先看日志里OOM发生时的批大小和输入长度然后调小max_batch_size或者对输入长度做截断。KV Cache是推理时显存的大头输入越长、输出越长KV Cache越大。可以用max_model_len参数限制总长度。提示留10%到20%的显存余量别把显存用满。GPU在显存接近满的时候性能会明显下降而且容易触发OOM。5.2 延迟忽高忽低的定位方法延迟不稳定通常是批处理等待或者资源竞争导致的。先看延迟的分布如果P50很低但P99很高说明大部分请求很快少数请求很慢。慢的请求可能是输入特别长或者恰好赶上了批处理等待。排查方法在日志里记录每个请求的输入长度、批大小、等待时间、推理时间。然后按输入长度分组看延迟如果长输入的延迟明显高那就是输入长度的问题考虑对长输入做特殊处理。如果等待时间波动大那就是批处理参数的问题调max_wait_ms。还有一种可能是GPU被其他进程占用。用nvidia-smi看看有没有别的进程在跑如果有要么杀掉要么隔离GPU。5.3 缓存命中率低的优化思路缓存命中率低先看是精确缓存命中低还是语义缓存命中低。精确缓存命中低说明请求的多样性高这很正常不用强求。语义缓存命中低通常是阈值设太高了。调阈值的方法拿一批真实请求两两算相似度看看相似请求的相似度分布。如果相似请求的相似度集中在0.9左右那阈值设0.95就太高了调到0.9。但要注意阈值降低会引入误命中返回不相关的结果。所以要在命中率和准确率之间找平衡。另一个思路是缓存粒度。如果整个回答缓存命中率低可以考虑缓存中间结果比如检索到的文档、生成的摘要等。粒度越细命中率越高但复用的逻辑也越复杂。5.4 常见问题速查表现象可能原因排查方法解决方向启动即OOM模型太大/量化未生效看加载日志换量化/换小模型推理时OOM批太大/输入太长看OOM时批大小调小批/截断输入P99延迟高长尾请求/批等待按输入长度分组限制长度/调批参数GPU利用率低批处理未生效看批大小分布调大max_wait_ms缓存命中率低阈值太高看相似度分布调低阈值服务无响应死锁/线程阻塞看线程栈检查异步逻辑5.5 几个我踩过的坑坑一tokenizer不一致。数据管道用的tokenizer和模型用的tokenizer不是同一个导致分块长度和模型实际处理长度对不上检索效果差。解决办法是全局统一用一个tokenizer从模型目录里加载。坑二异步里调同步代码。在FastAPI的异步接口里直接调了同步的推理函数导致整个事件循环被阻塞并发上不去。解决办法是用run_in_executor把同步调用扔到线程池里。坑三Prometheus指标重复注册。服务重启时如果没清理指标会重复注册报错。解决办法是用prometheus_client的REGISTRY做去重或者用多进程模式。坑四Redis连接池耗尽。高并发下Redis连接不够用请求排队。解决办法是调大连接池或者用异步Redis客户端。坑五模型热更新导致服务中断。直接替换模型文件正在处理的请求会失败。解决办法是双缓冲新模型加载好之后再切流量旧模型等请求处理完再卸载。6. 后续可以继续扩展的方向这套流水线搭起来之后能扩展的地方还有很多。比如多模型路由根据请求的复杂度自动选择不同大小的模型简单问题用小模型复杂问题用大模型能进一步省成本。再比如A/B测试框架同时跑两个版本的模型或提示词用真实流量对比效果。还有自动扩缩容根据QPS和GPU利用率动态调整实例数闲时缩容省钱忙时扩容保体验。我个人在实际操作中的体会是AI工程这件事难的不是某个单点技术而是把一堆单点串成一条稳定的流水线。每个模块单独看都不复杂但组合起来就会有各种意想不到的问题。所以我的建议是别追求一步到位先把最小可用版本跑起来然后根据实际遇到的问题逐个优化。这套ai-engineering-from-scratch的思路核心就是让你对每一层都有掌控力出了问题知道去哪查、怎么改。这种掌控力是调包式开发永远给不了的。最后分享一个小技巧给每个请求打一个trace_id贯穿数据、推理、缓存、监控全链路。出了问题拿trace_id一搜整条链路的信息都出来了排查效率能提升好几倍。这个习惯我从搭第一版服务的时候就养成了到现在受益无穷。