LangChain 多文件 RAG 应用

概述

本项目演示如何使用 LangChain 构建一个基于多文档的智能问答系统（Retrieval-Augmented Generation，RAG），能够从文档集合中检索相关信息并生成高质量的回答。

技术架构

代码分析

导入模块

import os
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.chat_models import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

导入所需的 LangChain 组件：

• DirectoryLoader, TextLoader: 用于从目录加载文档
• RecursiveCharacterTextSplitter: 递归文本分割器
• DashScopeEmbeddings: DashScope 文本嵌入模型
• FAISS: 向量检索引擎
• ChatTongyi: 通义千问语言模型
• ChatPromptTemplate: 聊天气泡提示模板
• StrOutputParser: 字符串输出解析器

API 配置

# 获取 API Key
DASHSCOPE_API_KEY = os.getenv('DASHSCOPE_API_KEY')
if not DASHSCOPE_API_KEY:
    raise ValueError("请设置环境变量 DASHSCOPE_API_KEY")

从环境变量获取 API 密钥并验证其有效性，这是安全的最佳实践，避免硬编码敏感信息。

文档加载与索引构建

def load_documents_and_create_index(file_dir: str = './docs', persist_dir: str = './langchain_storage'):
    # 创建嵌入模型 - 将文本转换为向量表示
    embeddings = DashScopeEmbeddings(
        model="text-embedding-v1",  # 阿里云文本嵌入模型
        dashscope_api_key=DASHSCOPE_API_KEY,  # API访问凭证
    )
    
    # 检查索引是否已存在 - 实现缓存机制，提升性能
    if os.path.exists(persist_dir):
        try:
            # 从存储中加载已训练好的索引
            vector_store = FAISS.load_local(
                persist_dir, 
                embeddings, 
                allow_dangerous_deserialization=True  # 允许反序列化（注意安全风险）
            )
            print("从存储加载索引成功")
            return vector_store
        except Exception as e:
            print(f"加载索引失败: {e}，将重新创建索引")
    
    # 如果索引不存在或加载失败，创建新索引
    if not os.path.exists(file_dir):
        print(f"文档目录 {file_dir} 不存在")
        return None
    
    # 加载目录下的所有txt文件 - 目前仅支持txt格式
    loader = DirectoryLoader(
        file_dir,          # 要扫描的目录
        glob="**/*.txt",   # 文件匹配模式（递归匹配所有txt文件）
        loader_cls=TextLoader,  # 使用TextLoader加载器
        loader_kwargs={"encoding": "utf-8"}  # 指定文件编码，保证中文正常处理
    )
    documents = loader.load()  # 执行文档加载
    print(f"加载了 {len(documents)} 个文档")
    
    if not documents:
        print("没有找到任何文档")  # 提示用户可能的配置问题
        return None
    
    # 文本分块处理 - 将长文档分割为适合模型处理的小块
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,      # 每个文本块的最大字符数
        chunk_overlap=200,    # 相邻块之间的重叠字符数（保持上下文连续性）
        length_function=len,  # 计算文本长度的方法
    )
    chunks = text_splitter.split_documents(documents)  # 执行分块
    print(f"文本被分割成 {len(chunks)} 个块")
    
    # 创建向量索引 - 将文本块转换为向量并建立索引
    vector_store = FAISS.from_documents(chunks, embeddings)
    
    # 保存索引到本地 - 实现持久化，下次可直接加载
    os.makedirs(persist_dir, exist_ok=True)  # 确保目录存在
    vector_store.save_local(persist_dir)     # 保存向量索引
    print(f"索引已保存到 {persist_dir}")
    
    return vector_store

这个函数实现了文档加载、文本分块、向量索引构建和持久化存储的完整流程。它具有缓存机制，避免重复计算，提升性能。

问答链构建

def create_qa_chain(llm):
    # QA Prompt模板 - 定义AI助手的行为和上下文处理方式
    qa_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个乐于助人的AI助手。
根据以下上下文内容回答用户的问题。如果上下文中没有相关信息，请如实说明。
你总是用中文回复用户。

上下文内容:
{context}"""),  # 系统角色设定，指导AI如何使用上下文信息
        ("human", "{question}")  # 用户问题占位符
    ])
    
    # 创建问答链（LCEL管道语法）
    # 执行流程：prompt -> llm -> output_parser
    # 1. 用qa_prompt模板格式化输入
    # 2. 传递给大语言模型处理
    # 3. 使用字符串解析器处理输出
    qa_chain = qa_prompt | llm | StrOutputParser()
    
    return qa_chain

使用 LCEL (LangChain Expression Language) 构建问答链，将提示模板、语言模型和输出解析器组合成一个处理管道。

主执行流程

def main():
    # 配置大语言模型 - 使用DeepSeek-V3模型
    llm = ChatTongyi(
        model_name="deepseek-v3",        # 指定使用的模型名称
        dashscope_api_key=DASHSCOPE_API_KEY  # API访问凭证
    )
    
    # 加载文档并创建索引 - 如果索引已存在则直接加载，否则重新创建
    vector_store = load_documents_and_create_index()
    if vector_store is None:
        print("无法创建索引，程序退出")  # 提供明确的错误提示
        return
    
    # 创建问答链 - 将prompt模板、LLM和输出解析器组合成完整处理链条
    qa_chain = create_qa_chain(llm)
    
    # 定义查询问题 - 这是用户想要咨询的内容
    query = "介绍下雇主责任险"
    print(f"\n用户查询: {query}\n")
    
    # 相似度搜索 - 从向量数据库中找到与查询最相关的文档片段
    # k=5表示返回最相似的5个文档片段
    docs = vector_store.similarity_search(query, k=5)
    
    # 显示召回的文档内容 - 让用户了解AI参考了哪些信息
    print("===== 召回的文档内容 =====")
    if docs:
        for i, doc in enumerate(docs):
            print(f"\n文档片段 {i+1}:")
            print(f"内容: {doc.page_content[:200]}...")  # 仅显示前200字符作为预览
            print(f"来源: {doc.metadata.get('source', '未知')}")  # 显示文档来源
    else:
        print("没有召回任何文档内容")  # 检索失败的提示
    print("===========================\n")
    
    # 格式化上下文 - 将检索到的文档片段合并为AI可理解的上下文
    context = "\n\n".join(doc.page_content for doc in docs)
    
    # 执行问答链 - 使用组装好的处理链条生成最终答案
    print("===== AI 回复 =====")
    response = qa_chain.invoke({"context": context, "question": query})  # 同步调用
    print(response)
    print("===================\n")

主函数执行完整的RAG问答流程：初始化模型、加载文档索引、构建问答链、执行查询并展示结果。

核心概念

1. RAG（检索增强生成）

RAG是一种结合信息检索和语言生成的技术，通过从外部知识源检索相关信息来增强语言模型的生成能力，特别是在处理特定领域或私有数据时非常有效。

2. 向量索引

将文档转换为向量表示，并建立索引以支持快速相似性搜索。FAISS是Facebook开发的高效相似性搜索库，能够快速找到与查询向量最相似的文档片段。

3. 文本分块

将长文档分割为较小的块，确保每个块都能适应语言模型的上下文窗口限制，同时保持内容的连贯性。

应用场景

这种多文件RAG系统适用于：

• 企业知识库问答
• 学术文献检索与摘要
• 法律文档查询
• 技术支持文档问答
• 医疗知识库查询

扩展思路

可以进一步扩展此示例：

• 支持更多文档格式（PDF、Word、HTML等）
• 实现更高级的文本分块策略
• 添加查询改写功能
• 实现多轮对话记忆
• 添加结果评估机制