1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人一个RAG问答系统用现成的框架搭起来可能只要一个下午一个Agent工作流拖拖拽拽也能跑通。但我观察到一个很有意思的现象能快速搭出Demo的人很多能把系统稳定跑在生产环境、出了问题能定位到根因的人少得可怜。这个差距恰恰就是“ai-engineering-from-scratch”这个方向要解决的核心问题。所谓from scratch不是让你从零实现一个Transformer、手写反向传播那是研究员的活而是指把AI工程链路里的每一个关键环节都亲手拆开、理解、复现一遍。比如向量检索你可以直接调一个向量数据库的API但如果你自己用numpy实现过一遍余弦相似度检索、自己算过一遍embedding的归一化、自己处理过一遍索引的持久化那么当线上召回效果突然变差时你脑子里是有画面的——你知道问题可能出在哪一层。这套东西适合谁我认为有三类人特别值得投入一是刚转行做AI应用的后端/全栈工程师你懂工程但不懂AI的那套数据流二是算法出身但工程能力偏弱的同学你懂模型但不懂怎么把它变成一个能扛住并发、能监控、能回滚的服务三是想真正搞懂AI系统内部机制的产品或技术负责人你不一定要写代码但你需要知道每个环节的成本、瓶颈和风险点在哪。我自己的经历是早期做AI项目时特别依赖框架遇到问题第一反应是去翻文档、搜issue效率极低。后来逼着自己把几个核心模块从零实现了一遍才发现很多“玄学问题”其实都有非常朴素的工程解释。这篇文章我就把这套from scratch的学习路径和实操细节完整拆一遍包括每一步为什么这么做、坑在哪、怎么验证你做对了。2. 整体学习路径设计把AI工程拆成五块可独立攻克的硬骨头2.1 为什么按“数据流”而不是按“技术栈”来拆很多人学AI工程习惯按技术栈来分先学PyTorch再学LangChain再学向量数据库。这种学法的问题是你学完一堆工具但脑子里没有一条完整的数据流遇到真实需求还是不知道从哪下手。我的建议是按数据流拆。一个典型的AI应用数据流大致是这样的原始数据进来 → 清洗和分块 → 转成向量或特征 → 存储和索引 → 检索或推理 → 后处理 → 返回结果。这条链路上的每一环都有它独立的技术难点和工程取舍。你按这条线去学每学一块都能立刻知道它在整个系统里的位置学完就能串起来。具体我把它拆成五块文本处理与分块、向量化与相似度计算、检索与索引、推理服务封装、评测与可观测性。这五块基本覆盖了一个AI应用从数据到服务的全生命周期而且每一块都可以脱离其他块独立练习。2.2 每块内容的难度梯度和预期产出我给自己定的标准是每一块都要有一个能跑、能测、能解释的最小实现。不是看完教程就算过而是合上教程能自己写出来并且能说清楚每个参数为什么这么设。模块核心难点预期产出建议投入时间文本处理与分块分块边界与语义完整性一个可配置的分块器支持重叠和元数据3-5天向量化与相似度embedding归一化与距离度量选择手写余弦相似度检索对比不同度量3-5天检索与索引索引结构与召回率权衡实现暴力检索简易倒排索引5-7天推理服务封装并发、超时、重试、降级一个带健康检查的推理API5-7天评测与可观测性评测集构建与指标定义一套自动化评测脚本日志埋点5-7天这个时间投入是按每天2-3小时有效学习算的如果你全职投入压缩到两三周完全可行。关键是不要跳步尤其是前两块它们是后面所有东西的地基。2.3 工具选型为什么我坚持用最朴素的库起步这一块我要特别说一下。from scratch阶段我强烈建议不要一上来就用LangChain、LlamaIndex这类高层框架。原因很简单这些框架帮你做了太多决策你学完只知道“这么调能跑”但不知道“为什么这么调”。我的选型原则是能用标准库就不用第三方能用轻量库就不用重框架。文本处理用Python自带的字符串操作加regex向量计算用numpy检索先用numpy暴力算服务用FastAPI这种轻量框架。等你把底层跑通了再去用高层框架你会发现你看框架源码的速度快得惊人因为每个抽象你都在底层见过。提示这不是说高层框架不好而是学习顺序的问题。先用底层建立直觉再用框架提升效率这个顺序反了你会一直停留在“调包侠”阶段。3. 核心模块拆解与实操要点每一块都亲手跑一遍3.1 文本分块别小看这个环节它决定了检索质量的上限文本分块看起来最简单但它是整个RAG系统里最容易被低估的环节。我见过太多项目检索效果差最后排查下来是分块策略有问题。分块的核心矛盾是块太大检索精度下降因为一个块里混了太多主题块太小语义不完整检索到了也没法用。常见的分块策略有固定长度、按句子、按段落、按语义相似度几种。我的实操建议是固定长度重叠边界优化三件套。固定长度好控制重叠保证跨块语义不断裂边界优化则是在切分时尽量不在句子中间断开。具体实现上我会先按标点符号切成句子然后贪心地往当前块里塞句子直到接近目标长度再开新块新块的开头会带上上一块的最后1-2个句子作为重叠。import re def split_into_sentences(text): # 按中英文标点切句保留标点 pattern r(?[。.!?])\s* sentences re.split(pattern, text) return [s.strip() for s in sentences if s.strip()] def chunk_text(text, max_len500, overlap_sentences1): sentences split_into_sentences(text) chunks [] current [] current_len 0 for sent in sentences: if current_len len(sent) max_len and current: chunks.append(.join(current)) # 保留最后overlap_sentences句作为下一块开头 current current[-overlap_sentences:] if overlap_sentences 0 else [] current_len sum(len(s) for s in current) current.append(sent) current_len len(sent) if current: chunks.append(.join(current)) return chunks这段代码有几个细节值得说。max_len设多少我的经验是中文300-500字英文200-400词这个区间在检索精度和语义完整性之间比较平衡。overlap_sentences设1-2句就够了设太多会导致重复内容过多检索时同一段内容反复出现。注意分块时一定要保留元数据比如来源文档、页码、章节标题。后面做引用溯源和结果过滤时这些元数据比正文还重要。3.2 向量化与相似度手写一遍你才知道embedding到底在算什么向量化这块很多人直接调API拿embedding就完事了。但如果你想真正理解检索我建议你至少手写一遍相似度计算并且对比不同度量方式的效果。embedding的本质是把文本映射到一个高维空间里的点语义相近的文本点之间的距离就近。常用的距离度量有余弦相似度、欧氏距离、点积。这里有个关键细节如果你的embedding没有做归一化点积和余弦相似度的结果是不一样的。归一化之后点积等价于余弦相似度计算还更快。import numpy as np def normalize(vectors): norms np.linalg.norm(vectors, axis1, keepdimsTrue) norms[norms 0] 1e-10 # 防止除零 return vectors / norms def cosine_similarity(query_vec, doc_vecs): # 假设都已归一化 return np.dot(doc_vecs, query_vec) def euclidean_distance(query_vec, doc_vecs): return np.linalg.norm(doc_vecs - query_vec, axis1)实测下来归一化后的余弦相似度在文本检索任务上最稳因为文本embedding的模长往往和文本长度相关不归一化的话长文本会天然占优这不是我们想要的。欧氏距离在归一化后和余弦相似度是单调等价的但计算量稍大所以我一般直接用点积。这里有个坑我踩过不同来源的embedding不能混用。你用A模型生成的文档向量用B模型生成的查询向量两者不在同一个语义空间里算出来的相似度毫无意义。所以生产环境一定要固定embedding模型并且把模型版本号写进索引元数据里。3.3 检索与索引从暴力检索到倒排索引理解召回率的代价检索这块from scratch阶段我建议分三步走先实现暴力检索再实现简易倒排索引最后对比两者的召回率和性能。暴力检索就是拿查询向量和所有文档向量算一遍相似度取Top-K。实现简单召回率100%在向量检索的意义上但文档一多就慢。倒排索引则是先做聚类或量化把文档分到不同的桶里查询时只算查询向量和桶中心的相似度选最近的几个桶再在桶内做精细检索。这样速度快但可能漏掉一些文档召回率下降。class BruteForceRetriever: def __init__(self, vectors, docs): self.vectors normalize(vectors) self.docs docs def search(self, query_vec, top_k5): query_vec query_vec / (np.linalg.norm(query_vec) 1e-10) scores np.dot(self.vectors, query_vec) top_indices np.argsort(scores)[::-1][:top_k] return [(self.docs[i], scores[i]) for i in top_indices]这个暴力检索器我建议你至少用一万条数据跑一遍感受一下延迟。一万条768维向量单次查询大概几毫秒到几十毫秒取决于机器。十万条就开始明显变慢百万条基本不可用。这时候你就理解为什么需要专门的向量索引了。实操心得做检索评测时一定要准备一个标注了正确答案的查询集。没有评测集你根本不知道你的检索是变好了还是变坏了。我一般会人工标注50-100个查询-文档对作为回归测试的基准。3.4 推理服务封装把模型变成一个能扛事的API模型跑通不等于服务可用。一个能上生产的推理服务至少要处理这几件事并发请求、超时控制、失败重试、降级策略、健康检查。我用FastAPI封装推理服务时有几个固定套路。第一模型加载放在启动时不要每次请求都加载。第二用异步接口处理IO密集的等待用线程池处理CPU密集的计算。第三给每个请求设超时超时直接返回降级结果不要让请求堆积。第四加一个/health接口返回模型是否加载成功、当前队列长度等。from fastapi import FastAPI, HTTPException from concurrent.futures import ThreadPoolExecutor import asyncio app FastAPI() executor ThreadPoolExecutor(max_workers4) model None app.on_event(startup) async def load_model(): global model model load_your_model() # 启动时加载 app.post(/predict) async def predict(request: dict): try: loop asyncio.get_event_loop() result await asyncio.wait_for( loop.run_in_executor(executor, model.predict, request[text]), timeout5.0 ) return {result: result} except asyncio.TimeoutError: raise HTTPException(status_code504, detail推理超时) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: ok if model is not None else loading}这里max_workers设多少我的经验是CPU核数左右因为推理通常是CPU密集的。如果是GPU推理那要考虑显存和batch size逻辑不一样。超时设5秒是个保守值具体要看你的模型延迟分布一般设成P99延迟的1.5倍比较合理。3.5 评测与可观测性没有度量就没有优化这一块是区分“玩具项目”和“工程项目”的分水岭。很多人做完系统就结束了从来不评测结果上线后效果好坏全靠感觉。评测的核心是定义指标构建评测集自动化跑。检索环节常用召回率、MRR、NDCG生成环节常用BLEU、ROUGE、以及现在更流行的人工评测或LLM-as-judge。我一般会先做检索评测因为检索是生成的上游检索不准生成再好也没用。可观测性则是日志指标追踪。日志记录每个请求的输入输出和中间结果指标记录延迟、QPS、错误率追踪记录一个请求经过了哪些环节、每个环节耗时多少。这三样东西能让你在出问题时快速定位。观测维度记录内容用途请求日志输入、输出、耗时、状态码问题复现与审计性能指标P50/P95/P99延迟、QPS、错误率容量规划与告警链路追踪各环节耗时、检索命中情况瓶颈定位提示日志里千万不要记录用户的敏感信息尤其是原始文本。我一般只记录文本的哈希值和长度需要复现时再根据哈希去查脱敏后的存储。4. 完整实操流程从零搭一个可评测的检索问答系统4.1 环境准备与依赖安装这一节我把整个流程串一遍你可以跟着做。环境用Python 3.10依赖尽量少pip install numpy fastapi uvicorn sentence-transformerssentence-transformers是用来生成embedding的如果你想完全from scratch也可以用HuggingFace的transformers自己写池化逻辑但那个属于模型层面的from scratch这里我们聚焦工程层面用现成的embedding模型就够了。数据我建议用一份公开的中文问答数据集或者你自己收集一批文档。我一般会准备三个文件docs.jsonl存文档queries.jsonl存查询qrels.jsonl存查询和文档的对应关系用于评测。4.2 数据加载与分块处理加载数据后先做分块。这里要注意分块前最好做一次清洗去掉HTML标签、多余空白、特殊字符。清洗不干净embedding质量会受影响。import json def load_docs(path): docs [] with open(path, r, encodingutf-8) as f: for line in f: item json.loads(line) docs.append(item) return docs def process_docs(docs, max_len400): processed [] for doc in docs: text clean_text(doc[content]) chunks chunk_text(text, max_lenmax_len) for i, chunk in enumerate(chunks): processed.append({ doc_id: doc[id], chunk_id: f{doc[id]}_{i}, text: chunk, source: doc.get(source, ), position: i }) return processedmax_len400是我在中文场景下的常用值。如果你的文档是技术文档、法律条文这种信息密度高的可以设小一点300左右如果是小说、新闻这种叙述性的可以设大一点500-600。4.3 向量化与索引构建分块完成后批量生成embedding。这里有个性能细节批量生成比逐条生成快得多因为模型可以并行处理。我一般batch size设32或64具体看显存。from sentence_transformers import SentenceTransformer import numpy as np model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def build_index(chunks, batch_size64): texts [c[text] for c in chunks] embeddings model.encode(texts, batch_sizebatch_size, show_progress_barTrue) embeddings normalize(embeddings) return embeddings # 保存索引 embeddings build_index(processed_chunks) np.save(embeddings.npy, embeddings) with open(chunks.json, w, encodingutf-8) as f: json.dump(processed_chunks, f, ensure_asciiFalse)索引保存后加载时要注意向量和chunk的对应关系不能乱。我一般用同一个顺序保存加载时按顺序读回来。如果中间有过滤或重排一定要同步维护这个映射。4.4 检索接口实现与参数调优检索接口的核心是查询向量化相似度计算Top-K返回。Top-K设多少这取决于你的下游怎么用。如果是给LLM做上下文K一般设3-5因为上下文长度有限如果是做召回阶段K可以设20-50后面再接一个精排。class Retriever: def __init__(self, embeddings, chunks): self.embeddings embeddings self.chunks chunks def search(self, query, top_k5): query_vec model.encode([query])[0] query_vec query_vec / (np.linalg.norm(query_vec) 1e-10) scores np.dot(self.embeddings, query_vec) top_indices np.argsort(scores)[::-1][:top_k] results [] for idx in top_indices: results.append({ text: self.chunks[idx][text], score: float(scores[idx]), source: self.chunks[idx][source] }) return results调优时我一般会试几个维度分块大小、Top-K、是否加重叠、embedding模型。每次只改一个变量用评测集跑一遍记录指标变化。这样你才能知道每个参数的实际影响。4.5 评测脚本编写与指标计算评测脚本的核心是对每个查询检索Top-K然后和标注的正确答案对比。常用的指标是RecallK和MRR。def evaluate(retriever, queries, qrels, top_k5): recall_sum 0 mrr_sum 0 for q in queries: results retriever.search(q[text], top_ktop_k) retrieved_ids [r[chunk_id] for r in results] relevant_ids qrels.get(q[id], []) # RecallK hits len(set(retrieved_ids) set(relevant_ids)) recall_sum hits / len(relevant_ids) if relevant_ids else 0 # MRR for rank, rid in enumerate(retrieved_ids, 1): if rid in relevant_ids: mrr_sum 1 / rank break n len(queries) return {recallk: recall_sum / n, mrr: mrr_sum / n}跑完评测你会得到一组基线数字。之后每次改动都跑一遍评测对比基线。没有评测的优化都是瞎猜这句话我在很多项目里验证过。5. 常见问题与排查技巧实录5.1 检索结果不相关从分块和embedding两头查检索不准是最常见的问题。我的排查顺序是先看分块再看embedding最后看相似度计算。分块问题通常是块太大或太小。块太大一个块里混了多个主题查询匹配到的是块里的次要内容块太小语义不完整embedding本身就表达不清。排查方法是把检索到的块打印出来人工看一眼如果块内容明显不相关那就是分块问题。embedding问题通常是模型不适合你的领域。通用embedding模型在专业领域医疗、法律、代码上效果会打折。排查方法是拿几个典型查询看它们的embedding和正确答案的embedding相似度如果明显偏低考虑换领域模型或做微调。相似度计算问题通常是没归一化或度量选错。这个用前面说的归一化点积基本能解决。5.2 服务延迟高先定位瓶颈在哪一环延迟高不要盲目优化先定位。我的做法是在请求链路里埋点记录每个环节的耗时查询向量化、检索、后处理、返回。哪个环节占比高就优化哪个。常见瓶颈和对策瓶颈环节典型原因对策查询向量化模型太大或没批处理换小模型、缓存高频查询检索暴力检索数据量大换近似索引、减少候选集后处理逻辑复杂或重复计算简化逻辑、加缓存网络跨服务调用多合并服务、本地缓存我遇到最多的是查询向量化慢因为每次请求都跑一遍模型。后来加了一个查询缓存高频查询直接命中延迟降了一个数量级。5.3 评测指标波动大检查评测集和随机性评测指标忽高忽低通常是两个原因评测集太小或标注不一致或者系统有随机性。评测集至少要50个查询以上最好100-200个而且标注要统一标准。如果两个人标注要先对齐标准否则指标没有可比性。系统随机性方面embedding模型一般没有随机性但如果你用了采样或dropout那要固定随机种子。实操心得我一般会把评测集分成开发集和测试集开发集用来调参测试集只在最后验证一次。这样能避免过拟合到评测集上。5.4 常见问题速查表现象可能原因排查方法解决方案检索结果完全不相关分块或embedding问题打印检索块人工检查调整分块或换模型检索结果部分相关Top-K太小或相似度阈值问题增大K看是否改善调K或加精排服务超时某环节耗时过长链路埋点定位优化瓶颈环节评测指标低评测集或系统问题检查标注和随机性扩充评测集、固定种子内存占用高索引或缓存过大监控内存曲线压缩索引、限制缓存6. 从能跑到好用几个让系统更稳的工程习惯6.1 版本管理模型、索引、代码要一起管AI系统有个特殊之处同一个代码配不同的模型或索引效果可能完全不同。所以版本管理不能只管代码模型版本、索引版本、评测集版本都要管。我的做法是给每次索引构建打一个版本号记录用了哪个embedding模型、什么分块参数、多少条数据。代码里读取索引时把版本号也读进来日志里打出来。这样出问题时你能快速定位是哪个版本的问题。6.2 灰度与回滚新版本先小流量验证任何改动包括换模型、调分块、改检索逻辑都不要全量上线。先切10%流量观察核心指标检索命中率、延迟、错误率稳定后再逐步放量。回滚方案要提前准备好索引和模型都要能一键切回旧版本。6.3 成本控制embedding和推理都是钱embedding调用和模型推理都是按量计费的量大了成本很可观。控制成本的手段有几个缓存高频查询的embedding和结果、对文档做去重、用更小的模型做粗筛。我一般会统计每天的调用量和成本设一个告警阈值超了就排查是不是有异常流量。6.4 持续评测把评测变成日常最后也是最重要的把评测脚本接入CI每次改动自动跑一遍。指标下降就阻断合并。这个习惯一旦建立你的系统质量会非常稳定因为任何退化都会在合并前被发现。我个人在实际操作中的体会是from scratch最大的价值不是让你造轮子而是让你在轮子出问题时知道怎么修。这套东西学下来你对AI系统的掌控感会完全不一样遇到问题不再是“试试这个试试那个”而是有逻辑地定位和解决。后续如果你想继续深入可以往两个方向扩展一是把检索换成更复杂的混合检索向量关键词二是把评测从检索扩展到生成质量。这两个方向都需要你已经有了扎实的底层理解否则很容易做成黑盒调参。