夜雨聆风文章目录
引言 1 核心架构与流程对比 1.1 原生RAG vs 企业级RAG横向对比 2 环境准备与依赖安装 2.1 软硬件要求 2.2 依赖安装 3 全流程核心代码实现 3.1 多源数据接入与预处理 3.2 嵌入模型选型与量化性能对比 3.3 向量库搭建（入门+企业级双版本） 入门版：Chroma本地向量库，适合快速测试 企业级版：PGVector基于PostgreSQL，支持事务和权限过滤 3.4 多路召回+重排序实现 3.5 大模型生成问答结果 4 企业级生产部署方案（含安全审计） 4.1 核心安全设计 4.2 FastAPI接口核心代码 5 技术前瞻性分析 6 附录：完整技术图谱 总结
引言
大模型落地企业私有业务场景时，始终绕不开两大核心痛点：一是大模型训练数据截止日期固定，无法获取企业内部最新的私有数据；二是大模型天生存在幻觉问题，回答私有领域问题容易编造内容，无法直接落地使用。RAG（检索增强生成）因为其改造成本低、效果稳定、可解释性强、数据更新方便，已经成为当前企业私有大模型落地最成熟主流的方案。
网上大部分教程仅能跑通玩具级Demo，并没有覆盖生产环境需要的权限控制、安全审计、性能优化等核心需求，本文从零开始，一步步带你搭建一个可以直接在生产环境运行的本地私有知识库RAG系统，新手小白也可以跟着步骤直接落地。
1 核心架构与流程对比
本文设计的企业级RAG架构和原生Demo级RAG有本质区别，我们先通过两个流程图看整体差异。
1.1 原生RAG vs 企业级RAG横向对比
2 环境准备与依赖安装
2.1 软硬件要求
本地开发测试：CPU 4核以上，内存16G以上，GPU显存8G以上即可流畅运行，CPU也可运行仅速度较慢 企业生产部署：服务器CPU 8核以上，内存32G以上，GPU 16G显存以上，支持百级并发请求
2.2 依赖安装
创建requirements.txt文件，内容如下：
langchain==0.1.20langchain-community==0.0.38langchain-chroma==0.1.2sentence-transformers==2.7.0pypdf==4.2.0python-docx==1.1.2bitsandbytes==0.43.1transformers==4.41.2torch==2.3.0psycopg2-binary==2.9.9pgvector==0.2.5fastapi==0.110.2uvicorn==0.29.0python-jose[cryptography]==3.3.0passlib[bcrypt]==1.7.4
执行安装命令：
pip install -r requirements.txt3 全流程核心代码实现
3.1 多源数据接入与预处理
支持PDF、Word、TXT、MD四种常见格式，自带去重清洗逻辑：
import osimport refrom typing import Listfrom langchain.schema import Documentfrom pypdf import PdfReaderfrom docx import Document as DocxDocumentdef load_single_pdf(file_path: str) -> List[Document]:pdf_reader = PdfReader(file_path)text = ""for page in pdf_reader.pages:text += page.extract_text() + "\n"return [Document(page_content=text, metadata={"source": file_path, "file_type": "pdf"})]def load_single_docx(file_path: str) -> List[Document]:doc = DocxDocument(file_path)text = "\n".join([para.text for para in doc.paragraphs])return [Document(page_content=text, metadata={"source": file_path, "file_type": "docx"})]def load_text_file(file_path: str) -> List[Document]:with open(file_path, "r", encoding="utf-8") as f:text = f.read()ext = os.path.splitext(file_path)[1]return [Document(page_content=text, metadata={"source": file_path, "file_type": ext})]def load_dir_documents(dir_path: str) -> List[Document]:documents = []for root, _, files in os.walk(dir_path):for file in files:file_path = os.path.join(root, file)ext = file.lower().split(".")[-1]if ext == "pdf":documents.extend(load_single_pdf(file_path))elif ext in ["docx", "doc"]:documents.extend(load_single_docx(file_path))elif ext in ["txt", "md"]:documents.extend(load_text_file(file_path))return documentsdef preprocess_documents(documents: List[Document]) -> List[Document]:processed = []seen_content = set()for doc in documents:content = doc.page_content.strip()content = "\n".join([line.strip() for line in content.splitlines() if line.strip()])content = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\xff]', '', content)if content and content not in seen_content:seen_content.add(content)doc.page_content = contentprocessed.append(doc)return processed
文档分块采用企业级常用的递归分块，兼顾语义完整性：
from langchain.text_splitter import RecursiveCharacterTextSplitterdef split_documents(documents: List[Document], chunk_size: int = 512, chunk_overlap: int = 50) -> List[Document]:text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size,chunk_overlap=chunk_overlap,add_start_index=True,separators=["\n\n", "\n", "。", "，", " ", ""])return text_splitter.split_documents(documents)
3.2 嵌入模型选型与量化性能对比
中文领域当前效果最好的开源嵌入模型是BAAI的BGE系列，下表是不同量化版本的性能对比，可根据自身需求选择：
本文选择兼顾速度和精度的bge-base-zh-v1.5 INT4版本，加载代码如下：
import torchfrom langchain_community.embeddings import HuggingFaceBgeEmbeddingsmodel_name = "BAAI/bge-base-zh-v1.5"model_kwargs = {"device": "cuda" if torch.cuda.is_available() else "cpu", "load_in_4bit": True}encode_kwargs = {"normalize_embeddings": True}embeddings = HuggingFaceBgeEmbeddings(model_name=model_name,model_kwargs=model_kwargs,encode_kwargs=encode_kwargs)
3.3 向量库搭建（入门+企业级双版本）
入门版：Chroma本地向量库，适合快速测试
from langchain_chroma import Chroma# 初始化向量库persist_directory = "./chroma_db"vector_db = Chroma(persist_directory=persist_directory,embedding_function=embeddings,collection_name="private_knowledge")# 插入分块后的文档split_docs = split_documents(preprocess_documents(load_dir_documents("./my_docs")))vector_db.add_documents(split_docs)# 持久化存储vector_db.persist()
企业级版：PGVector基于PostgreSQL，支持事务和权限过滤
首先用Docker启动PGVector服务，docker-compose.yml配置如下：
version: '3.8'services:postgres:image: pgvector/pgvector:pg16container_name: pgvector-ragenvironment:POSTGRES_USER: rag_userPOSTGRES_PASSWORD: rag_2024_secure_privatePOSTGRES_DB: rag_dbports:- "5432:5432"volumes:- ./pg_data:/var/lib/postgresql/datarestart: unless-stopped
启动命令：
docker-compose up -d代码连接：
from langchain_community.vectorstores import PGVectorconnection_string = "postgresql+psycopg2://rag_user:rag_2024_secure_private@localhost:5432/rag_db"collection_name = "private_knowledge"vector_db = PGVector(embedding=embeddings,collection_name=collection_name,connection_string=connection_string,use_jsonb=True,)split_docs = split_documents(preprocess_documents(load_dir_documents("./my_docs")))vector_db.add_documents(split_docs)
3.4 多路召回+重排序实现
企业级RAG通过融合稀疏和稠密召回，再用重排序提升精度，代码如下：
from langchain.retrievers import BM25Retriever, EnsembleRetrieverfrom langchain_community.cross_encoders import HuggingFaceCrossEncoderfrom langchain.retrievers import ContextualCompressionRetrieverfrom langchain.retrievers.document_compressors import CrossEncoderRerank# 初始化BM25稀疏召回bm25_retriever = BM25Retriever.from_documents(split_docs)bm25_retriever.k = 10# 初始化稠密向量召回vector_retriever = vector_db.as_retriever(search_kwargs={"k": 10})# 融合两路召回结果ensemble_retriever = EnsembleRetriever(retrievers=[bm25_retriever, vector_retriever],weights=[0.4, 0.6])# BGE重排序模型，筛选最相关的Top5rerank_model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")compressor = CrossEncoderRerank(model=rerank_model, top_n=5)compression_retriever = ContextualCompressionRetriever(base_retriever=ensemble_retriever,base_compressor=compressor)
3.5 大模型生成问答结果
本文采用本地量化Llama3-8B，8G显存即可运行，代码如下：
from langchain_community.llms import HuggingFacePipelinefrom langchain.chains import RetrievalQAfrom transformers import AutoTokenizer, AutoModelForCausalLM, pipelinefrom langchain.prompts import PromptTemplate# 私有知识库专属Prompt模板template = """你是一个专业的企业内部助手，你需要根据提供的上下文信息回答用户的问题，如果你不知道答案，就直接说你不知道，不要编造答案。上下文信息如下:{context}用户问题: {question}回答:"""prompt = PromptTemplate.from_template(template)# 加载4bit量化Llama3-8B-Instructmodel_name = "NousResearch/Meta-Llama-3-8B-Instruct"tokenizer = AutoTokenizer.from_pretrained(model_name)model = AutoModelForCausalLM.from_pretrained(model_name,load_in_4bit=True,device_map="auto")pipe = pipeline("text-generation",model=model,tokenizer=tokenizer,max_new_tokens=512,temperature=0.1,top_p=0.9,repetition_penalty=1.1)llm = HuggingFacePipeline(pipeline=pipe)# 构建RAG问答链qa_chain = RetrievalQA.from_chain_type(llm=llm,retriever=compression_retriever,chain_type_kwargs={"prompt": prompt},return_source_documents=True)# 测试问答question = "请问公司的年假申请规则是什么？"result = qa_chain.invoke({"query": question})print("回答：", result["result"])print("参考文档：", [doc.metadata["source"] for doc in result["source_documents"]])
到这里，你已经跑通了完整的RAG流程，可以正常问答了。
4 企业级生产部署方案（含安全审计）
4.1 核心安全设计
企业生产部署必须满足合规要求，我们实现了五大安全能力：
JWT身份认证：仅授权用户可访问接口 行级权限控制：用户只能查看自己有权限的文档 输入输出双重敏感内容检查：防止违规内容生成 PII信息自动脱敏：对手机号、身份证等敏感信息自动打码 全链路日志审计：所有请求都记录留存，满足等保要求
4.2 FastAPI接口核心代码
import loggingimport refrom fastapi import FastAPI, Depends, HTTPException, statusfrom jose import JWTError, jwtfrom pydantic import BaseModelfrom typing import Optional# 初始化日志审计logging.basicConfig(filename='./rag_audit.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')app = FastAPI(title="企业私有知识库RAG API", version="1.0")SECRET_KEY = "替换为你自己的生产级密钥"ALGORITHM = "HS256"class QueryRequest(BaseModel):question: strtoken: str# JWT认证依赖async def get_current_user(token: str) -> Optional[int]:credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="无效的访问凭证",)try:payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])user_id: int = payload.get("user_id")if user_id is None:raise credentials_exceptionexcept JWTError:raise credentials_exceptionreturn user_id# 敏感内容检查SENSITIVE_WORDS = ["替换为你的敏感词列表"]def sensitive_check(text: str) -> bool:for word in SENSITIVE_WORDS:if word in text:return Falsereturn True# PII脱敏def pii_mask(text: str) -> str:text = re.sub(r'(1[3-9]\d)\d{4}(\d{4})', r'\1****\2', text)text = re.sub(r'(\d{6})\d{8}(\d{4})', r'\1********\2', text)return text# 问答接口@app.post("/api/rag/query")async def query_rag(request: QueryRequest, user_id: int = Depends(get_current_user)):logging.info(f"用户ID: {user_id}, 请求问题: {request.question}")# 输入敏感检查if not sensitive_check(request.question):logging.warning(f"用户ID: {user_id}, 请求包含敏感内容: {request.question}")raise HTTPException(status_code=400, detail="请求包含违规内容")# 权限过滤：只返回当前用户有权限的文档result = qa_chain.invoke({"query": request.question, "filter": {"accessible_user": [user_id, -1]}})# 输出脱敏answer = pii_mask(result["result"])# 输出敏感检查if not sensitive_check(answer):logging.warning(f"用户ID: {user_id}, 生成回答包含敏感内容, 问题: {request.question}")return {"answer": "回答包含敏感内容，无法展示", "sources": []}sources = [doc.metadata["source"] for doc in result["source_documents"]]return {"answer": answer, "sources": sources}
生产环境可以用Nginx反向代理，容器化整体部署，支持水平扩展，这里不展开赘述。
5 技术前瞻性分析
RAG技术当前还在快速迭代，未来企业级落地的趋势主要有三个方向：
- 架构升级
：从普通RAG走向Graph RAG和多模态RAG，Graph RAG解决了长文档全局信息丢失的问题，适合大型知识库梳理场景；多模态RAG支持图片、表格、视频等非文本数据，拓展了RAG的应用边界。 - 自动化优化
：自适应RAG会成为主流，系统会自动根据问题类型选择分块大小、召回策略、模型大小，无需人工调参，进一步降低落地成本。 - 技术融合
：RAG不会替代微调，未来企业级落地会是RAG+参数高效微调的结合，RAG解决知识更新和可解释性问题，微调解决大模型对齐和风格统一问题，两者结合效果最优。
6 附录：完整技术图谱
总结
本文从零开始完整实现了企业级RAG全流程，新手按照步骤即可搭建自己的本地私有知识库，企业可以直接基于这套方案改造落地生产环境。
