Python RAG系统构建：从文本向量化到检索增强生成

1. 环境准备与项目概述

在开始构建RAG系统之前，我们需要先搭建好开发环境。这个项目使用Python 3.11+作为基础运行环境，并采用uv作为包管理工具。uv是Astral团队开发的新一代Python包管理器，比传统的pip快10-100倍，特别适合管理AI项目中的大量依赖。

将以下配置保存为项目根目录下的pyproject.toml文件：

python复制[project]
name = "embedding-rag-demo"
version = "0.1.0"
description = "Embedding与向量数据库应用——从文本向量化到RAG检索增强生成"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "beautifulsoup4>=4.14.3",
    "dotenv>=0.9.9",
    "faiss-cpu>=1.13.2",
    "langchain>=1.2.10",
    "langchain-chroma>=1.1.0",
    "langchain-community>=0.4.1",
    "langchain-huggingface>=1.2.1",
    "langchain-openai>=1.1.10",
    "numpy>=2.2.6",
    "openai>=2.24.0",
    "pandas>=2.2.3",
    "sentence-transformers>=5.2.3",
]

安装依赖只需在终端执行：

bash复制uv sync

这个配置包含了我们需要的所有关键组件：

• beautifulsoup4：用于网页内容解析
• faiss-cpu：Meta开源的向量检索库
• langchain系列：构建RAG的核心框架
• sentence-transformers：加载本地嵌入模型

2. RAG基础概念解析

2.1 什么是RAG？

RAG（Retrieval-Augmented Generation，检索增强生成）是大语言模型应用中的一项关键技术，它通过结合检索系统和生成模型，有效解决了传统LLM的两个核心痛点：

1. 知识时效性问题：LLM的训练数据有截止日期，无法回答最新事件。例如，用GPT-4直接问"2023年诺贝尔文学奖得主是谁"可能得不到正确答案。

2. 幻觉问题：当LLM不确定答案时，可能会编造看似合理实则错误的回答。RAG通过提供真实参考文档，大幅降低了幻觉概率。

RAG的工作流程可以简化为：

code复制用户提问 → 检索相关文档 → 将文档+问题一起发给LLM → 生成有依据的回答

2.2 RAG的核心组件

一个完整的RAG系统包含三个关键部分：

1. 文本向量化（Embedding）：将文本转换为高维向量，使语义相似的文本在向量空间中距离相近。这决定了检索质量的上限。

2. 向量数据库：高效存储和检索向量数据。当数据量大时，直接计算相似度效率极低，需要专门的向量索引。

3. 文档处理流水线：包括文档加载、清洗、分割等预处理步骤，确保存入向量数据库的内容质量。

3. 文本向量化实战

3.1 为什么需要Embedding？

传统关键词搜索的局限性很明显。考虑以下两个句子：

1. "我想吃多汁的烤肉"
2. "美味的烧烤"

虽然语义相近，但词汇重叠率极低。Embedding通过将文本映射到高维向量空间，使得语义相近的文本向量距离小，从而支持语义级别的搜索。

3.2 OpenAI Embedding快速体验

OpenAI提供了高质量的嵌入模型，使用非常简单：

python复制from openai import OpenAI
from env_utils import OPENAI_API_KEY, OPENAI_BASE_URL

client = OpenAI(
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_BASE_URL
)

text = "I like large language models."
resp = client.embeddings.create(
    model='text-embedding-3-large',
    dimensions=256,
    input=text
)

print(len(resp.data[0].embedding))  # 输出256

关键参数说明：

• model：推荐使用最新的text-embedding-3-large
• dimensions：输出向量维度，越高精度越好但成本也越高
• input：支持单个字符串或字符串列表

3.3 本地私有化部署方案

对于数据敏感或延迟要求高的场景，我们需要本地部署的嵌入模型。这里介绍两种主流方案：

方案一：使用sentence-transformers加载Qwen3

python复制from sentence_transformers import SentenceTransformer

model = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B")
embeddings = model.encode(["今天的天气真好！", "Hello world!"])
print(embeddings.shape)  # (2, 1024)

方案二：使用LangChain+HuggingFace加载BGE模型

python复制from langchain_huggingface import HuggingFaceEmbeddings

model_name = "BAAI/bge-small-zh-v1.5"
model_kwargs = {'device': 'cpu'}
encode_kwargs = {'normalize_embeddings': True}

embedding = HuggingFaceEmbeddings(
    model_name=model_name,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs
)

result = embedding.embed_documents(["今天的天气真好！", "Hello world!"])
print(len(result[0]))  # 1024

3.4 自定义Embedding集成

为了将自定义模型集成到LangChain生态，我们需要实现Embeddings接口：

python复制from langchain_core.embeddings import Embeddings
from sentence_transformers import SentenceTransformer

class CustomEmbeddings(Embeddings):
    def __init__(self, model_name):
        self.model = SentenceTransformer(
            model_name,
            device="auto",
            torch_dtype="float16"
        )
    
    def embed_query(self, text: str) -> List[float]:
        return self.embed_documents([text])[0]
    
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return self.model.encode(texts).tolist()

生产级实现还需要考虑：

• 设备自动选择（GPU/MPS/CPU）
• 半精度推理节省显存
• 分词器参数配置

4. 向量数据库实战

4.1 FAISS基础使用

FAISS是Meta开源的高性能向量检索库，适合大规模数据场景。

python复制from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

# 准备文档
documents = [
    Document(page_content="苹果是一种水果", metadata={"source": "wiki"}),
    Document(page_content="香蕉是热带水果", metadata={"source": "book"}),
]

# 创建向量库
vectorstore = FAISS.from_documents(
    documents=documents,
    embedding=embedding_model
)

# 相似度搜索
results = vectorstore.similarity_search("好吃的水果", k=1)
print(results[0].page_content)  # 苹果是一种水果

4.2 Chroma向量数据库

Chroma是轻量级的嵌入式向量数据库，开发体验更好：

python复制from langchain_chroma import Chroma

# 初始化并持久化
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embedding_model,
    persist_directory="./chroma_db"
)

# 从磁盘加载
loaded_store = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embedding_model
)

4.3 向量数据库选型对比

5. 文档处理流水线

5.1 文档加载

LangChain提供了多种文档加载器：

python复制from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://example.com")
docs = loader.load()

5.2 文本分割

合理的文本分割对RAG效果至关重要：

python复制from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = splitter.split_documents(docs)

分割策略选择建议：

• 通用文档：RecursiveCharacterTextSplitter
• Markdown：MarkdownHeaderTextSplitter
• 高质量需求：SemanticChunker

6. 完整RAG系统实现

6.1 系统架构

code复制用户提问 → 问题重写 → 向量检索 → 上下文注入 → LLM生成 → 返回答案

6.2 核心代码实现

python复制from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain

# 1. 构建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 2. 定义提示模板
template = """使用以下上下文回答问题：
{context}

问题：{input}"""
prompt = ChatPromptTemplate.from_template(template)

# 3. 创建RAG链
rag_chain = create_retrieval_chain(retriever, prompt | llm)

# 4. 调用
response = rag_chain.invoke({"input": "什么是RAG？"})
print(response["answer"])

6.3 上下文感知的多轮对话

python复制from langchain.chains import create_history_aware_retriever

# 上下文感知检索器
contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", "根据聊天历史重写问题"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_prompt
)

# 带历史的RAG链
conversational_rag_chain = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

7. 生产环境优化建议

1. 检索优化：

   • 混合检索（关键词+向量）
   • 查询扩展（同义词扩展）
   • 元数据过滤

2. 生成优化：

   • 提示工程优化
   • 结果后处理
   • 引用溯源

3. 性能优化：

   • 批量处理文档
   • 异步处理
   • 缓存机制

8. 常见问题排查

8.1 检索结果不相关

• 检查Embedding模型是否适合你的领域
• 调整文本分割策略（chunk_size/chunk_overlap）
• 添加元数据过滤条件

8.2 生成答案质量差

• 检查检索到的上下文是否相关
• 优化提示模板
• 尝试不同的LLM模型

8.3 性能瓶颈

• 对于大规模数据，考虑分布式向量数据库如Milvus
• 使用GPU加速Embedding计算
• 实现分级缓存机制

9. 扩展应用方向

1. 多模态RAG：支持图像、音频等非文本数据
2. Agent集成：让LLM自主决定何时检索
3. 增量更新：实现知识库的动态更新
4. 评估体系：构建自动化评估流水线

在实际项目中，RAG系统的效果很大程度上取决于Embedding质量和文档处理策略。建议先在小规模数据上验证各组件效果，再逐步扩展到全量数据。同时，要建立完善的评估机制，持续监控系统表现。