把一个RAG系统从Demo做到生产中间要解决5个问题。最初的版本就是标准版全量文档 embedding向量检索LLM生成。演示没出过问题但是翻车发生在数据留存政策的时候因为系统召回了两段2废弃条款和一段聊员工留存的HR文档然后把这三段内容揉成了一个看似完整实则全错的回答。这不是检索的问题也不纯粹是模型的问题。从分块方式到搜索策略从排序逻辑到异常兜底每一层都藏着独立的故障模式。Level 1Naive RAG文档做 embedding存向量按相似度取 top-k丢给模型生成。流程就这么简单from openai import OpenAI import chromadb client OpenAI() chroma chromadb.Client() collection chroma.create_collection(docs) def index_document(doc_id: str, text: str): response client.embeddings.create( modeltext-embedding-3-small, inputtext ) collection.add( ids[doc_id], embeddings[response.data[0].embedding], documents[text] ) def naive_rag(query: str, k: int 3) - str: # Embed query query_embedding client.embeddings.create( modeltext-embedding-3-small, inputquery ).data[0].embedding # Retrieve results collection.query( query_embeddings[query_embedding], n_resultsk ) # Generate context \n\n.join(results[documents][0]) response client.chat.completions.create( modelgpt-4, messages[ {role: system, content: fAnswer based on this context:\n\n{context}}, {role: user, content: query} ] ) return response.choices[0].message.content所有RAG教程教的就是这套大多数RAG系统也停在了这一步。问题出在哪语义相似度不等于相关性。查data retention policyembedding 模型会把employee retention programs也拉进来因为它看到了词汇上的重叠。两个概念八竿子打不着但向量空间里靠得很近。还有一种情况更隐蔽召回的 chunk 确实跟主题相关但根本没在回答你的问题。三个 chunk 都在聊数据留存可没一个提到你要查的那条具体政策。Demo之所以看着没问题是因为测试用的 query 本身就是你已经知道答案的。Level 2智能分块多数RAG故障看着像检索出了问题实际上是分块出了问题。按固定500 token切一刀会怎样一份政策声明被劈成两半问题在上半截答案在下半截。上下文和结论被强行拆开。切出来的 chunk 单独看根本读不通。分块尺寸这件事比想象中关键得多100–200 tokens太碎chunk缺少语境90天后删除这句话脱离了上下文根本不知道删的是什么1000 tokens又太长一个 chunk 里塞了好几个主题检索的时候噪声和有效信息一把抓300–500 tokens是个比较舒服的区间上下文够用主题又足够聚焦。但尺寸还不是最关键的。重叠overlap才是。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap100, # This is the key separators[\n\n, \n, . , , ] )设100 token的重叠区一个句子即使被切断了两个相邻 chunk 里都有它的完整内容。原本卡在边界上的答案现在从哪一侧都能检索到。还有一个元数据的小技巧不要只存文本本身把来源信息也一起存进去。def chunk_with_metadata(doc: str, source: str, doc_date: str) - list[dict]: chunks splitter.split_text(doc) return [ { text: chunk, source: source, date: doc_date, section: extract_section_header(chunk), } for chunk in chunks ]这样当2019年和2024年的 chunk 同时出现在召回结果里的时候一眼就能看得出来。Prompt 里可以加优先引用最新来源代码里也可以在生成前直接按时间过滤。光是这一步就解决了大约40%的检索故障。垃圾进垃圾出——chunk 质量上去了检索效果自然跟着上去。Level 3混合搜索假设这样一个查询“What’s our PTO policy for employees with 5 years tenure?”语义搜索能找到跟休假政策沾边的 chunk概念上确实接近。关键词搜索能精确命中包含5 years和tenure的 chunk。单独用哪一个都不够。两路合并就可以了。from rank_bm25 import BM25Okapi import numpy as np class HybridRetriever: def __init__(self, documents: list[str]): self.documents documents self.embeddings self._embed_all(documents) # BM25 for keyword matching tokenized [doc.lower().split() for doc in documents] self.bm25 BM25Okapi(tokenized) def _embed_all(self, docs: list[str]) - list[list[float]]: response client.embeddings.create( modeltext-embedding-3-small, inputdocs ) return [d.embedding for d in response.data] def search(self, query: str, k: int 5, alpha: float 0.5) - list[str]: # Semantic scores (normalized) q_emb client.embeddings.create( modeltext-embedding-3-small, inputquery ).data[0].embedding sem_scores np.dot(self.embeddings, q_emb) sem_scores (sem_scores - sem_scores.min()) / (sem_scores.max() - sem_scores.min() 1e-8) # BM25 scores (normalized) bm25_scores np.array(self.bm25.get_scores(query.lower().split())) if bm25_scores.max() 0: bm25_scores bm25_scores / bm25_scores.max() # Combine: alpha controls semantic vs keyword weight combined alpha * sem_scores (1 - alpha) * bm25_scores top_k np.argsort(combined)[::-1][:k] return [self.documents[i] for i in top_k]alpha 的调法如果语料里领域术语多法律、医学、公司内部缩写alpha 调低一些让 BM25 主导如果用户提的是自然语言问题alpha 调高让语义检索权重大一些。初始值设0.5然后看哪些 query 挂了再微调。BM25是很老的技术了也没人再专门为它写博客了。但它能兜住纯向量搜索漏掉的那些 case尤其是用户输入的恰好是文档里的原始表述时。Level 4Reranking检索回来5个 chunk跟主题都沾边。但哪些真正在回答问题Embedding 相似度是单独算的每份文档独立跟 query 打分。Reranker 不一样——它把 query 和文档放在一起看问的是“这份文档是不是在回答这个问题”from sentence_transformers import CrossEncoder class RerankedRetriever: def __init__(self, documents: list[str]): self.hybrid HybridRetriever(documents) self.reranker CrossEncoder(cross-encoder/ms-marco-MiniLM-L-6-v2) def search(self, query: str, k: int 3) - list[str]: # Get 20 candidates (cheap, fast) candidates self.hybrid.search(query, k20) # Rerank with cross-encoder (expensive, accurate) pairs [(query, doc) for doc in candidates] scores self.reranker.predict(pairs) # Return top k after reranking reranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) return [doc for doc, _ in reranked[:k]]Cross-encoder 没办法预先算好文档 embedding必须 query 和文档一起输入。所以拿它做全量检索不现实——一万篇文档逐条打分太慢了。但从20个候选里精选3个这个开销完全可以接受。加入 reranking 之后正确 chunk 出现在前3的命中率从68%提到了89%。其实相关的 chunk 一直被检索到了只是排名不够靠前。不过有一点要清楚reranking 救不了烂检索。如果正确的 chunk 根本不在那20个候选里reranker 也变不出来。先把 Level 2 和 Level 3 做扎实。Level 5生产级RAG前面几个级别都在提升检索质量。生产级RAG要处理的是另一件事检索已经尽力了但还是失败了怎么办因为它一定会失败用户会问文档里根本没覆盖的问题。分块策略会漏掉某个关键段落。或者问题本身就很模糊召回的几个 chunk 互相矛盾。真正该问的不是怎么杜绝检索失败而是检索失败的时候系统该怎么表现。护栏上下文不够的时候别让LLM自己编。Air Canada 在这件事上付出了代价——他们输了一场官司原因是聊天机器人编造了一条根本不存在的退款政策def guarded_rag(query: str, retriever, min_score: float 0.6) - str: results retriever.search_with_scores(query, k3) # Check: Do we have ANY confident results? top_score results[0][1] if results else 0 if top_score min_score: return ( I dont have enough information to answer that confidently. Could you rephrase, or is there a specific document I should look at? ) # Check: Are sources from different time periods? dates [r[date] for r, _ in results] date_warning if len(set(dates)) 1: newest max(dates) if any(d newest for d in dates): date_warning \n\n[Note: Some sources are older. The most recent policy takes precedence.] # Generate with explicit grounding instruction context \n\n---\n\n.join([r[text] for r, _ in results]) response client.chat.completions.create( modelgpt-4, messages[ { role: system, content: fAnswer based ONLY on the provided context. If the context doesnt contain enough information, say so explicitly. Never infer or make up information not directly stated. Context: {context} }, {role: user, content: query} ] ) return response.choices[0].message.content date_warning评估没法度量的东西就没法改进。先建一组测试 query每条都带上已知的正确答案test_cases [ { query: Whats our data retention policy for customer records?, must_retrieve: [data-retention-policy-2024.md], answer_must_contain: [7 years, deletion request], answer_must_not_contain: [2019, employee retention] }, # ... 50 more cases covering your actual use cases ]每次改动跑一遍。追踪检索精度拿到正确文档了吗和答案准确率关键事实对了吗。哪个指标掉了马上能定位到是哪一步出了问题。做到这一步仍然会有边缘 case。用户的表述方式超出预期文档里藏着你不知道的自相矛盾。边缘 case 漏不了。关键是让系统在拿不准的时候老实说不知道而不是胡编一个答案。什么时候该停不是所有场景都需要做到 Level 5。判断该不该升级看用户反馈就行用户在抱怨什么RAG就坏在哪里。从 Level 1 开始。记录并监控系统在哪翻车搞清楚原因之后再往上走。这才是构建一个真正能用的RAG系统的路径。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】