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_embeddingsTrue) return embeddings.tolist()这里有个容易被忽略的细节normalize_embeddingsTrue。因为后续计算余弦相似度时归一化后的向量可以直接用点积代替余弦计算性能更快数值也更稳定。我早期没加这个参数检索结果总是有点说不清的不稳定后来发现就是归一化的问题。索引构建和检索我选择用简单直接的暴力检索实现。数据量在几十万条以下时暴力检索的延迟完全可接受省去维护向量数据库的复杂度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兼容接口你可以替换成任意模型服务。关键是把检索到的上下文和用户问题组装成结构化的promptdef build_prompt(query: str, contexts: List[str]) - str: context_text \n\n.join([f[{i1}] {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_krequest.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_code500, detailstr(e))部署时有几个细节值得注意模型加载放在全局作用域避免每个请求重复加载模型不然延迟会从毫秒级变成秒级推理并发控制用信号量或队列限制并发数防止显卡显存被撑爆请求超时与重试大模型推理可能超过网关默认超时时间需要设置合适的超时阈值优雅下线部署新版本时先停止接收新请求再等待存量请求处理完毕否则会大量报错3.4 监控与持续改进AI工程的后视镜模型上线只是开始真正的挑战在运营阶段。这个项目的监控模块设计得很务实它没有追求花哨的可观测性平台而是先要求你回答三个基础问题系统现在健康吗效果有没有退化用户实际在问什么系统健康监控记录延迟、吞吐、错误率、显存占用。之前有一次线上事故用户反馈回答越来越慢我查监控发现是向量检索的暴力计算在数据量增长后导致CPU跑满。后来加了缓存把热门问题的结果直接缓存延迟从800ms降到150ms。效果退化监控需要定期在黄金测试集上跑分。模型服务版本升级、上游文档库更新都可能导致效果波动。我习惯每次变更后自动跑一遍测试集分数下降超过阈值就触发告警。用户需求洞察是从用户真实问题里提取高频主题和未满足需求。每周翻一翻问答日志你能发现很多产品迭代的方向。4. 常见问题与排查技巧实录4.1 检索效果差的排查清单这是整个AI工程链路里被问得最多的问题。我总结了一个排查顺序从最可能的原因开始chunk切分不合理chunk过大导致语义混杂、过小导致上下文缺失。检查切分结果是否保留了完整语义单元embedding区分度不够换一个领域适配性更好的embedding模型往往立竿见影query与文档粒度不匹配用户问的是年度总结文档切的是项目细节检索自然找不到。考虑多粒度索引混合检索权重失衡BM25和向量检索的结果融合比例不合适需要按实际数据调参4.2 生成结果胡编乱造的应对策略大模型常见的一本正经地胡说八道在RAG系统里主要有三个来源检索到的上下文本身错误源头纠错很重要检查文档库是否有过期或错误信息prompt约束不够强在prompt里明确要求只能基于参考信息回答信息不足时直接说不清楚模型过度发挥降低生成温度甚至让模型先判断是否有足够信息回答再决定是否生成4.3 本地模型和API模型怎么选这是个反复被问的问题。我的建议很直接先看业务需求再算成本账。如果业务涉及私有数据、高并发调用、离线环境大概率要部署开源模型如果数据安全要求不高、调用量不大API模型性价比更高。不要一开始就追求私有化部署除非你已经遇到明确的瓶颈。4.4 数据更新后系统效果为什么不升反降文档库更新后效果反而变差我碰到过几次。查下来最常见的原因是新文档的切分质量差格式和旧文档差异太大导致检索时返回了大量低质量片段。解决方法是文档更新后必须重跑数据质量评估新增内容不能绕过清洗和切分流程直接进索引。5. 实操心得与下一步扩展方向跟着这个项目完整走一遍我自己最大的收获不是某个具体技术点的突破而是建立了一套AI工程全局观。以前遇到问题只会盯着某一个环节猛调现在会习惯性地问数据源有没有问题检索链路有没有问题prompt设计有没有问题生成参数有没有问题这种系统化排查习惯比任何单个技巧都值钱。如果你也想动手实践我的建议是不要一上来就追求完美的架构。先花一两天时间用最简单的代码把加载数据、切分、向量化、检索、生成这条最小链路跑通再逐步加入评估、缓存、监控这些增强环节。每一步都亲自动手改代码、看结果、记笔记遇到问题再回头看这个项目里的对应章节收获会完全不一样。项目后续还可以往几个方向扩展把暴力检索升级为FAISS或向量数据库、加入多轮对话的记忆管理、引入query改写和意图识别、用更细粒度的指标评估生成质量。这些扩展我之前都逐一尝试过每一次都能让系统迈上一个台阶。最后分享一个我实践中特别受益的小技巧把每次调参的实验结果记录在一张表里哪怕只是chunk大小、embedding模型、top_k这几个参数。这个项目让我养成了这个习惯靠着这张表我在多个AI项目里都避免过用同一个配方在不同场景反复踩坑的低效循环。AI工程没有银弹唯一稳赚不赔的投入就是把实验做得可记录、可比对、可回溯。
返回列表