1. 从零搭建AI工程能力为什么我劝你别急着调包这两年AI应用开发的门槛肉眼可见地降低了随便拉个框架、调个API就能跑出一个能对话的Demo。但我见过太多团队Demo阶段惊艳四座一上生产环境就原形毕露——推理延迟飙到十几秒、显存说爆就爆、模型更新一次整个服务抖三抖。问题的根子不在模型本身而在于AI工程能力的缺失。ai-engineering-from-scratch这个项目标题说的就是从零开始把AI工程这套东西搭起来。它不是教你训一个更大的模型也不是教你调某个闭源接口而是把AI系统落地过程中那些脏活累活——数据处理、推理服务、性能压测、监控告警、版本管理——一层一层剥开讲清楚。适合谁看我认为有三类人最该认真读一是刚转行做AI应用、只会写Prompt不会写服务的开发者二是在团队里负责把算法模型推上线的工程同学三是技术负责人需要判断一套AI系统到底能不能扛住真实流量。我自己带过几个从零到一的AI项目踩过的坑足够写一本小册子。这篇文章就把ai-engineering-from-scratch这个主题拆透从整体设计思路到核心环节实现再到常见问题排查尽量给出一套可以直接抄作业的方案。全文不聊虚的只讲能落地的工程细节。2. 整体设计与思路拆解2.1 为什么从零反而比调包更靠谱很多人一上来就问现在框架这么多为什么还要从零搭直接用现成的不香吗我的回答是框架解决的是通用问题而AI工程的核心矛盾往往出在你的业务特有场景上。举个真实例子我们做过一个文档问答系统用现成的RAG框架跑Demo没问题但上线后发现两个致命问题一是长文档切分后检索召回率断崖式下跌二是并发上来后向量检索的延迟波动极大。这两个问题框架文档里根本没提因为它们是业务数据分布和部署环境共同决定的。从零搭建的意义不是让你重复造轮子而是让你清楚每个环节的边界和代价。你知道数据怎么流、显存怎么分、请求怎么排队出问题时才能定位到具体那一层。这就像开车你可以不会修发动机但至少得知道仪表盘上哪个灯亮了代表什么。2.2 分层架构把AI系统拆成五块积木我习惯把一套AI工程系统拆成五层从下往上分别是数据层负责原始数据的采集、清洗、切分、向量化以及向量库的构建和更新。模型层负责模型的加载、推理、批处理、量化以及多模型的路由调度。服务层负责API网关、请求队列、限流熔断、超时重试。观测层负责日志、指标、链路追踪尤其是推理延迟和显存占用的监控。迭代层负责模型版本管理、A/B测试、灰度发布、回滚机制。这五层不是拍脑袋分的而是对应了AI系统从能跑到跑得稳再到跑得久的三个阶段。很多团队只做了数据层和模型层服务层和观测层基本空白结果就是线上出问题只能靠重启大法。2.3 技术选型的取舍逻辑选型这块我踩过最大的坑就是过早追求先进。比如向量库一开始就上了分布式方案结果数据量才几十万条运维复杂度却翻了三倍。后来我总结了一个原则按数据规模和并发量选型而不是按技术热度选型。具体来说数据量在百万级以下、并发在几十QPS以内单机向量库加内存索引完全够用到了千万级再考虑分片上亿级别才需要认真设计分布式架构。模型推理也是同理小模型用CPU加ONNX Runtime就能跑没必要上来就上GPU集群。这个判断逻辑后面在实操部分会展开讲。3. 核心细节解析与实操要点3.1 数据层切分策略决定检索上限数据层最容易被低估的环节就是文本切分。很多人直接用固定长度切比如每500字一段结果把一句话切成两半检索出来的片段语义不完整模型回答自然驴唇不对马嘴。我的做法是按语义边界切分再按长度兜底。具体步骤先按段落、标题、列表项等自然边界切分保留结构信息。对超长段落按句子边界二次切分优先在句号、问号、分号处断开。设置重叠窗口一般取切分长度的10%到20%保证跨段语义连续。每个片段附带元数据来源文档、章节标题、位置索引。这里有个细节重叠窗口不是越大越好。我实测过重叠超过30%后检索结果重复率明显上升反而挤占了有效片段的召回空间。10%到20%是个比较稳的区间。注意切分后的片段一定要做去重和长度过滤太短的片段比如少于50字往往是噪声会拉低检索质量。3.2 模型层批处理与量化的平衡点模型推理的性能瓶颈通常不在计算本身而在显存带宽和批处理效率。我见过一个服务单条请求推理要800毫秒改成批处理32条后单条平均延迟降到120毫秒吞吐量翻了六倍多。批处理的核心参数是最大批大小和最大等待时间。这两个参数是一对矛盾批越大吞吐越高但单条延迟也越高等待时间越长越容易凑大批但用户感知的延迟也越大。我的经验值是场景最大批大小最大等待时间说明实时对话850ms优先保证低延迟文档问答32200ms吞吐和延迟兼顾离线批处理1281000ms优先保证吞吐量化方面INT8量化通常能带来2到3倍的推理加速精度损失在1%以内性价比很高。但要注意量化后的模型对输入长度更敏感长文本推理时精度下降会更明显。我的做法是短文本场景用INT8长文本场景用FP16极端性能要求下才考虑INT4。3.3 服务层限流和熔断是保命符服务层最关键的三个机制是限流、熔断、超时。这三个东西平时看不出价值一旦流量突增或模型异常就是保命符。限流我推荐用令牌桶算法因为它允许一定程度的突发流量。配置上令牌生成速率按平均QPS设置桶容量设为平均QPS的2到3倍。这样正常流量下不会误杀突发流量下也能平滑削峰。熔断的触发条件要结合业务定。我的经验是连续10次请求中失败超过5次或者平均延迟超过阈值3倍就触发熔断。熔断后进入半开状态放少量请求试探成功则恢复失败则继续熔断。超时设置有个坑不要只设一个总超时。我一般设三层连接超时2秒、首字节超时5秒、总超时30秒。这样能区分是网络问题、模型加载问题还是推理本身慢。3.4 观测层没有监控的AI系统等于裸奔观测层我重点讲推理延迟的分解监控。一个请求的总延迟可以拆成排队时间、预处理时间、推理时间、后处理时间。这四段分别监控出问题时才能快速定位。我用的方案是Prometheus加Grafana核心指标包括request_queue_duration请求在队列中的等待时间preprocess_duration文本清洗、向量化等预处理耗时inference_duration模型推理耗时postprocess_duration结果组装、格式化耗时gpu_memory_used显存占用batch_size_actual实际批大小这些指标按分钟粒度聚合设置告警阈值。比如推理时间P99超过500毫秒就告警显存占用超过80%就告警。提示日志里一定要记录请求ID和对应的批大小这样排查问题时能把慢请求和批处理参数关联起来。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先说明一下这套方案基于Python生态推理框架用ONNX Runtime向量库用FAISS服务框架用FastAPI。选ONNX Runtime是因为它跨平台、部署简单、对量化支持好选FAISS是因为单机性能足够强百万级数据毫无压力。环境准备步骤# 创建虚拟环境 python -m venv ai-env source ai-env/bin/activate # 安装核心依赖 pip install onnxruntime-gpu1.16.0 pip install faiss-gpu1.7.4 pip install fastapi0.104.0 pip install uvicorn0.24.0 pip install transformers4.35.0 pip install prometheus-client0.18.0这里有个细节ONNX Runtime的GPU版本要和CUDA版本匹配。我用的CUDA 11.8对应onnxruntime-gpu 1.16.0。版本不匹配会直接报错而且报错信息很不友好建议先查官方兼容性表。4.2 数据管道搭建数据管道负责把原始文档变成向量库。核心代码如下import faiss import numpy as np from sentence_transformers import SentenceTransformer class DataPipeline: def __init__(self, model_nameBAAI/bge-small-zh-v1.5): self.encoder SentenceTransformer(model_name) self.index None self.chunks [] def split_text(self, text, max_len500, overlap100): 按语义边界切分文本 paragraphs text.split(\n\n) chunks [] for para in paragraphs: if len(para) max_len: chunks.append(para) else: # 按句子切分 sentences para.replace(。, 。\n).split(\n) current for sent in sentences: if len(current) len(sent) max_len: current sent else: if current: chunks.append(current) current sent if current: chunks.append(current) # 添加重叠 overlapped [] for i, chunk in enumerate(chunks): if i 0: prev_tail chunks[i-1][-overlap:] overlapped.append(prev_tail chunk) else: overlapped.append(chunk) return overlapped def build_index(self, documents): 构建向量索引 all_chunks [] for doc in documents: all_chunks.extend(self.split_text(doc)) self.chunks all_chunks # 向量化 embeddings self.encoder.encode( all_chunks, batch_size32, show_progress_barTrue, normalize_embeddingsTrue ) # 构建FAISS索引 dim embeddings.shape[1] self.index faiss.IndexFlatIP(dim) # 内积索引配合归一化向量等价于余弦相似度 self.index.add(embeddings.astype(float32)) return self.index这段代码有几个关键点向量归一化配合内积索引等价于余弦相似度比直接算余弦快很多批大小设为32兼顾显存和速度重叠窗口取100字对应500字切分长度的20%。4.3 推理服务实现推理服务的核心是批处理调度。我用一个异步队列加后台批处理线程实现import asyncio import onnxruntime as ort from queue import Queue from threading import Thread import numpy as np class InferenceServer: def __init__(self, model_path, max_batch32, max_wait0.2): self.session ort.InferenceSession( model_path, providers[CUDAExecutionProvider, CPUExecutionProvider] ) self.max_batch max_batch self.max_wait max_wait self.queue Queue() self.results {} self._start_worker() def _start_worker(self): def worker(): while True: batch [] batch_ids [] # 收集批次 while len(batch) self.max_batch: try: item self.queue.get(timeoutself.max_wait) batch.append(item[input]) batch_ids.append(item[id]) except: break if not batch: continue # 批推理 inputs np.stack(batch) outputs self.session.run(None, {input: inputs})[0] # 分发结果 for bid, out in zip(batch_ids, outputs): self.results[bid] out t Thread(targetworker, daemonTrue) t.start() async def infer(self, input_data): req_id id(input_data) self.queue.put({id: req_id, input: input_data}) # 等待结果 while req_id not in self.results: await asyncio.sleep(0.01) return self.results.pop(req_id)这个实现里max_wait设为0.2秒对应文档问答场景。如果是实时对话改成0.05秒。注意self.results这个字典在高并发下会有竞争问题生产环境建议用线程安全的队列或者Redis。4.4 监控埋点接入监控埋点用Prometheus客户端库在关键路径上打点from prometheus_client import Histogram, Gauge, Counter import time # 定义指标 queue_duration Histogram(request_queue_duration_seconds, Queue wait time) inference_duration Histogram(inference_duration_seconds, Inference time) gpu_memory Gauge(gpu_memory_used_bytes, GPU memory usage) request_count Counter(request_total, Total requests) class MonitoredInference: def __init__(self, server): self.server server async def infer(self, input_data): request_count.inc() start time.time() # 记录排队时间 queue_start time.time() result await self.server.infer(input_data) queue_duration.observe(time.time() - queue_start) # 记录推理时间 inference_duration.observe(time.time() - start) return result这里有个实操心得Histogram的桶设置要贴合实际延迟分布。默认桶是0.005到10秒但AI推理延迟通常在0.1到2秒之间默认桶太粗。我一般自定义桶为[0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0]这样P99计算更准确。5. 常见问题与排查技巧实录5.1 推理延迟突然飙升怎么查这是最常见的问题。我的排查顺序是先看队列再看批大小最后看显存。第一步看request_queue_duration指标。如果排队时间涨了说明请求积压要么是并发太高要么是推理变慢导致消费不过来。第二步看batch_size_actual。如果批大小突然变小说明请求间隔变长可能是上游调用变慢。第三步看gpu_memory_used。如果显存接近上限GPU会开始频繁换页延迟自然飙升。我遇到过一次诡异的情况延迟飙升但所有指标都正常。最后发现是向量检索的索引文件被操作系统缓存淘汰了每次检索都要重新读盘。解决办法是把索引文件放在tmpfs内存盘上或者用mlock锁定内存。5.2 检索结果不相关怎么调检索质量差通常有三个原因切分不合理、向量模型不匹配、检索参数不当。切分问题前面讲过了重点说向量模型。中文场景我推荐BGE系列英文场景推荐E5系列。但要注意向量模型必须和检索任务匹配。比如你做的是问答检索就要用问答对训练的模型而不是通用语义模型。检索参数方面top_k不是越大越好。我实测下来top_k从5增加到20召回率提升不到5%但延迟增加了一倍多。一般设5到10就够了配合重排序模型效果更好。5.3 显存溢出怎么预防显存溢出基本是三个原因批太大、序列太长、模型太多。预防措施我总结了一个检查清单启动时先跑一次最大批大小的推理确认显存够用设置显存上限超过就拒绝新请求而不是硬撑长序列场景单独走一个队列避免和短序列混批多模型场景用懒加载不用的时候释放显存注意ONNX Runtime的显存管理比较粗放不会主动释放。如果模型频繁切换建议用独立的Session用完就销毁。5.4 常见问题速查表问题现象可能原因排查方法解决方案推理延迟P99飙升队列积压看queue_duration增加实例或降低批等待检索结果不相关切分不合理人工检查片段调整切分策略显存溢出批太大或序列太长看gpu_memory限制批大小或序列长度吞吐量上不去批处理没生效看batch_size_actual检查队列实现服务间歇性超时索引读盘看磁盘IO索引放内存盘模型加载慢模型太大看加载时间用量化模型或懒加载5.5 几个我踩过的坑第一个坑用默认参数跑生产。ONNX Runtime默认用CPU忘了指定CUDA结果推理慢十倍。一定要显式指定providers[CUDAExecutionProvider]。第二个坑向量归一化忘了做。用内积索引但不归一化检索结果完全乱套。归一化就一行代码但忘了就是灾难。第三个坑监控指标没设告警。有次显存泄漏跑了三天才被发现期间服务一直降级运行。后来我加了显存告警超过80%就发通知。第四个坑版本管理混乱。模型更新后没记录版本出问题回滚都不知道回滚到哪个版本。现在我用模型文件的哈希值作为版本号每次更新都记录。6. 迭代层让系统能持续进化6.1 模型版本管理与灰度发布迭代层是很多团队忽略的一层但它是系统能否长期运行的关键。模型版本管理我推荐用文件哈希加元数据的方式每个模型文件对应一个JSON元数据记录训练数据、评估指标、上线时间。灰度发布的策略我一般分三步先放1%流量观察一天再放10%观察三天最后全量。观察指标包括推理延迟、错误率、业务指标比如问答的满意度。任何一项异常就自动回滚。6.2 A/B测试的工程实现A/B测试的关键是流量分割要稳定。同一个用户每次请求都应该落到同一个版本否则体验会割裂。实现上用用户ID哈希取模而不是随机数。def route_version(user_id, versions): 根据用户ID稳定路由到某个版本 hash_val hash(user_id) % 100 cumulative 0 for version, ratio in versions.items(): cumulative ratio if hash_val cumulative: return version return list(versions.keys())[-1]这个函数保证同一用户始终路由到同一版本同时整体流量比例符合配置。6.3 反馈闭环的搭建AI系统和其他系统最大的区别是需要持续迭代。我建议在服务层加一个反馈收集接口用户可以对回答点赞或点踩。这些反馈数据定期回流到数据层用于优化检索策略或微调模型。反馈数据的处理要注意去偏。点赞多的回答不一定质量高可能只是位置靠前。我一般会做位置校正把位置因素从反馈中剥离出来。7. 一些个人体会这套从零搭建的方案我前后迭代了三个版本。第一版只做了数据层和模型层上线两周就崩了第二版加了服务层和观测层稳定运行了半年第三版补上迭代层才算真正能持续运转。最大的体会是AI工程的难点从来不在AI本身而在工程。模型可以换、框架可以换但数据怎么流、请求怎么调度、故障怎么定位这些底层能力是换不掉的。把这几层搭扎实了上面用什么模型都是锦上添花。另外一个小技巧压测一定要用真实数据。我用合成数据压测时一切正常换成真实数据后检索延迟翻了三倍因为真实数据的向量分布更分散索引效率更低。压测数据从生产环境采样脱敏后使用这样测出来的结果才有参考价值。最后说个扩展方向这套架构可以平滑迁移到多模态场景。图片、音频的向量化和文本类似只是编码器不同。服务层的批处理和调度逻辑基本不用改观测层的指标加几个模态相关的维度就行。我最近在做一个图文混合检索的项目就是在这套架构上直接扩展的改动量不到20%。