ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

从零手搓AI工程核心模块:RAG检索与上下文管理实战

从零手搓AI工程核心模块:RAG检索与上下文管理实战 1. 从零搭建AI工程能力为什么“手搓一遍”比调包更值钱这两年AI应用层的工具链成熟得吓人LangChain、LlamaIndex、各种Agent框架几乎把能封装的都封装了。打开文档几行代码就能跑通一个RAG问答复制一段示例一个带工具调用的智能体就活了。表面上看门槛被拉到了地板上谁都能说自己“做过AI项目”。但真到了线上环境问题就来了检索召回率上不去、上下文一长模型就开始胡言乱语、工具调用偶尔死循环、成本账单月底一看吓一跳。这时候你会发现那些被框架藏起来的细节恰恰是决定项目能不能落地的关键。ai-engineering-from-scratch这个标题说的就是这件事——不依赖现成的高级封装从最底层的原理和最小可运行单元开始把AI工程里那些核心环节亲手实现一遍。它不是一个具体的开源库名字而是一种学习路径和工程方法论从token怎么切、向量怎么算、注意力怎么加权到检索怎么排、上下文怎么拼、工具怎么调度每一层都自己写一遍最朴素的版本。适合谁看适合已经会用API调模型、但总感觉“心里没底”的开发者适合想从调包侠进阶到能独立设计AI系统的人也适合那些被框架的“黑盒”坑过、想搞清楚内部到底发生了什么的工程师。我自己走过这条路。最开始也是重度依赖框架后来一次线上事故逼着我回去读源码、手写检索逻辑才发现很多问题的根因都在最基础的地方。这篇文章就把我踩过的坑、总结的路径、以及每个环节“为什么要这么设计”拆开讲清楚。核心关键词就一个从零实现。不是让你重复造轮子去上线而是通过亲手实现来建立对AI系统的直觉这种直觉在排查问题和做技术选型时比任何文档都管用。2. 整体设计思路把AI系统拆成可独立验证的积木2.1 为什么选择“自底向上”而不是“自顶向下”大多数人学AI工程是从调用开始的先跑通一个完整demo再往下改。这条路的问题是你永远在别人的抽象层上打补丁遇到问题只能靠猜。自底向上的思路反过来先把每个基础模块用最朴素的方式实现出来验证它的输入输出符合预期再往上组装。这样做的好处是每个环节都是可解释、可调试的。举个具体例子。RAG系统里有个环节叫“文本切分”框架通常给你一个RecursiveCharacterTextSplitter参数一填就完事。但如果你自己写过切分逻辑就会知道chunk size和overlap的设置直接决定了检索质量切太大一个chunk里混了好几个主题向量表示被稀释切太小语义不完整检索出来答非所问。这些权衡只有亲手实现过、跑过对比实验才能真正理解。自底向上的代价是前期慢但后期排查问题的速度快得不是一点半点。2.2 核心模块的划分与依赖关系一个完整的AI工程系统从零实现的话我习惯把它拆成五个层次每层只依赖它下面那层基础层tokenization、embedding计算、向量相似度。这层是纯数学和字符串处理不涉及任何模型调用。检索层向量索引、相似度搜索、重排序。依赖基础层的embedding和相似度。上下文层prompt模板、上下文拼接、token预算管理。依赖检索层的结果。生成层模型调用、流式输出、结果解析。依赖上下文层。编排层工具调用、多步推理、状态管理。依赖生成层。这个划分的关键在于每一层都可以单独测试。比如你写完检索层不需要接模型就能验证“给定一个query返回的文档是不是相关的”。这种可独立验证性是自底向上最大的优势。2.3 技术选型的取舍逻辑从零实现不代表什么都要自己写。我的原则是核心逻辑自己写基础设施用现成的。比如向量相似度计算自己写余弦相似度就十几行代码值得写但向量存储的持久化和索引结构用FAISS或者hnswlib这种成熟库就行没必要自己实现HNSW图。再比如tokenization如果只是做实验用tiktoken或者transformers的tokenizer足够但如果你想理解BPE的合并过程那就自己写一个简化版。选型的判断标准很简单这个模块的细节会不会影响你对系统行为的理解会就自己写不会就用库。检索的排序逻辑会影响结果质量自己写HTTP请求的重试逻辑不影响核心理解用库。按这个标准筛下来真正需要手写的部分其实不多但每一处都是关键。3. 核心细节解析那些框架不会告诉你的实现要点3.1 Tokenization一切从切词开始Tokenization是AI工程的第一道关。很多人觉得这不就是把文本切成词吗有什么好讲的。但实际操作中tokenizer的选择直接影响三件事成本、上下文长度、以及模型对文本的理解。以BPEByte Pair Encoding为例它的核心思想是从字符开始不断合并出现频率最高的相邻对直到达到预设的词表大小。自己实现一个简化版BPE大概需要这几步# 简化版BPE训练过程伪代码思路 def train_bpe(corpus, vocab_size): # 1. 初始化把所有文本拆成字符每个字符是一个token vocab set(char for text in corpus for char in text) # 2. 统计相邻token对的频率 pairs count_adjacent_pairs(corpus) # 3. 循环合并频率最高的对直到词表达到目标大小 while len(vocab) vocab_size: best_pair max(pairs, keypairs.get) corpus merge_pair(corpus, best_pair) vocab.add(best_pair) pairs count_adjacent_pairs(corpus) return vocab这段逻辑看起来简单但有几个坑。第一合并顺序很重要先合并的pair会成为后续合并的基础所以频率统计必须每轮重新做。第二处理多语言时中文字符和英文字母的合并策略差异很大中文往往需要更大的词表才能覆盖常用词。第三特殊token如|endoftext|要预留位置不能参与合并。实操心得如果你在做中文场景不要直接用GPT系列的tokenizer它对中文的压缩率很低同样的文本会消耗更多token。自己训练一个领域相关的BPE词表能把token数降下来30%以上直接省钱。3.2 Embedding与相似度向量到底在表达什么Embedding是把文本映射到高维空间的一个点语义相近的文本在空间里距离更近。自己实现embedding计算有两种路径一是用预训练模型推理得到向量二是自己训练一个简单的词向量模型如word2vec的简化版。前者实用后者有助于理解。余弦相似度是最常用的度量方式公式是两向量点积除以模长乘积。自己写一遍就知道它的取值范围是[-1, 1]1表示方向完全一致0表示正交-1表示相反。但这里有个容易被忽略的点余弦相似度只关心方向不关心长度。这意味着一个短文本和一个长文本只要主题一致相似度就可能很高。在检索场景下这既是优点也是缺点——优点是主题匹配灵敏缺点是长度差异大的文档可能被误判为高度相关。import numpy as np def cosine_similarity(a, b): dot np.dot(a, b) norm_a np.linalg.norm(a) norm_b np.linalg.norm(b) if norm_a 0 or norm_b 0: return 0.0 return dot / (norm_a * norm_b)实际用的时候我通常会把embedding先做L2归一化这样余弦相似度就退化成点积计算更快。另外如果向量维度很高比如1536维直接用numpy算点积在数据量大时会有性能问题这时候要么用矩阵运算批量处理要么上FAISS。3.3 向量索引暴力搜索什么时候不够用数据量小的时候暴力搜索遍历所有向量算相似度完全够用几百上千条数据毫秒级返回。但数据量上万以后暴力搜索的延迟就上来了。这时候需要近似最近邻ANN索引。自己实现一个最简单的ANN可以用随机投影或者局部敏感哈希LSH的思路。LSH的核心是设计一组哈希函数让相似的向量有更高概率落到同一个桶里。查询时只搜索同一个桶里的向量大幅减少计算量。代价是可能漏掉一些真正相似的向量这就是“近似”的含义。# LSH的简化思路随机超平面划分 def random_projection_hash(vector, num_planes, planes): # planes是预先随机生成的超平面法向量 hash_bits [] for plane in planes[:num_planes]: hash_bits.append(1 if np.dot(vector, plane) 0 else 0) return tuple(hash_bits)这个实现很粗糙但能帮你理解ANN的本质用空间划分换查询速度。实际生产中hnswlib或FAISS的IVF索引都是这个思路的工程化版本。自己写一遍的意义在于你知道召回率和延迟之间的权衡点在哪里调参时心里有数。注意事项ANN索引的召回率不是100%在检索增强生成场景下漏掉关键文档可能导致模型答错。我的做法是先用ANN粗筛出top-50再用暴力搜索精排top-10兼顾速度和准确率。3.4 上下文拼接token预算怎么分配检索回来一堆文档不能全塞给模型因为上下文窗口有限。这时候需要做token预算管理。假设模型上下文窗口是8k token系统prompt占500用户问题占100那么留给检索文档的只有7400。如果每个文档平均500 token最多放14个文档。但简单按数量截断是有问题的。更好的做法是按相关性排序从高到低填充直到预算用完。这里有个细节文档之间要加分隔符分隔符本身也消耗token。我通常用\n---\n这种简单分隔既清晰又省token。def build_context(docs, max_tokens, tokenizer): context [] used 0 for doc in sorted(docs, keylambda d: d[score], reverseTrue): doc_tokens len(tokenizer.encode(doc[text])) if used doc_tokens max_tokens: break context.append(doc[text]) used doc_tokens return \n---\n.join(context)这个逻辑看起来简单但有个隐藏问题如果单个文档就超过了预算怎么办我的处理是截断文档但保留开头和结尾——开头通常是主题句结尾往往是结论中间可以牺牲。这个策略在长文档场景下效果不错。4. 实操过程从零搭一个最小可用的RAG系统4.1 环境准备与依赖选择从零实现不代表不用任何库。我的最小依赖清单是这样的numpy做向量运算tiktoken做token计数faiss-cpu做向量索引如果数据量小可以不用openai或任何模型API做生成。其他的一律自己写。pip install numpy tiktoken faiss-cpu openai为什么不用LangChain因为在这个练习里LangChain会把你需要理解的细节全部藏起来。等你手写一遍之后再用LangChain就是如虎添翼知道每个参数背后的含义。4.2 文档加载与切分实现假设我们有一批Markdown文档第一步是加载和切分。切分策略我试过好几种最后稳定用的是“按段落切分超长段落再按句子切”def split_text(text, max_chunk_size500, overlap50): paragraphs text.split(\n\n) chunks [] for para in paragraphs: if len(para) max_chunk_size: chunks.append(para) else: # 超长段落按句子切 sentences para.split(。) current for sent in sentences: if len(current) len(sent) max_chunk_size: current sent 。 else: if current: chunks.append(current) current sent 。 if current: chunks.append(current) # 加overlap final_chunks [] for i, chunk in enumerate(chunks): if i 0: prev_tail chunks[i-1][-overlap:] final_chunks.append(prev_tail chunk) else: final_chunks.append(chunk) return final_chunksoverlap的作用是防止关键信息刚好被切在边界上。50个字符的overlap在中文场景下大概是一句话的长度够用了。4.3 向量化与索引构建拿到chunks之后批量调embedding接口得到向量然后归一化、建索引import numpy as np def build_index(chunks, embed_fn): vectors [] for chunk in chunks: vec embed_fn(chunk) vec vec / np.linalg.norm(vec) # L2归一化 vectors.append(vec) return np.array(vectors) def search(query, vectors, chunks, embed_fn, top_k5): q_vec embed_fn(query) q_vec q_vec / np.linalg.norm(q_vec) scores vectors q_vec # 归一化后点积等于余弦相似度 top_indices np.argsort(scores)[::-1][:top_k] return [(chunks[i], scores[i]) for i in top_indices]这里用矩阵乘法一次性算所有相似度比循环快很多。数据量上万时vectors q_vec这行可能要几十毫秒可以接受。再大就上FAISS。4.4 生成与结果组装最后一步是把检索结果拼成prompt调模型生成def generate_answer(query, retrieved_docs, llm_fn): context \n---\n.join([doc for doc, _ in retrieved_docs]) prompt f基于以下资料回答问题。如果资料中没有相关信息直接说不知道。 资料 {context} 问题{query} 回答 return llm_fn(prompt)这个prompt模板我调了很多次。关键点是“如果资料中没有相关信息直接说不知道”这句话能显著降低模型胡编的概率。另外资料和问题之间用空行隔开模型对格式的敏感度比想象中高。实操心得检索结果不要直接按分数排序就完事我习惯在拼接前做一次去重。因为overlap的存在相邻chunk可能内容高度重复浪费token。简单的做法是如果两个chunk的相似度超过0.95只保留分数高的那个。5. 常见问题与排查技巧实录5.1 检索不准从query和文档两头找原因检索不准是最常见的问题。排查思路分两步先看query的embedding是否合理再看文档的切分是否合理。query的问题通常是太短或太模糊。比如用户问“怎么弄”embedding出来跟任何文档都不太像。解决办法是做query改写让模型先把问题扩写成一句完整的问句再拿去检索。文档的问题通常是切分粒度不对一个chunk里混了多个主题。解决办法是调整切分策略或者加一个重排序模型对粗筛结果做精排。现象可能原因排查方法解决方向检索结果完全不相关query太短或embedding模型不匹配打印query向量看与文档向量的相似度分布改写query或换embedding模型相关文档排在后边相似度度量方式不合适对比余弦相似度和点积的排序差异归一化后使用点积或加BM25混合检索同一文档反复出现overlap设置过大检查chunk之间的重叠比例减小overlap或做去重长文档检索效果差切分粒度过粗看单个chunk的token数减小chunk size增加切分密度5.2 生成质量差上下文和prompt都要背锅生成质量差很多时候不是模型不行而是喂给它的上下文有问题。我遇到过几种典型情况一是上下文里混入了不相关的文档模型被带偏二是上下文太长关键信息被淹没三是prompt指令不清晰模型不知道要干什么。排查的时候我会先把最终拼好的prompt打印出来人工读一遍。如果我自己都觉得信息杂乱模型肯定也答不好。解决办法包括加相关性阈值低于阈值的文档直接丢弃把最相关的文档放在最前面和最后面中间放次要的利用模型对首尾位置更敏感的特性prompt里明确输出格式比如“用三句话回答”。5.3 成本失控token消耗的监控与优化成本问题在项目上线后才会暴露。我见过一个问答系统单次请求消耗上万token账单直接爆炸。排查下来发现检索返回了20个文档每个文档平均500token光上下文就一万token。优化手段有几个一是限制检索文档数量top-5通常够用二是对文档做摘要压缩用一个小模型先把长文档压缩成短摘要再喂给大模型三是缓存高频问题的答案相同或相似的问题直接返回缓存结果。我实测下来top-5加摘要压缩能把token消耗降到原来的三分之一效果几乎无损。注意事项缓存相似问题的判断也要用embedding设置一个相似度阈值比如0.92超过就命中缓存。阈值太低会返回错误答案太高则命中率低需要根据业务场景调。5.4 工具调用死循环状态管理的重要性如果系统里加了工具调用比如让模型查数据库、调API很容易遇到死循环模型调工具拿到结果后又调同一个工具反复几次耗尽token。根因是模型没有记住“这个工具已经调过了”。解决办法是在上下文里显式记录工具调用历史并在prompt里加一句“不要重复调用已经成功执行过的工具”。另外设置最大调用轮数比如5轮超过就强制停止并返回当前结果。这个兜底机制必须有不然线上出问题很难及时发现。6. 从手写实现到工程落地的经验总结手写一遍AI工程的核心模块最大的收获不是代码本身而是对系统行为的直觉。你知道token是怎么切的就知道为什么某些文本特别费token你知道相似度是怎么算的就知道为什么检索结果时好时坏你知道上下文是怎么拼的就知道为什么模型有时候答非所问。这种直觉在排查线上问题和做技术选型时比任何文档都可靠。实际落地时我不会把所有手写代码都搬到生产环境。手写版本的价值在于验证和理解生产环境该用库还是用库。但选库的时候我会看它的实现思路跟我手写的是不是一致参数含义是不是清晰出问题时能不能追溯到源码。这种判断力是调包调不出来的。最后分享一个小技巧每次遇到检索或生成效果不好的case不要急着调参先把中间结果打印出来——query的向量、检索到的文档、拼好的prompt、模型的原始输出。把这四个东西摆在一起看问题通常一目了然。这个习惯帮我省下了大量瞎调参的时间。
返回列表