Redis向量数据库文本嵌入存储教程

1. 引言

在构建智能问答、语义搜索或推荐系统时，将非结构化文本数据转化为可计算的向量表示（即嵌入）并高效存储是核心步骤。Redis作为一款高性能的内存数据库，结合其向量搜索能力，成为了存储和检索文本嵌入的理想选择。本教程将深入探讨如何利用LangChain库，从本地文本文件加载数据，进行预处理，生成嵌入，并最终将其存储到Redis向量数据库中，以便后续进行高效的相似性搜索。

2. 核心概念解析

在开始实践之前，理解几个关键概念至关重要：

• 文本加载（Text Loading）：指从文件、网页或其他数据源读取原始文本内容的过程。对于本地文件，LangChain提供了TextLoader等工具。
• 文本切分（Text Splitting）：由于大型文档通常不适合直接嵌入或作为单个搜索单元，需要将其切分成更小、更具语义连贯性的块（chunks）。CharacterTextSplitter是常用的切分器之一。
• 文本嵌入（Text Embeddings）：将文本转化为高维浮点数向量的过程。这些向量能够捕捉文本的语义信息，使得语义相似的文本在向量空间中距离更近。OpenAI Embeddings是常用的嵌入模型之一。
• Redis向量数据库（Redis Vector Database）：Redis不仅是一个键值存储，通过Redis Stack中的RediSearch模块，它还能支持向量索引和相似性搜索，使其成为一个功能强大的向量数据库。LangChain提供了与Redis的集成，简化了操作。

3. 实践指南：构建自定义文本嵌入检索系统

本节将通过一个具体的示例，展示如何将本地union.txt文件中的内容加载、处理并存储到Redis中，并执行相似性搜索。

3.1 环境准备

在开始之前，请确保您已安装必要的Python库和Redis服务：

pip install langchain openai redis

确保您的本地或远程Redis服务正在运行，默认端口为6379。

3.2 加载与预处理文本数据

首先，我们需要从本地文件加载文本。假设您有一个名为union.txt的文本文件，其中包含您希望进行嵌入和搜索的内容。

union.txt 示例内容：

This is a comprehensive document about the history of the European Union.
It covers its formation, key milestones, and challenges faced over the decades.
The EU aims to promote peace, values, and the well-being of its peoples.
It has developed an internal single market through a standardised system of laws.
The European Union has faced various challenges, including economic crises and Brexit.
Despite these, it continues to play a significant role in global politics and trade.

接下来，使用TextLoader加载文件，并使用CharacterTextSplitter将文档切分成小块。切分是提高检索准确性和效率的关键步骤。

from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import Redis
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
import os

# 配置OpenAI API Key
# 请确保您的环境变量中已设置 OPENAI_API_KEY 或在此处直接赋值
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" 

# 1. 初始化嵌入模型
# 使用OpenAIEmbeddings，需要配置OpenAI API Key
embeddings = OpenAIEmbeddings()

# 2. 加载文本文件
# 假设您的文本文件名为 union.txt 且与脚本在同一目录下
loader = TextLoader("union.txt", encoding="utf-8")
documents = loader.load()

# 3. 切分文档
# chunk_size 定义每个文本块的最大字符数
# chunk_overlap 定义相邻文本块之间的重叠字符数，有助于保持上下文连贯性
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_documents(documents)

print(f"原始文档切分成了 {len(docs)} 个文本块。")
# 打印第一个文本块的内容
if docs:
    print(f"第一个文本块内容示例:\n{docs[0].page_content[:200]}...")

3.3 存储嵌入至Redis

切分后的文档列表docs现在可以与嵌入模型一起，通过Redis.from_documents方法存储到Redis向量数据库中。这个方法会自动为每个文档生成嵌入，并将文档内容、元数据和嵌入向量一同存储。

# 4. 存储文档和嵌入到Redis
# redis_url 指定Redis服务的地址
# index_name 是在Redis中创建的索引名称，用于组织和检索向量数据
vectorstore = Redis.from_documents(
    docs,
    embeddings,
    redis_url="redis://localhost:6379",
    index_name="users", # 建议使用更具描述性的索引名，例如 "eu_documents"
)

print(f"成功将 {len(docs)} 个文本块及其嵌入存储到Redis索引 '{vectorstore.index_name}' 中。")

3.4 执行相似性搜索

一旦数据存储在Redis中，您就可以执行相似性搜索，根据查询文本的语义相似性来检索相关的文档块。

# 5. 执行相似性搜索
# similarity_search_with_score 返回匹配的文档以及它们的相似度分数
# 分数越低表示相似度越高（通常是余弦距离或L2距离，取决于Redis索引配置）
query = "What are the main goals of the European Union?" # 更相关的查询
# query = "He met the Ukrainian people." # 使用原问题中的查询，如果文档内容不相关，结果可能不理想

print(f"\n执行查询: '{query}'")
results_with_score = vectorstore.similarity_search_with_score(query)

# 打印搜索结果
if results_with_score:
    print("搜索结果 (文档内容和相似度分数):")
    for doc, score in results_with_score:
        print(f"  文档内容: {doc.page_content[:150]}...")
        print(f"  相似度分数: {score}")
        print("-" * 20)
else:
    print("未找到相关结果。")

完整示例代码：

from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import Redis
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
import os

# --- 配置部分 ---
# 请确保您的环境变量中已设置 OPENAI_API_KEY
# 或者在此处直接赋值，但不推荐在生产环境直接硬编码
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" 

REDIS_URL = "redis://localhost:6379"
INDEX_NAME = "eu_documents_index" # 建议使用更具描述性的索引名
TEXT_FILE_PATH = "union.txt"

# --- 1. 初始化嵌入模型 ---
embeddings = OpenAIEmbeddings()

# --- 2. 加载文本文件 ---
try:
    loader = TextLoader(TEXT_FILE_PATH, encoding="utf-8")
    documents = loader.load()
    print(f"成功加载文件: {TEXT_FILE_PATH}")
except FileNotFoundError:
    print(f"错误: 文件 '{TEXT_FILE_PATH}' 未找到。请确保文件存在。")
    exit()
except Exception as e:
    print(f"加载文件时发生错误: {e}")
    exit()

# --- 3. 切分文档 ---
text_splitter = CharacterTextSplitter(chunk_size=500, chunk_overlap=50) # 调整块大小和重叠
docs = text_splitter.split_documents(documents)
print(f"原始文档切分成了 {len(docs)} 个文本块。")

# --- 4. 存储文档和嵌入到Redis ---
try:
    vectorstore = Redis.from_documents(
        docs,
        embeddings,
        redis_url=REDIS_URL,
        index_name=INDEX_NAME,
    )
    print(f"成功将 {len(docs)} 个文本块及其嵌入存储到Redis索引 '{INDEX_NAME}' 中。")
except Exception as e:
    print(f"存储到Redis时发生错误: {e}")
    print("请检查Redis服务是否运行，以及redis-py和RedisStack是否正确安装。")
    exit()

# --- 5. 执行相似性搜索 ---
query = "What are the main objectives of the European Union?" 
print(f"\n执行查询: '{query}'")

try:
    results_with_score = vectorstore.similarity_search_with_score(query)

    if results_with_score:
        print("搜索结果 (文档内容和相似度分数):")
        for doc, score in results_with_score:
            print(f"  文档内容: {doc.page_content.strip()[:200]}...")
            print(f"  相似度分数: {score:.4f}")
            print("-" * 20)
    else:
        print("未找到相关结果。")
except Exception as e:
    print(f"执行相似性搜索时发生错误: {e}")

4. 注意事项与进阶考量

• 文本切分策略：chunk_size和chunk_overlap参数对检索效果影响很大。较小的chunk_size可能导致上下文丢失，而较大的chunk_size可能包含过多不相关信息。chunk_overlap有助于在块边界处保持语义连贯性。根据您的数据特性和应用场景，可能需要进行多次试验以找到最佳参数。LangChain还提供了其他切分器，如RecursiveCharacterTextSplitter，可以根据不同的分隔符递归切分。
• 嵌入模型选择：本教程使用了OpenAI Embeddings，但您也可以选择其他嵌入模型，例如SentenceTransformerEmbeddings（来自langchain_community.embeddings.sentence_transformer），它允许您在本地运行模型，无需外部API调用。选择合适的模型取决于您的性能要求、成本预算和数据类型。
• Redis连接与索引：确保redis_url正确指向您的Redis实例。index_name是Redis中用于存储和检索向量的索引名称，每次调用from_documents如果索引已存在，会默认覆盖或追加，请根据需要管理索引生命周期。
• 数据生命周期管理（TTL）：原始问题提到了TTL。虽然LangChain的Redis.from_documents方法本身没有直接提供设置单个嵌入或整个索引TTL的参数，但Redis原生支持对键设置过期时间（TTL）。如果您需要为存储在Redis中的嵌入设置生命周期，可以在LangChain完成存储后，通过直接操作Redis客户端（如redis-py库）对相应的键设置TTL，或者在设计数据模型时考虑将TTL逻辑集成到应用程序层面。
• 性能优化：对于海量数据，批量处理和异步操作可以显著提升存储效率。此外，Redis的硬件配置（内存、CPU）和网络带宽也会影响读写性能。
• 元数据利用：Redis.from_documents支持传入文档的元数据。在实际应用中，丰富的元数据可以用于过滤、排序或增强检索结果，提高搜索的精确性和用户体验。

5. 总结

通过本教程，我们学习了如何利用LangChain框架，结合Redis向量数据库，构建一个从自定义文本文件到可搜索嵌入的完整流程。这包括文本的加载、智能切分、嵌入生成以及最终的向量存储和相似性搜索。掌握这些技术，您将能够为各种智能应用（如知识库检索、语义问答等）奠定坚实的基础，实现对非结构化文本数据的高效管理和智能利用。