Python + AI 实战：用 LangChain 搭建企业级 RAG 知识库

在 AI 大模型时代，如何让 AI "读懂"企业内部文档并精准回答？RAG（检索增强生成）是目前最成熟的落地方案。

请严格根据以下参考资料回答用户的问题。

要求： 1. 只基于参考资料回答，不要编造信息 2. 如果参考资料中没有相关内容，请明确告知用户 3. 回答时引用来源文档名称 4. 使用清晰的结构化格式 参考资料： {context} 用户问题：{question} 回答：""" def __init__(self): settings = get_settings() self.llm = ChatOpenAI( model=settings.llm_model, openai_api_key=settings.openai_api_key, openai_api_base=settings.openai_base_url, temperature=0.1, ) vectorstore_mgr = VectorStoreManager() self.retriever = vectorstore_mgr.as_retriever() self.chain = self._build_chain() def _format_docs(self, docs): """格式化检索结果""" formatted = [] for i, doc in enumerate(docs, 1): source = doc.metadata.get("source_file", "未知来源") formatted.append(f"[文档{i}: {source}]\n{doc.page_content}") return "\n\n---\n\n".join(formatted) def _build_chain(self): """构建 RAG 链""" prompt = ChatPromptTemplate.from_template(self.PROMPT_TEMPLATE) chain = ( RunnableParallel({ "context": self.retriever | self._format_docs, "question": RunnablePassthrough(), }) | prompt | self.llm | StrOutputParser() ) return chain def query(self, question: str) -> dict: """执行查询""" # 先检索，获取源文档 docs = self.retriever.invoke(question) sources = list({ doc.metadata.get("source_file", "未知") for doc in docs }) # 生成回答 answer = self.chain.invoke(question) return { "question": question, "answer": answer, "sources": sources, } 4.6 FastAPI 服务封装 # app/api/schemas.py from pydantic import BaseModel class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): question: str answer: str sources: list[str] class IngestRequest(BaseModel): directory: str = "./data" # app/api/routes.py from fastapi import APIRouter from app.api.schemas import QueryRequest, QueryResponse, IngestRequest from app.chain import RAGChain from app.document_loader import DocumentLoader from app.chunker import TextChunker from app.vectorstore import VectorStoreManager router = APIRouter() # 初始化组件 rag_chain = RAGChain() @router.post("/query", response_model=QueryResponse) async def query(request: QueryRequest): """知识库问答""" result = rag_chain.query(request.question) return QueryResponse(**result) @router.post("/ingest") async def ingest_documents(request: IngestRequest): """导入文档到知识库""" loader = DocumentLoader() documents = loader.load_directory(request.directory) chunker = TextChunker() chunks = chunker.split(documents) vectorstore = VectorStoreManager() vectorstore.add_documents(chunks) return { "status": "success", "documents_loaded": len(documents), "chunks_created": len(chunks), } # app/main.py from fastapi import FastAPI from app.api.routes import router app = FastAPI( title="企业级 RAG 知识库", description="基于 LangChain 的 RAG 知识库问答系统", version="1.0.0", ) app.include_router(router, prefix="/api/v1") @app.get("/health") async def health(): return {"status": "ok"} 五、使用流程 5.1 启动服务 # 1. 配置环境变量 cp .env.example .env # 编辑 .env，填入 OPENAI_API_KEY # 2. 导入文档 curl -X POST http://localhost:8000/api/v1/ingest \ -H "Content-Type: application/json" \ -d '{"directory": "./data"}' # 3. 启动服务 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload 5.2 问答测试 curl -X POST http://localhost:8000/api/v1/query \ -H "Content-Type: application/json" \ -d '{"question": "公司的年假政策是什么？"}' 5.3 完整调用流程图 用户提问 │ ▼ ┌──────────┐ │ FastAPI │ 接收 POST /query │ 路由层 │ └────┬─────┘ │ ▼ ┌──────────┐ │ RAG Chain │ 核心 RAG 链 └────┬─────┘ │ ├─── 1. 问题向量化 (Embedding) │ ├─── 2. 向量检索 (ChromaDB Top-K) │ │ │ ▼ │ ┌──────────┐ │ │ 相似文档块 │ ← source_file, content │ └──────────┘ │ ├─── 3. 组装 Prompt (context + question) │ ├─── 4. LLM 生成回答 (GPT-4o-mini) │ ▼ ┌──────────────┐ │ 结构化响应返回 │ {answer, sources} └──────────────┘ 六、进阶优化建议 优化方向方案效果检索精度使用 MMR 搜索 + 重排序（Reranker）减少冗余，提升相关性长文档处理父子文档检索（Parent-Child Retriever）兼顾精度与上下文完整性多轮对话加入对话历史管理（Memory）支持追问与上下文关联权限控制按部门/角色建立独立 Collection数据隔离，安全合规缓存层相似问题缓存（Semantic Cache）降低 API 调用成本评估体系RAGAS 框架评估检索/生成质量量化系统表现重排序示例（Reranker） from langchain_community.document_compressors import CohereRerank # 先多检索，再精排 retriever = vectorstore.as_retriever(search_kwargs={"k": 20}) compressor = CohereRerank(top_n=5) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=retriever, ) 七、总结 本文从零搭建了一套企业级 RAG 知识库系统，核心组件包括： 文档加载器 — 支持 PDF、Word、TXT、Markdown 多格式智能分块 — RecursiveCharacterTextSplitter 兼顾语义完整性和检索粒度向量存储 — ChromaDB 轻量级本地向量数据库RAG 链 — LangChain Runnable 接口，灵活可扩展API 服务 — FastAPI 封装，开箱即用 完整代码已开源，读者可根据企业实际需求进行定制化开发。

本文将手把手带你用 LangChain 搭建一套完整的企业级 RAG 知识库系统。

一、什么是 RAG？ RAG（Retrieval-Augmented Generation）的核心思想是：先检索，再生成。

", "！", "？", ".", "!", "?", " ", ""], ) def split(self, documents: list[Document]) -> list[Document]: chunks = self.splitter.split_documents(documents) print(f"[TextChunker] 分块完成: {len(documents)} → {len(chunks)} 个片段") return chunks 4.4 向量化与存储 # app/embedder.py from langchain_openai import OpenAIEmbeddings from app.config import get_settings def get_embeddings(): settings = get_settings() return OpenAIEmbeddings( model=settings.embedding_model, openai_api_key=settings.openai_api_key, openai_api_base=settings.openai_base_url, ) # app/vectorstore.py from langchain_community.vectorstores import Chroma from langchain_core.documents import Document from app.config import get_settings from app.embedder import get_embeddings class VectorStoreManager: """向量数据库管理器""" def __init__(self): self.settings = get_settings() self.embeddings = get_embeddings() self.db = self._load_or_create() def _load_or_create(self) -> Chroma: """加载已有数据库或创建新的""" return Chroma( collection_name=self.settings.collection_name, embedding_function=self.embeddings, persist_directory=self.settings.vectordb_path, ) def add_documents(self, documents: list[Document]) -> None: """将文档块向量化并存入数据库""" batch_size = 100 for i in range(0, len(documents), batch_size): batch = documents[i : i + batch_size] self.db.add_documents(batch) print(f"[VectorStore] 已入库 {min(i + batch_size, len(documents))}/{len(documents)}") self.db.persist() def as_retriever(self): """返回检索器""" return self.db.as_retriever( search_type="mmr", # 最大边际相关性，兼顾相关性和多样性 search_kwargs={"k": self.settings.top_k}, ) 4.5 RAG 链（核心） # app/chain.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableParallel from app.config import get_settings from app.vectorstore import VectorStoreManager class RAGChain: """RAG 问答链""" PROMPT_TEMPLATE = """你是一个专业的企业知识库助手。

将用户问题与文档库中的相关片段匹配，再将检索到的内容作为上下文喂给大模型，从而获得准确且有依据的回答。

传统 LLM vs RAG 对比 ┌─────────────────────────────────────────────────────────┐ │ 传统 LLM 问答 │ │ │ │ 用户提问 ──────────► 大模型 ──────────► 回答 │ │ ▲ │ │ │ │ │ 仅靠训练知识 │ │ (可能过时/缺乏企业信息) │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ RAG 问答 │ │ │ │ 用户提问 ──► 检索相关文档 ──► 组合 Prompt ──► 回答 │ │ ▲ │ │ │ │ │ 企业知识库(向量数据库) │ │ (实时、准确、可溯源) │ └─────────────────────────────────────────────────────────┘ 二、系统架构总览 ┌──────────┐ ┌───────────┐ ┌──────────────┐ ┌──────────┐ │ 文档输入 │───►│ 文档解析与 │───►│ 向量化与存储 │───►│ 向量数据库 │ │ PDF/Word │ │ 文本分块 │ │ (Embedding) │ │ ChromaDB │ │ /TXT/MD │ │ │ │ │ │ │ └──────────┘ └───────────┘ └──────────────┘ └──────────┘ │ ▼ ┌──────────┐ ┌───────────┐ ┌──────────────┐ ┌──────────────┐ │ 最终回答 │◄───│ 大模型生成 │◄───│ Prompt 组装 │◄───│ 相似度检索 │ │ (含引用源) │ │ (LLM) │ │ (上下文+问题) │ │ (Top-K 匹配) │ └──────────┘ └───────────┘ └──────────────┘ └──────────────┘ 三、环境准备 3.1 安装依赖 pip install langchain langchain-openai langchain-community pip install chromadb sentence-transformers pip install pypdf python-docx markdown pip install fastapi uvicorn 3.2 项目结构 enterprise-rag/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置管理 │ ├── document_loader.py # 文档加载与解析 │ ├── chunker.py # 文本分块 │ ├── embedder.py # 向量化 │ ├── vectorstore.py # 向量数据库管理 │ ├── retriever.py # 检索器 │ ├── chain.py # RAG 链 │ └── api/ │ ├── routes.py # API 路由 │ └── schemas.py # 数据模型 ├── data/ # 原始文档目录 ├── vectordb/ # 向量数据库持久化目录 ├── .env # 环境变量 └── requirements.txt 四、核心模块实现 4.1 配置管理 # app/config.py from pydantic_settings import BaseSettings from functools import lru_cache class Settings(BaseSettings): # LLM 配置 openai_api_key: str = "" openai_base_url: str = "https://api.openai.com/v1" llm_model: str = "gpt-4o-mini" # Embedding 配置 embedding_model: str = "text-embedding-3-small" # 向量数据库 vectordb_path: str = "./vectordb" collection_name: str = "enterprise_docs" # 分块参数 chunk_size: int = 1000 chunk_overlap: int = 200 # 检索参数 top_k: int = 5 class Config: env_file = ".env" @lru_cache() def get_settings() -> Settings: return Settings() 4.2 文档加载与解析 # app/document_loader.py from pathlib import Path from langchain_community.document_loaders import ( PyPDFLoader, Docx2txtLoader, TextLoader, UnstructuredMarkdownLoader, ) from langchain_core.documents import Document from typing import List class DocumentLoader: """多格式文档加载器""" LOADER_MAP = { ".pdf": PyPDFLoader, ".docx": Docx2txtLoader, ".txt": TextLoader, ".md": UnstructuredMarkdownLoader, } def load_directory(self, dir_path: str) -> List[Document]: """加载目录下所有支持的文档""" documents = [] path = Path(dir_path) for file_path in path.rglob("*"): if file_path.suffix.lower() in self.LOADER_MAP: loader_cls = self.LOADER_MAP[file_path.suffix.lower()] loader = loader_cls(str(file_path)) docs = loader.load() # 为每个文档添加元数据 for doc in docs: doc.metadata["source_file"] = file_path.name doc.metadata["source_path"] = str(file_path) documents.extend(docs) print(f"[DocumentLoader] 共加载 {len(documents)} 个文档片段") return documents 4.3 文本分块 # app/chunker.py from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_core.documents import Document from app.config import get_settings class TextChunker: """智能文本分块器""" def __init__(self): settings = get_settings() self.splitter = RecursiveCharacterTextSplitter( chunk_size=settings.chunk_size, chunk_overlap=settings.chunk_overlap, separators=["\n\n", "\n", "。

RAG 是 AI 落地最务实的路径之一，值得每位 Python 开发者掌握。