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”,既控制了词表规模,又保留了语义线索。
BPE(Byte 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[i+1]] += 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, key=pairs.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_size=16): all_vecs = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] inputs = tokenizer(batch, padding=True, truncation=True, max_length=128, return_tensors='pt') with torch.no_grad(): outputs = model(**inputs) # 自己做mean pooling,排除padding mask = inputs['attention_mask'].unsqueeze(-1).float() summed = (outputs.last_hidden_state * mask).sum(dim=1) counts = mask.sum(dim=1).clamp(min=1e-9) vecs = summed / counts # L2归一化 vecs = torch.nn.functional.normalize(vecs, p=2, dim=1) all_vecs.append(vecs) return torch.cat(all_vecs, dim=0).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_k=5): # query_vec和doc_vecs都已归一化 scores = doc_vecs @ query_vec idx = np.argsort(-scores)[:top_k] return idx, scores[idx]然后构造一批测试数据,比如5000条doc,100条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里排名靠前但标注为不相关的样本。指标上,检索看Recall@K和MRR,生成看人工评分或基于规则的匹配。评估集要定期更新,因为数据分布会漂移。我一般每两周抽一批线上真实query,人工标注后加入评估集,保持评估的时效性。
6. 从能跑到好用:工程化收尾的几个关键动作
把链路跑通只是起点,真正让它可用还需要几个收尾动作。日志要结构化,每次请求记录query、检索结果、生成结果、各环节耗时,方便事后分析。监控要覆盖关键指标,包括QPS、P99延迟、错误率、召回率(用评估集定期跑)。配置要外置,模型路径、索引路径、阈值这些不要硬编码,用配置文件管理,方便切换环境。版本要管理,模型版本、索引版本、代码版本三者要对应,出问题能回滚。
我个人在实际操作中的体会是,从零实现最大的收获不是写出了多牛的代码,而是建立了一套“可解释”的直觉。当线上出问题时,你知道该看哪一层、该调哪个参数,而不是对着黑盒干瞪眼。这套能力,是调包调不出来的。后续如果你想继续深入,可以往两个方向扩展:一是把生成环节也自己实现一遍采样逻辑,理解temperature和top-p的真实作用;二是把评估做成自动化流水线,每次代码变更自动跑一遍回归。这两件事做完,你对AI工程的理解会再上一个台阶。