使用 Redis 和 VectorFlow 构建向量嵌入注入管道

正在构建使用向量嵌入的应用程序？您可能已经经历了几个阶段：获取原始数据、尝试对其进行分块、将原始数据向量化，然后将向量上传到选择的向量数据库。

此过程涉及许多步骤，这就是 VectorFlow 之类的工具发挥作用的地方。

VectorFlow 是一个向量嵌入管道，有助于预处理数据并将其注入向量数据库。它减轻了手动执行数据分块和向量化所需的工作量和时间。
在本文中，我们将结合 VectorFlow 和 Redis 来构建一个向量嵌入管道。Redis 将用作我们的向量数据库，而 VectorFlow 将处理其余任务。

什么是 VectorFlow？

在深入研究构建管道之前，让我们更多地了解 VectorFlow。VectorFlow 是一个开源向量嵌入管道，其代码库托管在 Github 上，地址为 https://github.com/dgarnitz/vectorflow。目前，它适用于各种文件格式的图像和文本。VectorFlow 与多个嵌入模型兼容，包括 OpenAI 嵌入等第三方嵌入模型。它还可以随时执行任何 Hugging Face Sentence Transformer 模型，而无需任何 API 密钥。
您可以本地运行 VectorFlow 或使用其免费的云服务。在本文中，我将使用本地 VectorFlow。让我指导您完成设置过程。首先，我们将克隆其存储库

git clone https://github.com/dgarnitz/vectorflow.git

克隆存储库后，下一步是导航到其根文件夹并执行 setup.sh 命令。

cd vectorflow
./setup.sh

通过此过程，您将拥有本地设置的 VectorFlow。

管道概述

我们的非结构化文本数据将被馈送到 VectorFlow。VectorFlow 将把文本数据分割成块。然后，它将使用选择的向量化模型将这些分段部分转换为向量嵌入。为了演示，我将使用 OpenAI 的嵌入，它通常需要 API 密钥，但您可以选择任何 Hugging Face Sentence Transformer 模型。生成嵌入后，它们将存储在 Redis 向量数据库中。在本例中，我们使用的是保罗·格雷厄姆 (Paul Graham) 的文章，标题为 “我所从事的工作”，但您可以使用任何您喜欢的文本或图像数据。

将 Redis 设置为向量数据库

您可能以前使用过 Redis 作为内存中的键值存储，甚至作为 NoSQL 数据库。但是，您可能不知道的是，Redis 也可以用作向量数据库。开始使用 Redis 的最快方法是使用 Redis 云。创建一个 Redis 帐户并利用其免费计划。或者，您可以使用 Redis Stack 在 Docker 上本地运行它。设置好帐户后，请安装 Redis 客户端。您可以选择您喜欢的任何 Redis 客户端。在本文中，我将重点介绍如何使用 Redis Python 客户端。

!pip install redis

我们需要三个参数来连接到我们的 Redis 数据库。这些参数是

您的 Redis 实例主机名
您的 Redis 实例端口
您的 Redis 实例密码

以下是放置到变量中的参数

hostname = "redis-hostname"
port = 12345
password = "****"

让我们连接到我们的 Redis 数据库

r = redis.from_url(url = f'redis://{hostname}:{port}', password=password, decode_responses=True)

现在让我们测试我们的连接。

r.ping()

如果您正确设置了一切，这应该返回 True。

在 Redis 中创建索引

要准备我们的 Redis 实例以使用向量嵌入，我们需要建立一个索引名称并为其分配一个前缀。前缀至关重要，因为它使我们能够在 Redis 中找到它。

INDEX_NAME = "vectorflow_idx"
DOC_PREFIX = "vec:"

接下来，我们需要为我们的索引建立一个模式。VectorFlow 强制使用标准模式，包括以下内容

id：字符串
source_data：字符串
source_document：字符串
embeddings：浮点数组

此模式将使用 Redis 哈希数据结构存储在 Redis 中。id 将用作我们的哈希集的键，而 source_data 和 source_document 将是 TextField。同时，embeddings 将被归类为 VectorField。

from redis.commands.search.field import VectorField, TextField

schema = (
    TextField("source_data"),
    TextField("source_document"),
    VectorField("embeddings",
        "HNSW", {
            "TYPE": "FLOAT32",
            "DIM": 1536,
            "DISTANCE_METRIC": "COSINE",
        }
    ),
 )

鉴于我们的模式定义，让我们关注我们的 VectorField 定义。它首先获取字段的名称，在本例中为 “embeddings”。然后，它调用我们使用的算法，例如 HNSW（分层可导航小世界）。第三个参数表示我们嵌入的配置。TYPE 参数定义向量字段的数据类型，DIM 参数确定嵌入字段的维数。此值由您选择的向量化模型决定。我使用了 1536，因为我使用的是 OpenAI 嵌入。DISTANCE_METRIC 设置为余弦。这些值都取决于您的特定配置。

# index Definition
definition = IndexDefinition(prefix=[DOC_PREFIX],
index_type=IndexType.HASH)

接下来，我们创建索引定义。在此定义中，我们指定索引是一个哈希集。此外，我们指定了我们之前定义的索引前缀。

# create Index
r.ft(INDEX_NAME).create_index(fields=schema, 
definition=definition)

然后，我们通过指定名称、模式和先前建立的定义来创建索引。完成这些步骤后，我们的 Redis 数据库将准备好与 VectorFlow 协同工作。

将 VectorFlow 与 Redis 结合使用

现在我们已经设置了 Redis 向量数据库，让我们使用 VectorFlow 执行数据导入，这通过 HTTP 端点完成。用于数据导入的端点是 embed/，它是一个 POST 端点。

让我们配置我们的环境变量。我们需要三个秘密环境变量，这些变量不应该公开。

三个秘密环境变量是

VECTORFLOW_API_KEY：您的 VectorFlow 秘密密钥，您可以自行创建，或者在设置 VectorFlow 云帐户时提供给您。
EmbeddingAPI_Key：这代表嵌入提供者的 API 密钥。在我的情况下，它将是我的 OpenAI 密钥。
VectorDB-Key：这代表您的向量数据库密钥。在我们的例子中，它是我们的 Redis 实例密码。

%env VECTORFLOW_API_KEY=your-vectorflow-api-key
%env EmbeddingAPI_Key=your-openai-key
%env VectorDB_Key=your-redis-instance-password

然后，我们定义 VectorFlow URL。我使用的是本地版本的 VectorFlow，但您的部署可以位于任何地方。

import os

url = "http://localhost/embed"
headers = {
    "Content-Type": "multipart/form-data",
    headers: {
        "Authorization": os.getenv('VECTORFLOW_API_KEY'),
        "X-EmbeddingAPI-Key": os.getenv('EmbeddingAPI_Key')
        "X-VectorDB-Key": os.getenv('VectorDB_Key'),
    },
}

让我们定义我们的元数据。在元数据中，我们指定嵌入元数据和向量数据库元数据。

data = {
    'EmbeddingsMetadata': 
    '{"embeddings_type": "OPEN_AI", "chunk_size": 256, "chunk_overlap": 128}',
    'VectorDBMetadata': 
    '{"vector_db_type": "REDIS", "index_name": "vectorflow_idx", "environment": "redis://redis-13952.c14.us-east-1-3.ec2.cloud.redislabs.com:13952", "collection": "vect"}'
}

在嵌入元数据中，我指定了块大小为 256，重叠为 128。因为我使用的是 OpenAI 嵌入，所以我已经指定了嵌入类型为 “OPEN_AI”。VectorDBMetadata 包含我们所有向量数据库信息：向量数据库类型、索引名称、集合（索引前缀）和我们的 Redis URL（来自我们的环境）。

此设置对于提供有关处理数据导入过程的分块和嵌入的具体说明至关重要。

接下来，我们将下载我们的数据。让我们编写一个函数来执行此任务。

def download_file(url, save_path):
    response = requests.get(url)
    if response.status_code == 200:
        with open(save_path, 'wb') as file:
            file.write(response.content)
        print("File downloaded successfully")
    else:
        print(f"Failed to download the file. Status code: {response.status_code}")

然后，我们设置文件的 URL 目标。

data_url =  'https://raw.githubusercontent.com/EteimZ/LLM_Stack/main/llama_index/pg-what-i-worked-on/data/paul_graham_essay.txt'

通过此过程，我们应该已经获得了数据。接下来，我们需要将文件添加到包含 SourceData 键的字典中。

file = {"SourceData": open("paul_graham_essay.txt", "rb")}

现在我们已经收集了所有必要的组件，我们使用之前指定的 URL 向 embed 端点发送 POST 请求。我们将在此请求中传递我们的标头、数据和文件。

response = requests.post(url, headers=headers, data=data, files=file)

然后，我们检查响应文本

print(response.text)

我们应该期望收到类似于此的响应

{"JobID": 1, "message": "Sucessfully added 2 batches to the queue"}

当数据上传到 VectorFlow 时，它会被发送到一个后台工作程序，该工作程序管理所有向量化过程。

要检查任务的状态，请向端点 /jobs/<int:job_id>/status 发送 GET 请求。job_id 代表从上一个请求返回的 ID。

在我的情况下，我的请求将类似于以下内容

!curl localhost/jobs/1/status

如果请求仍在进行，您将收到类似于此的响应

{"JobStatus": "IN_PROGRESS"}

如果过程已完成，则响应将如下所示

{"JobStatus": "COMPLETED"}

通过此过程，您已成功将嵌入注入 Redis。您现在可以使用您的嵌入来执行您选择的任何任务。

让我们将我们完成的所有步骤整理在一起。

import os

url = "http://localhost/embed"

headers = {
    "Content-Type": "multipart/form-data",
    headers: {
        "Authorization": os.getenv('VECTORFLOW_API_KEY'),
        "X-EmbeddingAPI-Key": os.getenv('EmbeddingAPI_Key')
        "X-VectorDB-Key": os.getenv('VectorDB_Key'),
    },
}

data = {
    'EmbeddingsMetadata': 
    '{"embeddings_type": "OPEN_AI", "chunk_size": 256, "chunk_overlap": 128}',
    'VectorDBMetadata': 
    '{"vector_db_type": "REDIS", "index_name": "vectorflow_idx", "environment": "redis://redis-13952.c14.us-east-1-3.ec2.cloud.redislabs.com:13952", "collection": "vect"}'
}

file = {"SourceData": open("test_text.txt", "rb")}

response = requests.post(url, headers=headers, data=data, files=file)

print(response.text)

在执行任何项目时，选择合适的工具对于项目的成功至关重要。VectorFlow 是一个相对较新的项目，它是开源的。因此，请前往其 GitHub 存储库，为其加星，并查看其问题。也许有一个问题可以解决或为其做出贡献。