ARTICLE DETAIL

资讯详情

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

从零搭建AI工程链路:手写RAG核心模块的实践指南

从零搭建AI工程链路:手写RAG核心模块的实践指南 1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地降低了一个刚入门的开发者花一个下午就能用现成的框架跑通一个对话机器人。但我在带团队和做技术评审的过程中发现一个很普遍的现象很多人能跑通Demo却说不清楚一次推理请求背后到底发生了什么模型加载慢在哪里、显存为什么爆、Token是怎么被切分的、向量检索的召回率为什么上不去全都答不上来。这就是我特别想聊ai-engineering-from-scratch这个主题的原因——不是反对用现成工具而是主张你得先有能力从底层把一条链路亲手搭一遍再回头用框架你才知道每一步在替你做什么。所谓从零做AI工程不是让你去手写Transformer的每一个矩阵乘法也不是让你从零训练一个大模型那不现实也没必要。它真正指的是把AI应用从输入到输出的完整工程链路拆成一个个可以独立理解、独立实现、独立调试的环节用最朴素的代码把它们串起来。这条链路通常包括文本预处理与分词、嵌入向量生成、向量存储与检索、提示词组装、模型调用与流式输出、结果后处理与评估。你亲手实现过一遍哪怕每一环都用的是简化版方案你对整个系统的掌控力也会完全不一样。这篇文章适合三类人一是刚转行做AI应用、只会调API的开发者想补上工程底子二是有后端或算法基础、想系统梳理AI工程链路的工程师三是技术负责人需要判断团队里哪些环节该自研、哪些该用现成方案。我会把整条链路的选型逻辑、核心实现、参数计算、踩坑经验都摊开讲代码以Python为主尽量做到你照着就能复现。全文不涉及任何敏感内容纯粹是工程实践的分享。2. 整体架构设计与技术选型思路2.1 为什么坚持先手写再上框架我见过太多项目一开始就引入重型框架结果出了问题只能靠猜。举个真实的例子有个团队用某框架做RAG检索增强生成线上反馈回答质量忽高忽低。排查了两天最后发现是文本切分策略的问题——框架默认按固定字符数切分把一段完整的表格切成了两半检索时召回的片段语义不完整。如果他们自己手写过切分逻辑五分钟就能定位到问题。手写一遍的价值在于建立因果直觉。你知道每个环节的输入输出长什么样知道哪个参数会影响什么出了问题你能顺着链路往回找。框架是加速器不是黑箱前提是你得先知道箱子里装的是什么。所以我的建议是第一版原型全部手写用最基础的库跑通之后再决定哪些环节替换成成熟框架。2.2 链路拆解与模块划分我把整条链路拆成六个核心模块每个模块都可以独立测试文本预处理模块负责清洗、分句、切分输出结构化的文本块嵌入生成模块把文本块转成向量是检索的基础向量存储与检索模块存向量、算相似度、返回Top-K结果提示词组装模块把检索结果和用户问题拼成最终给模型的输入模型调用模块处理请求、流式输出、异常重试评估模块量化回答质量指导后续优化这个划分的好处是职责清晰。比如你发现检索结果不准问题一定出在前三个模块如果检索没问题但回答跑偏那大概率是提示词组装或模型调用的问题。定位问题的效率会高很多。2.3 技术选型的取舍逻辑选型上我遵循一个原则能用标准库就不用第三方能用轻量库就不用重型框架。不是为了炫技而是为了减少不确定性。环节手写方案成熟方案我的建议分词切分正则规则专用切分库先手写规则复杂文档再上库嵌入生成调用嵌入接口本地部署嵌入模型原型阶段用接口量大再考虑本地向量检索NumPy算余弦相似度专业向量数据库数据量小于10万条用NumPy足够模型调用原生HTTP请求官方SDK用SDK但要看懂它发的请求评估人工简单指标评估框架先建人工评估集再谈自动化这张表的核心思想是数据量小的时候简单方案完全够用别过早引入复杂度。我见过有人为了几千条数据上了一个分布式向量数据库运维成本远超收益。等数据量真的上来了再迁移也不迟因为你的接口设计是清晰的。3. 核心模块的细节实现与实操要点3.1 文本切分最容易被低估的环节文本切分看着简单实际上是整个链路里最影响效果的一环。切得太碎语义不完整切得太粗检索精度下降。我的经验是按语义边界切分而不是按固定长度。具体做法是先按段落切段落超过阈值再按句子切句子还超才硬切。阈值怎么定这跟你的嵌入模型有关。大多数嵌入模型的有效上下文在256到512个Token之间超过这个长度后面的内容会被截断或稀释。所以切分后的块最好控制在300到400个Token留一点余量。import re def split_text(text, max_tokens400, overlap50): # 先按段落切 paragraphs re.split(r\n\s*\n, text) chunks [] for para in paragraphs: para para.strip() if not para: continue # 估算Token数中文大致按字符数英文按空格分词 if estimate_tokens(para) max_tokens: chunks.append(para) else: # 按句子切 sentences re.split(r(?[。.!?]), para) current for sent in sentences: if estimate_tokens(current sent) max_tokens: if current: chunks.append(current) current sent else: current sent if current: chunks.append(current) return chunks注意overlap重叠参数很重要。相邻块之间保留一定重叠能避免关键信息正好卡在切分边界上被割裂。我一般设50个Token左右太多会导致检索结果冗余。这里有个坑我踩过中文和英文的Token估算方式完全不同。中文一个字大约对应1到2个Token英文一个单词大约1.3个Token。如果你用统一的字符数来切中文块会偏小、英文块会偏大。稳妥的做法是调用嵌入模型自带的分词器来精确计算原型阶段用估算也行但心里要有数。3.2 嵌入向量理解它的本质才能用好它嵌入向量说白了就是把一段文本映射到一个高维空间里的一个点语义相近的文本它们的点距离就近。这个距离通常用余弦相似度来衡量值越接近1越相似。很多人不知道的是嵌入模型是有领域偏好的。通用嵌入模型在通用语料上表现好但在法律、医疗、代码这些专业领域效果会打折扣。如果你的应用是垂直领域要么选领域专用的嵌入模型要么在领域数据上做微调。原型阶段先用通用模型跑通效果不满意再考虑换。生成嵌入的代码很简单import numpy as np def get_embedding(text, embed_client): # 调用嵌入接口返回一个向量 response embed_client.embed(text) vec np.array(response.vector, dtypenp.float32) # 归一化这样后续算余弦相似度就是点积 norm np.linalg.norm(vec) if norm 0: vec vec / norm return vec提示一定要做归一化。归一化之后余弦相似度就等于两个向量的点积计算量大幅下降。这个细节在数据量大时能省不少时间。还有个实操心得批量生成嵌入比逐条生成快得多。大多数嵌入接口支持一次传多条文本吞吐量能提升好几倍。但要注意批次大小太大可能触发接口限制我一般设32或64一批根据接口文档调整。3.3 向量检索NumPy也能撑起一片天数据量不大的时候用NumPy做向量检索完全够用而且透明可控。核心就是把所有向量存成一个矩阵查询时算一次矩阵乘法取Top-K。class SimpleVectorStore: def __init__(self): self.vectors None self.texts [] def add(self, vectors, texts): vectors np.array(vectors, dtypenp.float32) if self.vectors is None: self.vectors vectors else: self.vectors np.vstack([self.vectors, vectors]) self.texts.extend(texts) def search(self, query_vec, top_k5): # query_vec已归一化vectors也已归一化 scores self.vectors query_vec top_indices np.argsort(scores)[::-1][:top_k] return [(self.texts[i], float(scores[i])) for i in top_indices]这段代码简单到不能再简单但它揭示了一个关键事实向量检索的本质就是一次矩阵乘法加排序。理解了这一点你再看那些向量数据库就知道它们无非是在这个基础上加了索引优化、持久化、分布式这些工程能力。什么时候该换向量数据库我的经验阈值是数据量超过10万条或者需要频繁增删改或者需要多条件过滤。低于这个量级NumPy方案的延迟通常在毫秒级完全够用。3.4 提示词组装把检索结果喂给模型的正确姿势检索出来的片段不能直接扔给模型得组装成结构清晰的提示词。我的模板通常长这样你是一个严谨的助手请根据下面提供的资料回答问题。 如果资料中没有相关信息请明确说明资料中未提及不要编造。 资料 [1] {片段1} [2] {片段2} [3] {片段3} 问题{用户问题} 请给出回答并在引用资料时标注编号。这个模板有几个讲究。第一明确要求资料中没有就说没有能显著降低幻觉。第二给片段编号方便模型引用也方便你追溯。第三把资料放在问题前面符合大多数模型的注意力分布特点。注意片段数量不是越多越好。我实测下来3到5个片段是比较好的平衡点。太多会稀释关键信息还会占用宝贵的上下文长度增加成本。组装时还要注意Token预算。假设你的模型上下文是8K Token提示词模板占200问题占100那留给资料的大约是7000。如果每个片段400 Token最多放17个。但实际用不了这么多因为放太多反而降低质量。所以我会设一个上限比如最多5个片段超出的按相似度截断。4. 完整实操流程与关键环节落地4.1 环境准备与依赖安装先把环境搭起来。我习惯用虚拟环境隔离依赖避免污染全局。python -m venv ai_env source ai_env/bin/activate # Windows用 ai_env\Scripts\activate pip install numpy requests就这两个核心依赖。numpy做向量计算requests发HTTP请求。原型阶段不需要更多。等你确认要上框架了再按需安装。4.2 端到端流程串讲整个流程走一遍是这样的文档入库读取原始文档清洗切分成块生成嵌入批量调用嵌入接口得到每个块的向量存入向量库向量和原文一起存起来用户提问拿到问题生成问题的嵌入向量检索用问题向量在库里搜Top-K相似片段组装提示词把片段和问题拼成最终输入调用模型发请求拿回答后处理清理格式返回给用户这八步里第2步和第7步是耗时大头也是成本大头。优化的时候优先看这两步。4.3 参数计算与成本估算假设你有1万篇文档每篇平均切出10个块总共10万个块。每个块平均300 Token。嵌入成本10万块 × 300 Token 3000万Token。按主流嵌入接口的价格大概几美元到几十美元不等一次性投入。检索成本本地计算忽略不计。生成成本每次提问提示词约2000 Token含5个片段回答约300 Token合计2300 Token。如果每天1000次提问就是230万Token/天。这个成本要按你用的模型单价来算贵的模型和便宜的模型能差几十倍。提示成本优化的大头在生成环节。能用小模型解决的场景别用大模型。检索质量提上去片段数量就能降下来成本也跟着降。4.4 流式输出的实现用户体验上流式输出几乎是必须的。等模型把整段话生成完再返回用户会觉得卡。流式输出让用户看到字一个个蹦出来感知延迟大幅降低。def stream_chat(prompt, model_client): response model_client.chat_stream(prompt) for chunk in response: delta chunk.get(delta, ) if delta: yield delta实现上就是逐块读取响应逐块吐给前端。要注意处理网络中断和超时加个重试逻辑。我一般设3次重试每次间隔1秒超过就报错让用户重试。5. 常见问题排查与避坑经验实录5.1 检索不准的排查思路检索不准是最常见的问题排查要按顺序来。先看切分把检索出来的片段打印出来看看是不是语义完整。如果片段本身就不完整那是切分的问题。再看嵌入拿几个语义相近的句子算一下相似度如果相似度很低说明嵌入模型不适合你的领域。最后看检索参数Top-K是不是太小相似度阈值是不是设得太高。我整理了一个速查表现象可能原因排查方法召回片段不相关切分太碎或太粗打印片段人工检查相似问题得分低嵌入模型不匹配领域换模型或微调关键信息总漏掉Top-K太小或阈值太高调大K降低阈值结果重复冗余切分重叠太多减小overlap参数5.2 模型幻觉的抑制手段幻觉就是模型编造资料里没有的内容。抑制手段有几个层次。提示词层面明确要求没有就说没有这是最基础的。检索层面提高召回质量让模型有据可依。后处理层面可以做一个校验检查回答里的关键实体是否出现在检索片段中不在就标记为可疑。我实测下来提示词约束能解决大部分幻觉剩下的靠检索质量兜底。如果还有那可能是模型本身能力问题考虑换模型。5.3 性能与成本的平衡技巧性能和成本往往矛盾。我的做法是分层高频简单问题走小模型低频复杂问题走大模型。怎么判断简单复杂可以用检索结果的相似度分数做路由分数高说明资料匹配好小模型就能答分数低说明问题偏难交给大模型。还有个技巧是缓存。相同或相似的问题直接返回缓存结果省一次模型调用。缓存key可以用问题的嵌入向量做近似匹配相似度超过阈值就命中缓存。5.4 我踩过的几个坑第一个坑是忽略编码问题。读取文档时没指定编码中文全乱码嵌入出来全是噪声。后来统一用UTF-8并在读取时加异常处理。第二个坑是嵌入接口的批次限制。我一次传了200条接口直接报错。后来改成64一批稳定了。第三个坑是向量没归一化。检索结果忽好忽坏排查半天才发现是没归一化导致相似度计算错误。归一化之后稳定了。第四个坑是提示词里片段顺序。我一开始按检索分数从低到高排结果模型总忽略后面的高分片段。改成从高到低排效果好多了。6. 评估体系与持续优化方向6.1 先建人工评估集自动化评估听着美好但没有人工评估集做基准自动化指标就是空中楼阁。我的做法是从真实用户问题里挑100条覆盖不同难度和类型人工标注标准答案。这100条就是你的黄金测试集每次改动都跑一遍看效果是涨是跌。标注的时候要记录几个维度答案是否正确、是否完整、是否有幻觉、引用是否准确。这四个维度基本能反映回答质量。6.2 关键指标的定义与计算检索环节看召回率和精确率。召回率是标准答案涉及的片段有多少被检索出来精确率是检索出来的片段有多少是相关的。这两个指标要平衡光看一个会误导。生成环节看答案准确率和幻觉率。准确率是回答正确的比例幻觉率是编造内容的比例。这两个指标直接反映用户体验。计算召回率需要标注每个问题的相关片段工作量不小但值得。有了这个基准你调切分参数、换嵌入模型、改检索策略都能量化效果。6.3 迭代优化的优先级优化要有优先级别眉毛胡子一把抓。我的顺序是先修检索再调提示词最后换模型。因为检索是根基检索不准后面怎么调都白搭。提示词是性价比最高的优化点改几行字可能就有明显提升。换模型是最后手段成本高而且不一定解决问题。每次只改一个变量改完跑评估集确认有效再改下一个。同时改多个变量你根本不知道是哪个起了作用。7. 从原型到生产的扩展路径原型跑通之后往生产走要考虑几件事。稳定性方面加超时、重试、降级。模型接口挂了要有兜底方案比如返回缓存结果或友好提示。可观测性方面记录每次请求的耗时、Token消耗、检索结果、模型输出出问题能追溯。扩展性方面把各模块的接口设计清楚将来替换实现时不影响其他部分。数据量上来之后向量检索换成专业数据库嵌入生成考虑本地部署模型调用做多路负载。但这些都是在原型验证有效之后的事别提前做。我个人在实际操作中的体会是从零搭一遍最大的收获不是代码本身而是对整条链路的掌控感。你知道每个环节的边界在哪知道问题可能出在哪知道优化该往哪个方向走。这种掌控感是调包调不出来的。等你有了这个底子再用框架你会发现框架的每个设计你都能理解每个参数你都知道该不该调。这才是ai-engineering-from-scratch真正的价值所在。
返回列表