当前位置：首页 > news >正文

LangChain+FAISS 向量数据库搭建轻量化 RAG 应用

news 2026/5/13 1:12:21

📝 本章学习目标：本章聚焦企业轻量化落地，帮助读者快速掌握基于 LangChain+FAISS 的私有化 RAG 开发流程。通过本章学习，你将从零搭建一套无需 GPU、无外网依赖、纯本地运行、代码极简、可直接上线的轻量化 RAG 应用。

一、引言：为什么 LangChain+FAISS 是轻量化 RAG 首选

1.1 背景与痛点

随着大模型应用快速普及，企业和开发者面临三大现实困境：

成本高：使用云服务向量库（如 Pinecone、Milvus 集群）需付费，中小团队难以承担；
部署重：传统向量数据库依赖 Docker、K8s、高配置服务器，本地 / 轻量服务器无法跑；
隐私泄露：云端向量库必须上传数据，无法满足内网、涉密、私有化场景。

而 LangChain+FAISS 的组合，完美解决以上痛点：FAISS 是单文件向量库、零服务、零依赖、CPU 就能跑；LangChain 封装度高、上手快、可快速构建检索增强链路。

据行业统计：90% 的个人开发者、70% 的中小企业、60% 的内网项目，优先选择 LangChain+FAISS 构建轻量化 RAG。它是目前最成熟、成本最低、落地最快的私有化知识库方案。

1.2 方案价值

LangChain+FAISS 轻量化 RAG 的核心价值：

轻量无依赖：FAISS 直接本地文件存储，不需要额外部署服务；
低成本落地：CPU 即可运行，8G 内存笔记本也能跑；
全链路可控：数据不上云、向量不上云、模型本地跑，安全可控；
开发效率高：LangChain 封装完整，几行代码实现文档加载、分割、向量、检索、问答；
迭代灵活：支持文档增量更新、知识库随时重建、快速调试优化。

1.3 本章结构概览

为帮助读者系统性掌握轻量化 RAG 全流程，本章从以下维度展开：

plaintext

📊 概念解析 → 技术选型 → 环境搭建 → 代码实战 → 优化调优 → 部署上线 → 问题排查

二、核心概念解析

2.1 基本定义

概念一：RAG（检索增强生成）

RAG（Retrieval-Augmented Generation）=检索 + 大模型生成。流程：用户问题 → 文档检索 → 相关片段 → 拼接上下文 → 大模型生成答案。作用：解决大模型幻觉、知识过时、无法接入私有数据等问题。

概念二：LangChain

LangChain 是大模型应用开发框架，提供文档加载、文本分割、向量封装、检索、问答链、记忆、Agent等模块，大幅降低 RAG 开发难度。

概念三：FAISS（Facebook 向量数据库）

FAISS（Facebook AI Similarity Search）是 Facebook 开源的轻量级向量检索库，特点：

纯本地、零服务、单文件存储；
CPU 高效检索，支持百万级向量；
速度快、体积小、易集成；
支持保存 / 加载向量库，避免重复计算。

概念四：Embedding（向量嵌入）

将文本（句子 / 段落）映射成高维数值向量，语义越接近，向量距离越近。轻量化常用：BGE、all-MiniLM、m3e-small。

2.2 关键术语解释

⚠️ 以下术语必须掌握：

Chunk（文本块）：长文档切分成短片段，避免上下文超限；
向量库（Vector Store）：存储文本向量 + 原文映射；
Retriever（检索器）：从向量库召回相似片段；
LLM（大语言模型）：基于检索结果生成答案；
幻觉（Hallucination）：模型编造不存在信息，RAG 可抑制。

2.3 轻量化 RAG 技术架构

plaintext

┌───────────────────────────────┐ │ 用户输入（自然语言） │ ├───────────────────────────────┤ │ LangChain 核心调度 │ ├───────────┬───────────┬────────┤ │ 文档加载 │ 向量存储 │ 模型生成│ │(DirectoryLoader)│(FAISS)│(本地LLM)│ ├───────────┴───────────┴────────┤ │ 本地文件系统（PDF/Word）│ └───────────────────────────────┘

三、技术选型（轻量化 + 私有化优先）

3.1 整体技术栈

Python：3.9+
RAG 框架：LangChain
向量数据库：FAISS（CPU 版）
文档加载：PyPDF2、python-docx、TextLoader
文本分割：RecursiveCharacterTextSplitter
向量模型：BGE-small-zh-v1.5（中文、轻量、离线）
大模型：Qwen-7B-Chat（4bit 量化、CPU 可跑）
Web 界面：Gradio（轻量、几行代码）

3.2 选型理由

全离线：无任何外网依赖，全程本地运行；
轻量：FAISS 无需服务，8G 内存即可跑通；
中文友好：BGE+Qwen 对中文优化，效果远超国外模型；
代码简洁：LangChain 封装度高，新手易上手；
成本极低：全开源，无需付费 API。

四、环境搭建（Windows/Linux/Mac 通用）

4.1 硬件最低要求

CPU：2 核
内存：8G
硬盘：50G SSD
GPU：可选（非必需）

4.2 依赖安装

bash

运行

# 创建虚拟环境 python3 -m venv rag-env source rag-env/bin/activate # 安装核心依赖 pip install langchain==0.1.10 pip install langchain-community==0.0.25 pip install pypdf2==3.0.1 python-docx==1.1.2 pip install sentence-transformers==2.5.1 pip install faiss-cpu==1.7.4 pip install transformers==4.38.2 torch==2.2.1 pip install gradio==4.21.0

4.3 离线模型下载（关键）

向量模型：BGE-small-zh-v1.5
大模型：Qwen-7B-Chat-4bit

下载后放到项目目录：

plaintext

./models/ ├─ bge-small-zh-v1.5/ └─ Qwen-7B-Chat-4bit/

五、全链路代码实战（可直接复制运行）

5.1 项目目录

plaintext

lightweight_rag/ ├─ docs/ # 你的知识库文档（PDF/Word/TXT） ├─ models/ # 离线模型 ├─ vector_db/ # FAISS 向量库（自动生成） └─ app.py # 主程序

5.2 第一步：文档加载与分割

python

运行

# app.py import os from typing import List from langchain.document_loaders import DirectoryLoader, PyPDFLoader, Docx2txtLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # ---------------------- 配置区 ---------------------- DOCS_PATH = "./docs" VECTOR_DB_PATH = "./vector_db" CHUNK_SIZE = 512 CHUNK_OVERLAP = 50 # ---------------------------------------------------- def load_documents() -> List: """ 批量加载 PDF、Word、TXT """ print("加载文档中...") loaders = [ DirectoryLoader(DOCS_PATH, glob="**/*.pdf", loader_cls=PyPDFLoader), DirectoryLoader(DOCS_PATH, glob="**/*.docx", loader_cls=Docx2txtLoader), DirectoryLoader(DOCS_PATH, glob="**/*.txt", loader_cls=TextLoader) ] docs = [] for loader in loaders: docs.extend(loader.load()) print(f"加载完成，共 {len(docs)} 个文档") return docs def split_documents(docs: List) -> List: """ 中文友好分割 """ print("分割文档中...") splitter = RecursiveCharacterTextSplitter( chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separators=["\n\n", "\n", "。", "，", "、"] ) chunks = splitter.split_documents(docs) print(f"分割完成，共 {len(chunks)} 个文本块") return chunks # 测试 if __name__ == "__main__": raw_docs = load_documents() chunks = split_documents(raw_docs) print("示例文本块：\n", chunks[0].page_content[:200])

代码说明：

自动遍历 docs 目录所有 PDF/Word/TXT；
中文分隔符优先按句号、逗号分割，语义更完整；
支持子目录递归加载。

5.3 第二步：FAISS 向量库构建与加载

python

运行

from langchain.vectorstores import FAISS from langchain.embeddings import HuggingFaceEmbeddings def create_faiss_vector_db(chunks: List) -> FAISS: """ 构建 FAISS 向量库 """ print("加载向量模型并构建向量库...") embeddings = HuggingFaceEmbeddings( model_name="./models/bge-small-zh-v1.5", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True} ) vector_db = FAISS.from_documents(chunks, embeddings) os.makedirs(VECTOR_DB_PATH, exist_ok=True) vector_db.save_local(VECTOR_DB_PATH) print("向量库保存成功") return vector_db def load_faiss_vector_db() -> FAISS: """ 加载本地向量库 """ print("加载向量库...") embeddings = HuggingFaceEmbeddings( model_name="./models/bge-small-zh-v1.5", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True} ) vector_db = FAISS.load_local( folder_path=VECTOR_DB_PATH, embeddings=embeddings, allow_dangerous_deserialization=True ) print("向量库加载完成") return vector_db # 测试检索 if __name__ == "__main__": if not os.path.exists(os.path.join(VECTOR_DB_PATH, "index.faiss")): raw_docs = load_documents() chunks = split_documents(raw_docs) vector_db = create_faiss_vector_db(chunks) else: vector_db = load_faiss_vector_db() query = "公司报销流程是什么？" results = vector_db.similarity_search(query, k=3) for idx, res in enumerate(results): print(f"\n【检索结果{idx+1}】来源：{res.metadata['source']}") print(res.page_content[:300])

代码说明：

FAISS 以文件形式保存，下次直接加载，无需重复计算向量；
检索返回 top-k 相似文本，语义匹配，不是关键词匹配；
全程离线，不联网。

5.4 第三步：本地大模型加载

python

运行

from langchain.llms import HuggingFacePipeline from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline def load_local_llm(): """ 加载本地 Qwen-7B-Chat-4bit """ print("加载本地大模型...") tokenizer = AutoTokenizer.from_pretrained( "./models/Qwen-7B-Chat-4bit", trust_remote_code=True, local_files_only=True ) model = AutoModelForCausalLM.from_pretrained( "./models/Qwen-7B-Chat-4bit", trust_remote_code=True, local_files_only=True, load_in_4bit=True, device_map="auto" ) pipe = pipeline( "text-generation", model=model, tokenizer=tokenizer, max_new_tokens=512, temperature=0.7, repetition_penalty=1.1 ) llm = HuggingFacePipeline(pipeline=pipe) print("大模型加载完成") return llm # 测试 if __name__ == "__main__": llm = load_local_llm() print(llm.invoke("你好，请介绍自己"))

代码说明：

4bit 量化，8G 内存可流畅运行；
local_files_only=True，强制离线；
可替换为 ChatGLM、Llama3 等本地模型。

5.5 第四步：构建 RAG 问答链

python

运行

from langchain.chains import RetrievalQA def build_rag_chain(vector_db: FAISS, llm) -> RetrievalQA: """ 构建 RAG 链 """ print("构建 RAG 链...") retriever = vector_db.as_retriever(search_kwargs={"k": 3}) rag_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=True ) print("RAG 链构建完成") return rag_chain # 测试问答 if __name__ == "__main__": if not os.path.exists(os.path.join(VECTOR_DB_PATH, "index.faiss")): raw_docs = load_documents() chunks = split_documents(raw_docs) vector_db = create_faiss_vector_db(chunks) else: vector_db = load_faiss_vector_db() llm = load_local_llm() rag_chain = build_rag_chain(vector_db, llm) result = rag_chain.invoke({"query": "公司报销流程是什么？"}) print("\n答案：\n", result["result"]) print("\n来源：") for doc in result["source_documents"]: print("-", doc.metadata["source"])

代码说明：

chain_type="stuff"：直接把检索结果拼进 Prompt；
return_source_documents=True：返回来源，可溯源，抑制幻觉。

5.6 第五步：Gradio Web 界面

python

运行

import gradio as gr # 全局变量 vector_db = None llm = None rag_chain = None def init_system(): global vector_db, llm, rag_chain if not os.path.exists(os.path.join(VECTOR_DB_PATH, "index.faiss")): raw_docs = load_documents() chunks = split_documents(raw_docs) vector_db = create_faiss_vector_db(chunks) else: vector_db = load_faiss_vector_db() llm = load_local_llm() rag_chain = build_rag_chain(vector_db, llm) def answer_question(question): if not question.strip(): return "请输入问题" try: res = rag_chain.invoke({"query": question}) answer = res["result"] source = "\n来源：\n" + "\n".join([f"- {d.metadata['source']}" for d in res["source_documents"]]) return answer + source except Exception as e: return f"错误：{str(e)}" def upload_file(file): try: save_path = os.path.join(DOCS_PATH, os.path.basename(file.name)) with open(save_path, "wb") as f: f.write(file.read()) # 重建向量库 raw_docs = load_documents() chunks = split_documents(raw_docs) global vector_db, rag_chain vector_db = create_faiss_vector_db(chunks) rag_chain = build_rag_chain(vector_db, llm) return f"上传成功：{file.name}，知识库已更新" except Exception as e: return f"上传失败：{str(e)}" # 界面 with gr.Blocks(title="轻量化 RAG 知识库") as demo: gr.Markdown("# 🚀 LangChain+FAISS 轻量化 RAG 知识库") gr.Markdown("全离线、私有化、CPU 可跑、可溯源") question = gr.Textbox(label="输入问题", lines=3) answer = gr.Textbox(label="回答", lines=10) submit_btn = gr.Button("提问", variant="primary") submit_btn.click(answer_question, inputs=question, outputs=answer) gr.Markdown("### 📤 上传文档更新知识库") file_upload = gr.File(file_types=[".pdf", ".docx", ".txt"]) upload_btn = gr.Button("上传并更新") upload_result = gr.Textbox(label="上传结果") upload_btn.click(upload_file, inputs=file_upload, outputs=upload_result) # 启动 if __name__ == "__main__": init_system() demo.launch(server_name="0.0.0.0", server_port=7860, share=False)

代码说明：