☰
从零搭建AI工程链路:手写RAG核心模块的实践指南
2026/10/3 3:37:21 网站建设 项目流程

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_tokens=400, overlap=50): # 先按段落切 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, dtype=np.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, dtype=np.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_k=5): # 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 端到端流程串讲

整个流程走一遍是这样的:

  1. 文档入库:读取原始文档,清洗,切分成块
  2. 生成嵌入:批量调用嵌入接口,得到每个块的向量
  3. 存入向量库:向量和原文一起存起来
  4. 用户提问:拿到问题,生成问题的嵌入向量
  5. 检索:用问题向量在库里搜Top-K相似片段
  6. 组装提示词:把片段和问题拼成最终输入
  7. 调用模型:发请求,拿回答
  8. 后处理:清理格式,返回给用户

这八步里,第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"真正的价值所在。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询