ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

从零搭建AI工程能力:构建稳定可用的RAG问答系统实战

从零搭建AI工程能力:构建稳定可用的RAG问答系统实战 1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年“AI工程”这个词被说得太多了多到有点变味。打开任何一个技术社区满屏都是“三行代码调用大模型”“十分钟搭建RAG应用”“零基础转行AI工程师”。我身边不少做后端、做前端的朋友看得心痒觉得门槛也没那么高于是跟着教程跑通了几个demo简历上就敢写“熟悉AI应用开发”。结果一到真实项目就露馅——模型输出不稳定、检索召回率上不去、推理成本失控、上线之后延迟高得没法用。问题出在哪出在大家把“调包”当成了“工程”。ai-engineering-from-scratch这个标题我理解它想表达的核心不是“从零训练一个大模型”那玩意儿不是个人能碰的。它真正想说的是从工程视角把AI能力从“能跑”做到“能用、好用、稳定用”。这中间隔着的是数据管道、提示词管理、检索策略、评估体系、成本控制、可观测性这一整套东西。这些东西才是AI工程师和“调包侠”之间的分水岭。我写这篇东西不是要给你灌鸡汤也不是要复述官方文档。我想做的是把我在实际项目里踩过的坑、总结出来的方法按一个可复现的顺序摊开来讲。适合谁看如果你已经会写Python、能看懂基本的API文档但一提到“把AI集成到生产系统”就心里没底那这篇就是写给你的。如果你还在纠结要不要入行那也可以先看看AI工程到底在工程什么。2. 整体设计思路把AI当系统而不是当魔法2.1 为什么“从零”不等于“从模型开始”很多人对“从零搭建AI工程”有个误解觉得第一步应该是去HuggingFace上下个模型或者去研究Transformer的注意力机制怎么推导。这个方向不能说错但它不是工程视角。工程视角的第一步是定义清楚你要解决什么问题以及这个问题对系统的约束是什么。我举个具体的例子。假设你要做一个“智能客服知识库问答”。如果你一上来就选模型、搭向量库很容易陷入“技术选型焦虑”——用哪个embedding模型用哪个向量数据库chunk size设多少这些问题当然要回答但如果你先花半天时间把下面这张表填清楚后面的选型会快很多约束维度具体问题对设计的影响延迟要求用户能接受几秒的等待决定是否用流式输出、是否要缓存成本预算每次查询能花多少钱决定模型大小、是否用本地模型准确率底线答错的代价有多大决定是否要人工兜底、检索策略激进程度数据更新频率知识库多久变一次决定索引重建策略、增量更新方案并发量级峰值多少QPS决定部署架构、是否要批处理这张表填完你会发现很多“选型问题”其实变成了“约束下的最优解问题”。比如延迟要求是2秒内那你就别考虑那些参数量巨大的模型做在线推理成本预算极低那你就得认真评估本地小模型加缓存能不能扛住。工程的核心不是选最牛的技术而是在约束下做最合适的取舍。2.2 分层架构把变化的部分隔离出来AI应用和传统应用最大的区别在于不确定性。传统后端接口输入确定输出基本确定AI应用同样的输入输出可能每次都不一样。这种不确定性会渗透到系统的每一层所以架构设计的第一原则是把不确定的部分关进笼子里让确定的部分保持稳定。我习惯把AI应用分成四层接入层处理用户请求、鉴权、限流、日志。这层是传统的Web工程没什么特别的。编排层决定一次请求要经过哪些步骤——要不要检索要不要调用工具要不要多轮反思这层是AI工程的核心也是逻辑最复杂的地方。能力层模型推理、向量检索、外部API调用。这层是“不确定”的主要来源需要重点做超时、重试、降级。数据层知识库、对话历史、评估集、日志。这层是长期资产设计好坏直接影响迭代速度。这么分的好处是当模型效果不好时你能快速定位是编排逻辑的问题、检索的问题还是模型本身的问题。我见过太多项目把所有逻辑塞在一个函数里出了问题只能靠打印日志大海捞针。2.3 评估先行没有度量就没有优化这是我最想强调的一点也是大多数从零开始的项目最容易忽略的。在写第一行业务代码之前先建评估集。哪怕只有20条问答对也比没有强。为什么因为AI应用的迭代和传统软件不一样。传统软件改代码逻辑是确定的你大概知道改完会怎样。AI应用改一个提示词、换一个模型、调一个参数效果可能变好也可能变差而且你很难凭感觉判断。没有评估集你的迭代就是盲人摸象。评估集怎么建我的经验是分三步冷启动阶段手动构造20-50条覆盖核心场景的问答对包括简单问题、复杂问题、边界问题。运行阶段把线上真实请求中“用户点了不满意”的case补充进去。迭代阶段每次修改后跑一遍评估集记录准确率、召回率、平均延迟、平均成本。评估集的格式可以很简单就是一个JSON文件[ { question: 你们的退货政策是什么, expected_answer: 支持7天无理由退货商品需保持完好。, expected_source: 售后政策文档第3节 } ]跑评估的脚本也不复杂核心就是遍历、调用、对比、统计。但就是这么一个简单的东西能让你的迭代效率提升好几倍。3. 核心细节解析数据、检索、提示词、模型3.1 数据管道垃圾进垃圾出AI工程里有一句老话Garbage in, garbage out。你的知识库质量直接决定了最终效果的上限。我见过一个项目模型换了三轮提示词改了十几版效果就是上不去最后发现是原始文档里有一半是扫描件OCR出来的乱码。数据管道要解决三个问题采集、清洗、切分。采集相对简单无非是从数据库、文件系统、API拉数据。但要注意的是元数据一定要保留。什么是元数据就是这条数据来自哪个文件、哪个章节、更新时间是什么。这些信息在检索和引用时非常关键。我习惯在采集阶段就给每条数据打上唯一ID和来源标记。清洗是最容易被低估的环节。常见的脏数据包括HTML标签残留、多余空白字符、页眉页脚、乱码、重复内容。清洗规则要根据数据源定制但有几条通用原则统一编码为UTF-8去除连续空白和换行过滤掉长度过短的片段比如少于10个字符对重复内容做去重可以用简单的哈希也可以用向量相似度切分是决定检索效果的关键。切分太粗检索出来的内容包含太多无关信息模型容易被干扰切分太细上下文丢失模型理解不了完整语义。我的经验值是中文按300-500字切分英文按200-300词切分重叠部分设为10%-20%。但这个值不是固定的要根据你的文档类型调整。技术文档可以细一点叙事性文档可以粗一点。注意切分的时候尽量按语义边界切比如按段落、按标题。不要硬按字符数切否则很容易把一句话切成两半。3.2 检索策略不只是向量相似度很多人做RAG检索部分就是“把问题向量化然后去向量库找最相似的top-k”。这当然可以但效果往往不够好。原因很简单用户的提问方式和文档的表述方式往往不一致。用户问“怎么退钱”文档里写的是“退款流程”向量相似度可能就不高。我的做法是混合检索向量检索加关键词检索然后做融合排序。向量检索负责语义匹配关键词检索负责精确匹配。两者互补召回率会明显提升。具体实现上可以用BM25做关键词检索用embedding做向量检索然后用RRFReciprocal Rank Fusion做融合。RRF的公式很简单score sum(1 / (k rank_i))其中k一般取60rank_i是文档在第i个检索结果中的排名。这个公式的好处是不需要归一化分数直接看排名工程上很好实现。还有一个容易被忽略的点是重排序。检索出来的top-k文档可以再用一个交叉编码器cross-encoder做精排。交叉编码器把问题和文档一起输入直接输出相关性分数比向量相似度准得多。代价是计算量大所以一般只对top-20做重排选出top-5给模型。检索方式优点缺点适用场景向量检索语义匹配强精确匹配弱自然语言提问关键词检索精确匹配强语义泛化弱专有名词查询混合检索互补效果好实现复杂度高大多数生产场景重排序精度高延迟高对准确率要求高3.3 提示词管理别把提示词硬编码在代码里提示词是AI应用的“业务逻辑”但它又不像传统代码那样有明确的语法和测试。很多项目把提示词直接写在Python文件里改一次就要重新部署这太痛苦了。我的做法是提示词外置加版本管理。具体来说提示词存在单独的YAML或JSON文件里按场景分文件。每个提示词有版本号修改时新增版本而不是直接改。运行时根据配置加载对应版本的提示词。评估集跑的时候记录用了哪个版本方便对比。一个提示词文件的例子version: v3 description: 客服问答提示词强调引用来源 system: | 你是一个客服助手。请根据提供的参考资料回答用户问题。 如果参考资料中没有相关信息请明确告知用户你不知道。 回答时请引用参考资料的来源。 user: | 参考资料 {context} 用户问题{question}这么做的好处是当效果变差时你可以快速回滚到上一个版本而不是手忙脚乱地改代码。3.4 模型选型别只看排行榜模型选型是另一个容易踩坑的地方。很多人只看排行榜谁排第一就用谁。但排行榜上的评测集和你的实际场景往往差别很大。我的建议是用自己的评估集跑一遍重点看三个指标准确率、延迟、成本。准确率不用多说。延迟方面要注意区分首token延迟和总延迟。流式输出场景下首token延迟决定用户体验非流式场景下总延迟决定一切。成本方面要算清楚每百万token的输入输出价格然后乘以你的预估调用量。还有一个策略是模型分级。简单问题用小模型复杂问题用大模型。怎么判断简单还是复杂可以用规则也可以用一个小分类器。我做过一个项目用规则把问题分成“事实查询”和“推理分析”两类前者走小模型后者走大模型成本直接降了60%效果几乎没变。4. 实操过程从零搭一个可用的问答系统4.1 环境准备与依赖安装我假设你已经有一个Python环境版本3.10以上。依赖方面核心是这几个pip install fastapi uvicorn openai tiktoken pip install sentence-transformers faiss-cpu pip install rank-bm25 jieba pip install pyyaml python-dotenv简单解释一下FastAPI做Web服务tiktoken算token数sentence-transformers做向量化faiss做向量检索rank-bm25做关键词检索jieba做中文分词。这些都是轻量级方案本地就能跑。提示如果你用OpenAI的API需要设置环境变量。建议用.env文件管理不要硬编码在代码里。4.2 数据准备与索引构建假设你有一批Markdown格式的文档。第一步是读取和清洗import os import re def load_documents(doc_dir): docs [] for filename in os.listdir(doc_dir): if filename.endswith(.md): with open(os.path.join(doc_dir, filename), r, encodingutf-8) as f: content f.read() content clean_text(content) docs.append({ id: filename, content: content, source: filename }) return docs def clean_text(text): text re.sub(r[^], , text) text re.sub(r\n{3,}, \n\n, text) text re.sub(r[ \t]{2,}, , text) return text.strip()然后是切分def split_documents(docs, chunk_size400, overlap50): chunks [] for doc in docs: content doc[content] start 0 while start len(content): end start chunk_size chunk content[start:end] chunks.append({ id: f{doc[id]}_{start}, content: chunk, source: doc[source] }) start end - overlap return chunks接下来构建向量索引和BM25索引from sentence_transformers import SentenceTransformer import faiss import numpy as np from rank_bm25 import BM25Okapi import jieba model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) chunks split_documents(docs) texts [c[content] for c in chunks] embeddings model.encode(texts, normalize_embeddingsTrue) dimension embeddings.shape[1] index faiss.IndexFlatIP(dimension) index.add(embeddings.astype(float32)) tokenized [list(jieba.cut(t)) for t in texts] bm25 BM25Okapi(tokenized)4.3 混合检索与重排序实现检索的时候两路并行def hybrid_search(query, top_k5): query_vec model.encode([query], normalize_embeddingsTrue) _, vector_indices index.search(query_vec.astype(float32), top_k * 2) query_tokens list(jieba.cut(query)) bm25_scores bm25.get_scores(query_tokens) bm25_indices np.argsort(bm25_scores)[::-1][:top_k * 2] rrf_scores {} for rank, idx in enumerate(vector_indices[0]): rrf_scores[idx] rrf_scores.get(idx, 0) 1 / (60 rank) for rank, idx in enumerate(bm25_indices): rrf_scores[idx] rrf_scores.get(idx, 0) 1 / (60 rank) sorted_indices sorted(rrf_scores.keys(), keylambda x: rrf_scores[x], reverseTrue) return [chunks[i] for i in sorted_indices[:top_k]]重排序可以用交叉编码器但为了简化这里先用RRF的结果。实际项目中如果对准确率要求高建议加一个cross-encoder。4.4 提示词组装与模型调用import openai def build_prompt(question, contexts): context_text \n\n.join([ f[来源{c[source]}]\n{c[content]} for c in contexts ]) return f你是一个客服助手。请根据以下参考资料回答用户问题。 如果参考资料中没有相关信息请明确告知用户你不知道。 回答时请引用参考资料的来源。 参考资料 {context_text} 用户问题{question} def ask(question): contexts hybrid_search(question) prompt build_prompt(question, contexts) response openai.ChatCompletion.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.3 ) return response.choices[0].message.content4.5 评估脚本与迭代流程评估脚本的核心逻辑import json def evaluate(eval_file): with open(eval_file, r, encodingutf-8) as f: cases json.load(f) correct 0 total len(cases) for case in cases: answer ask(case[question]) if case[expected_answer] in answer: correct 1 else: print(f错误案例{case[question]}) print(f期望{case[expected_answer]}) print(f实际{answer}) print(---) print(f准确率{correct}/{total} {correct/total:.2%})每次修改提示词、检索参数、模型之后跑一遍这个脚本记录结果。我习惯用一个简单的CSV记录每次迭代的配置和指标方便回溯。迭代版本模型chunk_sizetop_k准确率平均延迟v1gpt-4o-mini400372%1.2sv2gpt-4o-mini300580%1.5sv3gpt-4o-mini300585%1.6s5. 常见问题与排查技巧实录5.1 检索召回率低怎么办这是最常见的问题。用户问了一个问题检索出来的文档完全不相关。排查思路检查切分粒度如果chunk太大检索出来的内容可能包含太多噪声如果太小可能丢失关键信息。试着调整chunk_size跑评估集看效果。检查embedding模型有些模型对中文支持不好。可以换一个多语言模型试试或者用专门的中文模型。检查查询改写用户的问题可能很口语化可以先用一个小模型把问题改写成更正式的表述再去检索。增加关键词检索权重如果问题里有专有名词向量检索可能匹配不好提高BM25的权重试试。5.2 模型回答不准确或胡编乱造模型幻觉是另一个高频问题。排查思路检查提示词有没有明确告诉模型“不知道就说不知道”有没有要求引用来源检查上下文检索出来的内容是不是真的相关如果上下文本身就不相关模型只能瞎编。降低temperaturetemperature越高模型越有创造力也越容易胡编。问答场景建议设0.1-0.3。加few-shot示例在提示词里加几个“不知道”的示例模型会更容易学会。5.3 延迟太高怎么优化延迟问题要从链路各环节找环节常见延迟优化手段向量检索10-50ms用GPU加速、减少索引大小关键词检索5-20ms预加载、缓存重排序100-500ms减少候选数量、用更小的模型模型推理500ms-5s用流式输出、换小模型、加缓存流式输出是最有效的体验优化手段。虽然总延迟没变但用户看到第一个字的时间大大缩短感知上快很多。5.4 成本失控怎么控制成本主要来自模型调用。控制手段缓存相同或相似的问题直接返回缓存结果。可以用向量相似度做模糊缓存。模型分级简单问题走小模型。限制上下文长度检索出来的内容不要全塞进去只取最相关的部分。压缩提示词去掉不必要的说明和示例。我踩过的一个坑早期为了效果每次检索top-10每个chunk 500字结果每次请求输入token超过5000成本直接爆炸。后来改成top-3chunk 300字效果几乎没降成本降了70%。5.5 常见问题速查表问题现象可能原因排查方向检索结果不相关切分粒度、embedding模型、查询表述调整chunk_size、换模型、查询改写模型胡编乱造提示词不明确、上下文不相关、temperature高加约束、检查检索、降temperature延迟高模型推理慢、检索慢、重排序慢流式输出、换小模型、减少候选成本高上下文太长、模型太大、无缓存限制上下文、模型分级、加缓存效果不稳定提示词版本混乱、评估集缺失提示词版本管理、建评估集6. 一些个人体会和后续扩展方向这套从零搭建的流程我在几个项目里跑过最大的感受是AI工程的难点不在AI在工程。模型能力是现成的API是现成的但怎么把模型能力稳定、高效、低成本地交付给用户这才是真正要花时间的地方。如果你已经跑通了上面这套流程后续可以往几个方向扩展加工具调用让模型能查数据库、调API处理更复杂的任务。加多轮对话维护对话历史处理指代和上下文。加人工反馈让用户对回答点赞点踩把反馈数据补充到评估集。加可观测性记录每次请求的完整链路方便排查问题。最后分享一个小技巧每次上线新版本之前先跑一遍评估集再找几个真实用户做小范围测试。评估集告诉你“有没有变差”真实用户告诉你“有没有变好”。两者缺一不可。
返回列表