加入我们的 Discord | 官网 | 联系我们 | 文档 | 免费托管版本

简介

VectorFlow 是一个开源、高吞吐、容错的向量嵌入流水线。只需发送一个简单的 API 请求，即可上传原始数据，这些数据将被分块、嵌入，并存储到任何向量数据库中，或者直接返回给您。

当前版本为 MVP。我们建议在生产环境中使用 Kubernetes 运行它（详情见下文）。对于文本文件，支持 TXT、PDF、HTML 和 DOCX 格式。

在本地运行

只需三条命令，您就可以在本地运行 VectorFlow：

git clone https://github.com/dgarnitz/vectorflow.git
cd vectorflow
./setup.sh

使用客户端嵌入文档

要在本地开始嵌入文档，请在您的 Python 应用程序的虚拟环境中安装 VectorFlow 客户端 Python 库。

pip install vectorflow-client

然后运行以下代码：

from vectorflow-client.client.vectorflow import Vectorflow

vectorflow = Vectorflow()
vectorflow.embeddings_api_key = os.getenv("OPEN_AI_KEY")
paths = ['your_file_path1', 'your_file_path2', ...]
response = vectorflow.upload(paths)

您无需克隆 VectorFlow 仓库即可通过 pip 使用客户端功能。更多说明请参阅 client 目录中的 README.md 文件。

有关如何使用 testing_clients 脚本的详细信息，请参阅附录。

Docker-Compose

运行 VectorFlow 的最佳方式是使用 docker compose。如果您在 Mac 上运行，请按照此处的说明授予 Docker 读取您“文档”文件夹的权限：链接。如果此操作失败，请从 docker-compose.yml 中移除 volume 部分。

1) 设置环境变量

首先在根目录下创建一个名为 env_scripts 的文件夹来存放所有环境变量，然后在该文件夹中创建 env_vars.env 文件，添加以下环境变量。只有当您在本地运行 Qdrant、Milvus 或 Weaviate 时，才需要设置 LOCAL_VECTOR_DB 变量。

INTERNAL_API_KEY=your-choice
POSTGRES_USERNAME=postgres
POSTGRES_PASSWORD=your-choice
POSTGRES_DB=vectorflow
POSTGRES_HOST=postgres
RABBITMQ_USERNAME=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_HOST=rabbitmq
LOCAL_VECTOR_DB=qdrant | weaviate
API_STORAGE_DIRECTORY=/tmp
MINIO_ACCESS_KEY=minio99
MINIO_SECRET_KEY=minio123
MINIO_ENDPOINT=minio:9000
MINIO_BUCKET=vectorflow

您可以为 INTERNAL_API_KEY、POSTGRES_PASSWORD 和 POSTGRES_DB 选择任意值，但必须设置它们。

2) 运行 Docker-Compose

确保您已将 Rabbit MQ、Postgres 和 Min.io 拉取到本地 Docker 仓库中。我们还建议在本地运行一个向量数据库，因此请务必拉取您正在使用的数据库镜像。我们的 docker-compose 文件默认会启动 Qdrant 并创建两个索引/集合。如果您计划运行 Milvus 或 Weaviate，则需要自行配置。

docker pull rabbitmq
docker pull postgres
docker pull qdrant/qdrant | docker pull semitechnologies/weaviate
docker pull minio/minio

然后运行：

docker-compose build --no-cache
docker-compose up -d

请注意，init 容器正在运行一个脚本，用于设置数据库模式、向量数据库和 Minio 对象存储。脚本完成后，这些容器会停止。对于 Qdrant，请确保拉取 1.9.1 版本，因为这是 Qdrant 客户端 Python 包所兼容的版本。

使用 VectorFlow

使用 VectorFlow 的最佳方式是通过 Python 客户端。

要用于开发目的，您可以向您的 API URL 发送 HTTP 请求——例如，从开发机器上访问 localhost:8000，或从另一个 Docker 容器内访问 vectorflow_api:8000。

请求与响应负载

所有请求都需要包含一个 Authorization 键的 HTTP 头，其值应与您之前定义的 INTERNAL_API_KEY 环境变量相同（见上文）。如果您连接的是基于云的向量数据库实例，则必须通过 HTTP 头 X-VectorDB-Key 传递您的向量数据库 API 密钥；如果使用 OpenAI，则需通过 X-EmbeddingAPI-Key 传递嵌入 API 密钥。HuggingFace Sentence Transformer 嵌入不需要 API 密钥，但您仍需按照上述步骤运行包含所需模型的容器。

目前，VectorFlow 支持 Pinecone、Qdrant 和 Weaviate 向量数据库。

对单个文件进行嵌入

要提交单个文件进行嵌入，请向 /embed 端点发送 POST 请求，附带文件、'Content-Type: multipart/form-data' 头以及以下负载：

{
    'SourceData=txt文件路径'
    'LinesPerBatch=4096'
    'EmbeddingsMetadata={
        "embeddings_type": "OPEN_AI",
        "chunk_size": 512,
        "chunk_overlap": 128,
        "chunk_strategy": "EXACT | PARAGRAPH | SENTENCE | CUSTOM",
        "model": "text-embedding-3-small | text-embedding-3-large | text-embedding-ada-002"
    }'
    'VectorDBMetadata={
        "vector_db_type": "PINECONE | QDRANT | WEAVIATE",
        "index_name": "索引名称",
        "environment": "环境名称"
    }'
    'DocumentID=您可选的内部跟踪ID'
}

这将创建一个 job，您将收到如下负载作为响应：

{
    'message': f"成功将 {batch_count} 个批次添加到队列中",
    'JobID': job_id
}

目前，该端点仅支持一次上传单个文件，且由于超时问题，文件大小上限为 25 MB。请注意，此功能可能会被弃用。

一次性嵌入多个文件

要提交多个文件进行嵌入，请向 /jobs 端点发送 POST 请求。负载与单个文件嵌入相同，不同之处在于多文件的附加方式：

{
    'files=[
        ('file',  ('test_pdf.pdf', open(file1_path, 'rb'), 'application/octet-stream')),
        ('file', ('test_medium_text.txt', open(file2_path, 'rb'), 'application/octet-stream'))
    ]'
}

注意： 必须以流式传输的方式将文件发送到端点，而不能以常规 POST 请求的形式发送，否则会失败。

该端点会为每个上传的文件创建一个 job。您将收到如下 JSON 负载：

{   
    'successful_uploads': 成功上传的文件列表,
    'failed_uploads': 上传失败的文件列表,
    'empty_files_count': 空文件数量,
    'duplicate_files_count': 重复文件数量
}

其中 successfully_uploaded_files 是一个包含 (文件名, job id) 元组的列表，而 failed_uploads 是上传失败的文件名列表，以便您可以重试这些文件。

获取单个作业状态

要检查某个 job 的状态，请向 /jobs/<int:job_id>/status 端点发送 GET 请求。响应格式如下：

{
    'JobStatus': 作业状态
}

获取多个作业状态

要检查多个 job 的状态，请向 /jobs/status 端点发送 POST 请求。请求体格式如下：

{
    'JobIDs': 作业 ID 列表
}

响应格式为：

{
    'Jobs': [{'JobID': 作业 ID, 'JobStatus': 作业状态}, ...]}

testing_clients/get_jobs_by_ids.py 中有一个示例。

向量数据库标准元数据模式

VectorFlow 强制执行用于向量存储上传数据的标准模式：

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

ID 可用于去重和幂等性操作。请注意，在 Weaviate 中，ID 称为 vectorflow_id。

我们计划在不久的将来弃用此模式，以便在未来支持动态检测和/或可配置的模式。

分块模式与自定义分块

VectorFlow 内置的分块器按 token 计数，而非按字符计数。VectorFlow 中的 chunk 是一个包含以下键的字典：

text: str
vector: list[float]

您可以通过在构建 worker 的 Docker 镜像之前，将一个名为 custom_chunker.py 的文件及其方法 chunker(source_data: list[str]) 添加到 src/worker 目录来运行自定义分块器。该分块器必须返回符合上述标准的 chunk 字典列表。

只要 _JSON 可序列化_，您就可以向 chunk 字典中添加任何所需的键，即不允许使用自定义类或函数、日期时间类型或循环引用代码。然后，您可以使用此自定义分块将元数据上传到向量数据库，并采用您所需的任何模式。

原始嵌入 Webhook

如果您希望仅使用 VectorFlow 进行分块和生成嵌入，请在 /embed 请求的正文中传递 WebhookURL 参数，并在头中添加 X-Webhook-Key。VectorFlow 假设需要 Webhook 密钥才能写回任何端点。嵌入结果会连同源分块一起以上述 chunk 字典的形式返回。数据以 JSON 格式发送，形式如下：

{
    'Embeddings': list[dict],
    'DocumentID': str,
    'JobID': int
}

分块验证 Webhook

如果您希望验证哪些分块可以进行嵌入，请在 /embed 请求的正文中传递 ChunkValidationURL 参数。这将发送包含以下 JSON 负载的请求：{"chunks": chunked_data}，其中 chunked_data 是一个 chunk 字典列表。预期返回的 JSON 将包含 valid_chunks 键，其中列出可用于嵌入的有效分块。该端点默认会在 30 秒后超时，但可以在应用程序代码中进行配置。

S3 端点

VectorFlow 已与 AWS S3 集成。您可以在 HTTP 请求体中传递预签名的 S3 URL，而不是直接上传文件。使用表单字段 PreSignedURL 并访问 /s3 端点。该端点具有与 /embed 端点相同的配置和限制。

遥测

VectorFlow 使用 PostHog 匿名收集使用数据。它不会收集任何个人身份信息。如果您想禁用此功能，只需在您的 env_vars.env 文件中添加以下环境变量：

TELEMETRY_DISABLED=True

Kubernetes

您可以在本地使用 minikube 在 Kubernetes 中运行 VectorFlow，方法是执行 ./kube/scripts/deploy-local-k8s.sh 脚本，该脚本会应用 kube/ 目录下的所有 YAML 文件。如果您尚未安装 Docker、minikube 和 kubectl，此脚本将无法正常工作。

该脚本首先会在本地构建镜像，然后将其传输到 minikube 中。如果您想查看 minikube 中有哪些可用的镜像，请运行以下命令：

eval $(minikube docker-env)
docker images

为了从您的开发机器访问集群中的资源，您需要运行 minikube tunnel 命令。设置脚本会将您本地 Docker 上下文中的镜像加载到 minikube 的镜像仓库中。

您可以以 kube/ 目录下的 YAML 文件为基础进行生产环境部署，但需要根据您特定集群的需求进行一些自定义配置。如需帮助，请联系我们。

贡献

我们非常欢迎社区的反馈。如果您有任何改进该项目的想法，欢迎您提交问题或加入我们的 Discord 社区。请务必标记 dgarnitz 和 danmeier2。

我们的路线图在下方章节中列出，我们非常希望得到大家的帮助来完善它。我们当前的公开问题是一个很好的起点，您可以在此处查看：https://github.com/dgarnitz/vectorflow/issues。如果您想处理未列出的任务，建议您先提交一个问题，说明您的计划方案，然后再提交 PR。

请在所有 PR 中标记 dgarnitz，并更新 README以反映您的更改。

测试

在提交 PR 时，请添加单元测试以覆盖您新增的功能。同时，请重新运行现有测试，确保没有引入回归性错误。请从 src 目录运行测试。要运行单个测试用例，可以使用以下命令：

python -m unittest module.tests.test_file.TestClass.test_method

要运行文件中的所有测试，可以使用：

python -m unittest module.tests.test_file

对于端到端测试，建议使用 docker-compose 进行构建和运行，但请停止正在修改的容器，并在您的开发机器上本地运行。这样可以避免不断重建镜像和重启容器。请确保在开发机器的终端中将环境变量更改为正确的值（例如，将 rabbitmq 或 postgres 替换为 localhost），以便 Docker 容器能够与您的开发机器通信。当本地运行正常后，您可以使用 docker-compose 再次进行最终测试。

验证

请在提交 PR 之前，验证所有更改是否都能通过 docker-compose 正常工作。

我们还建议您提供验证证据，例如截图，以展示您的代码在端到端流程中的实际效果。

路线图

• 支持从 Salesforce、Google Drive 等来源批量导入多文件或多目录数据
• 重试机制
• Langchain 集成
• 支持回调功能，用于将对象元数据写入独立存储
• 动态可配置的向量数据库模式
• 去重功能
• 向量版本控制
• “智能”分块器
• “智能”元数据提取器

附录

使用测试脚本嵌入数据

使用 VectorFlow 的一种简单方式是借助位于 testing_clients/ 目录下的测试客户端。该目录包含多个脚本，具有不同的配置，可用于快速上传数据。我们建议从 testing_clients/standard_upload_client.py 开始——运行此脚本会将单个文档提交给 VectorFlow，使用 OpenAI ADA 模型进行嵌入，并上传至本地的 Qdrant 实例。您可以根据自己的配置调整相关参数。如果需要一次上传多个文件，可以使用 testing_clients/streaming_upload_client.py。

请注意，TESTING_ENV 变量相当于 VectorDBMetadata 中的 environment 字段，它对应于 Pinecone 中的环境、Weaviate 中的类、Qdrant 中的集合等。testing_clients 目录中提供了示例脚本，您可以按照这些脚本运行 VectorFlow。将您的嵌入模型和数据库密钥添加到生成的 env_scripts/env_vars.sh 脚本中，并将 testing_clients/standard_upload_client.py 中的 filepath 变量设置为您想要嵌入的文件路径。然后运行：

source env_scrips/env_vars.sh
python testing-clients/standard_upload_client.py

如果需要一次上传多个文件，可以使用 testing_clients/streaming_upload_client.py。

有关如何手动设置和配置系统的更详细说明，请参阅上方内容。请注意，setup 脚本不会在您的机器上创建开发环境，它仅负责设置并运行 docker-compose。我们不建议在 Windows 系统上使用 VectorFlow。

请求

要执行搜索，需向 /images/search 端口发送 POST 请求，请求中需附带图像文件，并设置 Content-Type: multipart/form-data 头部，以及以下请求体：

{
    'ReturnVectors': boolean,
    'TopK': integer，小于1000，
    'VectorDBMetadata={
        "vector_db_type": "PINECONE | QDRANT | WEAVIATE",
        "index_name": "index_name",
        "environment": "env_name"
    }'
}

所有请求都需要带有 Authorization 键的 HTTP 头部，该键与您之前定义的 INTERNAL_API_KEY 环境变量相同（见上文）。如果您连接的是基于云的向量数据库实例，则必须通过 HTTP 头部 X-VectorDB-Key 提供您的向量数据库 API 密钥。

响应

图像相似度搜索将返回一个响应对象，其中包含前 K 个匹配项，如果请求了原始向量，则还会一并返回。响应格式如下：

{
    "similar_images": 匹配对象列表
    "vectors": 浮点数列表的列表
}

其中，匹配对象的定义如下：

{
    "id": 字符串，
    "score": 浮点数，
    "metadata": {"source_document" : 字符串}
}

使用 Docker 命令构建和运行单个镜像

如果您希望使用 docker build 和 docker run 来构建和运行单个镜像，而不是使用 docker-compose，请按照以下步骤操作：

1. cd src/
2. docker build --file api/Dockerfile -t vectorflow_api:latest . 构建镜像——别忘了最后的句点。
3. docker run --network=vectorflow --name=vectorflow_api -d --env-file=../env_scripts/env_vars.env -p 8000:8000 vectorflow_api:latest 运行 API。运行 worker 时无需指定端口。

VectorFlow 快速上手指南

VectorFlow 是一个开源、高吞吐量且具备容错能力的向量嵌入（Embedding）流水线工具。它通过简单的 API 接口，支持将大量原始数据（如 TXT, PDF, HTML, DOCX）进行分块、向量化，并存储到主流向量数据库或直接返回结果。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统: Linux, macOS 或 Windows (需安装 Docker Desktop)
• 核心依赖:
  • Git
  • Docker & Docker Compose
  • Python 3.x (用于运行客户端脚本)
• 前置服务镜像: 本地需拉取以下 Docker 镜像（若网络受限，建议配置国内 Docker 镜像加速器）：
  • rabbitmq
  • postgres
  • qdrant/qdrant (默认向量库，版本建议 1.9.1)
  • minio/minio
  • (可选) semitechnologies/weaviate 或其他支持的向量库

安装步骤

方式一：使用 Docker Compose 部署（推荐）

这是运行 VectorFlow 最完整的方式，包含所有依赖服务。

1. 克隆项目

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

2. 配置环境变量 在项目根目录创建 env_scripts 文件夹，并在其中创建 env_vars.env 文件，填入以下配置（请根据实际需求修改密码和 Key）：

mkdir -p env_scripts
cat > env_scripts/env_vars.env <<EOF
INTERNAL_API_KEY=your-choice
POSTGRES_USERNAME=postgres
POSTGRES_PASSWORD=your-choice
POSTGRES_DB=vectorflow
POSTGRES_HOST=postgres
RABBITMQ_USERNAME=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_HOST=rabbitmq
LOCAL_VECTOR_DB=qdrant
API_STORAGE_DIRECTORY=/tmp
MINIO_ACCESS_KEY=minio99
MINIO_SECRET_KEY=minio123
MINIO_ENDPOINT=minio:9000
MINIO_BUCKET=vectorflow
EOF

注意: 若仅本地测试，LOCAL_VECTOR_DB 设为 qdrant 即可；若连接云端向量库，后续需在请求头中配置对应 Key。

3. 拉取镜像并启动服务

# 拉取必要镜像
docker pull rabbitmq
docker pull postgres
docker pull qdrant/qdrant:1.9.1
docker pull minio/minio

# 构建并启动容器
docker-compose build --no-cache
docker-compose up -d

启动后，初始化容器会自动配置数据库 Schema 和对象存储，完成后会自动停止。

方式二：安装 Python 客户端

若只需在代码中调用已部署的 VectorFlow 服务，无需克隆源码，直接安装客户端库即可：

pip install vectorflow-client

基本使用

1. 使用 Python 客户端上传文件

这是最简单的使用方式。确保已设置 OpenAI API Key（或其他 Embedding 模型 Key）。

import os
from vectorflow_client.client.vectorflow import Vectorflow

# 初始化客户端
vectorflow = Vectorflow()

# 设置 Embedding API Key (例如 OpenAI)
vectorflow.embeddings_api_key = os.getenv("OPEN_AI_KEY")

# 指定本地文件路径
paths = ['path_to_your_file1.pdf', 'path_to_your_file2.txt']

# 上传并触发嵌入任务
response = vectorflow.upload(paths)

print(response)

2. 直接使用 HTTP API (进阶)

如果您想直接通过 curl 或自定义代码调用，需向 /embed 端点发送 POST 请求。

请求示例：

• URL: http://localhost:8000/embed
• Header:
  • Authorization: 填写 env_vars.env 中的 INTERNAL_API_KEY
  • X-EmbeddingAPI-Key: 填写您的 Embedding 服务商 Key (如 OpenAI)
  • Content-Type: multipart/form-data

Payload 参数说明：

• SourceData: 文件路径
• LinesPerBatch: 每批处理行数 (例如 4096)
• EmbeddingsMetadata: JSON 字符串，定义模型和分块策略

  {
      "embeddings_type": "OPEN_AI",
      "chunk_size": 512,
      "chunk_overlap": 128,
      "chunk_strategy": "EXACT",
      "model": "text-embedding-3-small"
  }
  

• VectorDBMetadata: JSON 字符串，定义目标向量库

  {
      "vector_db_type": "QDRANT",
      "index_name": "my_index",
      "environment": "local"
  }
  

响应示例： 成功提交后，系统将返回任务 ID：

{
    "message": "Successfully added 1 batches to the queue",
    "JobID": 12345
}

3. 查询任务状态

可通过 API 检查处理进度：

• 单个任务: GET /jobs/<job_id>/status
• 多个任务: POST /jobs/status (Body: {"JobIDs": [12345, ...]})

---

提示: 当前版本为 MVP，生产环境建议在 Kubernetes 上运行以获得更好的容错性和扩展性。支持的文件格式包括 TXT, PDF, HTML 和 DOCX。