RAG 检索增强生成实战：从零搭建企业级知识库问答系统 — LangChain + Chroma + BGE 全链路

2026 年如果你问企业里最火的 AI 落地场景是什么不是 Agent不是微调——是 RAG。原因很简单企业有一堆 PDF、Word、Confluence 文档扔给大模型直接问它要么胡说八道要么回答我无法访问这些文件。RAG 直接把外部知识塞进 prompt让模型的回答有了「出处」。上个月给一个客户搭了一套 RAG 系统用 LangChain Chroma BGE 技术栈总共不到 300 行核心代码。这篇文章就是那个项目的精简版你复制出来直接跑。1. 环境搭建# 创建项目目录 mkdir rag-qa-system cd rag-qa-system python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装核心依赖 pip install langchain langchain-community langchain-text-splitters pip install chromadb sentence-transformers pip install openai python-dotenv unstructured pip install unstructured[pdf] unstructured[docx]# config.py import os from dotenv import load_dotenv load_dotenv() LLM_API_KEY os.getenv(DEEPSEEK_API_KEY) LLM_BASE_URL https://api.deepseek.com/v1 LLM_MODEL deepseek-chat EMBEDDING_MODEL BAAI/bge-large-zh-v1.5 # BGE 中文嵌入模型 CHROMA_PERSIST_DIR ./chroma_db CHUNK_SIZE 500 CHUNK_OVERLAP 50 TOP_K_RETRIEVAL 5 if not LLM_API_KEY: raise RuntimeError(请设置 DEEPSEEK_API_KEY)2. 文档加载与智能分块RAG 系统的第一道坎怎么把 PDF 切成合适的块。太大模型塞不进上下文太小丢失上下文关系。我试了 4 种分块策略最后锁定了这个组合。# loader.py — 文档加载与分块 from langchain_community.document_loaders import ( PyPDFLoader, Docx2txtLoader, TextLoader, DirectoryLoader, ) from langchain_text_splitters import ( RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter, ) from typing import List class DocumentProcessor: 多格式文档加载 语义分块 SUPPORTED_EXTS { .pdf: PyPDFLoader, .docx: Docx2txtLoader, .txt: TextLoader, .md: TextLoader, } def __init__(self, chunk_size: int 500, chunk_overlap: int 50): self.chunk_size chunk_size self.chunk_overlap chunk_overlap def load_documents(self, doc_dir: str) - List: 加载目录下所有支持的文档 all_docs [] for ext, loader_cls in self.SUPPORTED_EXTS.items(): loader DirectoryLoader( doc_dir, globf**/*{ext}, loader_clsloader_cls, loader_kwargs{extract_images: False}, silent_errorsTrue, ) docs loader.load() all_docs.extend(docs) print(f [{ext}] 加载 {len(docs)} 个文档) print(f总计加载 {len(all_docs)} 个文档) return all_docs def split_documents(self, docs: List) - List: RecursiveCharacterTextSplitter Markdown 感知 # 主分割器按段落和句子边界切分 text_splitter RecursiveCharacterTextSplitter( chunk_sizeself.chunk_size, chunk_overlapself.chunk_overlap, separators[\n\n, \n, 。, , , , ., !, ?, ;, ], length_functionlen, ) chunks text_splitter.split_documents(docs) # 元数据增强给每个 chunk 标记来源 for i, chunk in enumerate(chunks): source chunk.metadata.get(source, unknown) chunk.metadata[chunk_id] i chunk.metadata[source_file] source.split(/)[-1] print(f分割完成{len(docs)} 个文档 → {len(chunks)} 个文本块) return chunks # 使用示例 processor DocumentProcessor(chunk_size500, chunk_overlap50) docs processor.load_documents(./company_docs) chunks processor.split_documents(docs)3. 向量嵌入与 Chroma 存储BGE 中文模型的嵌入质量在中文语义检索上比 OpenAI text-embedding-3 强出一截而且是免费的。Chroma 是轻量级向量数据库适合中小规模百万级以下的知识库。# vector_store.py — 向量嵌入与存储 from langchain_community.embeddings import HuggingFaceBgeEmbeddings from langchain_chroma import Chroma import os class VectorStoreManager: BGE 嵌入 Chroma 向量存储 def __init__( self, model_name: str BAAI/bge-large-zh-v1.5, persist_dir: str ./chroma_db, ): # BGE embedding 配置 self.embeddings HuggingFaceBgeEmbeddings( model_namemodel_name, model_kwargs{device: cuda}, # 有 GPU 用 cuda没有用 cpu encode_kwargs{ normalize_embeddings: True, # BGE 建议开启归一化 batch_size: 32, }, ) self.persist_dir persist_dir def create_from_documents(self, chunks: List, collection_name: str knowledge_base): 从文档块创建向量库 vector_store Chroma.from_documents( documentschunks, embeddingself.embeddings, persist_directoryself.persist_dir, collection_namecollection_name, ) print(f向量库创建完成{len(chunks)} 条向量 → {self.persist_dir}) return vector_store def load_existing(self, collection_name: str knowledge_base): 加载已有向量库 return Chroma( persist_directoryself.persist_dir, embedding_functionself.embeddings, collection_namecollection_name, ) # 使用示例 store_manager VectorStoreManager() vector_store store_manager.create_from_documents(chunks)4. 混合检索器向量 BM25纯向量检索有一个盲区它对精确关键词匹配不够敏感。比如你问「2025 年 Q3 的营收是多少」向量检索可能返回 Q2 或 Q4 的数据因为它理解「季度营收」这个语义但忽略了「Q3」这个精确约束。解决方案是加一层 BM25 关键词检索把两边的结果融合。# hybrid_retriever.py — 混合检索 from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever from langchain_core.documents import Document from typing import List import jieba class HybridRetriever: 向量检索 BM25 关键词检索RRF 融合 def __init__(self, vector_store, chunks: List[Document], top_k: int 5): self.top_k top_k # 向量检索器 self.vector_retriever vector_store.as_retriever( search_typemmr, # MMR 保证结果多样性 search_kwargs{ k: top_k * 2, fetch_k: top_k * 4, lambda_mult: 0.7, # 0最大多样性, 1最大相关性 }, ) # BM25 关键词检索器用 jieba 做中文分词 self.bm25_retriever BM25Retriever.from_documents( chunks, preprocess_funcself._chinese_tokenize, ) self.bm25_retriever.k top_k * 2 # 集成检索器RRF (Reciprocal Rank Fusion) self.ensemble EnsembleRetriever( retrievers[self.vector_retriever, self.bm25_retriever], weights[0.6, 0.4], # 向量权重 60%BM25 权重 40% ) def _chinese_tokenize(self, text: str) - str: jieba 分词预处理提升 BM25 中文匹配效果 return .join(jieba.cut(text)) def retrieve(self, query: str) - List[Document]: 执行混合检索 results self.ensemble.invoke(query)[:self.top_k] return results5. RAG 问答链检索 生成把检索到的文本块拼进 prompt让 LLM 基于这些上下文回答。这里有一个容易被忽略的坑如果不加score_threshold过滤低相关度的 chunk 反而会拖累答案质量。# rag_chain.py — 完整的 RAG 问答链 from openai import OpenAI from typing import List, Dict, Optional class RAGChain: RAG 问答检索 → 上下文组织 → LLM 生成 SYSTEM_PROMPT 你是一个企业知识库助手。 请严格基于以下【参考资料】回答用户问题。 规则 1. 如果资料中有答案直接引用并注明出处文档名 段落编号 2. 如果资料不包含相关信息明确说资料中未找到相关信息 3. 不要编造资料中没有的内容 4. 回答要简洁用中文 def __init__(self, retriever: HybridRetriever): self.retriever retriever self.client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com/v1, ) def _format_context(self, docs: List[Document]) - str: 将检索结果格式化为上下文 parts [] for i, doc in enumerate(docs, 1): source doc.metadata.get(source_file, unknown) chunk_id doc.metadata.get(chunk_id, i) parts.append( f【资料 {i}】来源: {source} | 段落: {chunk_id}\n{doc.page_content}\n ) return \n---\n.join(parts) def query(self, question: str, chat_history: Optional[List] None) - Dict: 执行一次 RAG 问答 # 1. 检索 relevant_docs self.retriever.retrieve(question) if not relevant_docs: return { answer: 未找到相关文档。请检查知识库是否包含该主题的文档。, sources: [], } # 2. 组织上下文 context self._format_context(relevant_docs) # 3. 构建消息 messages [{role: system, content: self.SYSTEM_PROMPT}] # 加入对话历史最近 3 轮 if chat_history: messages.extend(chat_history[-6:]) messages.append({ role: user, content: f【参考资料】\n{context}\n\n【用户问题】\n{question}, }) # 4. LLM 生成 response self.client.chat.completions.create( modeldeepseek-chat, messagesmessages, temperature0.3, # 低温度保证答案一致性 max_tokens2048, ) return { answer: response.choices[0].message.content, sources: [ { file: doc.metadata.get(source_file, ), chunk_id: doc.metadata.get(chunk_id), content_preview: doc.page_content[:100], } for doc in relevant_docs ], } # 完整使用示例 retriever HybridRetriever(vector_store, chunks, top_k5) rag RAGChain(retriever) result rag.query(公司2025年的研发投入是多少) print(f回答: {result[answer]}) print(f引用来源: {len(result[sources])} 篇文档) for src in result[sources]: print(f - {src[file]} (段落 {src[chunk_id]}))6. 带对话记忆的多轮问答上面那个只能一问一答。真实场景里用户会追问需要把上一轮检索到的上下文也带进去。# chat_rag.py — 带记忆的多轮 RAG class ChatRAG(RAGChain): 支持多轮对话的 RAG 系统 def __init__(self, retriever: HybridRetriever, max_history: int 10): super().__init__(retriever) self.sessions: Dict[str, List] {} self.max_history max_history def chat(self, session_id: str, question: str) - Dict: 带记忆的对话接口 if session_id not in self.sessions: self.sessions[session_id] [] history self.sessions[session_id] result self.query(question, chat_historyhistory) # 记录对话历史 history.append({role: user, content: question}) history.append({role: assistant, content: result[answer]}) # 限制历史长度 if len(history) self.max_history * 2: self.sessions[session_id] history[-self.max_history * 2:] return result # 多轮对话测试 chat_rag ChatRAG(retriever, max_history10) questions [ 公司的研发团队有多少人, 那他们的主要研究方向是什么, 去年发了多少篇论文, ] for q in questions: result chat_rag.chat(session_iduser_001, questionq) print(f\nQ: {q}) print(fA: {result[answer][:200]}...)7. 效果评估用 Ragas 打分光说「效果好」没意义。用 Ragas 框架对检索和生成质量做量化评估。# eval_rag.py — Ragas 评估 from ragas import evaluate from ragas.metrics import ( context_precision, context_recall, faithfulness, answer_relevancy, ) from datasets import Dataset def evaluate_rag(rag_chain, test_questions: List[Dict]): test_questions [ {question: 研发投入多少?, ground_truth: 2025年研发投入5.2亿}, ... ] eval_data {question: [], answer: [], contexts: [], ground_truth: []} for item in test_questions: result rag_chain.query(item[question]) eval_data[question].append(item[question]) eval_data[answer].append(result[answer]) eval_data[contexts].append([s[content_preview] for s in result[sources]]) eval_data[ground_truth].append(item[ground_truth]) dataset Dataset.from_dict(eval_data) scores evaluate( dataset, metrics[context_precision, context_recall, faithfulness, answer_relevancy], ) print(fContext Precision: {scores[context_precision]:.3f}) print(fContext Recall: {scores[context_recall]:.3f}) print(fFaithfulness: {scores[faithfulness]:.3f}) print(fAnswer Relevancy: {scores[answer_relevancy]:.3f}) return scores踩坑笔记踩了几个坑记下来省你时间BGE 模型首次下载— 默认从 HuggingFace 拉模型国内网络大概率超时。先export HF_ENDPOINThttps://hf-mirror.com换镜像源。Chroma 的 persist 时机—from_documents()之后 Chroma 会自动持久化但后续add_documents()必须要手动调persist()不然重启丢数据。Chunk size 不是越大越好— 我实测 500 字符的中文 chunk 在 BGE 上效果最好超过 800 检索精度反而下降。MMR 的 lambda_mult 要调到 0.7— 默认值 0.5 多样性太激进会把最相关的几个文档挤到后面去。金句RAG 不是什么高级技术它就是解决了一个朴素的问题——大模型没有看过你的内部文档。但朴素到极致就是生产级。如果你也在搭 RAG 系统或者遇到了分块/检索效果不好的问题评论区说说你的场景——我看看能不能给点建议。