基于LangChain实现RAG的离线部分

本文介绍了使用LangChain和OpenAIEmbeddings构建RAG（检索增强生成）离线处理流程的方法。主要内容包括：1.核心工作流程分为文档加载、分割、嵌入和存储四个步骤；2.详细说明了环境配置和依赖安装要求；3.提供了模块化代码实践方案，包括配置管理、文档处理和向量转换；4.展示了如何验证嵌入向量生成结果。文章还顺带提及了大模型学习资源，包括书籍、行业报告和教程等资料的获取方式。该方案

码农Q！

585人浏览 · 2025-10-23 17:07:36

码农Q！ · 2025-10-23 17:07:36 发布

基于LangChain实现RAG的离线部分

使用 LangChain 和 OpenAI Embeddings 构建 RAG 离线处理流程的问题。

RAG 的离线部分，这个任务的核心是索引（Indexing）过程。它包括加载文档、将其分割成小块、为每个小块创建向量嵌入，并最终将这些嵌入存储到一个专门的数据库（向量存储）中，以备后续的快速检索。

1. 核心概念与工作流程

RAG 的离线流程（索引）可以分解为以下四个步骤：

加载 (Load): 从数据源（这里是从 PDF、DOC/DOCX 等多源文件）加载文档内容。
分割 (Split): 将加载的长文档分割成更小的、语义完整的文本块 (Chunks)。这对于提高检索精度和适应模型上下文窗口至关重要。
嵌入 (Embed): 使用 OpenAI 的 Embedding 模型（如 text-embedding-3-small）将每个文本块转换成一个数值向量（Embedding）。
存储 (Store): 将文本块及其对应的向量嵌入存储到一个向量数据库 (Vector Store) 中，并创建索引以便快速进行相似度搜索。

2. 所需依赖安装

本文推荐使用 uv 进行项目依赖管理。当然也可以使用 pip 进行相应的替换。python版本为3.10。

uv init rag-index
cd rag-index
uv add langchain langchain-openai langchain-community langchain-text-splitters docx2txt python-dotenv

3. 环境配置

大多数大型语言模型（LLM）服务，如OpenAI、Anthropic等，都需要API Key才能通过其API进行调用。本课程将主要以OpenAI 的Embedding模型为例进行讲解，但概念适用于其他服务。

获取OpenAI API Key（读者可自行获取）
配置API Key为环境变量

为了安全起见，我们不应将API Key直接写在代码中。推荐使用环境变量来管理。

推荐安装python-dotenv：这是一个用于从.env文件中加载环境变量的库。

pip install python-dotenv
# 或者
uv add python-dotenv

创建.env文件：在你的项目根目录下创建一个名为.env的文件（注意前面的点）。

OPENAI_API_KEY="你的OpenAI API密钥"
# 如果你使用其他服务，也可以在这里配置
# HUGGINGFACEHUB_API_TOKEN="你的Hugging Face Hub API Token"

在代码中加载环境变量。

from dotenv import load_dotenv
import os

load_dotenv() # 这将加载.env文件中的所有环境变量

# 之后你可以通过os.getenv()访问它们
# api_key = os.getenv("OPENAI_API_KEY")

4. 最佳代码实践

为了遵循了模块化、可配置和可重用的原则，将可变参数（如路径、模型名称、分块大小）放在一个单独的文件 config.py 中，方便管理和修改。

# config.py

from langchain_openai import OpenAIEmbeddings
from dotenv import load_dotenv
import os
load_dotenv()
# 数据和索引路径
SOURCE_FILE_PATH = "xxx.docx"

# 文本分割参数
CHUNK_SIZE = 1000  # 每个文本块的最大字符数
CHUNK_OVERLAP = 100  # 相邻文本块的重叠字符数

# Embedding 模型配置
# 推荐使用 text-embedding-3-small，性价比高
EMBEDDING_MODEL_NAME = "text-embedding-3-small"

# 实例化 Embedding 模型
# 可以在这里统一配置，比如 API key, base_url 等
# model=EMBEDDING_MODEL_NAME, dimensions=1536 等
embeddings_model = OpenAIEmbeddings(model=EMBEDDING_MODEL_NAME,
                                    base_url=os.getenv("OPENAI_BASE_URL"),
                                    api_key=os.getenv("OPENAI_API_KEY"))

# build_index.py

import os
import time
from langchain_text_splitters import RecursiveCharacterTextSplitter
from config import SOURCE_FILE_PATH, CHUNK_SIZE, CHUNK_OVERLAP, embeddings_model
from langchain_community.document_loaders import Docx2txtLoader

def load_single_document(file_path: str):
    print(f"从 '{file_path}' 加载文档...")
    loader = Docx2txtLoader(file_path)
    documents = loader.load()
    print(f"成功加载 1 个文档，包含 {len(documents[0].page_content)} 个字符。")
    return documents

def split_text_into_chunks(documents: list):
    """
    将加载的文档分割成文本块。
    """
    print("开始分割文档...")
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        length_function=len
    )
    chunks = text_splitter.split_documents(documents)
    print(f"文档被分割成 {len(chunks)} 个文本块。")
    return chunks

def main():
    """
    主函数，执行完整的离线索引流程。
    """
    # 检查逻辑，现在检查文件是否存在
    if not os.path.exists(SOURCE_FILE_PATH):
        print(f"错误：源文件 '{SOURCE_FILE_PATH}' 不存在。")
        print("请检查 config.py 中的文件路径是否正确。")
        return

    # 1. 加载单个文档
    documents = load_single_document(SOURCE_FILE_PATH)

    # 2. 分割文档
    chunks = split_text_into_chunks(documents)

    # 直接调用 embedding 模型并输出形状
    print("正在计算文本块的嵌入向量...")
    start_time = time.time()

    # 获取所有文本块的内容
    texts = [chunk.page_content for chunk in chunks]

    # 批量计算嵌入向量
    embeddings = embeddings_model.embed_documents(texts)

    end_time = time.time()
    print(f"嵌入计算完成，耗时 {end_time - start_time:.2f} 秒。")

    if embeddings:
        print(f"嵌入向量形状: {len(embeddings)} 个文本块 × {len(embeddings[0])} 维")
        print(f"总嵌入向量数: {len(embeddings)}")
        print(f"每个向量的维度: {len(embeddings[0])}")

        # 可选：显示前几个向量的部分维度作为示例
        print("\n前3个文本块的前10维嵌入向量示例:")
        for i in range(min(3, len(embeddings))):
            print(f"文本块 {i + 1}: {embeddings[i][:10]}...")
    else:
        print("未生成任何嵌入向量。")

    print("\n离线处理流程完成！")

if __name__ == "__main__":
    main()

这里的代码示例展示了从原始文件到生成向量的过程，至于持久化索引，读者可自行选择合适的向量数据库如FAISS，Milvus等进行向量embedding的持久化存储。