ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

从零手搓AI工程:分词、注意力、RAG与Agent全链路实战

2026/10/4 12:42:32 拓冰建站 浏览量
从零手搓AI工程:分词、注意力、RAG与Agent全链路实战 1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人LangChain、LlamaIndex、各种Agent框架轮番上阵好像只要会写几行pip install就能做出一个智能应用。但我自己带过几轮团队、也面过不少号称“做过大模型项目”的候选人之后发现一个很尴尬的现实很多人能跑通Demo却说不清楚一次对话请求背后到底发生了什么token是怎么被切分的向量检索为什么召回不准上下文超了之后模型为什么会“失忆”。一旦线上出问题除了重启服务基本束手无策。这就是我特别想聊ai-engineering-from-scratch这个方向的原因。它不是一个具体的库或者框架而是一种学习路径和工程态度——把AI工程里那些被封装起来的核心环节自己动手实现一遍。从最基础的文本分词、词嵌入到注意力机制的矩阵运算再到检索增强生成RAG的完整链路、Agent的工具调用循环最后到推理服务的性能优化。这套东西听起来硬核但它解决的恰恰是“只会调API、不懂原理、无法排障”这个最要命的问题。这篇文章适合三类人一是刚转行做AI应用、被各种框架绕晕的开发者二是有后端或算法基础、想系统补齐AI工程链路的工程师三是想带团队做技术选型、需要判断“哪些轮子该自己造、哪些可以直接用”的技术负责人。我会按照从零构建的思路把每个核心模块拆开讲清楚为什么这么设计、关键参数怎么算、实操中会踩哪些坑。全程不堆砌术语尽量用生活化的类比把原理讲透代码和配置都给到能直接抄的程度。2. 整体学习路径与工程架构设计2.1 为什么选择“自底向上”而不是“自顶向下”市面上大部分AI教程是自顶向下的先教你用现成框架搭一个聊天机器人跑通了再往下讲原理。这种方式上手快但有个致命问题——你学到的其实是“框架的使用方法”而不是“AI工程本身”。框架一升级你的知识就过期了框架没覆盖的场景你就卡住了。ai-engineering-from-scratch走的是相反的路自底向上。先用纯Python和NumPy实现一个最小的分词器再实现词嵌入和注意力然后拼出一个能跑的小型Transformer接着在这个基础上加检索、加工具调用、加服务化。每一步你都知道数据长什么样、张量形状怎么变、显存花在哪里。等你自己手搓过一遍之后再回头看LangChain那些封装你会发现它们不过是把你写过的东西包装得更通用而已用起来心里有底出问题也知道去哪找。我个人的经验是自底向上学一遍大概需要三到四周的业余时间但带来的收益是长期的你排查问题的速度会快一个数量级做技术选型时也不会被营销话术带偏。2.2 分层架构把AI工程拆成五层为了不让学习过程变成一团乱麻我习惯把AI工程拆成五个层次从下往上依次是层级核心内容典型产出是否建议手搓数据与分词层文本清洗、分词、词表构建、token化一个能处理中文的分词器强烈建议表示与模型层词嵌入、注意力、Transformer块一个能训练的小模型建议手搓核心检索与增强层向量化、索引、召回、重排一个RAG检索链路建议手搓编排与Agent层工具调用、循环控制、状态管理一个能调工具的Agent建议手搓服务与优化层推理服务、批处理、缓存、量化一个可压测的API部分手搓这个分层的好处是每一层都可以独立验证。你不需要等整个系统搭完才知道对不对每完成一层就能写测试、看输出。而且层与层之间的接口是清晰的比如检索层只关心“给我一段文本还我一组相关片段”不关心上层是聊天还是问答。2.3 技术选型的取舍逻辑既然是“from scratch”选型上就要克制。我的建议是语言Python。生态最全NumPy、PyTorch都能用调试也方便。不要一上来就追求C或Rust的性能那是优化阶段的事。数值计算前期用NumPy手写理解矩阵运算后期用PyTorch享受自动求导和GPU加速。两者不要混着学先NumPy后PyTorch。模型规模自己训练时用tiny级别几百万参数目的是验证流程不是追求效果。真正要效果时再加载预训练权重。存储向量索引用FAISS或hnswlib别自己写近似最近邻那是另一个专业领域。但要知道它们大概怎么工作。服务FastAPI起步够用且轻量。等有性能瓶颈了再考虑更重的方案。提示手搓的目的是理解原理和建立排障能力不是重复造所有轮子。像FAISS这种高度专业的库用现成的完全没问题但你要能说清楚它内部大概在做什么。3. 核心模块拆解与实操要点3.1 分词器AI工程的第一道门槛很多人觉得分词是个小问题直到他们遇到中文。英文按空格切就行中文你得考虑词边界。我见过太多项目在分词这一步埋雷训练时用一种分词方式推理时用另一种结果模型表现莫名其妙地差。自己实现一个分词器核心是搞清楚三种主流方案的区别BPE字节对编码从字符开始不断合并出现频率最高的相邻对。GPT系列用的就是这套。优点是能处理未登录词词表大小可控。WordPiece和BPE类似但合并时看的是似然增益而不是频率。BERT用的这套。SentencePiece把空格也当成普通字符处理对中文、日文这种没有天然分隔的语言特别友好。实操上我建议先用BPE手写一遍。核心逻辑不复杂统计所有相邻字符对的频率合并最高频的那对重复直到达到目标词表大小。写完之后你会对“为什么有些词会被切成奇怪的片段”有直观感受。# 极简BPE核心逻辑示意 from collections import Counter def get_stats(corpus): pairs Counter() for word, freq in corpus.items(): symbols word.split() for i in range(len(symbols) - 1): pairs[symbols[i], symbols[i1]] freq return pairs def merge_vocab(pair, corpus): new_corpus {} bigram .join(pair) replacement .join(pair) for word in corpus: new_word word.replace(bigram, replacement) new_corpus[new_word] corpus[word] return new_corpus这段代码跑一遍你就能看到词表是怎么从单个字符逐步合并成词的。注意几个坑一是要处理特殊token如unk、pad二是要固定随机种子保证可复现三是中文要先做字符级切分再合并。3.2 注意力机制把矩阵形状搞清楚就成功了一半注意力机制是Transformer的核心也是最多人卡住的地方。卡住的原因往往不是数学难而是张量形状对不上。我教别人的时候第一件事就是让他们把每个张量的形状写在纸上。缩放点积注意力的公式是Attention(Q,K,V) softmax(QK^T / sqrt(d_k)) V。假设batch size是B序列长度是L每个头的维度是d_k那么Q、K、V的形状都是(B, L, d_k)QK^T的形状是(B, L, L)这就是注意力分数矩阵除以sqrt(d_k)是为了防止点积过大导致softmax梯度消失softmax之后还是(B, L, L)每一行加起来等于1最后乘V得到(B, L, d_k)为什么要除以sqrt(d_k)因为当d_k很大时点积的方差会随维度线性增长数值太大会让softmax输出接近one-hot梯度几乎为零。除以sqrt(d_k)相当于把方差拉回1附近训练才稳定。这个细节自己实现一遍就再也不会忘。多头注意力就是把d_k拆成h个头并行做最后拼接。手写的时候建议先写单头验证形状对了再加多头。位置编码也别跳过正弦位置编码虽然现在用得少了但理解它有助于理解为什么后来会有RoPE这类相对位置编码。3.3 RAG检索链路召回不准的锅往往不在模型RAG检索增强生成是当前落地最多的AI应用形态但也是坑最多的。很多人以为RAG就是“把文档切块、向量化、存库、检索、拼进prompt”跑起来发现答非所问然后开始怀疑模型不行。实际上问题八成出在检索环节。自己搭一遍RAG你会被迫面对这些决策切块策略按固定长度切还是按语义切重叠多少我实测下来中文文档按300-500字切、重叠50-100字比较稳。切太碎会丢上下文切太大检索精度下降。向量模型选择中文场景下通用多语言模型往往不如专门的中文模型。但要注意向量模型和生成模型是两回事别混为一谈。相似度度量余弦相似度最常用但内积在向量归一化后等价。欧氏距离在高维空间会失效慎用。重排召回top-20再用交叉编码器重排到top-5效果通常比直接召回top-5好很多。这一步很多人省了是效果差的主因之一。# 检索链路的核心步骤示意 def retrieve(query, index, top_k20, rerank_k5): q_vec embed(query) candidates index.search(q_vec, top_k) # 粗召回 scored [(doc, cross_encoder(query, doc)) for doc in candidates] scored.sort(keylambda x: x[1], reverseTrue) return [doc for doc, _ in scored[:rerank_k]] # 精排注意粗召回和精排用的模型通常不是一个。粗召回要快精排要准。别用生成模型去做精排那是杀鸡用牛刀且效果未必好。3.4 Agent工具调用循环控制比工具本身更重要Agent的本质是一个“思考-行动-观察”的循环。模型根据当前状态决定调用哪个工具拿到结果后继续思考直到任务完成或达到最大轮数。听起来简单但自己实现一遍会发现难点全在循环控制上。几个必须处理的边界情况最大轮数限制不设上限模型可能陷入死循环烧钱又耗时。一般设5-10轮。工具调用失败工具报错时要把错误信息作为观察结果喂回模型让它决定重试还是换工具。参数校验模型生成的工具参数经常格式不对要在调用前做校验和修正。状态管理多轮对话中历史消息会越来越长需要做截断或摘要。我自己的做法是先实现一个只有两个工具的Agent比如计算器和搜索把循环跑通再逐步加工具。工具的描述要写得非常清楚包括参数类型、取值范围、返回格式模型才能正确调用。4. 完整实操流程从空目录到一个可用的RAG服务4.1 环境准备与依赖安装先把环境搭起来。我习惯用conda建独立环境避免污染系统Python。conda create -n ai-scratch python3.10 conda activate ai-scratch pip install numpy torch transformers fastapi uvicorn faiss-cpu sentencepiece这里解释下每个依赖的用途numpy做数值计算torch做模型训练和推理transformers加载预训练权重fastapi和uvicorn做服务faiss-cpu做向量检索sentencepiece做分词。如果要用GPU把faiss-cpu换成faiss-gputorch装对应CUDA版本。提示不要一上来就装一堆框架。先把这几个核心依赖跑通后面需要什么再加。依赖越多版本冲突越难排查。4.2 数据准备与分词器训练准备一份中文语料几MB就够练手。我用的是公开的新闻语料清洗掉HTML标签和特殊符号后训练一个BPE分词器。import sentencepiece as spm spm.SentencePieceTrainer.train( inputcorpus.txt, model_prefixtokenizer, vocab_size8000, model_typebpe, character_coverage0.9995, pad_id0, unk_id1, bos_id2, eos_id3 )参数说明vocab_size是词表大小中文一般8000-32000character_coverage是字符覆盖率中文建议0.9995以上否则生僻字会变成unkmodel_type选bpe或unigram都行中文场景unigram有时更稳。训练完用sp.encode()和sp.decode()验证一下看看中文句子切出来是不是合理。如果发现大量单字说明词表太小或语料不够。4.3 向量化与索引构建把文档切块后用向量模型编码存进FAISS索引。import faiss import numpy as np def build_index(chunks, embed_model): vectors np.array([embed_model.encode(c) for c in chunks]).astype(float32) faiss.normalize_L2(vectors) # 归一化后用内积等价余弦 index faiss.IndexFlatIP(vectors.shape[1]) index.add(vectors) return index这里用IndexFlatIP做精确检索数据量小的时候够用。数据量大了换成IndexIVFFlat但要先训练索引。归一化这步别省否则内积和余弦对不上。4.4 服务化与接口设计用FastAPI包一层暴露两个接口一个做检索一个做问答。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Query(BaseModel): question: str top_k: int 5 app.post(/retrieve) def retrieve(q: Query): docs retrieve_pipeline(q.question, q.top_k) return {docs: docs} app.post(/chat) def chat(q: Query): docs retrieve_pipeline(q.question, q.top_k) answer generate(q.question, docs) return {answer: answer, sources: docs}接口设计上检索和生成分开方便单独测试和替换。返回结果带上来源方便排查和展示引用。4.5 压测与性能观察服务跑起来后用简单的脚本压一下看QPS和延迟。ab -n 100 -c 10 -p query.json -T application/json http://localhost:8000/retrieve重点关注P99延迟而不是平均值。平均值好看但P99爆炸的情况太常见了。如果延迟高先看是检索慢还是生成慢再针对性优化。检索慢就换索引类型生成慢就上批处理或量化。5. 常见问题与排查技巧实录5.1 模型输出乱码或重复这是新手最常遇到的问题。原因通常有三个一是分词器和模型不匹配比如用A分词器训练却用B分词器推理二是解码策略有问题贪心解码容易重复可以加重复惩罚或换束搜索三是输入里有特殊字符没清洗干净。排查顺序先打印token id看是不是预期的那串再换解码策略试最后检查输入清洗。5.2 检索召回不相关先别怪向量模型。按这个顺序查切块是不是太碎或太大查询和文档是不是用了同一个向量模型相似度度量是不是选错了有没有做归一化。我遇到过最离谱的一次是查询用了归一化向量文档没归一化结果内积算出来全是乱的。5.3 显存溢出自己训练小模型时也会遇到。先算一下参数量乘以4字节float32就是模型大小再加上优化器状态通常是参数的2-3倍和激活值。如果超了减小batch size、用梯度累积、或者上混合精度。别一上来就买大显卡先把batch size调到1试试能不能跑。5.4 服务并发上不去FastAPI默认是单进程的并发高了就排队。用uvicorn多worker启动uvicorn main:app --workers 4。但要注意如果模型加载在进程内每个worker都会加载一份显存会翻倍。这时候要么用模型服务单独部署要么用共享内存。问题现象可能原因排查动作输出乱码分词器不匹配打印token id对比输出重复解码策略问题加重复惩罚召回不准切块或归一化问题检查切块粒度和向量归一化显存溢出batch过大或精度过高减小batch上混合精度并发低单worker或模型重复加载多worker或模型服务化5.5 几个独家避坑心得第一永远先跑通最小闭环。别一上来就搞多路召回、多模型融合先用最简单的方案跑通再逐步加复杂度。第二每个模块都要有独立的测试别等整个系统搭完才测。第三日志要打全尤其是输入输出和中间结果出问题时能省大量时间。第四版本要锁死requirements.txt里写清楚版本号不然过两个月环境就复现不了了。6. 从能跑到好用性能优化与工程化收尾6.1 推理加速的几条实用路径模型能跑之后下一步是让它跑得快。按投入产出比排序我推荐这几条批处理把多个请求攒一批一起推理GPU利用率能翻几倍。但要注意延迟和吞吐的权衡攒批会增大单请求延迟。量化把float32降到int8显存减半速度提升明显精度损失通常可接受。用bitsandbytes或torch的量化工具。KV缓存自回归生成时缓存已计算的key和value避免重复计算。这是生成加速的标配。模型蒸馏用大模型教小模型推理时用小模型。适合对延迟敏感的场景。6.2 缓存策略省钱又提速很多查询是重复的尤其是FAQ类场景。加一层语义缓存相似问题直接返回缓存结果能省大量推理成本。实现上用向量检索做相似度匹配超过阈值就命中缓存。def semantic_cache(query, cache_index, threshold0.95): q_vec embed(query) score, idx cache_index.search(q_vec, 1) if score[0][0] threshold: return cache_store[idx[0][0]] return None阈值别设太低否则会返回不相关的缓存结果体验反而更差。我一般从0.95开始调。6.3 监控与可观测性线上服务没有监控就是裸奔。至少要采集这几个指标请求量、延迟分布、错误率、token消耗、缓存命中率。用Prometheus加Grafana是常见组合轻量场景用日志加简单统计也行。关键是出问题时你要能快速定位是哪个环节慢、哪个环节错。所以每个环节都要打点检索耗时、生成耗时、总耗时分开记。6.4 后续可以怎么扩展这套从零搭建的框架跑通之后扩展方向很多。往深了走可以研究更高效的注意力实现如FlashAttention、更先进的检索算法如多向量检索、更复杂的Agent规划如树搜索。往广了走可以接入多模态、加语音交互、做多租户隔离。我个人在实际操作中的体会是ai-engineering-from-scratch最大的价值不是让你造出一个比现成框架更好的轮子而是让你在遇到问题时脑子里有一张清晰的地图知道每个环节大概在做什么、可能哪里出问题、该往哪个方向查。这种底气是调包调不出来的。最后再分享一个小技巧每学完一个模块试着用三句话给一个不懂技术的人讲清楚它在干嘛讲不清楚就说明还没真懂。