从零构建企业级 RAG 系统：LangChain + 向量数据库实战

2024 年RAGRetrieval-Augmented Generation检索增强生成成为企业 AI 落地的首选方案。相比微调大模型RAG 成本更低、效果更好、数据更新更快。本文将带你从零构建一个生产级的 RAG 系统。一、为什么选择 RAG传统方案的困境企业想让 AI懂自己的业务数据传统方案有两种方案 1微调大模型❌ 成本高GPT-3.5 微调一次 10,000❌ 数据更新慢需要重新训练❌ 容易过拟合❌ 黑盒问题不知道 AI 从哪学到的方案 2Prompt Engineering❌ Context 长度限制4K-128K tokens❌ 无法处理大规模知识库❌ 容易幻觉胡说八道RAG 的优势RAG 结合了检索和生成的优点核心流程1.用户提问2.向量检索找相关文档3.注入 Prompt4.LLM 生成答案核心优势✅ 成本可控无需训练只需向量检索 API 调用✅ 实时更新新数据立即生效✅ 可解释知道答案来自哪些文档✅ 准确性高基于事实生成减少幻觉✅ 数据安全敏感数据可本地部署二、技术架构设计系统架构1 graph LR 2 A[文档上传] -- B[文本分块] 3 B -- C[向量化 Embedding] 4 C -- D[向量数据库] 5 E[用户提问] -- F[问题向量化] 6 F -- G[向量检索] 7 D -- G 8 G -- H[构建 Prompt] 9 H -- I[LLM 生成] 10 I -- J[返回答案]技术栈选择核心组件LangChainLLM 应用开发框架向量数据库Pinecone / Weaviate / Milvus / ChromaEmbedding 模型OpenAI text-embedding-3 / BGE / M3ELLMGPT-4 / Claude 3 / DeepSeek企业级功能文档解析Unstructured / LlamaParse重排序Cohere Rerank / BGE Reranker缓存Redis / GPTCache监控LangSmith / Langfuse三、实战构建 RAG 系统Step 1文档预处理1.1 加载文档1 from langchain_community.document_loaders import ( 2 PyPDFLoader, 3 TextLoader, 4 UnstructuredMarkdownLoader, 5 DirectoryLoader 6 ) 7 8 # 加载单个文档 9 pdf_loader PyPDFLoader(company_manual.pdf) 10 documents pdf_loader.load() 11 12 # 批量加载目录 13 loader DirectoryLoader( 14 ./knowledge_base, 15 glob/*.pdf, 16 loader_clsPyPDFLoader 17 ) 18 all_docs loader.load() 19 20 print(f加载了 {len(all_docs)} 个文档片段)1.2 文本分块Chunking分块是 RAG 的关键步骤影响检索质量1 from langchain.text_splitter import RecursiveCharacterTextSplitter 2 3 text_splitter RecursiveCharacterTextSplitter( 4 chunk_size1000, # 每块 1000 字符 5 chunk_overlap200, # 重叠 200 字符避免信息断裂 6 length_functionlen, 7 separators[/n/n, /n, 。, , , , ] 8 ) 9 10 chunks text_splitter.split_documents(documents) 11 print(f分割成 {len(chunks)} 个文本块)分块策略固定长度简单但有断裂风险语义分块按段落、句子分割递归分块优先按段落再按句子推荐自定义分块Markdown 标题、代码函数等Step 2向量化Embedding2.1 选择 Embedding 模型模型维度语言性能成本OpenAI text-embedding-3-small1536多语言⭐⭐⭐⭐0.02/1M tokensOpenAI text-embedding-3-large3072多语言⭐⭐⭐⭐⭐0.13/1M tokensBGE-large-zh1024中文⭐⭐⭐⭐免费M3E-large1024中英⭐⭐⭐⭐免费2.2 生成向量1 from langchain_openai import OpenAIEmbeddings 2 3 # 使用 OpenAI Embedding 4 embeddings OpenAIEmbeddings( 5 modeltext-embedding-3-small, 6 dimensions1536 7 ) 8 9 # 批量向量化 10 texts [chunk.page_content for chunk in chunks] 11 vectors embeddings.embed_documents(texts) 12 13 print(f生成了 {len(vectors)} 个向量维度 {len(vectors[0])})性能优化1 # 批量处理避免 API 限流 2 from tqdm import tqdm 3 4 batch_size 100 5 all_vectors [] 6 7 for i in tqdm(range(0, len(texts), batch_size)): 8 batch texts[i:ibatch_size] 9 batch_vectors embeddings.embed_documents(batch) 10 all_vectors.extend(batch_vectors)Step 3向量数据库3.1 Chroma本地开发1 from langchain_community.vectorstores import Chroma 2 3 # 创建向量库 4 vectorstore Chroma.from_documents( 5 documentschunks, 6 embeddingembeddings, 7 persist_directory./chroma_db 8 ) 9 10 # 持久化 11 vectorstore.persist() 12 13 # 加载已存在的向量库 14 vectorstore Chroma( 15 persist_directory./chroma_db, 16 embedding_functionembeddings 17 )3.2 Pinecone生产环境1 from langchain_community.vectorstores import Pinecone 2 import pinecone 3 4 # 初始化 5 pinecone.init( 6 api_keyyour-api-key, 7 environmentus-east-1 8 ) 9 10 # 创建索引 11 index_name company-knowledge 12 if index_name not in pinecone.list_indexes(): 13 pinecone.create_index( 14 nameindex_name, 15 dimension1536, 16 metriccosine 17 ) 18 19 # 插入向量 20 vectorstore Pinecone.from_documents( 21 chunks, 22 embeddings, 23 index_nameindex_name 24 )3.3 Milvus大规模部署1 from langchain_community.vectorstores import Milvus 2 3 vectorstore Milvus.from_documents( 4 documentschunks, 5 embeddingembeddings, 6 connection_args{ 7 host: localhost, 8 port: 19530 9 }, 10 collection_namerag_knowledge 11 )Step 4检索与生成4.1 基础检索1 # 相似度检索 2 query 如何申请年假 3 results vectorstore.similarity_search(query, k3) 4 5 for i, doc in enumerate(results): 6 print(f[{i1}] {doc.page_content[:200]}...)4.2 构建完整的 RAG Chain1 from langchain_openai import ChatOpenAI 2 from langchain.chains import RetrievalQA 3 from langchain.prompts import PromptTemplate 4 5 # 定义 Prompt 模板 6 template 你是一个专业的客服助手。基于以下上下文回答用户问题。 7 如果上下文中没有相关信息请诚实说我不知道不要编造。 8 9 上下文 10 {context} 11 12 问题{question} 13 14 答案 15 16 PROMPT PromptTemplate( 17 templatetemplate, 18 input_variables[context, question] 19 ) 20 21 # 创建 RAG Chain 22 llm ChatOpenAI(model_namegpt-4-turbo, temperature0) 23 qa_chain RetrievalQA.from_chain_type( 24 llmllm, 25 chain_typestuff, 26 retrievervectorstore.as_retriever(search_kwargs{k: 3}), 27 return_source_documentsTrue, 28 chain_type_kwargs{prompt: PROMPT} 29 ) 30 31 # 提问 32 result qa_chain.invoke({query: 公司的年假政策是什么}) 33 print(f答案{result[result]}) 34 print(f/n来源文档) 35 for doc in result[source_documents]: 36 print(f- {doc.metadata[source]})Step 5进阶优化5.1 混合检索Hybrid Search结合关键词检索和向量检索提升召回率1 from langchain.retrievers import EnsembleRetriever 2 from langchain_community.retrievers import BM25Retriever 3 4 # 向量检索 5 vector_retriever vectorstore.as_retriever(search_kwargs{k: 5}) 6 7 # 关键词检索 8 bm25_retriever BM25Retriever.from_documents(chunks) 9 bm25_retriever.k 5 10 11 # 混合检索 12 ensemble_retriever EnsembleRetriever( 13 retrievers[bm25_retriever, vector_retriever], 14 weights[0.4, 0.6] # 权重分配 15 ) 16 17 results ensemble_retriever.get_relevant_documents(年假申请流程)5.2 重排序Reranking对检索结果重排序提升相关性1 from langchain.retrievers import ContextualCompressionRetriever 2 from langchain.retrievers.document_compressors import CohereRerank 3 4 # 使用 Cohere Rerank 5 compressor CohereRerank( 6 modelrerank-multilingual-v2.0, 7 top_n3 8 ) 9 10 compression_retriever ContextualCompressionRetriever( 11 base_compressorcompressor, 12 base_retrievervectorstore.as_retriever(search_kwargs{k: 10}) 13 ) 14 15 # 检索 重排序 16 results compression_retriever.get_relevant_documents(如何申请加班)5.3 多轮对话带记忆1 from langchain.memory import ConversationBufferMemory 2 from langchain.chains import ConversationalRetrievalChain 3 4 # 添加对话记忆 5 memory ConversationBufferMemory( 6 memory_keychat_history, 7 return_messagesTrue 8 ) 9 10 qa_chain ConversationalRetrievalChain.from_llm( 11 llmllm, 12 retrievervectorstore.as_retriever(), 13 memorymemory, 14 return_source_documentsTrue 15 ) 16 17 # 多轮对话 18 questions [ 19 公司的年假政策是什么, 20 可以累积到明年吗, # 依赖上下文 21 需要提前几天申请 22 ] 23 24 for q in questions: 25 result qa_chain({question: q}) 26 print(fQ: {q}/nA: {result[answer]}/n)四、生产级优化4.1 性能优化缓存策略1 from langchain.cache import InMemoryCache 2 from langchain.globals import set_llm_cache 3 4 # 启用 LLM 缓存 5 set_llm_cache(InMemoryCache()) 6 7 # 或使用 Redis 缓存 8 from langchain.cache import RedisCache 9 from redis import Redis 10 11 set_llm_cache(RedisCache(redis_Redis()))批量处理1 # 批量向量化提升 10x 速度 2 from concurrent.futures import ThreadPoolExecutor 3 4 def embed_batch(texts, batch_size100): 5 with ThreadPoolExecutor(max_workers5) as executor: 6 futures [] 7 for i in range(0, len(texts), batch_size): 8 batch texts[i:ibatch_size] 9 futures.append(executor.submit(embeddings.embed_documents, batch)) 10 return [f.result() for f in futures]4.2 监控与调试使用 LangSmith1 import os 2 os.environ[LANGCHAIN_TRACING_V2] true 3 os.environ[LANGCHAIN_API_KEY] your-key 4 5 # 所有 Chain 调用都会被自动追踪 6 result qa_chain({query: 问题}) 7 # 在 LangSmith 中查看完整的执行链路、耗时、Token 消耗4.3 评估与测试1 from ragas import evaluate 2 from ragas.metrics import ( 3 faithfulness, # 忠实度 4 answer_relevancy, # 答案相关性 5 context_recall, # 召回率 6 context_precision # 精确率 7 ) 8 9 # 准备测试数据 10 test_data { 11 question: [问题1, 问题2], 12 answer: [答案1, 答案2], 13 contexts: [[文档1], [文档2]], 14 ground_truths: [[标准答案1], [标准答案2]] 15 } 16 17 # 评估 18 results evaluate(test_data, metrics[ 19 faithfulness, 20 answer_relevancy, 21 context_recall, 22 context_precision 23 ]) 24 25 print(results) 26 # 输出{faithfulness: 0.85, answer_relevancy: 0.92, ...}五、成本分析5.1 成本构成企业级 RAG 系统月成本10万文档100万次查询项目成本向量化一次性200 (10万文档 × 5K tokens × 0.0004)向量数据库70/月 (Pinecone Starter)LLM 调用500/月 (100万查询 × 平均 1K tokens)服务器200/月 (4核16G)总计770/月对比微调微调成本10,000一次性 2,000/月推理RAG 成本200一次性 $770/月节省 80%5.2 成本优化技巧1.缓存热门问题减少 30% LLM 调用2.使用更便宜的模型GPT-3.5 代替 GPT-43.本地 EmbeddingBGE/M3E 代替 OpenAI4.自建向量库Milvus 代替 Pinecone六、最佳实践总结✅ 推荐做法1.分块策略1000-1500 字符200 字符重叠2.混合检索向量 关键词权重 6:43.重排序召回 Top-10重排取 Top-34.Prompt 设计明确约束不知道就说不知道5.监控追踪检索质量、答案相关性6.迭代优化根据用户反馈持续改进❌ 常见错误1.分块太大/太小影响检索质量2.忽略元数据丢失文档来源信息3.过度依赖向量检索关键词同样重要4.没有评估机制不知道系统好坏5.数据质量差Garbage In, Garbage Out七、开源项目推荐想要快速上手这些开源项目值得一试1.LangChain - 最流行的 LLM 框架2.LlamaIndex - 专注 RAG 的数据框架3.Dify - 可视化 RAG 平台4.FastGPT - 开源知识库问答5.Quivr - 第二大脑支持多模态八、总结RAG 是企业 AI 落地的最佳路径 精准基于事实生成减少幻觉 经济无需训练成本可控⚡ 灵活数据实时更新 安全敏感数据本地部署核心要点1.分块是基础决定检索质量2.混合检索 重排序是标配3.监控评估不能少4.持续迭代是关键 实战资源LangChain 官方文档https://python.langchain.comPinecone 学习中心https://www.pinecone.io/learnRAGAS 评估框架https://github.com/explodinggradients/ragas本项目完整代码[GitHub 仓库]这里给大家精心整理了一份全面的AI大模型学习资源包括AI大模型全套学习路线图从入门到实战、精品AI大模型学习书籍手册、视频教程、实战学习、面试题等资料免费分享扫码免费领取全部内容1. 成长路线图学习规划要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。这里我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。可以说是最科学最系统的学习成长路线。2. 大模型经典PDF书籍书籍和学习文档资料是学习大模型过程中必不可少的我们精选了一系列深入探讨大模型技术的书籍和学习文档它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。书籍含电子版PDF3. 大模型视频教程对于很多自学或者没有基础的同学来说书籍这些纯文字类的学习教材会觉得比较晦涩难以理解因此我们提供了丰富的大模型视频教程以动态、形象的方式展示技术概念帮助你更快、更轻松地掌握核心知识。4. 2026行业报告行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。5. 大模型项目实战学以致用当你的理论知识积累到一定程度就需要通过项目实战在实际操作中检验和巩固你所学到的知识同时为你找工作和职业发展打下坚实的基础。6. 大模型面试题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我们将提供精心整理的大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。7. 资料领取全套内容免费抱走学 AI 不用再找第二份不管你是 0 基础想入门 AI 大模型还是有基础想冲刺大厂、了解行业趋势这份资料都能满足你现在只需按照提示操作就能免费领取扫码免费领取全部内容