1. 从零构建AI工程能力为什么“手搓”比调包更值得投入这两年AI应用层的工具链成熟得吓人LangChain、LlamaIndex、各种Agent框架几乎把能封装的都封装了。打开文档三行代码就能跑通一个RAG问答复制粘贴十分钟搭出一个聊天机器人。表面上看门槛被踩到了地板上但我自己带过几批人之后发现一个很尴尬的现象很多人能跑通Demo却说不清楚一次检索到底发生了什么向量维度为什么是那个数Token是怎么被切分的模型返回的logits怎么变成最终那句话。一旦线上出了badcase比如检索召回一堆不相关的内容、回答开始胡言乱语、延迟突然飙高他们完全没有排查的抓手只能靠猜和反复重启。ai-engineering-from-scratch这个标题说的就是另一条路——不满足于当API的搬运工而是从最底层把AI工程的关键环节自己实现一遍。它不是一个具体的开源库而是一种学习与实践的范式从分词、嵌入、注意力机制、向量检索、推理服务到评估每一层都亲手写一遍最小可用版本理解每个参数背后的取舍。这件事的价值不在于“重复造轮子”而在于造过轮子的人才知道轮子什么时候会爆胎。这篇文章适合三类人一是刚入行AI应用开发、被各种框架绕晕的新人想搞清楚黑盒里到底装了什么二是有一定后端或算法基础、想系统补齐AI工程链路的工程师三是带团队的技术负责人想给团队设计一套真正能提升工程能力的训练路径。我会把整条链路拆成可操作的模块讲清楚每一步为什么这么做、参数怎么算、坑在哪里尽量让你看完就能动手复现。2. 整体设计思路把AI工程拆成可独立验证的六层2.1 为什么选择“自底向上”而不是“自顶向下”大多数人学AI工程是从调用开始的先跑通一个完整应用再往下钻。这条路的问题是你看到的永远是别人抽象好的接口遇到问题时不知道往哪一层看。自底向上的逻辑正好相反——先把每一层的最小实现写出来再往上组装。这样做的好处是每一层你都有独立的测试手段分词器对不对拿几个句子跑一下看切分结果嵌入模型好不好算一下相似句和无关句的余弦距离检索准不准构造一批query-doc对看召回率。每一层都可验证整个系统的行为就不再是玄学。我一般把AI工程拆成六层文本预处理与分词、嵌入表示、向量索引与检索、模型推理与生成控制、服务化与性能、评估与监控。这六层基本覆盖了一个典型AI应用从输入到输出的全链路。下面这张表是我给团队做培训时用的分层说明你可以对照着看自己卡在哪一层。层级核心职责常见黑盒工具自建最小实现的关键点文本预处理清洗、切分、归一化各框架内置tokenizer理解BPE/WordPiece的合并规则嵌入表示把文本映射为向量OpenAI Embedding API理解维度、归一化、相似度度量向量检索快速找最近邻FAISS、Milvus理解索引类型与召回/精度权衡推理生成控制模型输出各家推理API理解温度、top-p、采样逻辑服务化稳定对外提供能力FastAPI封装理解并发、批处理、超时评估监控量化效果与稳定性各种eval框架理解指标定义与数据构造2.2 每一层的最小可用标准是什么“从零实现”不等于“实现一个生产级系统”那样成本太高也没必要。我的原则是每一层实现到能解释清楚关键参数、能复现核心行为、能定位常见故障即可。比如分词层你不需要实现一个支持几万词表的工业级BPE但你需要写一个能处理几百个词、能展示合并过程的小版本这样你才能理解为什么中文分词和英文分词策略不同、为什么子词切分能缓解OOV问题。再比如检索层你不需要实现HNSW的完整工程优化但你需要实现一个暴力检索作为baseline再对比加索引后的召回变化理解“近似”到底近似在哪。这个标准很重要因为它决定了你的时间投入产出比。我见过有人一上来就想复现一个完整的Transformer训练结果卡在反向传播的数学推导上两周热情直接耗尽。正确的做法是分层推进每层花半天到一天先跑通再优化。2.3 技术选型的几个取舍语言上我推荐Python生态最全numpy和torch足够支撑所有底层实现。但有一个例外如果你主要做服务化Go或Rust在并发和内存上更有优势不过学习阶段没必要给自己加难度。依赖上我建议只允许numpy、torch这类基础库禁止直接调用高层封装否则就失去了“from scratch”的意义。比如做嵌入你可以用预训练模型加载权重但池化、归一化、相似度计算必须自己写。做检索可以用numpy做矩阵运算但索引结构要自己搭。提示不要一开始就追求性能。先用最笨的方法比如双重循环算相似度把逻辑跑通再逐步替换成向量化、加索引。性能优化是理解之后的奖励不是理解的前提。3. 核心细节解析分词、嵌入与检索的底层逻辑3.1 分词器为什么子词切分是绕不开的一课文本进入模型之前第一步就是变成数字。最朴素的做法是按空格切词但这样会遇到两个死结一是词表爆炸英语词汇量几十万加上专业术语和拼写变体根本管不住二是OOV问题没见过的词只能映射成UNK信息直接丢失。子词切分subword tokenization就是来解决这个矛盾的——把罕见词拆成常见子词比如“tokenization”拆成“token”“ization”既控制了词表规模又保留了语义线索。BPEByte Pair Encoding是最经典的实现。它的核心思想特别简单从字符开始统计相邻字符对的出现频率把最高频的一对合并成新单元重复这个过程直到达到目标词表大小。我建议你亲手写一遍这个合并过程哪怕只处理几百个句子。写完之后你会突然明白几个之前想不通的问题为什么有些模型的词表里会有奇怪的带前缀符号比如GPT系列的Ġ表示空格为什么同一个词在不同上下文里可能被切成不同的子词为什么中文的token效率普遍比英文低因为汉字本身信息密度高一个汉字往往就是一个token但词表里中文词条有限。实操上我一般用一个小语料库比如几千条英文句子加几百条中文句子来训练自己的BPE目标词表设成1000左右。训练完拿几个典型句子测试英文的“unbelievable”、中文的“人工智能工程师”、中英混排的“用Python做NLP”。观察切分结果你会发现很多有意思的细节。这里有个容易踩的坑训练语料的领域要和实际使用场景接近否则切分策略会偏。比如你用新闻语料训练的BPE去切医学文本专业术语会被切得很碎影响后续嵌入质量。3.2 嵌入表示维度、归一化与相似度的三角关系嵌入就是把文本变成一串数字向量让语义相近的文本在向量空间里距离也近。这里有几个关键决策点每一个都会影响最终效果。第一个是维度选择。维度太低表达能力不够不同语义挤在一起维度太高计算和存储成本上升还容易过拟合。常见的有128、256、384、768、1024、1536。我的经验是对于中小规模检索任务384到768是个甜点区。你可以做个简单实验用同一批文本分别降到128维和升到1024维看检索召回率的变化。通常你会发现从128到384提升明显从768到1024提升就很小了边际收益递减。第二个是归一化。向量归一化到单位长度之后余弦相似度就等价于点积计算更快。更重要的是归一化能消除向量模长带来的干扰——有些文本天然容易产生大模长向量不归一化的话相似度会被模长主导。我一般默认做L2归一化除非有特殊理由。第三个是相似度度量。余弦相似度看方向欧氏距离看绝对位置点积两者都掺一点。对于归一化后的向量余弦和点积等价。实际用的时候如果向量没归一化优先用余弦归一化了就随便。这里有个细节不同模型产出的向量空间不可直接比较你不能拿模型A的query向量去检索模型B的doc向量哪怕维度一样也不行。换模型必须重建整个索引这是很多人迁移时踩的坑。3.3 向量检索暴力检索是所有优化的基准线检索的本质是最近邻搜索给定一个query向量在doc向量集合里找最相似的K个。最笨的方法是暴力检索——query和每个doc算一次相似度排序取TopK。复杂度是O(N*D)N是doc数量D是维度。当N只有几千几万时暴力检索完全够用而且结果精确。我强烈建议你先实现暴力检索作为baseline记录下它的延迟和召回再去对比加索引后的变化。当N上到百万级暴力检索就扛不住了这时候需要近似最近邻ANN索引。常见的有几类基于树的如KD-Tree高维下退化严重、基于量化的如PQ把向量分段量化压缩、基于图的如HNSW构建近邻图做贪心搜索。HNSW是目前综合表现最好的但它的参数不少比如每层的连接数M、构建时的候选集大小efConstruction、搜索时的候选集大小efSearch。M和efConstruction越大索引质量越高但构建越慢efSearch越大召回越高但查询越慢。这几个参数的调优没有银弹只能根据你的数据分布和延迟要求去试。我一般会做一个召回率-延迟曲线固定其他参数逐步增大efSearch记录召回率和P99延迟找到拐点。通常efSearch在64到256之间能覆盖大部分场景。还有一个容易忽略的点索引需要定期重建因为doc集合会变。增量更新对某些索引类型支持不好HNSW虽然支持插入但频繁插入会降低图的质量定期重建更稳妥。4. 实操过程从零搭一个可验证的检索问答链路4.1 环境准备与依赖约束先把环境搭起来。我用的是一台普通开发机不需要GPU也能跑通大部分环节只有加载预训练嵌入模型时用一下CPU推理。依赖清单我刻意压到最少pip install numpy torch transformers fastapi uvicorn注意这里没有装任何向量数据库或检索框架检索逻辑全部自己写。transformers只用来加载预训练权重不调用它的pipeline高层接口。这样做的目的是逼自己处理每一个中间步骤。目录结构我习惯这样组织每一层一个模块方便单独测试ai-from-scratch/ tokenizer/ bpe.py embedding/ encoder.py retrieval/ index.py serving/ app.py eval/ metrics.py4.2 手写一个最小BPE分词器先实现BPE的核心逻辑。思路是统计词频把每个词拆成字符序列然后迭代合并最高频的相邻对。下面是一个精简版实现重点看合并循环from collections import Counter def get_stats(vocab): pairs Counter() for word, freq in vocab.items(): symbols word.split() for i in range(len(symbols) - 1): pairs[symbols[i], symbols[i1]] freq return pairs def merge_vocab(pair, vocab): new_vocab {} bigram .join(pair) replacement .join(pair) for word, freq in vocab.items(): new_word word.replace(bigram, replacement) new_vocab[new_word] freq return new_vocab def train_bpe(corpus, num_merges): vocab Counter() for sentence in corpus: for word in sentence.split(): vocab[ .join(list(word)) /w] 1 merges [] for i in range(num_merges): pairs get_stats(vocab) if not pairs: break best max(pairs, keypairs.get) vocab merge_vocab(best, vocab) merges.append(best) return merges, vocab跑一遍之后把merges打印出来你会看到合并顺序其实反映了语料里的构词规律。比如高频的“ing”“tion”会较早被合并。这个观察对理解子词很重要。测试的时候我建议准备三组句子纯英文、纯中文、中英混排分别看切分结果。中文因为没有空格需要先按字切分再合并效果和英文差异很大这个对比能让你直观感受到不同语言的token效率差异。注意这个实现没有处理特殊符号和大小写生产环境需要加normalization。但学习阶段保持简单把注意力放在合并逻辑上。4.3 嵌入与相似度从模型输出到可用向量加载一个预训练嵌入模型把句子编码成向量。关键是自己实现池化和归一化不要用模型自带的sentence-transformers封装import torch from transformers import AutoTokenizer, AutoModel tokenizer AutoTokenizer.from_pretrained(bert-base-chinese) model AutoModel.from_pretrained(bert-base-chinese) def encode(texts, batch_size16): all_vecs [] for i in range(0, len(texts), batch_size): batch texts[i:ibatch_size] inputs tokenizer(batch, paddingTrue, truncationTrue, max_length128, return_tensorspt) with torch.no_grad(): outputs model(**inputs) # 自己做mean pooling排除padding mask inputs[attention_mask].unsqueeze(-1).float() summed (outputs.last_hidden_state * mask).sum(dim1) counts mask.sum(dim1).clamp(min1e-9) vecs summed / counts # L2归一化 vecs torch.nn.functional.normalize(vecs, p2, dim1) all_vecs.append(vecs) return torch.cat(all_vecs, dim0).numpy()这里有几个细节值得说。mean pooling要排除padding否则短句会被padding的向量稀释语义偏移。归一化放在池化之后顺序不能反。max_length要设合理太长浪费算力太短截断信息128对大多数句子够用。编码完之后拿几个句子算余弦相似度验证一下语义相近的应该明显高于无关的。如果发现相似度普遍偏高比如都在0.9以上可能是模型不适合你的领域或者归一化出了问题。4.4 暴力检索与索引检索的对比实验先实现暴力检索作为精确基准import numpy as np def brute_force_search(query_vec, doc_vecs, top_k5): # query_vec和doc_vecs都已归一化 scores doc_vecs query_vec idx np.argsort(-scores)[:top_k] return idx, scores[idx]然后构造一批测试数据比如5000条doc100条query每条query有标注的正确doc。跑暴力检索记录召回率和延迟。接着实现一个简单的PQ量化索引做对比把向量分成若干段每段做kmeans聚类用聚类中心ID代替原始向量检索时先算query到各中心的距离再查表累加。PQ的召回会低于暴力检索但延迟大幅下降。这个对比能让你直观理解“近似”的代价。我实测下来5000条doc、384维的情况下暴力检索单次查询在几毫秒级别完全可用。真正需要索引是到十万级以上。所以不要过早优化先把baseline跑通。4.5 服务化把链路封装成API用FastAPI把整个链路包起来暴露一个检索接口和一个问答接口。这里的关键是批处理和超时控制。检索接口接收query文本内部完成编码、检索、返回TopK文档。问答接口在检索基础上加一层生成把检索到的文档拼进prompt。服务化阶段最容易出问题的是并发下的资源竞争比如模型推理不是线程安全的需要用锁或者队列串行化。我一般用一个简单的信号量控制并发数超过就排队或直接返回繁忙。from fastapi import FastAPI from pydantic import BaseModel import asyncio app FastAPI() sem asyncio.Semaphore(4) class QueryReq(BaseModel): text: str top_k: int 5 app.post(/search) async def search(req: QueryReq): async with sem: vec encode([req.text]) idx, scores brute_force_search(vec[0], doc_vecs, req.top_k) return {results: [{doc_id: int(i), score: float(s)} for i, s in zip(idx, scores)]}超时控制用asyncio.wait_for包一层避免单个慢请求拖垮整个服务。这些细节看起来琐碎但线上稳定性就靠它们。5. 常见问题与排查技巧实录5.1 检索召回不准的排查路径召回不准是最常见的问题排查要按链路顺序来不要跳步。第一步确认query编码是否正确把query向量和它应该匹配的doc向量算一下相似度如果这个相似度本身就很低说明编码环节有问题可能是模型不匹配领域或者文本预处理把关键信息洗掉了。第二步确认doc向量是否更新如果doc集合变了但索引没重建检索结果会错乱。第三步检查相似度度量是否一致编码时归一化了检索时也要用归一化后的向量否则量纲不一致。第四步看TopK的分数分布如果Top1和Top5分数很接近说明区分度不够可能需要换模型或加rerank。我整理了一个速查表遇到问题按这个顺序过一遍基本能定位到八成以上的故障现象可能原因排查方法召回内容完全不相关编码模型领域不匹配人工算query-doc相似度召回内容部分相关索引未更新或度量不一致检查索引时间戳和归一化TopK分数都很接近向量区分度不足换模型或加rerank层延迟突然升高并发过高或索引退化看P99延迟和并发数结果不稳定采样随机性或竞态固定随机种子加锁5.2 生成环节的幻觉与截断问题生成环节的坑主要集中在两点幻觉和截断。幻觉是指模型编造了检索内容里没有的信息。缓解办法是在prompt里明确约束“只根据提供的资料回答资料没有就说不知道”并且把检索到的原文完整放进去不要做摘要。截断是指回答被max_tokens截断说了一半就没了。这个要在服务层做处理检测到finish_reason是length时要么提示用户要么自动续写。续写要注意拼接处的连贯性我一般会在续写时把已生成的部分作为上下文传回去。还有一个隐蔽的坑是prompt注入用户输入里如果包含类似“忽略之前的指令”这样的内容可能让模型偏离任务。防御办法是在拼接时对用户输入做转义或加分隔符让模型清楚哪部分是资料、哪部分是问题。5.3 性能瓶颈的定位方法性能问题不要凭感觉猜要打点。我在链路的每个环节都加了计时编码耗时、检索耗时、生成耗时、总耗时。跑一批请求看哪个环节占比最大。通常编码和生成是大头检索相对轻。编码优化可以用批处理、量化、换更小的模型生成优化可以用流式输出、限制max_tokens、缓存常见问题的回答。检索优化前面说过先确认是不是真的需要索引。提示缓存是把双刃剑。语义缓存用相似度判断是否命中能大幅降低重复问题的延迟但阈值设不好会返回错误答案。我一般把阈值设得很高比如0.95以上宁可多算几次也不返回错答案。5.4 评估数据怎么造才靠谱没有评估就没有优化。评估数据要覆盖三类正例query和doc确实相关、难负例看起来相关其实不相关、随机负例。难负例最能暴露问题构造方法是拿检索TopK里排名靠前但标注为不相关的样本。指标上检索看RecallK和MRR生成看人工评分或基于规则的匹配。评估集要定期更新因为数据分布会漂移。我一般每两周抽一批线上真实query人工标注后加入评估集保持评估的时效性。6. 从能跑到好用工程化收尾的几个关键动作把链路跑通只是起点真正让它可用还需要几个收尾动作。日志要结构化每次请求记录query、检索结果、生成结果、各环节耗时方便事后分析。监控要覆盖关键指标包括QPS、P99延迟、错误率、召回率用评估集定期跑。配置要外置模型路径、索引路径、阈值这些不要硬编码用配置文件管理方便切换环境。版本要管理模型版本、索引版本、代码版本三者要对应出问题能回滚。我个人在实际操作中的体会是从零实现最大的收获不是写出了多牛的代码而是建立了一套“可解释”的直觉。当线上出问题时你知道该看哪一层、该调哪个参数而不是对着黑盒干瞪眼。这套能力是调包调不出来的。后续如果你想继续深入可以往两个方向扩展一是把生成环节也自己实现一遍采样逻辑理解temperature和top-p的真实作用二是把评估做成自动化流水线每次代码变更自动跑一遍回归。这两件事做完你对AI工程的理解会再上一个台阶。