1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人。想做个RAG问答LangChain几行代码就能跑通想微调个模型LoRA脚本满天飞想部署个推理服务vLLM一条命令起服务。表面上看AI工程的门槛已经被拉到了历史最低点。但我观察到一个很有意思的现象越是工具链成熟的时候真正能把线上AI系统扛住的人反而越稀缺。原因很简单——调包能让你跑通Demo但跑不通生产环境里那些千奇百怪的坑。“ai-engineering-from-scratch”这个标题我理解它想表达的核心诉求是不依赖高层封装从最底层的组件开始把AI工程链路里的每一环亲手实现一遍。这不是反对用框架而是主张一种学习路径——先知道轮子怎么造再去用轮子。适合谁看我觉得有三类人最该走这条路一是刚转行做AI应用、只会调API但说不清背后原理的开发者二是在小团队里被迫“全栈”扛AI系统、什么都得自己上的工程师三是面试时被问到“Transformer的KV Cache怎么优化”就卡壳的求职者。这篇文章不会给你一个“三天速成AI工程师”的幻觉。我会按照一条真实的从零构建路径把数据管道、模型推理、向量检索、服务编排、评测监控这几个核心环节拆开讲每个环节都告诉你为什么这个环节值得手搓一遍、手搓的时候关键决策点在哪、以及我实际踩过的坑。全文基于我在多个中小规模AI项目里的实践总结代码和配置都是可复现的但更重要的是背后的取舍逻辑。2. 数据管道从原始文本到模型能吃的格式中间隔着一堆脏活2.1 为什么数据清洗值得你亲手写一遍很多人做AI项目的第一步是找现成的数据集然后直接喂给模型。这在学术benchmark上没问题但真实业务里的数据源——用户上传的PDF、爬下来的网页、内部wiki导出的markdown——格式之混乱超出想象。我见过一个项目团队花了三周调模型效果最后发现是数据里混入了大量HTML标签和重复段落模型在学噪声。手写数据清洗管道的好处是你会被迫面对每一个脏数据的具体形态。比如PDF解析PyPDF2和pdfplumber的差异在哪扫描件和文字版PDF的处理路径完全不同前者需要OCR后者直接抽文本。我一般会先用pdfplumber做一版快速抽取统计一下每页的字符数和空白页比例如果发现某份PDF平均每页字符数低于50基本可以判定是扫描件得走OCR路线。import pdfplumber def diagnose_pdf(path): with pdfplumber.open(path) as pdf: stats [] for i, page in enumerate(pdf.pages): text page.extract_text() or stats.append({ page: i, chars: len(text.strip()), has_table: len(page.extract_tables()) 0 }) avg_chars sum(s[chars] for s in stats) / len(stats) return avg_chars, stats这段诊断代码看着简单但它能帮你省下大量盲目调试的时间。先诊断再处理这是我做数据管道的第一原则。2.2 分块策略固定长度是最省事但最坑的选择文本分块chunking是RAG系统里最容易被低估的环节。大多数教程教你按固定字符数切比如每500字一块重叠50字。这个策略在demo里能用但在真实文档里会切碎语义单元——一个完整的表格被切成两半一段代码的上下文被拦腰截断。我现在的做法是按结构分块先用正则或解析器识别文档的标题层级、段落边界、代码块、表格然后以“语义完整”为优先长度约束为次要。比如markdown文档我会按##和###标题切分如果某个章节超过800字再在段落级别二次切分。这样切出来的块每一块都是一个自包含的语义单元。import re def structural_chunk(md_text, max_len800): # 按二级标题切分 sections re.split(r\n(?## ), md_text) chunks [] for sec in sections: if len(sec) max_len: chunks.append(sec) else: # 超长章节按段落二次切分 paras sec.split(\n\n) buf for p in paras: if len(buf) len(p) max_len and buf: chunks.append(buf) buf p else: buf \n\n p if buf else p if buf: chunks.append(buf) return chunks提示分块长度没有万能值。我的经验是问答类场景块可以小一些300-500字摘要类场景块可以大一些800-1200字。关键是先确定下游任务再定分块参数而不是反过来。2.3 去重与质量过滤别让模型学废话数据里最常见的两种噪声是近似重复和低信息密度内容。近似重复用MinHash或者简单的SimHash就能检测低信息密度则可以通过“有效字符比例”“停用词占比”“是否包含大量特殊符号”来过滤。我一般会设一个简单的规则组合有效字符中文、英文、数字占比低于60%的块直接丢弃连续重复字符超过10个的块标记为可疑与已有块SimHash相似度超过0.9的块去重这些规则不复杂但能过滤掉大部分“看起来有内容、实际是垃圾”的数据。数据质量的上限决定了模型效果的上限这句话在AI工程里怎么强调都不过分。3. 推理层不调API自己把模型跑起来才能看清的细节3.1 本地推理环境搭建显存是硬约束从零做AI工程绕不开自己跑模型。不管是做embedding还是做生成本地推理能让你对延迟、显存、吞吐有第一手的感知。我一般用HuggingFace的transformers做基础推理配合accelerate做设备管理。第一步永远是算显存账模型规模FP16权重KV Cache2048上下文建议显存1.5B~3GB~0.5GB6GB7B~14GB~2GB20GB13B~26GB~4GB36GB这张表是估算实际会因batch size和实现细节浮动。但核心逻辑是权重占大头KV Cache随上下文线性增长。如果你只有一张消费级显卡7B模型用4-bit量化是性价比最高的选择。from transformers import AutoModelForCausalLM, AutoTokenizer import torch model_name Qwen/Qwen2.5-7B-Instruct tokenizer AutoTokenizer.from_pretrained(model_name) model AutoModelForCausalLM.from_pretrained( model_name, torch_dtypetorch.float16, device_mapauto, load_in_4bitTrue # 需要bitsandbytes ) inputs tokenizer(解释一下什么是KV Cache, return_tensorspt).to(model.device) outputs model.generate(**inputs, max_new_tokens256) print(tokenizer.decode(outputs[0], skip_special_tokensTrue))这段代码跑通不难难的是理解device_mapauto背后发生了什么——它会自动把不同层分配到可用设备上如果显存不够会offload到CPU但offload会导致推理速度断崖式下降。我踩过的坑是以为auto能解决一切结果模型跑起来了但每token要等好几秒排查半天才发现是部分层被offload了。后来我养成了一个习惯加载完模型后打印一下各设备的显存占用for i, device in enumerate(model.hf_device_map.values()): if isinstance(device, int): print(fLayer {i}: GPU {device}) else: print(fLayer {i}: {device})3.2 KV Cache推理加速的核心也是显存杀手KV Cache的原理不复杂自回归生成时每个新token的注意力计算需要前面所有token的Key和Value。如果每次都重算复杂度是O(n²)缓存下来每步只算新token的KV复杂度降到O(n)。但代价是显存——缓存大小 2 × 层数 × 注意力头数 × head_dim × 序列长度 × 精度字节数。以7B模型为例32层、32头、head_dim128、FP162048上下文KV Cache大约是 2×32×32×128×2048×2 ≈ 1GB。看着不大但如果你要并发处理10个请求就是10GB。这就是为什么生产环境里batch size和上下文长度是一对矛盾。手写一遍KV Cache的管理逻辑你会真正理解vLLM的PagedAttention在解决什么问题——内存碎片和动态分配。我建议至少实现一个简化版的KV Cache复用逻辑class SimpleKVCache: def __init__(self, num_layers, max_seq_len, num_heads, head_dim, dtypetorch.float16): self.cache [ { k: torch.zeros(1, num_heads, max_seq_len, head_dim, dtypedtype), v: torch.zeros(1, num_heads, max_seq_len, head_dim, dtypedtype) } for _ in range(num_layers) ] self.seq_len 0 def update(self, layer_idx, new_k, new_v): n new_k.shape[2] self.cache[layer_idx][k][:, :, self.seq_len:self.seq_lenn] new_k self.cache[layer_idx][v][:, :, self.seq_len:self.seq_lenn] new_v if layer_idx len(self.cache) - 1: self.seq_len n这个简化版没有做内存池化但能让你直观感受到“缓存”到底缓存了什么、什么时候会爆。3.3 批处理与流式输出用户体验的关键单条推理跑通之后下一步是批处理。批处理的核心是padding对齐——不同长度的输入需要pad到同一长度才能并行计算。但pad会产生无效计算所以更优的方案是连续批处理continuous batching也就是vLLM和TGI采用的策略不等所有请求到齐而是动态地把新请求插入到正在运行的batch里。手写连续批处理比较复杂但你可以先实现一个简单的动态batchimport asyncio from queue import Queue class BatchScheduler: def __init__(self, max_batch_size8, max_wait_ms50): self.queue Queue() self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms async def add_request(self, prompt): self.queue.put(prompt) # 等待batch凑齐或超时 await asyncio.sleep(self.max_wait_ms / 1000) batch [] while not self.queue.empty() and len(batch) self.max_batch_size: batch.append(self.queue.get()) return batch流式输出则是另一个体验关键点。用户不想等模型生成完才看到结果而是希望token一个个蹦出来。这需要把生成循环拆开每生成一个token就yield一次def stream_generate(model, tokenizer, prompt, max_new_tokens256): inputs tokenizer(prompt, return_tensorspt).to(model.device) generated inputs[input_ids] for _ in range(max_new_tokens): outputs model(generated) next_token outputs.logits[:, -1, :].argmax(dim-1, keepdimTrue) generated torch.cat([generated, next_token], dim-1) yield tokenizer.decode(next_token[0], skip_special_tokensTrue) if next_token.item() tokenizer.eos_token_id: break这段代码效率不高每步都重算但逻辑清晰。理解了它你再看vLLM的流式实现就会觉得理所当然。4. 向量检索RAG的命门手写一遍才懂ANN的取舍4.1 从暴力检索到近似检索什么时候该上ANN向量检索是RAG系统的核心组件。最朴素的实现是暴力检索——把查询向量和库里所有向量算一遍余弦相似度取Top-K。这在几千条数据时完全够用延迟也就几十毫秒。但数据量上到十万、百万级别暴力检索就扛不住了。这时候需要近似最近邻ANN算法。常见的几种算法核心思想优点缺点IVF聚类分桶只搜最近几个桶实现简单内存省召回率依赖聚类质量HNSW多层图结构逐层逼近召回率高速度快内存占用大PQ向量分段量化内存极省精度损失明显我一般先用暴力检索跑通全流程等数据量上来再换HNSW。不要一上来就上最复杂的方案这是我从多个项目里总结的血泪教训。过早优化检索层结果发现瓶颈其实在embedding质量上白折腾。4.2 手写一个简易HNSW理解图检索的直觉HNSW的核心是构建一个多层图底层包含所有节点上层是下层的子集。检索时从顶层开始贪心地向最近邻移动逐层下降。手写一个简化版能帮你理解“为什么HNSW快”import numpy as np import heapq class SimpleHNSW: def __init__(self, dim, M16): self.dim dim self.M M self.layers [[]] # 第0层包含所有节点 self.vectors [] def insert(self, vec): idx len(self.vectors) self.vectors.append(vec) self.layers[0].append(idx) # 简化随机决定是否上浮到更高层 level 0 while np.random.random() 0.5 and level 3: level 1 if len(self.layers) level: self.layers.append([]) self.layers[level].append(idx) def search(self, query, k5, ef50): # 从最高层开始贪心下降 entry self.layers[-1][0] if self.layers[-1] else 0 for level in range(len(self.layers)-1, -1, -1): entry self._greedy_search(query, entry, level) # 底层做ef宽度的搜索 candidates self._search_layer(query, entry, ef, 0) return heapq.nsmallest(k, candidates, keylambda x: x[0]) def _greedy_search(self, query, entry, level): current entry current_dist np.linalg.norm(query - self.vectors[current]) improved True while improved: improved False for neighbor in self._get_neighbors(current, level): d np.linalg.norm(query - self.vectors[neighbor]) if d current_dist: current, current_dist neighbor, d improved True return current def _get_neighbors(self, idx, level): # 简化返回同层随机M个邻居 layer self.layers[level] pos layer.index(idx) start max(0, pos - self.M // 2) return layer[start:start self.M] def _search_layer(self, query, entry, ef, level): visited {entry} candidates [(np.linalg.norm(query - self.vectors[entry]), entry)] results list(candidates) while candidates: dist, idx heapq.heappop(candidates) if dist results[-1][0] and len(results) ef: break for neighbor in self._get_neighbors(idx, level): if neighbor not in visited: visited.add(neighbor) d np.linalg.norm(query - self.vectors[neighbor]) if len(results) ef or d results[-1][0]: heapq.heappush(candidates, (d, neighbor)) results.append((d, neighbor)) results heapq.nsmallest(ef, results, keylambda x: x[0]) return results这个实现省略了很多工程细节比如邻居选择的启发式、删除支持但它把HNSW的核心逻辑——多层贪心下降 底层宽度搜索——完整呈现出来了。跑一遍这个代码你对“为什么HNSW能在对数时间内找到近似最近邻”会有肌肉记忆般的理解。4.3 检索质量调优embedding模型比索引结构更重要我见过太多团队在索引结构上反复折腾却忽略了embedding模型的选择。实际上检索质量的天花板由embedding模型决定索引结构只影响速度和召回率的权衡。一个差的embedding模型配再好的HNSW也搜不出相关结果。选embedding模型时我一般看三个维度领域匹配度、维度、推理成本。通用场景下BGE系列和GTE系列都不错中文场景优先选在中文语料上训练过的模型如果领域特殊比如法律、医疗最好用领域数据做一版微调。维度方面768维和1024维在实际效果上差异不大但1024维的存储和计算成本高不少我一般优先选768维。注意embedding模型换版本时必须重新索引全库。我踩过一次坑只对新数据用了新模型旧数据还是旧模型的向量结果检索时新旧向量空间不对齐召回率直接崩了。这个坑排查起来很隐蔽因为系统不报错只是效果变差。5. 服务编排与评测把零件组装成能扛住的系统5.1 用FastAPI把推理和检索串起来前面几层都是零件服务编排是把它们组装成产品的环节。我一般用FastAPI做服务层因为它异步支持好、生态成熟、调试方便。一个典型的RAG服务接口长这样from fastapi import FastAPI from pydantic import BaseModel import numpy as np app FastAPI() class QueryRequest(BaseModel): question: str top_k: int 5 class QueryResponse(BaseModel): answer: str sources: list app.post(/query, response_modelQueryResponse) async def query(req: QueryRequest): # 1. 查询向量化 q_vec embed(req.question) # 2. 向量检索 hits vector_store.search(q_vec, kreq.top_k) # 3. 组装上下文 context \n\n.join([h[text] for h in hits]) prompt f基于以下资料回答问题\n{context}\n\n问题{req.question} # 4. 生成回答 answer generate(prompt) return QueryResponse(answeranswer, sources[h[id] for h in hits])这个接口看着简单但有几个工程细节值得注意。第一embedding和生成最好分开部署因为它们的资源需求不同——embedding是计算密集型生成是显存密集型混在一起容易互相拖累。第二检索和生成之间要有超时控制检索慢了不能把生成也拖死。第三返回结果要带来源方便排查和用户验证。5.2 评测没有评测的AI系统就是盲盒AI系统最怕的是“感觉效果还行”。我坚持每个项目上线前必须有一套自动化评测。评测分两层检索层评测和生成层评测。检索层看RecallK和MRR。构造方式是从真实问题出发人工标注每个问题对应的正确文档块然后看检索结果里有没有命中。生成层看忠实度faithfulness和相关性relevance。忠实度是回答是否基于检索到的资料相关性是回答是否切题。这两个指标可以用另一个LLM来打分但要注意打分模型和被评模型不能是同一个否则会有自我偏好偏差。def evaluate_retrieval(questions, ground_truth, retriever, k5): recalls [] mrrs [] for q, gt in zip(questions, ground_truth): hits retriever.search(q, kk) hit_ids [h[id] for h in hits] # RecallK recall len(set(hit_ids) set(gt)) / len(gt) recalls.append(recall) # MRR for rank, hid in enumerate(hit_ids): if hid in gt: mrrs.append(1 / (rank 1)) break else: mrrs.append(0) return {recallk: np.mean(recalls), mrr: np.mean(mrrs)}这套评测跑下来你才能说“系统效果如何”而不是“我觉得还行”。评测集要持续维护每次发现bad case就补进去慢慢就积累成一套有价值的回归测试。5.3 监控与迭代上线只是开始AI系统上线后监控比传统软件更重要因为它的失败模式更隐蔽。我一般监控这几个指标延迟P50/P99、检索命中率、生成拒绝率、用户反馈。延迟突增可能是检索库膨胀或生成排队检索命中率下降可能是数据分布漂移生成拒绝率上升可能是prompt需要调整。迭代节奏上我倾向于小步快跑每周做一次bad case复盘挑出top 3问题针对性优化。优化方向无非是数据、检索、prompt、模型四选一。不要同时改多个变量否则你无法归因。我吃过这个亏一次上线同时换了embedding模型和prompt模板结果效果变差了排查了两天才定位到是prompt的问题。6. 手搓一遍之后我对AI工程的理解变了走完这一整条从零构建的路径最大的收获不是“我会手写HNSW了”或者“我能自己实现KV Cache了”而是对每个环节的边界和取舍有了直觉。知道什么时候该用现成框架、什么时候该自己写知道一个参数改动会影响链路的哪一环知道效果不好时该往哪个方向排查。这种直觉是调包调不出来的。你可以用LangChain三天搭一个RAG demo但当你遇到“检索结果明明相关、生成却答非所问”的问题时只有理解embedding空间、注意力机制、prompt构造这三者之间的关系才能快速定位。AI工程的竞争力不在于你会用多少工具而在于你能在工具失效时自己造一个。最后分享一个我自己的习惯每学一个新组件我都会先用最朴素的方式实现一个能跑的最小版本然后再去看成熟框架是怎么做的。对比之下框架里那些“看起来多余”的设计——vLLM的PagedAttention、HNSW的启发式邻居选择、连续批处理的调度策略——都会变得理所当然。这个习惯让我在技术选型时少了很多盲目也让我在面试和方案评审时能说出“为什么”而不只是“是什么”。