wordpress安装路径和站点地址的设置,东莞麻涌网站建设,厦门网站建设公,网站开发属于什么资产本文深入解析了RAG#xff08;检索增强生成#xff09;技术#xff0c;旨在解决大语言模型“一本正经胡说八道”的幻觉问题。首先分析了LLM幻觉的三大根源#xff0c;并提供了七种解决方案对比。接着#xff0c;详细介绍了四大向量数据库的实战配置#xff0c;包括FAISS、…本文深入解析了RAG检索增强生成技术旨在解决大语言模型“一本正经胡说八道”的幻觉问题。首先分析了LLM幻觉的三大根源并提供了七种解决方案对比。接着详细介绍了四大向量数据库的实战配置包括FAISS、Pinecone、腾讯云VectorDB和Weaviate并提供了选型指南。此外还深入解析了Embedding模型、文档加载与智能分割、检索器核心原理、中文分词与关键词提取、重排序Rerank原理与实践等关键内容。最后分享了RAG优化策略大全包括多查询融合、问题分解、混合检索、逻辑路由、父文档检索和CRAG自我纠正RAG等策略并展示了完整的RAG系统架构和快速开始模板。万字长文一次性掌握RAG核心技术大语言模型“一本正经胡说八道”的毛病终于有解了这就是今天要讲的RAG检索增强生成技术。它能让AI的回答有据可查避免幻觉真正成为你的智能助手。 第一关搞懂LLM为什么会“幻觉”幻觉的三大根源训练数据偏差模型只在训练数据中出现过的知识上表现可靠参数记忆限制知识被压缩在权重中细节容易丢失概率生成本质每次都是“猜”下一个词不是“回忆”事实7大解决方案对比方案原理适用场景复杂度提示工程在Prompt中明确要求“不要编造”简单问答★☆☆Few-shot示例提供正确回答示例格式固定场景★★☆知识检索实时查外部知识库动态知识问答★★★思维链推理让模型展示推理过程复杂问题★★★自我验证让模型检查自己回答高准确性要求★★★不确定性量化输出置信度分数风险控制★★★集成方法多个模型投票关键决策★★★★# 实战减少幻觉的Prompt模板 anti_hallucination_prompt 请你基于提供的知识回答问题遵循以下规则 1. 只使用提供的参考信息 2. 如果信息不足明确说“根据已有信息无法确定” 3. 不要添加额外信息 4. 如果信息冲突说明冲突点 参考信息{context} 问题{question} 回答 ️ 第二关四大向量数据库实战配置FAISS - Facebook开源本地首选# 安装pip install faiss-cpu import faiss import numpy as np # 创建索引 dimension 768 # embedding维度 index faiss.IndexFlatL2(dimension) # 添加向量 embeddings np.random.rand(1000, dimension).astype(float32) index.add(embeddings) # 搜索 query np.random.rand(1, dimension).astype(float32) k 5 distances, indices index.search(query, k) print(f找到最相似的 {k} 个文档索引{indices})FAISS进阶用法# IVF索引加速 nlist 100 # 聚类中心数 quantizer faiss.IndexFlatL2(dimension) index faiss.IndexIVFFlat(quantizer, dimension, nlist) index.train(embeddings) # 先训练 index.add(embeddings) # 保存和加载 faiss.write_index(index, my_index.faiss) loaded_index faiss.read_index(my_index.faiss)Pinecone - 云端托管简单易用# 安装pip install pinecone-client import pinecone from langchain.vectorstores import Pinecone from langchain.embeddings import OpenAIEmbeddings # 初始化 pinecone.init(api_keyyour-api-key, environmentus-west1-gcp) # 创建索引 index_name my-knowledge-base if index_name not in pinecone.list_indexes(): pinecone.create_index( nameindex_name, dimension1536, # text-embedding-ada-002维度 metriccosine ) # 连接索引 index pinecone.Index(index_name) # 通过LangChain使用 embeddings OpenAIEmbeddings() vectorstore Pinecone(index, embeddings, text)腾讯云VectorDB - 国产新星from tcvectordb.model.enum import FieldType, IndexType from tcvectordb.model.index import VectorIndex, FilterIndex from tcvectordb.model.document import Document, SearchParams from tcvectordb.model.collection import Embedding # 创建连接 client VectorDBClient( urlhttp://xxxxx.xxxxx.xxxxx.tencentcloudapi.com, usernameroot, keyyour-key, timeout30 ) # 创建集合 db client.database(db-test) coll db.create_collection( namemy-collection, shard1, replicas0, description知识库文档, indexIndex( VectorIndex( vector, 768, IndexType.HNSW, MetricType.COSINE ), FilterIndex(id, FieldType.String, IndexType.PRIMARY_KEY) ) )Weaviate - 开源图向量混合import weaviate from weaviate.classes.init import AdditionalConfig, Timeout import weaviate.classes as wvc # 客户端配置 client weaviate.WeaviateClient( additional_configAdditionalConfig( timeoutTimeout(init2, query45, insert120) # 超时设置 ) ) # 创建集合 client.collections.create( nameDocument, vectorizer_configwvc.config.Configure.Vectorizer.text2vec_openai(), properties[ wvc.config.Property( nametitle, data_typewvc.config.DataType.TEXT ), wvc.config.Property( namecontent, data_typewvc.config.DataType.TEXT ) ] ) 向量数据库选型指南维度FAISSPinecone腾讯云VectorDBWeaviate部署方式本地全托管全托管自托管/云托管开源✅❌❌✅最大数据量内存限制无限制PB级无限制多模态❌✅✅✅成本免费按用量按用量免费/自建成本适合场景中小规模企业级国内企业图向量混合 第三关Embedding模型深入解析文本向量化原理from sentence_transformers import SentenceTransformer # 加载模型 model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) # 生成向量 sentences [我爱编程, Python是最好的语言, 机器学习很有趣] embeddings model.encode(sentences) print(f向量维度{embeddings.shape}) # (3, 384) # 计算相似度 from sklearn.metrics.pairwise import cosine_similarity similarity cosine_similarity( [embeddings[0]], # 我爱编程 [embeddings[1]] # Python是最好的语言 ) print(f语义相似度{similarity[0][0]:.3f})主流Embedding模型对比embedding_models { OpenAI text-embedding-3: { dimension: 1536, 价格: $0.13/1M tokens, 特点: 效果最好英文优化 }, BGE-M3: { dimension: 1024, 价格: 免费开源, 特点: 多语言检索SOTA }, Jina Embeddings: { dimension: 768, 价格: 免费开源, 特点: 中文优化8K上下文 }, 阿里通义千问: { dimension: 1024, 价格: API调用, 特点: 中文场景优化 } } 第四关文档加载与智能分割LangChain文档加载器大全from langchain.document_loaders import ( TextLoader, # 文本文件 PyPDFLoader, # PDF UnstructuredWordDocumentLoader, # Word CSVLoader, # CSV UnstructuredHTMLLoader, # HTML JSONLoader, # JSON SeleniumURLLoader, # 动态网页 YoutubeLoader, # YouTube视频 GitLoader, # Git仓库 NotionDirectoryLoader # Notion导出 ) # 加载PDF loader PyPDFLoader(docs/report.pdf) pages loader.load() print(f加载了 {len(pages)} 页PDF) # 加载网页 loader SeleniumURLLoader([https://example.com]) docs loader.load() # 批量加载目录 from langchain.document_loaders import DirectoryLoader loader DirectoryLoader( ./docs, glob**/*.pdf, loader_clsPyPDFLoader ) documents loader.load()递归字符分割器 - 深入源码from langchain.text_splitter import RecursiveCharacterTextSplitter # 核心分割逻辑 def split_text(self, text: str) - List[str]: 递归分割文本的源码逻辑 final_chunks [] # 按分隔符优先级分割 separators self._separators if self._separators is not None else [\n\n, \n, , ] for separator in separators: if separator: splits text.split(separator) else: splits list(text) # 合并小片段 good_splits [] current_chunk for s in splits: if len(current_chunk) len(s) self._chunk_size: current_chunk s else: if current_chunk: good_splits.append(current_chunk) current_chunk s if current_chunk: good_splits.append(current_chunk) if len(good_splits) 1: final_chunks.extend(good_splits) break return final_chunks智能分割策略# 按语义分割 from langchain.text_splitter import SemanticChunker from langchain.embeddings import OpenAIEmbeddings text_splitter SemanticChunker( embeddingsOpenAIEmbeddings(), breakpoint_threshold_typepercentile # 按分位数分割 ) # 按Markdown结构分割 from langchain.text_splitter import MarkdownTextSplitter markdown_splitter MarkdownTextSplitter( chunk_size1000, chunk_overlap200 ) # 按代码结构分割 from langchain.text_splitter import ( Language, RecursiveCharacterTextSplitter ) python_splitter RecursiveCharacterTextSplitter.from_language( languageLanguage.PYTHON, chunk_size2000, chunk_overlap200 ) 第五关检索器核心原理基础检索器实现from langchain.vectorstores import FAISS from langchain.embeddings import OpenAIEmbeddings from langchain.retrievers import BM25Retriever, EnsembleRetriever class SmartRetriever: def __init__(self, vectorstore, bm25_retriever): # 混合检索器 self.ensemble_retriever EnsembleRetriever( retrievers[vectorstore.as_retriever(), bm25_retriever], weights[0.7, 0.3] # 向量检索70%关键词检索30% ) def retrieve(self, query, k5, score_threshold0.7): 智能检索文档 # 1. 并行检索 vector_results self.vector_retriever.get_relevant_documents(query) bm25_results self.bm25_retriever.get_relevant_documents(query) # 2. 结果融合 all_results self._merge_results(vector_results, bm25_results) # 3. 去重 unique_results self._deduplicate(all_results) # 4. 重排序 reranked self._rerank(query, unique_results) # 5. 分数过滤 filtered [doc for doc in reranked if doc.metadata.get(score, 0) score_threshold] return filtered[:k] def _merge_results(self, vec_results, bm25_results): 合并两种检索结果 # 使用RRF倒数排名融合算法 fused_scores {} for i, doc in enumerate(vec_results): doc_id doc.metadata.get(id, str(i)) fused_scores[doc_id] fused_scores.get(doc_id, 0) 1/(60 i) for i, doc in enumerate(bm25_results): doc_id doc.metadata.get(id, str(i)) fused_scores[doc_id] fused_scores.get(doc_id, 0) 1/(60 i) # 按融合分数排序 return sorted(vec_results bm25_results, keylambda x: fused_scores.get( x.metadata.get(id, 0), 0), reverseTrue)✂️ 第六关中文分词与关键词提取import jieba import jieba.analyse from collections import Counter class ChineseTextProcessor: def __init__(self, user_dict_pathNone): if user_dict_path: jieba.load_userdict(user_dict_path) # 停用词 self.stopwords set([ 的, 了, 在, 是, 我, 有, 和, 就, 不, 人, 都, 一, 一个, 上, 也, 很, 到, 说, 要, 去, 你, 会, 着, 没有, 看, 好, 自己, 这 ]) def extract_keywords(self, text, top_k10, with_weightTrue): 提取关键词 # TF-IDF算法 keywords jieba.analyse.extract_tags( text, topKtop_k, withWeightwith_weight, allowPOS(n, vn, v, ns, nr) # 名词、动名词、动词、地名、人名 ) # TextRank算法 textrank_keywords jieba.analyse.textrank( text, topKtop_k, withWeightwith_weight, allowPOS(n, vn, v) ) # 融合结果 combined self._combine_keywords(keywords, textrank_keywords) return combined def segment_with_pos(self, text): 分词词性标注 words jieba.posseg.cut(text) return [(word, flag) for word, flag in words] def _combine_keywords(self, tfidf_kws, textrank_kws): 融合TF-IDF和TextRank结果 keyword_scores {} # 加权融合 for word, score in tfidf_kws: keyword_scores[word] keyword_scores.get(word, 0) score * 0.6 for word, score in textrank_kws: keyword_scores[word] keyword_scores.get(word, 0) score * 0.4 return sorted(keyword_scores.items(), keylambda x: x[1], reverseTrue) # 使用示例 processor ChineseTextProcessor() text 人工智能是计算机科学的一个分支它企图了解智能的实质... keywords processor.extract_keywords(text, top_k5) print(提取的关键词, keywords) # 输出[(人工智能, 1.0), (计算机科学, 0.8), (智能, 0.6), ...] 第七关重排序Rerank原理与实践Cross-Encoder重排序from sentence_transformers import CrossEncoder class Reranker: def __init__(self, model_nameBAAI/bge-reranker-large): self.model CrossEncoder(model_name) def rerank(self, query, documents, top_k5): 对检索结果重排序 # 构造query-doc对 pairs [[query, doc.page_content] for doc in documents] # 计算相关性分数 scores self.model.predict(pairs) # 组合结果 reranked_results [] for i, (doc, score) in enumerate(zip(documents, scores)): doc.metadata[rerank_score] float(score) reranked_results.append((score, doc)) # 按分数排序 reranked_results.sort(keylambda x: x[0], reverseTrue) return [doc for _, doc in reranked_results[:top_k]] # 使用示例 reranker Reranker() initial_results vectorstore.similarity_search(query, k20) final_results reranker.rerank(query, initial_results, top_k5)多阶段重排序策略class MultiStageReranker: def __init__(self): # 第一层快速粗排 self.fast_ranker CrossEncoder(cross-encoder/ms-marco-MiniLM-L-6-v2) # 第二层精准精排 self.precise_ranker CrossEncoder(BAAI/bge-reranker-large) # 第三层LLM判断 self.llm_ranker OpenAI(modelgpt-3.5-turbo) def rerank(self, query, documents): 三级重排序 # 第一阶段快速过滤前50-20 stage1 self._fast_rerank(query, documents[:50], keep20) # 第二阶段精确排序20-10 stage2 self._precise_rerank(query, stage1, keep10) # 第三阶段LLM验证10-5 stage3 self._llm_rerank(query, stage2, keep5) return stage3 def _fast_rerank(self, query, docs, keep20): 快速粗排 pairs [[query, doc.page_content[:500]] for doc in docs] scores self.fast_ranker.predict(pairs, batch_size32) return self._select_top_k(docs, scores, keep) def _llm_rerank(self, query, docs, keep5): LLM智能判断 prompt f请判断以下文档与问题的相关性给出0-10分 问题{query} 请按以下格式输出 文档1: 分数 文档2: 分数 ... 文档列表 {self._format_docs(docs)} response self.llm_ranker.predict(prompt) scores self._parse_llm_scores(response) return self._select_top_k(docs, scores, keep) 第八关RAG优化策略大全策略1多查询融合def multi_query_expansion(original_query, llm, n3): 生成多个相关问题 prompt f基于以下问题生成 {n} 个不同角度的相关问题 原始问题{original_query} 生成的问题应该 1. 从不同角度理解原问题 2. 使用同义词替换 3. 分解复杂问题 4. 用不同方式提问 输出格式每行一个问题 response llm.predict(prompt) queries [original_query] [q.strip() for q in response.split(\n) if q.strip()] return queries[:n1] # 并行检索多个查询 expanded_queries multi_query_expansion(original_query, llm, n3) all_results [] for q in expanded_queries: results vectorstore.similarity_search(q, k3) all_results.extend(results)策略2问题分解def query_decomposition(complex_query, llm): 分解复杂问题 prompt f将复杂问题分解为多个子问题 复杂问题{complex_query} 请分解为可以独立检索和回答的子问题。 输出格式 1. 子问题1 2. 子问题2 ... response llm.predict(prompt) sub_queries [q.strip() for q in response.split(\n) if q.strip()] return sub_queries # 分别回答子问题然后整合 sub_queries query_decomposition(complex_query, llm) sub_answers [] for sub_q in sub_queries: context retrieve_context(sub_q) answer answer_question(sub_q, context) sub_answers.append(answer) final_answer integrate_answers(sub_answers, complex_query, llm)策略3混合检索class HybridRetriever: def __init__(self, vector_store, keyword_retriever): self.vector_store vector_store self.keyword_retriever keyword_retriever def hybrid_search(self, query, alpha0.7, k10): 混合检索 # 向量检索 vector_results self.vector_store.similarity_search_with_score(query, kk*2) # 关键词检索 keyword_results self.keyword_retriever.get_relevant_documents(query) # 归一化分数 vector_scores self._normalize_scores([score for _, score in vector_results]) keyword_scores self._normalize_scores( [doc.metadata.get(bm25_score, 0) for doc in keyword_results] ) # 融合结果 fused_results {} # 融合向量检索结果 for (doc, score), norm_score in zip(vector_results, vector_scores): doc_id doc.metadata.get(id, hash(doc.page_content)) fused_results[doc_id] { doc: doc, score: alpha * norm_score (1-alpha) * fused_results.get(doc_id, {}).get(score, 0) } # 融合关键词检索结果 for doc, norm_score in zip(keyword_results, keyword_scores): doc_id doc.metadata.get(id, hash(doc.page_content)) if doc_id in fused_results: fused_results[doc_id][score] (1-alpha) * norm_score else: fused_results[doc_id] { doc: doc, score: (1-alpha) * norm_score } # 排序返回 sorted_results sorted( fused_results.values(), keylambda x: x[score], reverseTrue ) return [item[doc] for item in sorted_results[:k]]策略4逻辑路由class LogicalRouter: def __init__(self, llm): self.llm llm def route_query(self, query, chat_history): 路由查询到不同处理流程 prompt f判断问题类型并选择处理策略 问题{query} 对话历史{chat_history} 可选的策略 1. vector_rag - 需要检索知识库的问题 2. direct_llm - 通用知识或逻辑推理 3. calculator - 数学计算 4. search_web - 需要实时信息 5. clarification - 需要澄清的问题 请只返回策略名称 response self.llm.predict(prompt).strip() routing_rules { vector_rag: self.handle_vector_rag, direct_llm: self.handle_direct_llm, calculator: self.handle_calculator, search_web: self.handle_search_web, clarification: self.handle_clarification } return routing_rules.get(response, self.handle_vector_rag)策略5父文档检索class ParentDocumentRetriever: def __init__(self, vector_store, parent_docs): self.vector_store vector_store self.parent_docs parent_docs # 父文档映射 def retrieve_with_parent(self, query, k5): 检索子文档返回父文档 # 检索子文档 child_docs self.vector_store.similarity_search(query, kk*3) # 获取父文档 parent_ids set() parent_docs [] for doc in child_docs: parent_id doc.metadata.get(parent_id) if parent_id and parent_id not in parent_ids: parent_docs.append(self.parent_docs[parent_id]) parent_ids.add(parent_id) if len(parent_docs) k: break return parent_docs策略6CRAG自我纠正RAGclass CRAGSystem: def __init__(self, retriever, llm, judge_llm): self.retriever retriever self.llm llm self.judge_llm judge_llm def answer_with_correction(self, query): 带自我纠正的RAG # 1. 初步检索 docs self.retriever.retrieve(query) # 2. 生成初步答案 context \n\n.join([doc.page_content for doc in docs]) answer self.llm.predict(f基于以下信息\n{context}\n\n回答{query}) # 3. 置信度评估 confidence self._evaluate_confidence(query, answer, docs) if confidence 0.7: # 低置信度 # 4. 扩大检索 expanded_docs self._expand_retrieval(query, docs) # 5. 重新生成 new_context \n\n.join([doc.page_content for doc in expanded_docs]) corrected_answer self.llm.predict(f基于以下更全信息\n{new_context}\n\n重新回答{query}) # 6. 验证改进 improvement self._verify_improvement( query, answer, corrected_answer, docs, expanded_docs ) if improvement: return corrected_answer, expanded_docs return answer, docs 完整RAG系统架构class ProductionRAGSystem: def __init__(self, config): # 1. 初始化组件 self.embeddings self._init_embeddings(config) self.vector_store self._init_vector_store(config) self.llm self._init_llm(config) self.reranker Reranker() self.query_analyzer QueryAnalyzer() # 2. 多检索器 self.retrievers { vector: VectorRetriever(self.vector_store), keyword: KeywordRetriever(), hybrid: HybridRetriever(self.vector_store, self.keyword_retriever) } # 3. 缓存 self.cache RedisCache() def answer_question(self, query, session_id, use_cacheTrue): 完整问答流程 # 检查缓存 if use_cache: cached self.cache.get(query, session_id) if cached: return cached # 1. 查询分析 query_type, decomposed self.query_analyzer.analyze(query) # 2. 智能路由 retriever_type self._route_retriever(query_type) retriever self.retrievers[retriever_type] # 3. 多策略检索 all_docs [] for sub_query in decomposed: docs retriever.retrieve(sub_query, k10) all_docs.extend(docs) # 4. 去重和重排序 unique_docs self._deduplicate(all_docs) reranked_docs self.reranker.rerank(query, unique_docs, top_k8) # 5. 上下文构建 context self._construct_context(reranked_docs, query) # 6. 生成答案 answer self._generate_answer(query, context) # 7. 引用溯源 answer_with_citations self._add_citations(answer, reranked_docs) # 8. 缓存结果 self.cache.set(query, session_id, answer_with_citations) return answer_with_citations 快速开始模板# requirements.txt langchain0.1.0 langchain-openai0.0.1 sentence-transformers2.2.2 faiss-cpu1.7.4 rank-bm250.2.1 jieba0.42.1 redis4.5.4 # 快速启动 from rag_system import ProductionRAGSystem config { embedding_model: BAAI/bge-large-zh, llm_model: gpt-4, vector_store: faiss, cache: redis } rag ProductionRAGSystem(config) # 问答 answer rag.answer_question( query什么是机器学习, session_iduser_123, use_cacheTrue ) print(answer) 性能监控与优化class RAGMonitor: def __init__(self): self.metrics { retrieval_time: [], generation_time: [], cache_hit_rate: 0, accuracy_scores: [] } def log_retrieval(self, query, docs, retrieval_time): 记录检索指标 self.metrics[retrieval_time].append(retrieval_time) # 计算检索相关性 relevance self._calculate_relevance(query, docs) self.metrics[accuracy_scores].append(relevance) def get_performance_report(self): 生成性能报告 return { avg_retrieval_time: np.mean(self.metrics[retrieval_time]), avg_generation_time: np.mean(self.metrics[generation_time]), cache_hit_rate: self.metrics[cache_hit_rate], avg_accuracy: np.mean(self.metrics[accuracy_scores]), total_queries: len(self.metrics[retrieval_time]) } 核心要点总结幻觉问题通过RAG从根本上解决让AI回答有据可查向量数据库根据场景选择小项目用FAISS大项目用Pinecone/腾讯云文档处理合理分割是关键递归分割器语义分割结合使用检索优化混合检索重排序是效果提升的关键中文优化使用BGE、Jina等中文优化模型RAG进阶多查询、问题分解、逻辑路由等策略组合使用假如你从2026年开始学大模型按这个步骤走准能稳步进阶。接下来告诉你一条最快的邪修路线3个月即可成为模型大师薪资直接起飞。阶段1:大模型基础阶段2:RAG应用开发工程阶段3:大模型Agent应用架构阶段4:大模型微调与私有化部署配套文档资源全套AI 大模型 学习资料朋友们如果需要可以微信扫描下方二维码免费领取【保证100%免费】配套文档资源全套AI 大模型 学习资料朋友们如果需要可以微信扫描下方二维码免费领取【保证100%免费】