1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人一个RAG问答机器人从向量库到编排框架拼拼凑凑半天就能跑起来。但我发现一个很尴尬的现象很多人能把LangChain的链式调用写得飞起却说不清楚一次embedding到底把文本变成了什么形状的张量能熟练调用推理接口却不知道KV Cache是怎么把显存吃掉的。这就是“ai-engineering-from-scratch”这个方向真正要解决的问题——不是教你怎么用现成的轮子而是逼着你把轮子拆开看清楚每一根辐条是怎么装上去的。我自己带过几个刚入行的同学最典型的问题就是“知其然不知其所以然”。模型输出不对第一反应是换prompt换完还不行就换模型最后实在没辙了才去翻文档。这种调试方式在简单场景下能蒙混过关一旦遇到需要控制延迟、控制成本、控制输出稳定性的生产环境立刻就抓瞎。所以这篇内容我想聊的是如何从零开始用最朴素的方式把AI工程里几个核心环节亲手实现一遍包括但不限于文本怎么变成向量、注意力机制到底在算什么、推理服务怎么组织、检索增强的每一环怎么串起来。适合谁来读如果你已经会用现成的框架做demo但总觉得心里没底想搞清楚底层发生了什么那这篇就是写给你的。如果你是完全的新手也没关系我会尽量用生活化的类比把原理讲透再给可以直接跑的代码。整篇内容不依赖任何特定平台的教程纯粹是从工程实践角度出发的经验总结你可以把它当成一份“AI工程自学路线图”来用。2. 整体设计思路为什么我选择“先造轮子再开车”2.1 从调用者到实现者的思维转变大部分人学AI工程是从“调用”开始的调一个API拿到结果调一个库完成检索调一个框架搭起服务。这种路径上手快但天花板也低。因为你所有的认知都建立在“别人已经封装好的抽象”之上一旦抽象泄漏——比如某个库的默认参数不适合你的场景——你就束手无策了。“from scratch”的核心思路是反过来的先不管有没有现成的库假设什么都没有你要怎么把这件事做出来。这个过程中你会被迫面对很多被封装隐藏掉的细节。举个最简单的例子文本相似度计算。用现成库就是一行cosine_similarity(a, b)但如果你自己实现就会遇到几个问题向量要不要归一化归一化之后余弦相似度和点积是不是等价的为什么归一化能加速计算这些问题想清楚了你对整个检索链路的理解就上了一个台阶。我自己的经验是每亲手实现一个核心组件你对整个系统的掌控力就强一分。这种掌控力在排查线上问题时尤其值钱。当检索结果不理想时你能快速定位是embedding的问题、索引的问题还是排序的问题而不是对着黑盒干瞪眼。2.2 技术选型为什么用Python而不是其他语言这个方向我强烈建议用Python来做。原因很实在AI工程生态里绝大多数参考实现、论文代码、工具库都是Python写的你用Python能直接对照着看省去大量翻译成本。而且Python的交互式环境特别适合做这种“边写边验证”的探索性工作你可以在REPL里一行行试看到中间结果这对理解原理帮助极大。具体到库的选择我的原则是能用标准库就不用第三方库能用轻量库就不用重型框架。比如数值计算NumPy是底线但不要一上来就用PyTorch的高级API先用NumPy把矩阵乘法、softmax这些手写一遍。等你对手写版本的计算量和内存占用有了体感再去用框架的优化版本你才会真正 appreciate 那些优化有多值钱。2.3 分阶段推进从单点突破到系统集成整个学习路径我建议分成三个阶段。第一阶段是单点原理实现把tokenization、embedding、attention、采样这几个核心环节各自用最小代码实现一遍不追求性能只追求逻辑正确。第二阶段是链路串联把上面这些单点串成一个能跑通的最小推理流程比如一个简单的文本生成pipeline。第三阶段是工程化改造加入批处理、缓存、并发控制这些生产环境必须的东西。这个顺序很重要。很多人一上来就想搭一个完整的RAG系统结果每个环节都是调包出了问题根本不知道从哪查。反过来如果你先把每个环节都手写过再组装的时候心里就有张清晰的地图知道数据在每一步长什么样、该是什么量级。3. 核心细节解析几个必须亲手实现的关键环节3.1 Tokenization文本到数字的第一道关卡Tokenization看起来简单实际上坑非常多。最朴素的实现是按空格切分但中文怎么办标点符号怎么处理子词怎么切这些问题在调用现成tokenizer的时候都被隐藏了但你自己实现一遍就会明白为什么BPEByte Pair Encoding会成为主流。我建议的第一个练习是实现一个最简单的字符级tokenizer。就是把每个字符映射成一个ID反过来也能映射回去。这个实现虽然简陋但能让你直观感受到几个关键问题词表大小怎么定未登录字符怎么处理特殊token比如padding、起始、结束怎么加class CharTokenizer: def __init__(self, text): chars sorted(set(text)) self.vocab {ch: i for i, ch in enumerate(chars)} self.inverse_vocab {i: ch for ch, i in self.vocab.items()} self.vocab_size len(self.vocab) def encode(self, text): return [self.vocab.get(ch, self.vocab.get(unk, 0)) for ch in text] def decode(self, ids): return .join(self.inverse_vocab.get(i, unk) for i in ids)这段代码跑起来之后你可以拿一段中英文混合的文本试试看看词表有多大编码解码是否可逆。然后你再去看GPT用的BPE tokenizer就会明白它为什么要做子词切分——因为字符级词表太大词级词表又覆盖不了新词子词是折中方案。注意自己实现tokenizer时一定要保证encode和decode是可逆的这是很多下游任务的基本假设。我见过有人写的tokenizer把连续空格合并了导致解码后文本和原文对不上排查了半天才发现是这里的问题。3.2 Embedding把离散符号变成连续向量Tokenization之后每个token是一个整数ID。但神经网络处理不了整数它需要连续的向量。Embedding层的作用就是维护一个查找表把每个ID映射成一个固定维度的向量。自己实现一个Embedding层核心就是初始化一个形状为(vocab_size, embed_dim)的矩阵然后前向传播就是按ID取行。听起来简单但有几个细节值得琢磨。第一初始化方式。用全零初始化会导致所有token的梯度更新完全一样模型学不到东西。用标准正态分布初始化是常见做法但方差要控制好太大容易梯度爆炸太小又学不动。第二embed_dim怎么选。太小表达能力不够太大参数爆炸且容易过拟合。实践中一般从128或256起步根据任务复杂度调整。import numpy as np class Embedding: def __init__(self, vocab_size, embed_dim): self.weight np.random.randn(vocab_size, embed_dim) * 0.01 def forward(self, token_ids): return self.weight[token_ids] def backward(self, token_ids, grad_output, lr0.01): for idx, token_id in enumerate(token_ids): self.weight[token_id] - lr * grad_output[idx]这个实现虽然简陋但它揭示了embedding的本质它就是一个可学习的查找表。理解了这一点你就能明白为什么embedding层往往占模型参数量的很大一部分词表大也能明白为什么微调embedding模型能显著提升特定领域的检索效果。3.3 Attention机制从公式到代码的完整推导Attention是Transformer的核心也是很多人觉得最难理解的部分。我试过很多种讲法最有效的是从“加权平均”这个角度切入。想象你有一个查询Query你手里有一堆键值对Key-Value你想根据Query和每个Key的相似度对Value做加权平均。这就是attention的直觉。自己实现的时候关键步骤是计算Q和K的点积除以根号d防止梯度消失做softmax归一化然后加权求和V。每一步都有存在的理由我建议你在实现的时候把中间结果都打印出来看看。def scaled_dot_product_attention(Q, K, V, maskNone): d_k Q.shape[-1] scores np.matmul(Q, K.transpose(0, 1, -1, -2)) / np.sqrt(d_k) if mask is not None: scores np.where(mask 0, -1e9, scores) attention_weights softmax(scores, axis-1) output np.matmul(attention_weights, V) return output, attention_weights def softmax(x, axis-1): x_max np.max(x, axisaxis, keepdimsTrue) exp_x np.exp(x - x_max) return exp_x / np.sum(exp_x, axisaxis, keepdimsTrue)跑通这段代码之后你可以做几个实验来加深理解。第一把mask去掉看看padding token是怎么影响结果的。第二把缩放因子去掉观察softmax输出的分布变化。第三把Q和K换成相同的矩阵看看会发生什么这其实就是self-attention的一种特殊情况。这些实验能让你对attention的数值行为有直观感受。提示实现softmax时一定要减去最大值否则指数运算容易溢出。这个技巧在数值计算里叫“log-sum-exp trick”是工程实现的基本功。3.4 采样策略为什么贪心解码往往不是最优模型输出的是每个位置上的logits也就是未归一化的分数。怎么从这些分数里选出一个token就是采样策略要解决的问题。最朴素的是贪心每次选分数最高的。但贪心解码有个问题它容易陷入重复循环生成“复读机”式的文本。自己实现几种采样策略对比效果是理解生成行为的好方法。Temperature scaling是调整分布的平滑程度温度趋近0就退化成贪心温度大于1分布更平坦、更随机。Top-k是只保留分数最高的k个token其余置零。Top-p也叫nucleus sampling是保留累积概率达到p的最小token集合。def sample_with_temperature(logits, temperature1.0): logits logits / temperature probs softmax(logits) return np.random.choice(len(probs), pprobs) def sample_top_k(logits, k10): top_k_indices np.argsort(logits)[-k:] top_k_logits logits[top_k_indices] probs softmax(top_k_logits) chosen np.random.choice(len(probs), pprobs) return top_k_indices[chosen] def sample_top_p(logits, p0.9): sorted_indices np.argsort(logits)[::-1] sorted_logits logits[sorted_indices] probs softmax(sorted_logits) cumulative_probs np.cumsum(probs) cutoff np.searchsorted(cumulative_probs, p) 1 top_p_indices sorted_indices[:cutoff] top_p_probs probs[:cutoff] top_p_probs top_p_probs / np.sum(top_p_probs) return np.random.choice(top_p_indices, ptop_p_probs)实测下来top-p通常比top-k更稳因为它能自适应地调整候选集大小。当分布很尖锐时候选集自动变小当分布很平坦时候选集自动变大。这个特性在生成长文本时特别有用。4. 实操过程从零搭一个最小可用的推理服务4.1 环境准备与依赖管理动手之前先把环境理清楚。我的建议是建一个干净的虚拟环境只装最必要的依赖。NumPy是必须的用来做数值计算。如果要做HTTP服务Flask或FastAPI选一个我个人偏好FastAPI因为它的异步支持和自动文档生成对调试很友好。除此之外暂时不要装任何AI框架我们要的就是“裸写”的感觉。python -m venv ai-from-scratch source ai-from-scratch/bin/activate # Windows用 ai-from-scratch\Scripts\activate pip install numpy fastapi uvicorn依赖管理有个小技巧把版本号固定下来写进requirements.txt。我踩过的坑是本地跑得好好的换台机器因为NumPy版本差异导致矩阵乘法结果有微小不同排查了很久才发现是BLAS后端不一样。固定版本能避免这类问题。4.2 实现一个极简的文本生成流程现在我们把前面实现的组件串起来做一个能跑的文本生成demo。为了不引入模型权重我们用一个随机初始化的“玩具模型”来演示流程。这个模型只有一个embedding层和一个线性输出层虽然生成的内容没有意义但整个数据流和真实模型是一样的。class ToyModel: def __init__(self, vocab_size, embed_dim, hidden_dim): self.embedding Embedding(vocab_size, embed_dim) self.W_out np.random.randn(embed_dim, vocab_size) * 0.01 def forward(self, token_ids): embeds self.embedding.forward(token_ids) # 简单地对序列维度取平均模拟一个“上下文向量” context np.mean(embeds, axis0) logits np.matmul(context, self.W_out) return logits def generate(model, tokenizer, prompt, max_new_tokens20, temperature0.8): token_ids tokenizer.encode(prompt) generated list(token_ids) for _ in range(max_new_tokens): logits model.forward(np.array(generated)) next_token sample_with_temperature(logits, temperature) generated.append(next_token) return tokenizer.decode(generated)跑通这个流程之后你可以把ToyModel替换成真实的模型权重加载逻辑整个pipeline的结构是不变的。这就是“from scratch”的价值你搭的架子是通用的换什么模型都能套进去。4.3 加入检索增强一个最小RAG实现有了生成流程我们再往前一步加入检索增强。RAG的核心思想是在生成之前先从知识库里检索相关文档把文档内容拼接到prompt里让模型基于这些内容生成回答。自己实现RAG关键环节有三个文档向量化、相似度检索、结果拼接。文档向量化可以用我们前面实现的embedding层也可以用现成的sentence-transformers。相似度检索就是计算查询向量和所有文档向量的余弦相似度取top-k。结果拼接要注意token预算不能把检索到的内容全塞进去要按相关性排序后截断。def retrieve(query, documents, doc_vectors, query_vector, top_k3): similarities np.dot(doc_vectors, query_vector) / ( np.linalg.norm(doc_vectors, axis1) * np.linalg.norm(query_vector) 1e-8 ) top_indices np.argsort(similarities)[-top_k:][::-1] return [(documents[i], similarities[i]) for i in top_indices] def build_prompt(query, retrieved_docs, max_context_tokens500): context for doc, score in retrieved_docs: if len(context) len(doc) max_context_tokens: break context doc \n return f基于以下信息回答问题\n{context}\n问题{query}\n回答这个实现虽然简单但已经包含了RAG的所有核心要素。你可以在此基础上逐步优化用更好的embedding模型、加入重排序、做查询改写、处理多轮对话。每一步优化你都能清楚地知道它影响了链路的哪个环节。5. 常见问题与排查技巧实录5.1 数值不稳定NaN和Inf的排查思路自己实现数值计算最常见的问题就是NaN和Inf。我整理了一个排查顺序基本能覆盖90%的情况。现象可能原因排查方法前向传播出现NaN除零、log(0)、exp溢出检查分母是否加了epsilonsoftmax是否减了最大值梯度出现Inf学习率过大、梯度爆炸打印梯度范数加入梯度裁剪损失不下降初始化不当、学习率过小检查初始化方差尝试不同学习率输出全相同权重全零初始化、死区检查初始化加入随机性我踩过最坑的一次是在实现softmax时忘了减最大值结果在logits数值较大时exp直接溢出整个训练崩掉。后来养成习惯任何涉及exp、log、除法的操作都先想想数值范围。5.2 维度不匹配矩阵乘法的对齐检查矩阵乘法要求前一个矩阵的列数等于后一个矩阵的行数。自己实现时维度错误非常常见。我的经验是在每次矩阵乘法之前把两个矩阵的shape打印出来确认对齐。另外批量维度batch dimension很容易搞混建议统一约定batch维度永远在最前面。def safe_matmul(A, B): assert A.shape[-1] B.shape[-2], f维度不匹配: {A.shape} x {B.shape} return np.matmul(A, B)这个简单的断言能帮你省下大量调试时间。等你对维度关系足够熟悉之后可以去掉断言但在学习阶段它是你的安全网。5.3 性能瓶颈从O(n²)到可接受自己实现的attention是O(n²)复杂度的序列长度一上去就慢得没法用。这不是你实现错了而是attention本身的特性。真实模型通过FlashAttention、稀疏attention等技术来优化但那些优化很复杂。在学习阶段我建议你先接受O(n²)把逻辑跑通然后再去了解优化技术。一个实用的技巧是用小序列长度做功能验证用大序列长度做性能测试。功能验证时序列长度设为8或16就够了性能测试时再逐步加大观察耗时增长曲线。这样你能直观感受到O(n²)的含义。注意不要一上来就优化性能。先让代码正确再让代码快。这个顺序反了的话你会陷入“优化了但不知道对不对”的困境。5.4 常见问题速查表问题快速定位解决方案生成结果重复采样策略太贪心提高temperature改用top-p检索结果不相关embedding质量差换更好的embedding模型加入重排序服务响应慢没有批处理实现动态批处理加缓存显存不够序列太长或batch太大减小batch用梯度累积输出格式不稳定prompt不够明确加入格式约束用few-shot示例6. 工程化改造从能跑到好用还差什么6.1 批处理与动态批处理单条推理的效率极低因为GPU或CPU的并行能力没被利用起来。批处理是把多条请求打包成一个batch一起计算。但简单的批处理有个问题如果请求是陆续到达的你要等多久才能凑够一个batch等太久延迟高等太短batch小效率低。动态批处理的做法是维护一个请求队列每隔一个很短的时间窗口比如10毫秒就把队列里的请求打包成一个batch送进去。这样既能保证一定的batch大小又能控制延迟。自己实现的话核心是一个带超时机制的队列。import time from collections import deque class DynamicBatcher: def __init__(self, max_batch_size8, max_wait_ms10): self.queue deque() self.max_batch_size max_batch_size self.max_wait_ms max_wait_ms def add_request(self, request): self.queue.append((time.time(), request)) def get_batch(self): if not self.queue: return [] batch [] start_time time.time() while self.queue and len(batch) self.max_batch_size: timestamp, request self.queue[0] if batch and (time.time() - start_time) * 1000 self.max_wait_ms: break batch.append(self.queue.popleft()[1]) return batch这个实现虽然简单但已经能显著提升吞吐量。实测下来在请求密集的场景下动态批处理能把吞吐量提升5到10倍。6.2 缓存策略哪些结果值得缓存AI推理有个特点同样的输入往往得到同样的输出除非采样有随机性。这就给缓存提供了空间。但不是所有结果都值得缓存要考虑缓存命中率和缓存失效成本。我一般会缓存三类东西embedding结果同样的文本embedding不变、检索结果同样的查询在知识库不变时结果不变、高频查询的生成结果如果采样温度设为0。缓存键的设计要注意要把所有影响输出的参数都包含进去比如模型版本、温度、top-p等。import hashlib def make_cache_key(text, model_version, temperature, top_p): content f{text}|{model_version}|{temperature}|{top_p} return hashlib.md5(content.encode()).hexdigest()缓存失效是个容易被忽视的问题。知识库更新了检索缓存要清模型更新了生成缓存要清。我建议给缓存加一个版本号版本号变了整体失效简单可靠。6.3 监控与可观测性生产环境和实验环境最大的区别是生产环境出问题时你不能停下来慢慢调试。所以监控和日志必须提前做好。我一般会记录这几个指标请求延迟P50、P95、P99、吞吐量、错误率、缓存命中率、各环节耗时占比。各环节耗时占比特别有用。如果检索耗时占了80%那优化重点就在检索如果生成耗时占大头那就去优化生成。没有这个数据优化就是盲人摸象。import time from contextlib import contextmanager contextmanager def timer(name, metrics): start time.time() yield elapsed time.time() - start metrics.setdefault(name, []).append(elapsed)这个简单的计时上下文管理器能帮你快速定位瓶颈。我习惯在每个关键环节都套一个跑一段时间后把统计结果打出来看。7. 我在这条路上踩过的几个坑第一个坑是过早优化。刚开始实现attention的时候我觉得O(n²)太慢了花了两天去研究怎么用矩阵分块优化结果功能还没跑通优化了个寂寞。后来想明白了学习阶段最重要的是建立正确的心理模型性能优化是后面的事。第二个坑是忽视数值精度。用float32和float16跑同样的代码结果可能差很多。特别是在累加操作中float16的精度损失会累积。我现在的习惯是关键计算用float32存储和传输可以用float16但累加前先转回float32。第三个坑是不做单元测试。自己实现的组件一定要写测试。比如softmax的输出和是否为1attention的mask是否生效tokenizer的encode-decode是否可逆。这些测试写起来很快但能帮你省下大量排查时间。第四个坑是盲目追求“完整”。一开始我想把所有组件都实现一遍结果每个都浅尝辄止。后来调整策略一次只深入一个组件把它吃透再换下一个。慢就是快这句话在底层学习上特别成立。8. 后续可以继续深入的方向把上面这些跑通之后你已经对AI工程的核心链路有了扎实的理解。接下来可以往几个方向深入。一是性能优化学习FlashAttention、量化、蒸馏这些技术理解它们分别优化了哪个环节。二是分布式推理当单机扛不住时怎么把模型切分到多张卡上怎么做流水线并行。三是领域适配怎么用LoRA这类方法低成本地让模型适应特定领域。每个方向都够写一篇长文但有了“from scratch”打下的底子你再去学这些高级技术时就不会觉得它们是空中楼阁。你知道它们是在解决什么问题也知道它们牺牲了什么换来了什么。这种判断力才是AI工程师真正的护城河。我个人在实际操作中的体会是手写一遍带来的认知提升比读十篇论文都管用。论文告诉你“是什么”手写让你知道“为什么”和“哪里会出问题”。如果你正在这条路上别急着跳过基础去追新框架把底层的几个环节亲手实现一遍后面的路会好走很多。
