LangChain数据连接与检索：企业级知识库构建全攻略

@[toc]

1. 为何企业级RAG应用必须实现数据连接与检索？

快速编写一个调用LLM的脚本并不复杂，但若用户询问“今日股市行情”或“公司最新产品发布信息”，模型要么回复“不知道”，要么基于过时数据给出看似合理实则谬误的答案。

这正是LangChain数据连接与检索模块（即RAG范式）所要攻克的痛点：让大语言模型能够实时访问并利用外部知识。通用大模型的知识边界止于训练数据截止日期，而现实世界的信息持续更新。企业内部私有数据、实时市场动态、特定领域专业知识——这些都无法通过重新训练模型来获取。

LangChain的数据连接与检索模块（通常称为RAG, Retrieval Augmented Generation）提供了一整套标准解决方案，将外部知识库与LLM智能结合，从而构建出真正可落地的企业级AI应用。

2. RAG核心架构：四步构建企业知识库系统

graph TDA[原始资料] --> B[文档加载器]B --> C[统一文档格式]C --> D[文本分割器]D --> E[文本块 chunks]E --> F[嵌入模型]F --> G[向量表示]G --> H[向量存储]H --> I[相似性搜索]I --> J[最相关文档]J --> K[LLM 生成回答]K --> L[最终答案]style A fill:#e1f5festyle H fill:#f3e5f5style K fill:#e8f5e8

2.1 文档加载器：统一数据入口

文档加载器作为整个数据管道的起点，其核心价值在于数据格式的统一。无论数据源自网页、PDF、Word、Excel、数据库，还是Slack聊天记录，文档加载器都能将其转换为统一的Document对象。

为什么必须统一格式？

• 标准化处理：后续的文本分割、嵌入、检索均基于同一数据结构，避免适配成本。
• 元数据保留：来源、创建时间、作者等关键信息得以完整保留。
• 扩展性：新增数据源只需实现对应的加载器接口，无需改动下游流程。

# 实战：加载多种格式的企业文档
from langchain_community.document_loaders import (
    TextLoader,
    PyPDFLoader,
    UnstructuredWordDocumentLoader,
    WebBaseLoader
)

# 加载本地文本文件（如产品说明书）
text_loader = TextLoader("data/requirements.txt")
text_docs = text_loader.load()

# 加载 PDF 文件（保留页面信息，适合财报/合同）
pdf_loader = PyPDFLoader("data/report.pdf")
pdf_docs = pdf_loader.load()

# 加载网页内容（如公司公告页）
web_loader = WebBaseLoader(["https://example.com/blog"])
web_docs = web_loader.load()

print(f"已加载 {{len(text_docs)}} 个文本文档")
print(f"已加载 {{len(pdf_docs)}} 个PDF页面")
print(f"已加载 {{len(web_docs)}} 个网页文档")

架构决策记录：LangChain 提供了 100+ 种文档加载器，优先选用 langchain-community 中的加载器而非自研，原因有三：社区维护的加载器经过充分测试；自动处理编码、格式转换等复杂问题；拥有统一的错误处理机制。

2.2 文本分割器：智能分块的艺术

文本分割器负责将长文档切割为较小的块（chunks），这是RAG系统中最关键也最容易出错的一环。

为什么需要分块？

1. 上下文限制：LLM 有固定的上下文窗口（如 GPT-4 的 128K），但单篇文档可能远超此限。
2. 检索精度：过大的文本块夹杂大量无关信息，反而降低检索相关性。
3. 计算效率：小块嵌入计算和存储更高效，便于大规模应用。

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建文本分割器 —— 这是最常用的递归分割策略
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,          # 每块最大字符数（中文场景推荐800-1200）
    chunk_overlap=200,        # 块之间重叠字符数，防止关键信息被切断
    length_function=len,        # 长度计算函数
    separators=["nn", "n", "。", "！", "？", "；", "，", " ", ""]  # 中文友好的分隔符优先级
)

# 对文档进行分割
documents = [...]  # 从加载器获取的文档列表
chunks = text_splitter.split_documents(documents)

print(f"原始文档数量: {{len(documents)}}")
print(f"分割后文本块数量: {{len(chunks)}}")
print(f"平均每块长度: {{sum(len(chunk.page_content) for chunk in chunks) / len(chunks):.0f}} 字符")

关键参数调优经验：

• chunk_size=1000：平衡检索精度与上下文完整性，中文场景可适当调整至800-1200。
• chunk_overlap=200：避免重要信息被切分到两个块之间而丢失关联。
• 中文特殊处理：默认分隔符针对英文优化，需添加“。”、“！”等中文标点。

常见踩坑：初次使用时若设置chunk_size=5000，以为能保留更多上下文，结果检索相关性下降30%（块太大包含不相关信息），嵌入计算时间增加3倍，LLM处理时也容易“淹没”在大量冗余信息中。

2.3 嵌入模型：从文字到向量的魔法

嵌入模型将文本转换为高维向量，这些向量能够精准捕捉语义信息。这是让计算机“理解”文本含义的关键步骤。

语义相似性的威力：

• “苹果”和“水果”：词面不同，但语义接近 → 向量距离近
• “苹果”和“iPhone”：词面不同，但上下文相关 → 向量距离近
• “苹果”和“橙子”：都是水果 → 向量距离中等
• “苹果”和“汽车”：语义无关 → 向量距离远

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
import numpy as np

# 方案1：使用 OpenAI 嵌入（效果最优，但需 API 调用）
openai_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key=os.getenv("OPENAI_API_KEY")   # ⚠️ 切勿硬编码 API Key
)

# 方案2：使用本地 HuggingFace 模型（免费且支持离线）
hf_embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",   # 专为中文优化的嵌入模型
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 测试语义相似性
texts = ["苹果是一种水果", "iPhone 是苹果公司的产品", "汽车需要汽油"]
embeddings = hf_embeddings.embed_documents(texts)

def cosine_similarity(vec1, vec2):
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

sim1 = cosine_similarity(embeddings[0], embeddings[1])   # 苹果 vs iPhone
sim2 = cosine_similarity(embeddings[0], embeddings[2])   # 苹果 vs 汽车

print(f"语义相似度 - 苹果/iPhone: {{sim1:.3f}}")
print(f"语义相似度 - 苹果/汽车: {{sim2:.3f}}")

模型选择决策：

• 生产环境：优先使用 text-embedding-3-small，1536 维，效果稳定。
• 中文场景：BAAI/bge-small-zh-v1.5 专门针对中文语义优化。
• 成本敏感：all-MiniLM-L6-v2 英文效果好，仅 384 维。

2.4 向量存储：高效检索的基石

向量存储负责保存嵌入向量及元数据，并提供高效的相似性搜索能力。可以把它理解为一个专为语义搜索设计的数据库。

from langchain_chroma import Chroma
from langchain_community.vectorstores import FAISS
import chromadb

# 方案1：ChromaDB —— 轻量级，适合快速原型验证
vectorstore_chroma = Chroma.from_documents(
    documents=chunks,
    embedding=openai_embeddings,
    persist_directory="./chroma_db"   # 持久化存储路径
)

# 方案2：FAISS —— Facebook 开源，性能碾压
vectorstore_faiss = FAISS.from_documents(
    documents=chunks,
    embedding=hf_embeddings
)
vectorstore_faiss.sa ve_local("faiss_index")

# 相似性搜索示例
query = "如何配置 LangChain 的环境变量？"
results = vectorstore_chroma.similarity_search(query, k=3)
print("最相关的 3 个文档块：")
for i, doc in enumerate(results, 1):
    print(f"n{{i}}. 相似度: {{doc.metadata.get('score', 'N/A')}}")
    print(f"  来源: {{doc.metadata.get('source', '未知')}}")
    print(f"  内容片段: {{doc.page_content[:200]}}...")

向量存储选型对比：

3. 完整实战：构建企业知识库问答系统

现在，将四个模块串联起来，搭建一个完整的企业知识库问答系统。

3.1 环境准备与依赖安装

# requirements.txt —— 建议使用 0.3+ 版本
# LangChain 核心库
langchain>=0.3.0
langchain-community>=0.3.0
langchain-openai>=0.1.0

# 文档加载器依赖
pypdf>=3.17.0               # PDF 解析
unstructured>=0.15.0         # 多种办公文档格式
beautifulsoup4>=4.12.0        # HTML 解析

# 向量存储
chromadb>=0.5.0               # ChromaDB
faiss-cpu>=1.7.4             # FAISS（CPU版）

# 嵌入模型
openai>=1.12.0               # OpenAI 嵌入
sentence-transformers>=2.2.0  # HuggingFace 嵌入

# 其他工具
python-dotenv>=1.0.0          # 环境变量管理

# 一键安装
pip install -r requirements.txt

# 创建 .env 文件并填入 API Key
echo "OPENAI_API_KEY=your-api-key-here" > .env

3.2 完整代码实现

import os
from dotenv import load_dotenv
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# 加载环境变量
load_dotenv()

class KnowledgeBaseQA:
    def __init__(self, data_dir="./data", persist_dir="./chroma_db"):
        """初始化企业知识库问答系统"""
        self.data_dir = data_dir
        self.persist_dir = persist_dir
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.llm = ChatOpenAI(
            model="gpt-4o-mini",
            temperature=0,   # 设为 0 保证回复的确定性
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.vectorstore = None

    def build_knowledge_base(self):
        """构建知识库：加载→分割→嵌入→存储，四步一气呵成"""
        print("? 开始构建知识库...")

        # 1. 加载文档（支持 PDF，可扩展为更多格式）
        loader = DirectoryLoader(
            self.data_dir,
            glob="**/*.pdf",   # 可追加 **/*.txt, **/*.docx 等
            loader_cls=PyPDFLoader,
            show_progress=True
        )
        documents = loader.load()
        print(f"✅ 已加载 {{len(documents)}} 个文档")

        # 2. 智能文本分割
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["nn", "n", "。", "！", "？", "；", "，", " ", ""]
        )
        chunks = text_splitter.split_documents(documents)
        print(f"✅ 分割为 {{len(chunks)}} 个文本块")

        # 3. 创建向量存储并持久化
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_dir
        )
        print(f"✅ 向量存储已保存至 {{self.persist_dir}}")
        return self

    def load_existing_knowledge_base(self):
        """加载已存在的知识库，避免重复构建"""
        if os.path.exists(self.persist_dir):
            self.vectorstore = Chroma(
                persist_directory=self.persist_dir,
                embedding_function=self.embeddings
            )
            print(f"✅ 从 {{self.persist_dir}} 加载已有知识库")
            return True
        return False

    def ask_question(self, question, k=4):
        """向知识库提出问题，返回增强后的答案及参考来源"""
        if not self.vectorstore:
            print("❌ 请先构建或加载知识库")
            return None

        # 创建检索增强生成链
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",   # 简单合并多个上下文片段
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": k}),   # 返回最相关的 k 个文档
            return_source_documents=True
        )

        # 执行查询
        result = qa_chain.invoke({"query": question})

        # 输出结构化的结果
        print(f"n? 问题: {{question}}")
        print(f"n? 回答: {{result['result']}}")
        print(f"n? 参考来源:")
        for i, doc in enumerate(result['source_documents'], 1):
            print(f"{{i}}. {{doc.metadata.get('source', '未知')}} (页 {{doc.metadata.get('page', 'N/A')}})")
            print(f"   相关片段: {{doc.page_content[:150]}}...")
        return result

# 使用示例
if __name__ == "__main__":
    # 初始化系统
    kb_qa = KnowledgeBaseQA()

    # 先尝试加载已有知识库，若不存在则新建
    if not kb_qa.load_existing_knowledge_base():
        print("未找到已有知识库，开始构建...")
        kb_qa.build_knowledge_base()

    # 示例问答
    questions = [
        "LangChain 的文档加载器支持哪些格式？",
        "文本分割时 chunk_size 设置多少合适？",
        "如何选择嵌入模型？"
    ]
    for q in questions:
        kb_qa.ask_question(q)
        print("n" + "="*50 + "n")

3.3 运行输出与行为分析

? 开始构建知识库...
✅ 已加载 15 个文档
✅ 分割为 127 个文本块
✅ 向量存储已保存至 ./chroma_db

? 问题: LangChain 的文档加载器支持哪些格式？
? 回答: LangChain 内置了百余种文档加载器，覆盖以下常见格式：
1. **文本格式**：TXT、CSV、JSON、Markdown
2. **办公文档**：PDF、Word (.docx)、Excel (.xlsx)、PowerPoint (.pptx)
3. **网页内容**：HTML、在线文章、博客
4. **代码文件**：Python、Java、JavaScript 等源文件
5. **社交媒体**：Twitter、Slack、Discord 导出记录
6. **数据库**：SQL 查询结果、MongoDB 文档
建议根据实际数据源选用对应加载器，私有格式可自行实现加载器接口。

? 参考来源:
1. langchain_docs.pdf (页 23)
   相关片段: LangChain 文档加载器模块支持 100+ 种数据源格式转换...
2. best_practices.pdf