LLM 应用开发：RAG 与知识增强

LLM 应用开发RAG 与知识增强1. 技术分析1.1 RAG 概述RAG (Retrieval-Augmented Generation) 将检索与生成结合RAG 架构 知识库检索 → 生成回答 流程: 1. 问题向量化 2. 检索相关文档 3. 构建提示词 4. LLM 生成回答1.2 RAG 组件组件功能常用工具知识库存储文档FAISS、Pinecone向量化文本转向量Sentence-BERT、OpenAI Embedding检索查询匹配BM25、向量检索生成回答生成GPT-4、Llama-21.3 RAG 类型RAG 类型 Basic RAG: 简单检索生成 Advanced RAG: 多步检索、重排序 Hybrid RAG: 关键词向量混合检索2. 核心功能实现2.1 知识库构建import torch import faiss from sentence_transformers import SentenceTransformer class VectorStore: def __init__(self, model_nameall-MiniLM-L6-v2): self.model SentenceTransformer(model_name) self.index None self.documents [] def add_documents(self, documents): self.documents.extend(documents) embeddings self.model.encode(documents, convert_to_tensorTrue) if self.index is None: dimension embeddings.shape[1] self.index faiss.IndexFlatL2(dimension) self.index.add(embeddings.cpu().numpy()) def search(self, query, k5): query_embedding self.model.encode([query], convert_to_tensorTrue) distances, indices self.index.search(query_embedding.cpu().numpy(), k) results [] for i, idx in enumerate(indices[0]): results.append({ document: self.documents[idx], distance: distances[0][i] }) return results class HybridSearch: def __init__(self): self.vector_store VectorStore() self.bm25_index None def build_bm25(self, documents): from rank_bm25 import BM25Okapi tokenized_docs [doc.split() for doc in documents] self.bm25_index BM25Okapi(tokenized_docs) def search(self, query, k5): vector_results self.vector_store.search(query, k) vector_indices set([self.vector_store.documents.index(r[document]) for r in vector_results]) bm25_scores self.bm25_index.get_scores(query.split()) bm25_results sorted(enumerate(bm25_scores), keylambda x: x[1], reverseTrue)[:k] bm25_indices set([idx for idx, _ in bm25_results]) combined_indices vector_indices.union(bm25_indices) results [] for idx in combined_indices: results.append({ document: self.vector_store.documents[idx], source: both if idx in vector_indices and idx in bm25_indices else vector if idx in vector_indices else bm25 }) return results[:k]2.2 RAG 流水线class RAGPipeline: def __init__(self, llm, vector_store): self.llm llm self.vector_store vector_store def generate_prompt(self, query, documents): context \n\n.join([f文档{i1}: {doc[document]} for i, doc in enumerate(documents)]) prompt f基于以下文档回答问题 {context} 问题{query} 请根据文档内容给出详细的回答。如果文档中没有相关信息请说明。 return prompt def answer(self, query): documents self.vector_store.search(query) prompt self.generate_prompt(query, documents) answer self.llm.generate(prompt) return answer, documents class AdvancedRAG: def __init__(self, llm, vector_store): self.llm llm self.vector_store vector_store def refine_query(self, query): refinement_prompt f将以下问题优化为更适合检索的形式 原始问题{query} 优化后的问题 refined self.llm.generate(refinement_prompt) return refined.strip() def rerank(self, query, documents): rerank_prompt f根据与问题的相关性对以下文档进行排序 问题{query} 文档列表 {chr(10).join([f{i1}. {doc[document][:100]}... for i, doc in enumerate(documents)])} 请返回排序后的文档编号用逗号分隔 ranking self.llm.generate(rerank_prompt) indices [int(i.strip()) - 1 for i in ranking.split(,)] return [documents[i] for i in indices if 0 i len(documents)] def answer(self, query): refined_query self.refine_query(query) documents self.vector_store.search(refined_query) reranked_docs self.rerank(query, documents) prompt self._generate_prompt(query, reranked_docs) answer self.llm.generate(prompt) return answer, reranked_docs2.3 文档处理class DocumentProcessor: def __init__(self): pass def load_pdf(self, path): from PyPDF2 import PdfReader reader PdfReader(path) text for page in reader.pages: text page.extract_text() \n return text def split_text(self, text, chunk_size512, overlap100): chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) chunks.append(text[start:end]) if end len(text): break start end - overlap return chunks def process_document(self, path): text self.load_pdf(path) chunks self.split_text(text) return chunks class KnowledgeBaseManager: def __init__(self): self.vector_store VectorStore() self.processor DocumentProcessor() def add_document(self, path): chunks self.processor.process_document(path) self.vector_store.add_documents(chunks) def add_directory(self, directory): import os for filename in os.listdir(directory): if filename.endswith(.pdf): path os.path.join(directory, filename) self.add_document(path)3. 性能对比3.1 检索方法对比方法速度准确性适用场景BM25快中关键词匹配Vector中高语义匹配Hybrid中很高综合场景3.2 向量化模型对比模型维度速度准确性all-MiniLM-L6-v2384快中all-mpnet-base-v2768中高text-embedding-ada-0021536中很高3.3 RAG 效果对比场景纯LLMBasic RAGAdvanced RAG事实问答70%85%90%技术文档60%80%88%多文档推理50%75%85%4. 最佳实践4.1 RAG 配置选择def select_rag_config(use_case): configs { fact_qa: { embedding_model: all-mpnet-base-v2, chunk_size: 512, overlap: 100, top_k: 3 }, tech_docs: { embedding_model: text-embedding-ada-002, chunk_size: 1024, overlap: 200, top_k: 5 }, general: { embedding_model: all-MiniLM-L6-v2, chunk_size: 512, overlap: 50, top_k: 3 } } return configs.get(use_case, configs[general]) class RAGConfigGenerator: staticmethod def from_use_case(use_case): return select_rag_config(use_case)4.2 RAG 部署class RAGDeployer: def __init__(self, config): self.config config def build(self): vector_store VectorStore(self.config[embedding_model]) llm self._load_llm() kb_manager KnowledgeBaseManager() kb_manager.vector_store vector_store kb_manager.add_directory(self.config[documents_dir]) pipeline RAGPipeline(llm, vector_store) return pipeline def _load_llm(self): from transformers import AutoModelForCausalLM, AutoTokenizer model AutoModelForCausalLM.from_pretrained(self.config[llm_model]) tokenizer AutoTokenizer.from_pretrained(self.config[llm_model]) return LLMInference(model, tokenizer)5. 总结RAG 是 LLM 应用开发的重要范式知识库构建文档向量化存储检索找到相关信息生成基于检索结果回答优化查询优化、重排序对比数据如下Advanced RAG 比 Basic RAG 提升 5-10% 准确率Hybrid 检索结合了关键词和语义的优势推荐使用 512-1024 token 的 chunk 大小向量数据库选择影响检索速度和成本