拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

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

AI工程从零开始:构建生产级RAG系统的完整实践指南

AI工程从零开始:构建生产级RAG系统的完整实践指南

1. 项目概述与核心思路

1.1 这个项目到底在做什么

"ai-engineering-from-scratch"这个标题乍看像是又一个AI学习仓库,但真正动手做过的人会明白,它瞄准的其实是今天AI落地过程中最扎手的那一环——从零开始搭建一套能用于实际业务的技术体系。不是调个接口、跑个模型demo,而是把数据、模型、评估、部署、监控这条链路完整地走通,每一步都不靠现成的"全家桶"糊弄过去。

我接触这个项目是在一次内部技术分享上。当时团队正在做一个知识库问答系统,市面上有各种成熟的RAG框架,但真到生产环境就发现:框架帮你掩盖了太多本该由你掌控的细节。比如召回结果里混着大量低质量片段,你根本不知道问题出在chunk切分还是embedding模型选择上。后来我们干脆拆掉框架,从最底层一步步搭,反而把问题看得清清楚楚——这正是"from scratch"的价值所在。

这个项目适合三类人:刚入门AI工程、不想只会调包的同学;被生产环境各种诡异问题折磨过的开发者;以及需要给团队搭建内部技术体系的技术管理者。它的核心不是教你某个框架怎么用,而是帮你建立一套"理解原理、掌控细节、能排查问题"的工程能力。

1.2 为什么"从零开始"比直接用框架更重要

举个真实的例子。我之前用某个流行的RAG框架搭demo,跑通只花了一个下午。但上线后遇到一个怪问题:用户问"去年的营收数据"时,系统老是召回前年的报表。我翻框架源码,发现它的默认chunk大小是512字符,而我们的财务表格经常把一个季度的数据塞进一个超长单元格,切分后语义全被截断了。你要是不理解chunking原理,这种问题排查起来就像大海捞针。

"from scratch"的意义就在这:当你自己实现过文本切分、向量化、检索排序这些基础组件,你会清楚地知道每个环节的假设是什么、边界在哪里、参数调整会影响什么。框架给你的是便利,但代价是黑盒。黑盒在demo阶段没问题,到生产环境就是事故高发区。

这个项目把整个AI工程链路拆成了六大模块:数据工程与预处理、模型选型与微调、检索增强生成(RAG)实现、评估体系搭建、部署与服务化、监控与持续改进。每一块都有对应的最小实现和实验记录,完全不依赖重型框架,用最朴素的代码把原理讲明白。

1.3 项目要解决的实际痛点

我在几个不同规模的团队里都见过类似的困境:

  • 框架依赖症:项目代码里到处都是框架特有的抽象概念,换个场景就束手束脚
  • 评估缺失:模型上线全靠"感觉还行",没有量化指标,出了问题说不清是哪个环节的锅
  • 调试困难:数据、模型、检索、生成,每个环节都像独立黑盒,串联起来排查效率极低
  • 知识断层:新同学上手就是"调框架",底层原理一片空白,出了问题只能靠试错

这个项目的目标就是把这四个痛点一次解决掉。它不是那种"收藏即学会"的资料合集,而是一个需要你跟着动手实现的工程实践指南。

2. 核心模块拆解与实操解析

2.1 数据工程:整个AI系统的地基

数据工程在AI项目里的地位,怎么强调都不过分。我见过太多团队花大精力调模型,却对数据质量视而不见。这个项目的第一课就是让你亲手搭建一个数据预处理流水线,处理清洗、去重、格式转换、质量评估这几类基础任务。

清洗环节最容易踩的坑是规则写得太激进。比如你用正则去删除HTML标签,结果把代码块里的"<" ">"也给误删了。我后来改成两步走:先用解析器提取正文,再对提取结果做规则清洗,误伤率大幅下降。去重要用minhash这类近似算法,精确去重在千万级数据上根本跑不动。格式转换要区分数据源头是结构化数据库还是非结构化文档,处理逻辑完全不同。

数据质量评估是我特别想强调的一环。很多人直接跳过这步,结果模型训完才发现训练集里全是噪音。这个项目里实现了一个简单的质量打分器,从完整性、一致性、时效性、可读性四个维度给每条数据打分,低于阈值的自动进入人工审核队列。单这一项,就能帮团队省下大量清洗时间。

2.2 模型选型:别被"暴力美学"带偏

模型选型是AI工程里最容易走向极端的环节。要么无脑上最大参数量的模型,要么迷信"小模型调优吊打大模型"的传说。我从这个项目里学到的最重要一课是:选型不是选最强的,而是选约束条件下最合适的。

做选型决策时要先明确几个约束条件:推理预算上限是多少?目标延迟能否接受?部署环境是GPU集群还是边缘设备?数据规模是否支撑微调?把这四个问题答案列出来,你大概就能筛掉80%的候选模型。

我自己的经验是:通用任务优先考虑API调用成熟模型,但一旦涉及私有数据或专业领域,本地部署开源模型反而更可控。比如之前做医疗文献问答,直接用通用大模型回答,专业名词经常胡说八道。后来用领域数据微调了一个7B参数的开源模型,配合检索增强,效果反而明显更好。关键是把"何时用大模型、何时用小模型、何时用检索增强"这套决策逻辑想清楚。

2.3 RAG实现:检索和生成的"双人舞"

RAG是当前AI应用落地最主流的架构之一,但这个项目里对RAG的实现拆解得很克制,没有堆叠各种花哨技巧,而是先把最核心的检索链路讲透。

RAG的核心链路是:文档加载、文本切分、向量化、索引构建、检索排序、上下文组装、生成回答。每一环都有大量可以优化的空间,但前提是你得先有一个可靠的基线版本。

文本切分是最容易被低估的环节。固定长度切分虽然简单,但会把语义完整的段落拦腰截断。我尝试过基于标题层级和段落结构做智能切分,加上少量重叠,检索效果明显提升。你可以先用固定长度跑基线,再逐步迭代优化。

向量化要区分稠密向量和稀疏向量。稠密向量对语义相似度敏感,但容易忽略关键词匹配;稀疏向量(比如BM25)正好相反。这个项目里做了一个简单的混合检索:BM25和向量检索各出一部分结果,再用RRF算法融合排序。实测下来,混合检索的召回质量比单独用任何一种都稳定。

我建议大家一定要动手把RAG的每一步都自己实现一遍,哪怕只是一个几百行代码的最小版本。只有自己写一遍,你才能真正理解"为什么chunk大小会显著影响检索效果""为什么embedding模型的选择会造成那么大的召回差异"。

3. 实操过程与核心环节实现

3.1 最小RAG系统的完整实现

我按照这个项目的思路,一步步搭建了一个最小可用的RAG系统。这里把核心代码和设计思路分享出来,大家可以直接参考和修改。

首先是文档加载与切分模块。为了兼顾代码可读性和实际效果,我用递归字符切分器,按标题层级优先切分,并在相邻块之间保留少量重叠:

import re from typing import List, Dict class RecursiveChunker: def __init__(self, chunk_size: int = 800, overlap: int = 100): self.chunk_size = chunk_size self.overlap = overlap # 切分优先级:段落标题 > 换行 > 句号 > 逗号 self.splitters = [r"\n#{1,6}\s", r"\n\n", r"(?<=[。!?])", r"(?<=[,;])"] def split(self, text: str) -> List[str]: if len(text) <= self.chunk_size: return [text] for splitter in self.splitters: parts = re.split(splitter, text) if len(parts) > 1: break chunks = [] current = "" for part in parts: if len(current) + len(part) > self.chunk_size: if current: chunks.append(current) current = part else: current += part if current: chunks.append(current) # 处理重叠:在相邻chunk之间拼接tail和head final_chunks = [] for i, chunk in enumerate(chunks): if i > 0: tail = chunks[i-1][-self.overlap:] chunk = tail + chunk final_chunks.append(chunk) return final_chunks

这段代码看似简单,但体现了一个关键设计:切分优先级从粗粒度到细粒度,先按标题、再按段落、最后按句子,这样能最大程度保留语义边界。重叠机制则是为了弥补边界信息丢失的问题,避免一个完整语义片段被硬生生切开。

向量化这一步,我选择了开源的国产模型,用HuggingFace的sentence-transformers加载:

from sentence_transformers import SentenceTransformer # 加载embedding模型,维度是768 model = SentenceTransformer("BAAI/bge-large-zh-v1.5") def embed_texts(texts: List[str]) -> List[List[float]]: embeddings = model.encode(texts, normalize_embeddings=True) return embeddings.tolist()

这里有个容易被忽略的细节:normalize_embeddings=True。因为后续计算余弦相似度时,归一化后的向量可以直接用点积代替余弦计算,性能更快,数值也更稳定。我早期没加这个参数,检索结果总是有点说不清的不稳定,后来发现就是归一化的问题。

索引构建和检索我选择用简单直接的暴力检索实现。数据量在几十万条以下时,暴力检索的延迟完全可接受,省去维护向量数据库的复杂度:

import numpy as np from typing import List, Tuple class VectorStore: def __init__(self): self.vectors = [] self.texts = [] def add(self, text: str, vector: List[float]) -> None: self.texts.append(text) self.vectors.append(np.array(vector)) def search(self, query_vector: List[float], top_k: int = 5) -> List[Tuple[str, float]]: query = np.array(query_vector) scores = [] for vec in self.vectors: score = np.dot(query, vec) scores.append(score) top_indices = np.argsort(scores)[-top_k:][::-1] return [(self.texts[i], scores[i]) for i in top_indices]

如果你数据量超过百万,再考虑换成FAISS这类向量检索库,原理是一样的。前期用暴力检索跑通整个链路,后期换库的成本很低,这是保持简单性的一个重要策略。

最后是生成环节。项目默认对接OpenAI兼容接口,你可以替换成任意模型服务。关键是把检索到的上下文和用户问题组装成结构化的prompt:

def build_prompt(query: str, contexts: List[str]) -> str: context_text = "\n\n".join([f"[{i+1}] {c}" for i, c in enumerate(contexts)]) return f"""基于以下参考信息回答问题。如果信息不足,请明确说明。 参考信息: {context_text} 用户问题:{query} 请给出准确、简洁的回答。"""

3.2 评估体系:没有指标优化的AI工程就是"自嗨"

AI项目最怕的就是"感觉还行"。这个项目把评估体系提到很高的优先级,我非常认同。我的建议是至少从三个维度搭建评估指标:检索质量、生成质量、端到端效果。

检索质量主要看召回率、命中率、MRR(平均倒数排名)。具体做法是构建一个测试集,每个问题关联一个或几个标准答案对应的文档片段。检索后检查正确片段是否在召回结果里,位置越靠前越好。

生成质量相对难评估一些。自动化层面可以看答案与标准答案的文本相似度(ROUGE、BLEU),但这套指标对开放域问答没那么可靠。我实践中还会加一层规则校验:答案是否包含必须提到的关键实体,是否引用了检索上下文中的内容。这些规则能抓出一大类"编造答案"的问题。

端到端效果才是最终决策依据。我会把真实用户问题抽样出来,人工打分或者用更强模型打分。这个环节虽然费时,但最能反映真实体验。项目里给了一个很务实的建议:先建一个20~50条样本的黄金测试集,迭代初期不要过度追求评估方法的复杂度,重点是快速发现问题、快速修正。

这里分享一个我踩过的坑:曾经只关注检索召回率,把召回率从60%优化到85%,但用户反馈并没有明显改善。后来查了日志,发现召回的片段虽然对了,但排序混乱,正确答案排在第三第四位,生成的答案被前面几个错误片段干扰了。所以评估一定要看"端到端效果",单点指标好看没有用。

3.3 部署与服务化:从"能跑"到"能扛"跨越

部署这一步,"from scratch"意味着你要理解服务化底层原理,而不是只会用现成的部署框架。这个项目推荐了一条很清晰的路径:先用FastAPI把推理封装成HTTP服务,再逐步加批处理、缓存、限流。

FastAPI封装模型推理的代码很简洁:

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): query: str top_k: int = 5 @app.post("/ask") def ask(request: QueryRequest): try: query_vector = embed_texts([request.query])[0] contexts = vector_store.search(query_vector, top_k=request.top_k) prompt = build_prompt(request.query, [c[0] for c in contexts]) answer = generate(prompt) return {"answer": answer, "contexts": [c[0] for c in contexts]} except Exception as e: raise HTTPException(status_code=500, detail=str(e))

部署时有几个细节值得注意:

  • 模型加载放在全局作用域:避免每个请求重复加载模型,不然延迟会从毫秒级变成秒级
  • 推理并发控制:用信号量或队列限制并发数,防止显卡显存被撑爆
  • 请求超时与重试:大模型推理可能超过网关默认超时时间,需要设置合适的超时阈值
  • 优雅下线:部署新版本时,先停止接收新请求,再等待存量请求处理完毕,否则会大量报错

3.4 监控与持续改进:AI工程的"后视镜"

模型上线只是开始,真正的挑战在运营阶段。这个项目的监控模块设计得很务实,它没有追求花哨的可观测性平台,而是先要求你回答三个基础问题:系统现在健康吗?效果有没有退化?用户实际在问什么?

系统健康监控记录延迟、吞吐、错误率、显存占用。之前有一次线上事故,用户反馈"回答越来越慢",我查监控发现是向量检索的暴力计算在数据量增长后导致CPU跑满。后来加了缓存,把热门问题的结果直接缓存,延迟从800ms降到150ms。

效果退化监控需要定期在黄金测试集上跑分。模型服务版本升级、上游文档库更新,都可能导致效果波动。我习惯每次变更后自动跑一遍测试集,分数下降超过阈值就触发告警。

用户需求洞察是从用户真实问题里提取高频主题和未满足需求。每周翻一翻问答日志,你能发现很多产品迭代的方向。

4. 常见问题与排查技巧实录

4.1 检索效果差的排查清单

这是整个AI工程链路里被问得最多的问题。我总结了一个排查顺序,从最可能的原因开始:

  1. chunk切分不合理:chunk过大导致语义混杂、过小导致上下文缺失。检查切分结果是否保留了完整语义单元
  2. embedding区分度不够:换一个领域适配性更好的embedding模型,往往立竿见影
  3. query与文档粒度不匹配:用户问的是"年度总结",文档切的是"项目细节",检索自然找不到。考虑多粒度索引
  4. 混合检索权重失衡:BM25和向量检索的结果融合比例不合适,需要按实际数据调参

4.2 生成结果胡编乱造的应对策略

大模型常见的"一本正经地胡说八道",在RAG系统里主要有三个来源:

  • 检索到的上下文本身错误:源头纠错很重要,检查文档库是否有过期或错误信息
  • prompt约束不够强:在prompt里明确要求"只能基于参考信息回答,信息不足时直接说不清楚"
  • 模型过度发挥:降低生成温度,甚至让模型先判断"是否有足够信息回答",再决定是否生成

4.3 本地模型和API模型怎么选

这是个反复被问的问题。我的建议很直接:先看业务需求,再算成本账。如果业务涉及私有数据、高并发调用、离线环境,大概率要部署开源模型;如果数据安全要求不高、调用量不大,API模型性价比更高。不要一开始就追求"私有化部署",除非你已经遇到明确的瓶颈。

4.4 数据更新后系统效果为什么不升反降

文档库更新后,效果反而变差,我碰到过几次。查下来最常见的原因是新文档的切分质量差,格式和旧文档差异太大,导致检索时返回了大量低质量片段。解决方法是:文档更新后必须重跑数据质量评估,新增内容不能绕过清洗和切分流程直接进索引。

5. 实操心得与下一步扩展方向

跟着这个项目完整走一遍,我自己最大的收获不是某个具体技术点的突破,而是建立了一套"AI工程全局观"。以前遇到问题只会盯着某一个环节猛调,现在会习惯性地问:数据源有没有问题?检索链路有没有问题?prompt设计有没有问题?生成参数有没有问题?这种系统化排查习惯,比任何单个技巧都值钱。

如果你也想动手实践,我的建议是:不要一上来就追求完美的架构。先花一两天时间,用最简单的代码把"加载数据、切分、向量化、检索、生成"这条最小链路跑通,再逐步加入评估、缓存、监控这些增强环节。每一步都亲自动手改代码、看结果、记笔记,遇到问题再回头看这个项目里的对应章节,收获会完全不一样。

项目后续还可以往几个方向扩展:把暴力检索升级为FAISS或向量数据库、加入多轮对话的记忆管理、引入query改写和意图识别、用更细粒度的指标评估生成质量。这些扩展我之前都逐一尝试过,每一次都能让系统迈上一个台阶。

最后分享一个我实践中特别受益的小技巧:把每次调参的实验结果记录在一张表里,哪怕只是chunk大小、embedding模型、top_k这几个参数。这个项目让我养成了这个习惯,靠着这张表,我在多个AI项目里都避免过"用同一个配方在不同场景反复踩坑"的低效循环。AI工程没有银弹,唯一稳赚不赔的投入就是把实验做得可记录、可比对、可回溯。

返回列表