1. 为什么我要从零手搓一套AI工程流水线第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是某个具体框架而是一连串踩过的坑。过去两年我参与过推荐系统特征工程、做过RAG知识库、也帮朋友调过本地小模型的推理服务。每次项目启动团队里总有人提议“直接上LangChain吧”“用现成的AutoML平台不香吗”结果往往是原型跑得飞快一上生产就各种玄学问题——检索召回率忽高忽低、向量库内存爆炸、推理延迟从200ms飙到3秒。后来我意识到问题不在于工具不好而在于我们对“AI工程”这四个字的理解太表层了。所谓ai-engineering-from-scratch我的定义是不依赖任何高度封装的AI应用框架从数据管道、特征存储、向量索引、模型服务到监控反馈每一层都亲手搭一遍哪怕先用最笨的办法。它解决的不是“能不能跑通demo”的问题而是“当流量涨十倍、数据分布漂移、模型版本迭代时你还能不能睡得着觉”的问题。这套内容适合两类人一是刚入行AI应用开发、被各种框架API绕晕的新人二是有后端或数据工程背景、想补上AI系统设计这一课的资深工程师。我会把每个模块的选型逻辑、参数计算、避坑经验全部摊开讲代码能抄思路能复用。2. 整体架构设计与技术选型逻辑2.1 为什么我坚持“分层解耦”而不是“全家桶”很多教程一上来就让你pip install langchain然后十行代码搞定一个问答机器人。我不否认这很爽但爽完之后呢你根本不知道检索器内部是怎么做分块的不知道embedding模型在CPU和GPU上的吞吐差异更不知道当你想换一个重排模型时该动哪一行。ai-engineering-from-scratch的核心思路就是把AI应用拆成五个独立层数据接入层、特征与向量化层、检索与推理层、服务编排层、可观测层。每一层之间用明确的接口通信比如向量化层只负责把文本转成向量并写入索引检索层只负责根据查询向量召回候选集。这样做的代价是前期代码量翻倍但收益在第二次迭代时就显现了。我试过在一个项目里把向量库从FAISS换成Qdrant因为接口定义清晰只改了一个适配器文件上层业务代码零改动。反过来如果一开始用全家桶换组件往往意味着重写整个链路。选型上我建议新手从SQLite FAISS FastAPI这个组合起步理由是SQLite零配置、FAISS单机性能足够、FastAPI自带异步和文档。等单机QPS超过50再考虑上Milvus或PgVector这是后话。2.2 数据管道的“脏活累活”才是分水岭AI工程和传统后端最大的区别在于数据不是干净的。你拿到的PDF可能是扫描件HTML里混着广告用户query里全是错别字。我在ai-engineering-from-scratch里把数据管道单独拎出来讲因为它决定了整个系统的上限。具体来说管道要处理四件事格式解析、文本清洗、分块策略、元数据抽取。格式解析我用unstructured库它能统一处理PDF、DOCX、PPT虽然速度慢但省心文本清洗我写了一套正则规则重点去掉页眉页脚、连续空行、特殊符号分块策略我默认用递归字符分割块大小512 token重叠64 token这个参数后面会详细算。元数据抽取是最容易被忽略的。比如一份产品手册你得把“章节标题”“页码”“产品型号”存进向量库的payload里这样检索时才能做过滤。我见过太多项目把所有文本一股脑塞进去结果用户问“A型号的保修期”系统召回了一堆B型号的内容。所以我的原则是宁可分块小一点也要保证每个块携带足够的上下文元数据。这部分代码我会在第三节给出完整实现。2.3 向量化与索引别迷信“越大越好”Embedding模型的选择直接决定检索质量。我实测下来bge-small-zh-v1.5在中文场景性价比最高维度512单条推理在CPU上约15msGPU上2ms。如果你追求极致效果可以上bge-large-zh-v1.5维度1024但内存占用翻倍检索延迟也增加。我的建议是先用small跑通全流程等评估指标卡住了再换large。索引方面FAISS的IndexFlatIP适合一万条以内的数据超过一万条就换IndexIVFFlat需要设置nlist参数。这里有个经验公式nlist 4 * sqrt(N)N是向量总数。比如十万条数据nlist设为1265左右查询时nprobe设为nlist/10能在召回率和速度之间取得平衡。注意FAISS的IVF索引需要训练训练数据必须和实际数据同分布。我踩过的坑是拿随机文本训练索引结果真实query的召回率只有30%。正确做法是从你的业务数据里随机采样至少nlist * 39条向量做训练。3. 核心模块拆解与实操要点3.1 文本分块512 token不是拍脑袋定的分块大小直接影响检索粒度和生成质量。块太大噪声多LLM容易跑偏块太小上下文断裂答案不完整。我做过一组对比实验在技术文档问答场景下块大小从256到1024 token召回率5的变化是256→0.72512→0.81768→0.791024→0.74。512 token是峰值。为什么因为大多数技术段落的信息密度在300-500字之间512 token约350中文字刚好覆盖一个完整概念。重叠64 token是为了防止句子被切断比如“向量数据库的索引结构”被切成“向量数据库的”和“索引结构”重叠后两个块都能包含完整短语。实操中我推荐用RecursiveCharacterTextSplitter分隔符优先级设为[\n\n, \n, 。, , , , , , ]。这样它会先按段落切段落太长再按句子切保证语义完整性。代码示例如下from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap64, separators[\n\n, \n, 。, , , , , , ], length_functionlen, ) chunks splitter.split_text(raw_text)提示如果你的文档里有大量表格或代码建议单独处理不要和正文混在一起分块。表格转成Markdown后再分块代码块保持完整。3.2 向量化批处理别一条一条调APIEmbedding推理是典型的吞吐型任务逐条调用会浪费大量时间在IO等待上。我实测过单条调用bge-small处理1000条文本需要约18秒而批量调用batch_size32只需要3.2秒提升近6倍。批量大小的选择取决于显存GPU上可以设64甚至128CPU上建议16-32。另外一定要做归一化因为FAISS的內积索引要求向量是单位向量。归一化代码很简单import numpy as np def normalize(vectors): norms np.linalg.norm(vectors, axis1, keepdimsTrue) return vectors / norms还有一个坑不同embedding模型的输出维度不同换模型时记得重建索引。我见过有人换了模型但没重建结果检索结果全是乱的。所以我的做法是在向量库的元数据里记录model_name和dimension启动时校验不匹配直接报错。3.3 检索重排两阶段召回才是工业级做法单靠向量相似度召回Top-5里经常混入不相关的内容。工业级方案是两阶段第一阶段用向量召回Top-50第二阶段用交叉编码器Cross-Encoder重排取Top-5。重排模型我推荐bge-reranker-base它会把query和每个候选块拼在一起打分精度比向量相似度高出一大截。代价是延迟增加50个候选重排大约需要200msGPU。如果你的QPS要求高可以只对Top-20重排或者用更小的bge-reranker-small。这里有个参数需要计算召回数量K1和重排后数量K2。K1太小重排没得选K1太大延迟受不了。我的经验是K1 10 * K2K2根据LLM的上下文窗口定。比如LLM上下文4096 token每个块512 token最多放6个块那K25K150。如果上下文只有2048K23K130。这个公式不是绝对的但能帮你快速起步。3.4 服务编排FastAPI 异步队列推理服务最怕阻塞。如果检索和生成都在同一个同步接口里一个慢请求会拖垮整个服务。我的方案是FastAPI定义两个端点/retrieve只做检索和重排返回候选块/generate接收候选块和query调用LLM生成答案。前端先调/retrieve拿到结果后再调/generate这样检索和生成可以并行处理不同请求。如果非要一个端点搞定那就用asyncio把检索和生成包成协程但要注意LLM的流式输出和检索的阻塞IO混用容易出问题。注意FastAPI的默认线程池大小是40如果并发超过这个数请求会排队。可以通过--workers参数启动多个进程或者用uvicorn的--limit-concurrency控制。我一般设--workers 4 --limit-concurrency 100配合Nginx做负载均衡。4. 完整实操流程从零搭建一个可用的RAG服务4.1 环境准备与依赖安装我习惯用conda管理环境因为FAISS和PyTorch的版本兼容性比较敏感。以下是经过验证的依赖组合Python 3.10conda create -n ai-eng python3.10 conda activate ai-eng pip install fastapi uvicorn[standard] faiss-cpu sentence-transformers unstructured langchain pypdf如果你有GPU把faiss-cpu换成faiss-gpusentence-transformers会自动使用CUDA。注意unstructured需要系统依赖libmagicUbuntu下apt install libmagic-devMac下brew install libmagic。这一步经常卡住新手我建议先跑一个最小示例验证环境from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5) vec model.encode(测试文本) print(vec.shape) # 应该输出 (512,)如果输出维度不对说明模型下载有问题检查网络或手动下载模型到~/.cache/torch/sentence_transformers/。4.2 数据接入与清洗的完整代码假设我们有一批PDF格式的产品手册放在./docs目录下。第一步是解析和清洗import os from unstructured.partition.pdf import partition_pdf def load_and_clean(pdf_dir): all_text [] for filename in os.listdir(pdf_dir): if not filename.endswith(.pdf): continue elements partition_pdf( filenameos.path.join(pdf_dir, filename), strategyfast, # 快速模式适合文本型PDF include_page_breaksFalse, ) for elem in elements: text str(elem).strip() if len(text) 10: # 过滤太短的碎片 continue # 清洗规则去掉连续空格、特殊符号 text .join(text.split()) all_text.append({ text: text, source: filename, page: elem.metadata.page_number if hasattr(elem.metadata, page_number) else 0, }) return all_text这里strategyfast只适用于文本型PDF如果是扫描件需要改成hi_res并配合OCR速度会慢很多。我一般先用fast跑一遍如果发现某页文本为空再单独对该页做OCR。元数据里的source和page非常重要后面检索时可以展示引用来源。4.3 向量化与索引构建拿到清洗后的文本块接下来做向量化和索引import faiss import numpy as np from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5) dimension 512 def build_index(chunks): texts [c[text] for c in chunks] # 批量编码batch_size根据显存调整 embeddings model.encode(texts, batch_size32, show_progress_barTrue) # 归一化 faiss.normalize_L2(embeddings) # 构建IVF索引 nlist int(4 * np.sqrt(len(embeddings))) quantizer faiss.IndexFlatIP(dimension) index faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT) # 训练索引 index.train(embeddings) index.add(embeddings) index.nprobe max(1, nlist // 10) return index, texts注意faiss.normalize_L2是原地操作会修改embeddings数组。如果你后续还要用原始向量记得先拷贝一份。另外nlist的计算公式在小数据集上可能得到0所以用max(1, ...)兜底。索引构建完成后建议保存到磁盘避免每次重启都重新编码faiss.write_index(index, faiss.index) np.save(texts.npy, np.array(texts, dtypeobject))4.4 检索与重排的接口实现检索接口需要接收query返回Top-K候选块。这里我加上重排逻辑from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-base) def retrieve(query, index, texts, top_k5, recall_k50): # 第一阶段向量召回 query_vec model.encode([query]) faiss.normalize_L2(query_vec) distances, indices index.search(query_vec, recall_k) candidates [texts[i] for i in indices[0] if i ! -1] # 第二阶段重排 pairs [[query, cand] for cand in candidates] scores reranker.predict(pairs) # 按分数排序取Top-K ranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) return [{text: t, score: float(s)} for t, s in ranked[:top_k]]这里有个细节index.search返回的indices可能包含-1当索引中向量数量不足时所以要过滤。重排的predict方法返回的是logits不是概率但排序用足够了。如果你想要归一化的分数可以加一个sigmoid。4.5 FastAPI服务封装与启动最后把检索和生成封装成HTTP接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): query: str top_k: int 5 app.post(/retrieve) async def retrieve_endpoint(req: QueryRequest): results retrieve(req.query, index, texts, top_kreq.top_k) return {results: results} app.post(/generate) async def generate_endpoint(req: QueryRequest): # 先检索 results retrieve(req.query, index, texts, top_kreq.top_k) context \n\n.join([r[text] for r in results]) # 这里调用LLM省略具体实现 prompt f基于以下资料回答问题\n{context}\n\n问题{req.query} # answer llm.generate(prompt) return {answer: 生成的答案, sources: results}启动命令uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4。如果你用GPU注意多个worker会各自加载一份模型显存可能不够。这时候要么用单worker加异步要么用模型服务化方案如Triton单独部署embedding和reranker。5. 常见问题与排查技巧实录5.1 检索结果不相关先查分块再查模型这是最高频的问题。我的排查顺序是第一打印出召回块的原文看是否语义完整。如果块被切得七零八落调整分块参数。第二检查query和文档的语言是否一致。我遇到过用英文模型编码中文文档的情况相似度全是乱的。第三看是否需要重排。如果向量召回Top-50里有相关块但Top-5没有说明重排模型没起作用检查reranker的输入格式是否正确。第四考虑query改写。用户问“怎么退货”文档里写的是“退款流程”这时候需要同义词扩展或query改写。5.2 内存爆炸FAISS索引和模型加载是大头单机部署时内存主要消耗在三处embedding模型约400MB、reranker模型约1.1GB、FAISS索引每条向量512维float32约2KB十万条就是200MB。加起来不到2GB一般机器都能扛住。但如果你的数据量到百万级FAISS索引会膨胀到2GB以上这时候建议用IndexIVFPQ做量化压缩能把内存降到1/4代价是召回率下降2-3个百分点。另一个内存杀手是unstructured解析大PDF它会一次性加载整个文件建议分批处理。5.3 推理延迟高从批处理和缓存入手延迟优化有几个层次第一embedding和reranker都用GPU比CPU快5-10倍。第二开启批处理把多个query攒在一起编码。第三加缓存对相同的query直接返回缓存结果。我用functools.lru_cache做过简单缓存命中率在客服场景下能达到30%。第四如果LLM生成是瓶颈考虑流式输出让用户先看到部分结果。第五减少重排候选数从50降到20延迟能降一半召回率只降1-2个百分点。5.4 常见问题速查表问题现象可能原因排查方法解决方案检索结果完全不相关模型语言不匹配检查模型名称是否含zh换用中文模型召回率突然下降索引未重建检查模型版本和索引维度重建索引服务启动报错依赖版本冲突查看pip list按本文依赖组合重装内存持续增长缓存未清理监控进程内存加LRU缓存上限并发高时超时worker数不足查看uvicorn日志增加workers或限流重排后结果变差重排模型不适用对比重排前后Top-5换模型或关闭重排提示我习惯在每次修改分块参数或换模型后跑一个固定的评估集50-100条query-answer对记录召回率5和MRR。没有评估集你根本不知道改动是变好还是变坏。6. 可观测性与迭代让系统自己说话6.1 日志埋点记录每一次检索的“心电图”AI系统最怕黑盒。我在retrieve函数里埋了三个关键指标召回耗时、重排耗时、Top-1分数。这些数据写入日志文件每天用脚本聚合一次。如果发现Top-1分数持续低于0.5说明用户query和文档分布出现了漂移需要补充数据或调整模型。日志格式我用JSON Lines方便后续用pandas分析import json import time def retrieve_with_log(query, index, texts, top_k5): start time.time() results retrieve(query, index, texts, top_k) elapsed time.time() - start log_entry { query: query, top1_score: results[0][score] if results else 0, elapsed: elapsed, timestamp: time.time(), } with open(retrieval.log, a) as f: f.write(json.dumps(log_entry, ensure_asciiFalse) \n) return results6.2 反馈闭环用户点击就是最好的标注如果做的是面向用户的产品一定要加“有用/没用”按钮。用户点“没用”的query就是负样本点“有用”的就是正样本。积累几百条后可以用这些数据微调reranker效果提升非常明显。我试过在一个项目里用200条反馈数据微调bge-reranker-baseTop-1准确率从68%提升到79%。微调代码用sentence-transformers的CrossEncoder.fit方法几行就能搞定。6.3 版本管理与回滚别让新模型背锅每次换embedding模型或reranker都要记录版本号并且保留旧索引。我的做法是索引文件名带日期和模型名比如faiss_bge-small_20250101.index。服务启动时通过环境变量指定索引路径回滚只需改环境变量重启。另外模型文件也建议固定版本不要用latest标签因为上游更新可能导致行为变化。7. 一些踩坑后的个人体会这套ai-engineering-from-scratch的搭建过程我前后迭代了三个版本。第一版用LangChain全家桶两周跑通但第三周就遇到了检索结果无法解释的问题排查了两天发现是分块器默认参数把表格切碎了。第二版拆掉了LangChain自己写分块和检索代码量多了三倍但每个环节都可控。第三版加了重排和日志才算真正能上生产。我的体会是AI工程的门槛不在模型而在数据管道和系统设计。你花在清洗数据、调分块参数、设计评估集上的时间应该占总时间的70%以上。模型选型反而是最简单的因为社区已经帮你 benchmark 过了。另外不要追求一步到位。我见过太多人一开始就想上分布式向量库、Kubernetes、模型量化结果连单机版都没跑稳。我的建议是先用SQLiteFAISSFastAPI把单机版跑通压测到QPS 20再考虑横向扩展。单机版能撑住的场景比你想象的多得多。最后分享一个小技巧如果你的文档更新频繁不要每次全量重建索引可以用FAISS的add_with_ids和remove_ids做增量更新配合一个定时任务每天合并一次。这样能把索引重建时间从小时级降到分钟级。