这次我们来看一个 RAG 实战项目重点不是讲概念而是直接动手搭建一个完整的检索增强生成系统。RAGRetrieval-Augmented Generation技术现在在企业知识库、智能问答、文档分析等场景应用广泛但很多教程只讲理论缺乏可落地的实操指导。本文基于RAG实战第4课从 Retrieval 到 Generation主题将带大家完整走通从文档处理、向量检索到文本生成的整个流程。我们会用到主流的 Embedding 模型、Chroma 向量数据库以及实际可运行的代码示例。无论你是想搭建企业知识库还是需要为 LLM 增加外部知识源这篇文章都能提供直接的参考价值。最值得关注的是这个实战方案可以在普通开发环境下运行不需要高端显卡CPU 环境也能正常工作。我们会重点测试检索准确性、生成质量以及如何优化整个流程的性能。1. 核心能力速览能力项说明技术栈RAG检索增强生成核心组件Embedding 模型 向量数据库 LLM硬件需求CPU 可运行GPU 可加速 Embedding向量数据库Chroma轻量级无需单独部署Embedding 模型BGE、Sentence-BERT 等可选LLM 接入支持 OpenAI API、本地模型等多种方式主要功能文档检索、知识问答、内容生成适合场景企业知识库、智能客服、文档分析2. 适用场景与使用边界RAG 技术特别适合需要结合外部知识库的应用场景。比如企业内部的规章制度查询、产品文档检索、技术知识问答等。通过 RAG可以让大语言模型在回答问题时参考具体的文档内容避免凭空编造信息。但需要注意几个边界首先RAG 的效果严重依赖检索质量如果文档质量差或者检索不准生成的内容也会有问题。其次涉及敏感信息时要注意数据安全确保检索的文档都是经过审核的。最后RAG 不能完全解决幻觉问题还需要结合其他校验机制。对于企业使用建议先在小范围测试验证检索准确性和生成质量再逐步扩大应用范围。3. 环境准备与前置条件开始之前需要准备好以下环境操作系统要求Windows 10/11、macOS 10.15 或 Linux Ubuntu 18.04建议使用 Linux 或 macOS 以获得更好的兼容性Python 环境Python 3.8-3.11版本推荐使用 conda 或 venv 创建虚拟环境硬件要求内存至少 8GB推荐 16GB 以上存储至少 10GB 可用空间用于模型文件GPU可选如有 GPU 可加速 Embedding 计算网络要求需要能访问 Hugging Face 下载模型如果使用 OpenAI API需要能够访问相应服务4. 安装部署与启动方式首先创建并激活 Python 虚拟环境# 创建虚拟环境 python -m venv rag_env source rag_env/bin/activate # Linux/macOS # 或 rag_env\Scripts\activate # Windows # 安装核心依赖 pip install chromadb sentence-transformers openai如果需要使用本地 LLM可以额外安装pip install transformers torch接下来创建项目目录结构mkdir rag_project cd rag_project mkdir -p data/documents data/embeddings scripts5. 文档处理与向量化RAG 的第一步是将文档转换为向量并存入数据库。我们使用 Chroma 作为向量数据库它轻量且易于使用。5.1 文档加载与预处理创建文档处理脚本scripts/document_processor.pyimport os import re from typing import List, Dict import chromadb from sentence_transformers import SentenceTransformer class DocumentProcessor: def __init__(self, persist_directory: str ./data/embeddings): self.client chromadb.PersistentClient(pathpersist_directory) self.collection self.client.get_or_create_collection(documents) self.embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) def chunk_documents(self, text: str, chunk_size: int 500, chunk_overlap: int 50) - List[str]: 将长文本分割为小块 chunks [] words text.split() for i in range(0, len(words), chunk_size - chunk_overlap): chunk .join(words[i:i chunk_size]) chunks.append(chunk) if i chunk_size len(words): break return chunks def process_documents(self, document_dir: str): 处理目录中的所有文档 documents [] metadatas [] ids [] for filename in os.listdir(document_dir): if filename.endswith(.txt): filepath os.path.join(document_dir, filename) with open(filepath, r, encodingutf-8) as f: content f.read() chunks self.chunk_documents(content) for i, chunk in enumerate(chunks): documents.append(chunk) metadatas.append({source: filename, chunk_id: i}) ids.append(f{filename}_{i}) # 生成嵌入向量 embeddings self.embedding_model.encode(documents).tolist() # 存入向量数据库 self.collection.add( embeddingsembeddings, documentsdocuments, metadatasmetadatas, idsids ) print(f成功处理 {len(documents)} 个文档块) if __name__ __main__: processor DocumentProcessor() processor.process_documents(./data/documents)5.2 准备测试文档在data/documents目录下创建示例文档example.txtRAG技术通过结合检索和生成两个阶段来提升大语言模型的效果。 在检索阶段系统从知识库中查找与问题相关的文档片段。 在生成阶段大语言模型基于检索到的内容生成最终答案。 这种方法可以有效减少模型幻觉提高回答的准确性。运行文档处理python scripts/document_processor.py6. 检索系统实现检索是 RAG 的核心环节直接影响到最终生成质量。6.1 基础检索器实现创建scripts/retriever.pyimport chromadb from sentence_transformers import SentenceTransformer from typing import List, Dict class Retriever: def __init__(self, persist_directory: str ./data/embeddings): self.client chromadb.PersistentClient(pathpersist_directory) self.collection self.client.get_collection(documents) self.embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) def retrieve(self, query: str, n_results: int 3) - List[Dict]: 检索相关文档 # 生成查询向量 query_embedding self.embedding_model.encode([query]).tolist() # 执行检索 results self.collection.query( query_embeddingsquery_embedding, n_resultsn_results ) retrieved_docs [] for i in range(len(results[documents][0])): retrieved_docs.append({ content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] }) return retrieved_docs def hybrid_retrieve(self, query: str, n_results: int 3) - List[Dict]: 混合检索结合向量检索和关键词检索 # 向量检索 vector_results self.retrieve(query, n_results) # 这里可以添加关键词检索逻辑 # 实际项目中可以使用 BM25 等算法 return vector_results if __name__ __main__: retriever Retriever() results retriever.retrieve(RAG技术如何工作) for result in results: print(f内容: {result[content]}) print(f距离: {result[distance]}) print(---)6.2 检索质量测试测试不同查询的检索效果# 测试检索效果 test_queries [ 什么是RAG技术, 检索阶段的作用是什么, 如何减少模型幻觉 ] retriever Retriever() for query in test_queries: print(f查询: {query}) results retriever.retrieve(query) for i, result in enumerate(results): print(f{i1}. {result[content][:100]}...) print(\n *50 \n)7. 生成阶段集成检索到相关文档后需要将其与问题结合交给 LLM 生成最终答案。7.1 基于 OpenAI API 的生成器创建scripts/generator.pyimport openai from typing import List, Dict import os class OpenAIGenerator: def __init__(self, api_key: str None): self.api_key api_key or os.getenv(OPENAI_API_KEY) if self.api_key: openai.api_key self.api_key def generate(self, query: str, contexts: List[str], model: str gpt-3.5-turbo) - str: 基于检索结果生成答案 # 构建上下文 context_text \n\n.join([f参考文档 {i1}: {context} for i, context in enumerate(contexts)]) prompt f基于以下参考文档回答问题。如果文档中没有相关信息请明确说明。 参考文档 {context_text} 问题{query} 请根据上述文档内容回答问题 try: response openai.ChatCompletion.create( modelmodel, messages[ {role: system, content: 你是一个有帮助的助手基于提供的文档内容回答问题。}, {role: user, content: prompt} ], temperature0.3, max_tokens500 ) return response.choices[0].message.content except Exception as e: return f生成失败: {str(e)} class LocalGenerator: 本地模型生成器示例 def __init__(self, model_path: str None): # 这里可以集成本地模型如 ChatGLM、Qwen 等 self.model None # 实际项目中加载本地模型 def generate(self, query: str, contexts: List[str]) - str: # 本地模型生成逻辑 context_text \n.join(contexts) prompt f基于以下信息回答问题\n{context_text}\n\n问题{query}\n答案 # 这里调用本地模型生成 # 返回生成的答案 return 本地模型生成示例需要实际模型支持 if __name__ __main__: # 使用示例 generator OpenAIGenerator() # 模拟检索结果 test_contexts [ RAG技术通过检索和生成结合来提升效果。, 检索阶段从知识库查找相关文档。, 生成阶段基于检索内容产生答案。 ] answer generator.generate(RAG如何工作, test_contexts) print(生成结果:, answer)7.2 完整的 RAG 流水线创建完整的 RAG 系统scripts/rag_pipeline.pyfrom retriever import Retriever from generator import OpenAIGenerator import os class RAGPipeline: def __init__(self, use_local: bool False): self.retriever Retriever() if use_local: self.generator LocalGenerator() else: self.generator OpenAIGenerator() def query(self, question: str, n_retrieve: int 3) - dict: 完整 RAG 流程 # 1. 检索阶段 retrieved_docs self.retriever.retrieve(question, n_retrieve) contexts [doc[content] for doc in retrieved_docs] # 2. 生成阶段 answer self.generator.generate(question, contexts) return { question: question, retrieved_contexts: retrieved_docs, answer: answer } def batch_query(self, questions: list) - list: 批量处理问题 results [] for question in questions: result self.query(question) results.append(result) return results if __name__ __main__: # 初始化 RAG 系统 rag RAGPipeline() # 测试查询 test_questions [ RAG技术的主要优势是什么, 检索阶段在RAG中起什么作用, 如何评估RAG系统的效果 ] for question in test_questions: result rag.query(question) print(f问题: {result[question]}) print(f答案: {result[answer]}) print(检索到的文档:) for i, context in enumerate(result[retrieved_contexts]): print(f{i1}. {context[content][:100]}...) print(\n *80 \n)8. 效果验证与性能测试8.1 检索准确性测试创建评估脚本scripts/evaluate.pyimport json from rag_pipeline import RAGPipeline class Evaluator: def __init__(self): self.rag RAGPipeline() def evaluate_retrieval(self, test_cases: list) - dict: 评估检索效果 results [] for case in test_cases: question case[question] expected_keywords case[expected_keywords] # 执行检索 retrieved_docs self.rag.retriever.retrieve(question) contexts [doc[content] for doc in retrieved_docs] # 检查是否包含期望的关键词 hit_keywords [] for keyword in expected_keywords: if any(keyword in context for context in contexts): hit_keywords.append(keyword) recall len(hit_keywords) / len(expected_keywords) if expected_keywords else 0 results.append({ question: question, recall: recall, hit_keywords: hit_keywords, retrieved_count: len(retrieved_docs) }) avg_recall sum(r[recall] for r in results) / len(results) return { average_recall: avg_recall, detailed_results: results } def test_response_quality(self, questions: list) - list: 测试生成质量 responses [] for question in questions: result self.rag.query(question) responses.append({ question: question, answer: result[answer], contexts_used: len(result[retrieved_contexts]) }) return responses # 测试用例 test_cases [ { question: RAG技术如何减少模型幻觉, expected_keywords: [检索, 文档, 减少幻觉, 准确性] }, { question: 检索阶段的主要任务是什么, expected_keywords: [查找, 相关文档, 知识库, 向量] } ] evaluator Evaluator() retrieval_results evaluator.evaluate_retrieval(test_cases) print(检索评估结果:, json.dumps(retrieval_results, indent2, ensure_asciiFalse))8.2 性能监控添加性能监控功能import time from functools import wraps def timing_decorator(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() execution_time end_time - start_time print(f{func.__name__} 执行时间: {execution_time:.2f}秒) return result return wrapper class MonitoredRAGPipeline(RAGPipeline): timing_decorator def query(self, question: str, n_retrieve: int 3) - dict: return super().query(question, n_retrieve) def performance_test(self, questions: list, iterations: int 5): 性能压力测试 times [] for i in range(iterations): start_time time.time() for question in questions: self.query(question) end_time time.time() times.append(end_time - start_time) avg_time sum(times) / len(times) print(f平均处理时间: {avg_time:.2f}秒) print(f单问题平均时间: {avg_time/len(questions):.2f}秒)9. 高级优化技巧9.1 查询重写与扩展class QueryOptimizer: 查询优化器 def rewrite_query(self, original_query: str) - str: 重写查询以提高检索效果 # 简单的查询扩展 expansion_rules { 如何: 方法 步骤 流程, 什么是: 定义 概念 含义, 为什么: 原因 理由 目的 } rewritten original_query for key, value in expansion_rules.items(): if key in original_query: rewritten value return rewritten def generate_alternative_queries(self, query: str) - list: 生成替代查询 alternatives [query] # 同义词替换简化版 synonyms { 技术: 方法 方案 手段, 优势: 优点 好处 强项 } for word, replacement in synonyms.items(): if word in query: for syn in replacement.split(): new_query query.replace(word, syn) alternatives.append(new_query) return alternatives # 使用优化器 optimizer QueryOptimizer() original_query RAG技术的优势是什么 rewritten optimizer.rewrite_query(original_query) alternatives optimizer.generate_alternative_queries(original_query) print(原始查询:, original_query) print(重写后:, rewritten) print(替代查询:, alternatives)9.2 重排序策略class Reranker: 重排序器 def __init__(self): # 可以集成更复杂的重排序模型 pass def rerank(self, query: str, documents: list) - list: 对检索结果进行重排序 # 简单的基于关键词匹配的排序 scored_docs [] for doc in documents: score self.calculate_relevance_score(query, doc[content]) scored_docs.append((score, doc)) # 按分数降序排列 scored_docs.sort(keylambda x: x[0], reverseTrue) return [doc for score, doc in scored_docs] def calculate_relevance_score(self, query: str, document: str) - float: 计算查询与文档的相关性分数 query_words set(query.split()) doc_words set(document.split()) # 简单的 Jaccard 相似度 intersection query_words.intersection(doc_words) union query_words.union(doc_words) return len(intersection) / len(union) if union else 010. 实际部署考虑10.1 配置管理创建配置文件config.yamlrag_config: embedding: model: BAAI/bge-small-zh-v1.5 chunk_size: 500 chunk_overlap: 50 retrieval: n_results: 3 use_hybrid: true generation: model: gpt-3.5-turbo temperature: 0.3 max_tokens: 500 database: persist_directory: ./data/embeddings collection_name: documents10.2 API 服务封装创建简单的 Web APIfrom flask import Flask, request, jsonify from rag_pipeline import RAGPipeline import yaml app Flask(__name__) # 加载配置 with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) # 初始化 RAG 系统 rag_system RAGPipeline() app.route(/api/query, methods[POST]) def query(): 处理查询请求 data request.json question data.get(question, ) n_retrieve data.get(n_retrieve, 3) if not question: return jsonify({error: 问题不能为空}), 400 try: result rag_system.query(question, n_retrieve) return jsonify(result) except Exception as e: return jsonify({error: str(e)}), 500 app.route(/api/batch_query, methods[POST]) def batch_query(): 批量查询 data request.json questions data.get(questions, []) if not questions: return jsonify({error: 问题列表不能为空}), 400 try: results rag_system.batch_query(questions) return jsonify({results: results}) except Exception as e: return jsonify({error: str(e)}), 500 if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)11. 常见问题与排查方法问题现象可能原因排查方式解决方案检索结果不相关文档分块不合理或 Embedding 模型不适合检查文档分块大小测试不同 Embedding 模型调整分块策略尝试其他 Embedding 模型生成答案质量差检索内容不足或 LLM 参数不当检查检索到的文档数量和质量增加检索数量调整生成参数系统运行缓慢硬件资源不足或模型太大监控 CPU/内存使用情况使用更小的模型优化代码向量数据库连接失败路径错误或文件损坏检查数据库路径和文件权限重新创建向量数据库API 调用失败网络问题或密钥错误检查网络连接和 API 密钥验证网络配置检查密钥有效性12. 最佳实践与使用建议文档预处理是关键确保文档质量高分块大小适中通常 300-800 字逐步测试验证先测试检索效果再测试生成质量最后测试整体流程监控系统性能记录查询响应时间、检索准确率等指标版本控制配置保存不同版本的配置便于回滚和对比安全考虑涉及敏感信息时确保数据加密和访问控制定期更新随着业务发展定期更新知识库和模型这个 RAG 实战方案提供了从零搭建完整系统的完整路径核心价值在于可落地性。读者可以基于这个框架快速搭建自己的 RAG 系统再根据具体需求进行优化扩展。 RAG实战：从文档处理到文本生成的完整检索增强生成系统搭建
这次我们来看一个 RAG 实战项目重点不是讲概念而是直接动手搭建一个完整的检索增强生成系统。RAGRetrieval-Augmented Generation技术现在在企业知识库、智能问答、文档分析等场景应用广泛但很多教程只讲理论缺乏可落地的实操指导。本文基于RAG实战第4课从 Retrieval 到 Generation主题将带大家完整走通从文档处理、向量检索到文本生成的整个流程。我们会用到主流的 Embedding 模型、Chroma 向量数据库以及实际可运行的代码示例。无论你是想搭建企业知识库还是需要为 LLM 增加外部知识源这篇文章都能提供直接的参考价值。最值得关注的是这个实战方案可以在普通开发环境下运行不需要高端显卡CPU 环境也能正常工作。我们会重点测试检索准确性、生成质量以及如何优化整个流程的性能。1. 核心能力速览能力项说明技术栈RAG检索增强生成核心组件Embedding 模型 向量数据库 LLM硬件需求CPU 可运行GPU 可加速 Embedding向量数据库Chroma轻量级无需单独部署Embedding 模型BGE、Sentence-BERT 等可选LLM 接入支持 OpenAI API、本地模型等多种方式主要功能文档检索、知识问答、内容生成适合场景企业知识库、智能客服、文档分析2. 适用场景与使用边界RAG 技术特别适合需要结合外部知识库的应用场景。比如企业内部的规章制度查询、产品文档检索、技术知识问答等。通过 RAG可以让大语言模型在回答问题时参考具体的文档内容避免凭空编造信息。但需要注意几个边界首先RAG 的效果严重依赖检索质量如果文档质量差或者检索不准生成的内容也会有问题。其次涉及敏感信息时要注意数据安全确保检索的文档都是经过审核的。最后RAG 不能完全解决幻觉问题还需要结合其他校验机制。对于企业使用建议先在小范围测试验证检索准确性和生成质量再逐步扩大应用范围。3. 环境准备与前置条件开始之前需要准备好以下环境操作系统要求Windows 10/11、macOS 10.15 或 Linux Ubuntu 18.04建议使用 Linux 或 macOS 以获得更好的兼容性Python 环境Python 3.8-3.11版本推荐使用 conda 或 venv 创建虚拟环境硬件要求内存至少 8GB推荐 16GB 以上存储至少 10GB 可用空间用于模型文件GPU可选如有 GPU 可加速 Embedding 计算网络要求需要能访问 Hugging Face 下载模型如果使用 OpenAI API需要能够访问相应服务4. 安装部署与启动方式首先创建并激活 Python 虚拟环境# 创建虚拟环境 python -m venv rag_env source rag_env/bin/activate # Linux/macOS # 或 rag_env\Scripts\activate # Windows # 安装核心依赖 pip install chromadb sentence-transformers openai如果需要使用本地 LLM可以额外安装pip install transformers torch接下来创建项目目录结构mkdir rag_project cd rag_project mkdir -p data/documents data/embeddings scripts5. 文档处理与向量化RAG 的第一步是将文档转换为向量并存入数据库。我们使用 Chroma 作为向量数据库它轻量且易于使用。5.1 文档加载与预处理创建文档处理脚本scripts/document_processor.pyimport os import re from typing import List, Dict import chromadb from sentence_transformers import SentenceTransformer class DocumentProcessor: def __init__(self, persist_directory: str ./data/embeddings): self.client chromadb.PersistentClient(pathpersist_directory) self.collection self.client.get_or_create_collection(documents) self.embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) def chunk_documents(self, text: str, chunk_size: int 500, chunk_overlap: int 50) - List[str]: 将长文本分割为小块 chunks [] words text.split() for i in range(0, len(words), chunk_size - chunk_overlap): chunk .join(words[i:i chunk_size]) chunks.append(chunk) if i chunk_size len(words): break return chunks def process_documents(self, document_dir: str): 处理目录中的所有文档 documents [] metadatas [] ids [] for filename in os.listdir(document_dir): if filename.endswith(.txt): filepath os.path.join(document_dir, filename) with open(filepath, r, encodingutf-8) as f: content f.read() chunks self.chunk_documents(content) for i, chunk in enumerate(chunks): documents.append(chunk) metadatas.append({source: filename, chunk_id: i}) ids.append(f{filename}_{i}) # 生成嵌入向量 embeddings self.embedding_model.encode(documents).tolist() # 存入向量数据库 self.collection.add( embeddingsembeddings, documentsdocuments, metadatasmetadatas, idsids ) print(f成功处理 {len(documents)} 个文档块) if __name__ __main__: processor DocumentProcessor() processor.process_documents(./data/documents)5.2 准备测试文档在data/documents目录下创建示例文档example.txtRAG技术通过结合检索和生成两个阶段来提升大语言模型的效果。 在检索阶段系统从知识库中查找与问题相关的文档片段。 在生成阶段大语言模型基于检索到的内容生成最终答案。 这种方法可以有效减少模型幻觉提高回答的准确性。运行文档处理python scripts/document_processor.py6. 检索系统实现检索是 RAG 的核心环节直接影响到最终生成质量。6.1 基础检索器实现创建scripts/retriever.pyimport chromadb from sentence_transformers import SentenceTransformer from typing import List, Dict class Retriever: def __init__(self, persist_directory: str ./data/embeddings): self.client chromadb.PersistentClient(pathpersist_directory) self.collection self.client.get_collection(documents) self.embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) def retrieve(self, query: str, n_results: int 3) - List[Dict]: 检索相关文档 # 生成查询向量 query_embedding self.embedding_model.encode([query]).tolist() # 执行检索 results self.collection.query( query_embeddingsquery_embedding, n_resultsn_results ) retrieved_docs [] for i in range(len(results[documents][0])): retrieved_docs.append({ content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] }) return retrieved_docs def hybrid_retrieve(self, query: str, n_results: int 3) - List[Dict]: 混合检索结合向量检索和关键词检索 # 向量检索 vector_results self.retrieve(query, n_results) # 这里可以添加关键词检索逻辑 # 实际项目中可以使用 BM25 等算法 return vector_results if __name__ __main__: retriever Retriever() results retriever.retrieve(RAG技术如何工作) for result in results: print(f内容: {result[content]}) print(f距离: {result[distance]}) print(---)6.2 检索质量测试测试不同查询的检索效果# 测试检索效果 test_queries [ 什么是RAG技术, 检索阶段的作用是什么, 如何减少模型幻觉 ] retriever Retriever() for query in test_queries: print(f查询: {query}) results retriever.retrieve(query) for i, result in enumerate(results): print(f{i1}. {result[content][:100]}...) print(\n *50 \n)7. 生成阶段集成检索到相关文档后需要将其与问题结合交给 LLM 生成最终答案。7.1 基于 OpenAI API 的生成器创建scripts/generator.pyimport openai from typing import List, Dict import os class OpenAIGenerator: def __init__(self, api_key: str None): self.api_key api_key or os.getenv(OPENAI_API_KEY) if self.api_key: openai.api_key self.api_key def generate(self, query: str, contexts: List[str], model: str gpt-3.5-turbo) - str: 基于检索结果生成答案 # 构建上下文 context_text \n\n.join([f参考文档 {i1}: {context} for i, context in enumerate(contexts)]) prompt f基于以下参考文档回答问题。如果文档中没有相关信息请明确说明。 参考文档 {context_text} 问题{query} 请根据上述文档内容回答问题 try: response openai.ChatCompletion.create( modelmodel, messages[ {role: system, content: 你是一个有帮助的助手基于提供的文档内容回答问题。}, {role: user, content: prompt} ], temperature0.3, max_tokens500 ) return response.choices[0].message.content except Exception as e: return f生成失败: {str(e)} class LocalGenerator: 本地模型生成器示例 def __init__(self, model_path: str None): # 这里可以集成本地模型如 ChatGLM、Qwen 等 self.model None # 实际项目中加载本地模型 def generate(self, query: str, contexts: List[str]) - str: # 本地模型生成逻辑 context_text \n.join(contexts) prompt f基于以下信息回答问题\n{context_text}\n\n问题{query}\n答案 # 这里调用本地模型生成 # 返回生成的答案 return 本地模型生成示例需要实际模型支持 if __name__ __main__: # 使用示例 generator OpenAIGenerator() # 模拟检索结果 test_contexts [ RAG技术通过检索和生成结合来提升效果。, 检索阶段从知识库查找相关文档。, 生成阶段基于检索内容产生答案。 ] answer generator.generate(RAG如何工作, test_contexts) print(生成结果:, answer)7.2 完整的 RAG 流水线创建完整的 RAG 系统scripts/rag_pipeline.pyfrom retriever import Retriever from generator import OpenAIGenerator import os class RAGPipeline: def __init__(self, use_local: bool False): self.retriever Retriever() if use_local: self.generator LocalGenerator() else: self.generator OpenAIGenerator() def query(self, question: str, n_retrieve: int 3) - dict: 完整 RAG 流程 # 1. 检索阶段 retrieved_docs self.retriever.retrieve(question, n_retrieve) contexts [doc[content] for doc in retrieved_docs] # 2. 生成阶段 answer self.generator.generate(question, contexts) return { question: question, retrieved_contexts: retrieved_docs, answer: answer } def batch_query(self, questions: list) - list: 批量处理问题 results [] for question in questions: result self.query(question) results.append(result) return results if __name__ __main__: # 初始化 RAG 系统 rag RAGPipeline() # 测试查询 test_questions [ RAG技术的主要优势是什么, 检索阶段在RAG中起什么作用, 如何评估RAG系统的效果 ] for question in test_questions: result rag.query(question) print(f问题: {result[question]}) print(f答案: {result[answer]}) print(检索到的文档:) for i, context in enumerate(result[retrieved_contexts]): print(f{i1}. {context[content][:100]}...) print(\n *80 \n)8. 效果验证与性能测试8.1 检索准确性测试创建评估脚本scripts/evaluate.pyimport json from rag_pipeline import RAGPipeline class Evaluator: def __init__(self): self.rag RAGPipeline() def evaluate_retrieval(self, test_cases: list) - dict: 评估检索效果 results [] for case in test_cases: question case[question] expected_keywords case[expected_keywords] # 执行检索 retrieved_docs self.rag.retriever.retrieve(question) contexts [doc[content] for doc in retrieved_docs] # 检查是否包含期望的关键词 hit_keywords [] for keyword in expected_keywords: if any(keyword in context for context in contexts): hit_keywords.append(keyword) recall len(hit_keywords) / len(expected_keywords) if expected_keywords else 0 results.append({ question: question, recall: recall, hit_keywords: hit_keywords, retrieved_count: len(retrieved_docs) }) avg_recall sum(r[recall] for r in results) / len(results) return { average_recall: avg_recall, detailed_results: results } def test_response_quality(self, questions: list) - list: 测试生成质量 responses [] for question in questions: result self.rag.query(question) responses.append({ question: question, answer: result[answer], contexts_used: len(result[retrieved_contexts]) }) return responses # 测试用例 test_cases [ { question: RAG技术如何减少模型幻觉, expected_keywords: [检索, 文档, 减少幻觉, 准确性] }, { question: 检索阶段的主要任务是什么, expected_keywords: [查找, 相关文档, 知识库, 向量] } ] evaluator Evaluator() retrieval_results evaluator.evaluate_retrieval(test_cases) print(检索评估结果:, json.dumps(retrieval_results, indent2, ensure_asciiFalse))8.2 性能监控添加性能监控功能import time from functools import wraps def timing_decorator(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() execution_time end_time - start_time print(f{func.__name__} 执行时间: {execution_time:.2f}秒) return result return wrapper class MonitoredRAGPipeline(RAGPipeline): timing_decorator def query(self, question: str, n_retrieve: int 3) - dict: return super().query(question, n_retrieve) def performance_test(self, questions: list, iterations: int 5): 性能压力测试 times [] for i in range(iterations): start_time time.time() for question in questions: self.query(question) end_time time.time() times.append(end_time - start_time) avg_time sum(times) / len(times) print(f平均处理时间: {avg_time:.2f}秒) print(f单问题平均时间: {avg_time/len(questions):.2f}秒)9. 高级优化技巧9.1 查询重写与扩展class QueryOptimizer: 查询优化器 def rewrite_query(self, original_query: str) - str: 重写查询以提高检索效果 # 简单的查询扩展 expansion_rules { 如何: 方法 步骤 流程, 什么是: 定义 概念 含义, 为什么: 原因 理由 目的 } rewritten original_query for key, value in expansion_rules.items(): if key in original_query: rewritten value return rewritten def generate_alternative_queries(self, query: str) - list: 生成替代查询 alternatives [query] # 同义词替换简化版 synonyms { 技术: 方法 方案 手段, 优势: 优点 好处 强项 } for word, replacement in synonyms.items(): if word in query: for syn in replacement.split(): new_query query.replace(word, syn) alternatives.append(new_query) return alternatives # 使用优化器 optimizer QueryOptimizer() original_query RAG技术的优势是什么 rewritten optimizer.rewrite_query(original_query) alternatives optimizer.generate_alternative_queries(original_query) print(原始查询:, original_query) print(重写后:, rewritten) print(替代查询:, alternatives)9.2 重排序策略class Reranker: 重排序器 def __init__(self): # 可以集成更复杂的重排序模型 pass def rerank(self, query: str, documents: list) - list: 对检索结果进行重排序 # 简单的基于关键词匹配的排序 scored_docs [] for doc in documents: score self.calculate_relevance_score(query, doc[content]) scored_docs.append((score, doc)) # 按分数降序排列 scored_docs.sort(keylambda x: x[0], reverseTrue) return [doc for score, doc in scored_docs] def calculate_relevance_score(self, query: str, document: str) - float: 计算查询与文档的相关性分数 query_words set(query.split()) doc_words set(document.split()) # 简单的 Jaccard 相似度 intersection query_words.intersection(doc_words) union query_words.union(doc_words) return len(intersection) / len(union) if union else 010. 实际部署考虑10.1 配置管理创建配置文件config.yamlrag_config: embedding: model: BAAI/bge-small-zh-v1.5 chunk_size: 500 chunk_overlap: 50 retrieval: n_results: 3 use_hybrid: true generation: model: gpt-3.5-turbo temperature: 0.3 max_tokens: 500 database: persist_directory: ./data/embeddings collection_name: documents10.2 API 服务封装创建简单的 Web APIfrom flask import Flask, request, jsonify from rag_pipeline import RAGPipeline import yaml app Flask(__name__) # 加载配置 with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) # 初始化 RAG 系统 rag_system RAGPipeline() app.route(/api/query, methods[POST]) def query(): 处理查询请求 data request.json question data.get(question, ) n_retrieve data.get(n_retrieve, 3) if not question: return jsonify({error: 问题不能为空}), 400 try: result rag_system.query(question, n_retrieve) return jsonify(result) except Exception as e: return jsonify({error: str(e)}), 500 app.route(/api/batch_query, methods[POST]) def batch_query(): 批量查询 data request.json questions data.get(questions, []) if not questions: return jsonify({error: 问题列表不能为空}), 400 try: results rag_system.batch_query(questions) return jsonify({results: results}) except Exception as e: return jsonify({error: str(e)}), 500 if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)11. 常见问题与排查方法问题现象可能原因排查方式解决方案检索结果不相关文档分块不合理或 Embedding 模型不适合检查文档分块大小测试不同 Embedding 模型调整分块策略尝试其他 Embedding 模型生成答案质量差检索内容不足或 LLM 参数不当检查检索到的文档数量和质量增加检索数量调整生成参数系统运行缓慢硬件资源不足或模型太大监控 CPU/内存使用情况使用更小的模型优化代码向量数据库连接失败路径错误或文件损坏检查数据库路径和文件权限重新创建向量数据库API 调用失败网络问题或密钥错误检查网络连接和 API 密钥验证网络配置检查密钥有效性12. 最佳实践与使用建议文档预处理是关键确保文档质量高分块大小适中通常 300-800 字逐步测试验证先测试检索效果再测试生成质量最后测试整体流程监控系统性能记录查询响应时间、检索准确率等指标版本控制配置保存不同版本的配置便于回滚和对比安全考虑涉及敏感信息时确保数据加密和访问控制定期更新随着业务发展定期更新知识库和模型这个 RAG 实战方案提供了从零搭建完整系统的完整路径核心价值在于可落地性。读者可以基于这个框架快速搭建自己的 RAG 系统再根据具体需求进行优化扩展。