Haystack × Supabase 集成实战：SupabasePgvectorDocumentStore、双检索器与 Storage 下载器完全指南

Haystack × Supabase 集成实战：SupabasePgvectorDocumentStore、双检索器与 Storage 下载器完全指南

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

导读

本文以 Haystack 官方 API 参考文档（integrations-api/supabase.md）为主体，系统讲解 Haystack 与 Supabase 的四个集成组件：SupabasePgvectorDocumentStore（基于 PostgreSQL + pgvector 的向量文档存储）、SupabasePgvectorEmbeddingRetriever（稠密向量检索器）、SupabasePgvectorKeywordRetriever（PostgreSQL 全文关键词检索器）以及SupabaseBucketDownloader（从 Supabase Storage 下载文件为ByteStream）。读完本文，你将能够完成从环境配置、文档索引、双路检索到文件下载接入索引管线的完整 Supabase RAG 方案搭建，并理解每个参数在源码层面的实际作用。

一、组件全景：Haystack 为 Supabase 提供什么

Supabase 是一个基于 PostgreSQL 的开源后端平台。Haystack 的 Supabase 集成围绕 Supabase 的两项核心能力展开：

• pgvector 扩展（Supabase 预装）：支撑向量相似度搜索；
• Supabase Storage：支撑对象存储文件的批量下载与索引。

集成共提供四个组件，全部位于haystack_integrations命名空间下：

其中前三个组件是对 pgvector 集成（PgvectorDocumentStore及其检索器）的薄封装（thin wrapper），仅调整了 Supabase 专属默认值；第四个组件则是独立的文件下载器。更多官方用法可参考仓库中的 supabasedocumentstore.mdx。

二、安装与环境准备

2.1 安装集成包

pip install supabase-haystack

官方示例使用 Sentence Transformers 嵌入器，它们已迁移至独立包sentence-transformers-haystack，如需运行下文示例，请一并安装：

pip install sentence-transformers-haystack

2.2 配置两个环境变量

Supabase 集成依赖两个环境变量：

# 1. 数据库连接串（文档存储与检索器使用） export SUPABASE_DB_URL="postgresql://postgres.[project-ref]:[password]@aws-0-[region].pooler.supabase.com:5432/postgres" # 2. Service Role Key（Storage 下载器使用，私有桶必须使用服务角色密钥） export SUPABASE_SERVICE_KEY="your-service-role-key"

SUPABASE_DB_URL由SupabasePgvectorDocumentStore通过Secret.from_env_var("SUPABASE_DB_URL")自动读取；SUPABASE_SERVICE_KEY由SupabaseBucketDownloader通过Secret.from_env_var("SUPABASE_SERVICE_KEY")读取，两个默认值均可显式覆盖。

2.3 连接注意事项：选择 session 模式端口

Supabase 提供两个连接池端口：

• 事务模式（transaction mode）：端口 6543；
• 会话模式（session mode）：端口 5432。

官方文档明确指出：为了与 pgvector 操作获得最佳兼容性，应使用会话模式（端口 5432）或直连。原因在于事务模式下的连接池代理可能干扰 pgvector 这类需要稳定会话上下文的操作。此外，使用 URI 格式连接串时，密码中的特殊字符需要进行百分号编码（例如p=ssword应写为p%3Dssword），否则可能触发psycopg.OperationalError连接错误。

三、SupabasePgvectorDocumentStore：Supabase 上的向量文档存储

3.1 设计定位

SupabasePgvectorDocumentStore继承自PgvectorDocumentStore，是一个"薄封装"，只改变两处 Supabase 专属默认值：

1. 连接串默认从SUPABASE_DB_URL环境变量读取；
2. create_extension默认为False——因为 pgvector 已在 Supabase 上预装，无需再创建扩展。

其官方使用前置说明为"It should be used with Supabase installed"，即假设你已有可用的 Supabase 项目。基础初始化示例：

from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore document_store = SupabasePgvectorDocumentStore( embedding_dimension=768, vector_function="cosine_similarity", recreate_table=True, )

3.2 构造参数详解

完整构造函数签名如下：

__init__( *, connection_string: Secret = Secret.from_env_var("SUPABASE_DB_URL"), create_extension: bool = False, schema_name: str = "public", table_name: str = "haystack_documents", language: str = "english", embedding_dimension: int = 768, vector_type: Literal["vector", "halfvec"] = "vector", vector_function: Literal[ "cosine_similarity", "inner_product", "l2_distance" ] = "cosine_similarity", recreate_table: bool = False, search_strategy: Literal[ "exact_nearest_neighbor", "hnsw" ] = "exact_nearest_neighbor", hnsw_recreate_index_if_exists: bool = False, hnsw_index_creation_kwargs: dict[str, int] | None = None, hnsw_index_name: str = "haystack_hnsw_index", hnsw_ef_search: int | None = None, keyword_index_name: str = "haystack_keyword_index" ) -> None

各参数含义与默认值一览：

3.3 三种相似度函数的选择要点

vector_function决定向量相似度度量，直接影响检索排序语义：

• cosine_similarity：余弦相似度，适合文本嵌入场景，分值越高越相似；
• inner_product：内积，分值越高越相似；
• l2_distance：欧氏距离，返回向量间直线距离，分值越小越相似（与其他两者方向相反）。

重要：当search_strategy="hnsw"时，检索时使用的vector_function应与建索引时使用的一致，才能有效利用 HNSW 索引；否则索引可能无法命中，退化为全表扫描。

3.4 写入文档

配合SentenceTransformersDocumentEmbedder生成嵌入后写入，使用DuplicatePolicy控制重复文档行为：

from haystack import Document from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document(content="Elephants have been observed to behave in a way that indicates..."), Document(content="In certain places, you can witness the phenomenon of bioluminescent waves."), ] document_embedder = SentenceTransformersDocumentEmbedder() documents_with_embeddings = document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get("documents"), policy=DuplicatePolicy.OVERWRITE, )

四、SupabasePgvectorEmbeddingRetriever：稠密向量检索

4.1 定位与用法

SupabasePgvectorEmbeddingRetriever基于PgvectorEmbeddingRetriever，从SupabasePgvectorDocumentStore中按稠密向量检索文档，是 RAG 查询链路的核心组件。典型用法是嵌入查询文本后检索：

from haystack import Document, Pipeline from haystack.document_stores.types.policy import DuplicatePolicy # Requires: pip install sentence-transformers-haystack from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore from haystack_integrations.components.retrievers.supabase import SupabasePgvectorEmbeddingRetriever document_store = SupabasePgvectorDocumentStore( embedding_dimension=768, vector_function="cosine_similarity", recreate_table=True, ) # —— 索引阶段 —— documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document(content="Elephants have been observed to behave in a way that indicates..."), Document(content="In certain places, you can witness the phenomenon of bioluminescent waves."), ] document_embedder = SentenceTransformersDocumentEmbedder() documents_with_embeddings = document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get("documents"), policy=DuplicatePolicy.OVERWRITE, ) # —— 查询阶段 —— query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder()) query_pipeline.add_component("retriever", SupabasePgvectorEmbeddingRetriever(document_store=document_store)) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") query = "How many languages are there?" res = query_pipeline.run({"text_embedder": {"text": query}}) print(res["retriever"]["documents"][0].content) # >> "There are over 7,000 languages spoken around the world today."

4.2 构造参数

__init__( *, document_store: SupabasePgvectorDocumentStore, filters: dict[str, Any] | None = None, top_k: int = 10, vector_function: ( Literal["cosine_similarity", "inner_product", "l2_distance"] | None ) = None, filter_policy: str | FilterPolicy = FilterPolicy.REPLACE ) -> None

异常：当document_store不是SupabasePgvectorDocumentStore实例，或vector_function不在合法选项中时，抛出ValueError。

4.3 filter_policy：过滤器如何生效

FilterPolicy定义于 haystack/document_stores/types/filter_policy.py，是包含两个成员的枚举：

• REPLACE（"replace"）：运行时传入的过滤器替换初始化时设置的过滤器；
• MERGE（"merge"）：运行时过滤器与初始化过滤器合并，字段冲突时运行时值覆盖初始化值。

合并逻辑在源码的apply_filter_policy中实现，会依据比较过滤器（field/operator/value）与逻辑过滤器（operator/conditions，支持AND/OR/NOT）的不同组合执行相应的合并函数。实际使用时注意：filters仅作用于文档meta字段，例如：

retriever = SupabasePgvectorEmbeddingRetriever( document_store=document_store, filters={"field": "meta.category", "operator": "==", "value": "article"}, top_k=5, )

五、SupabasePgvectorKeywordRetriever：PostgreSQL 关键词检索

5.1 定位与排名机制

SupabasePgvectorKeywordRetriever基于PgvectorKeywordRetriever，按关键词从文档存储中检索文档。与向量检索不同，它不需要查询嵌入，直接对关键词查询执行 PostgreSQL 全文搜索。其排序使用 PostgreSQL 的ts_rank_cd函数，该函数综合考量：

• 查询词在文档中出现的频率；
• 查询词在文档中彼此靠近的程度；
• 查询词出现位置所属文档片段的重要性（例如标题通常比正文更重要）。

5.2 使用示例

from haystack import Document from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore from haystack_integrations.components.retrievers.supabase import SupabasePgvectorKeywordRetriever document_store = SupabasePgvectorDocumentStore( embedding_dimension=768, recreate_table=True, ) documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document(content="Elephants have been observed to behave in a way that indicates..."), Document(content="In certain places, you can witness the phenomenon of bioluminescent waves."), ] document_store.write_documents(documents, policy=DuplicatePolicy.OVERWRITE) retriever = SupabasePgvectorKeywordRetriever(document_store=document_store) result = retriever.run(query="languages") print(result["documents"][0].content) # >> "There are over 7,000 languages spoken around the world today."

5.3 构造参数

__init__( *, document_store: SupabasePgvectorDocumentStore, filters: dict[str, Any] | None = None, top_k: int = 10, filter_policy: str | FilterPolicy = FilterPolicy.REPLACE ) -> None

注意：关键词检索的语言由document_store初始化时的language参数（默认"english"）控制，全文索引名由keyword_index_name（默认"haystack_keyword_index"）指定，检索器自身不重复设置。

六、SupabaseBucketDownloader：从 Supabase Storage 下载文件

6.1 定位与设计

SupabaseBucketDownloader从 Supabase Storage 的 bucket 中下载文件，并以ByteStream对象形式在内存中返回。它专为索引管线前置步骤设计：下载得到的ByteStream可以直接交给DocumentConverter（如TextFileToDocument、PyPDFToDocument等，参见>from haystack_integrations.components.downloaders.supabase import SupabaseBucketDownloader from haystack.utils import Secret downloader = SupabaseBucketDownloader( supabase_url="https://<project-ref>.supabase.co", supabase_key=Secret.from_env_var("SUPABASE_SERVICE_KEY"), bucket_name="my-documents", ) result = downloader.run(sources=["reports/report.pdf", "data/notes.txt"]) streams = result["streams"]

6.3 构造参数

__init__( *, supabase_url: str, supabase_key: Secret = Secret.from_env_var("SUPABASE_SERVICE_KEY"), bucket_name: str, file_extensions: list[str] | None = None ) -> None

6.4 warm_up 与 run

• warm_up() -> None：初始化 Supabase 客户端。首次调用run()时会自动执行，也可在 Pipeline 中显式调用（配合Pipeline.warm_up()预热机制）。
• run(sources: list[str]) -> dict[str, list[ByteStream]]：sources为 bucket 内的文件路径列表，例如["folder/file.pdf", "notes.txt"]。

返回值结构：

{ "streams": [ByteStream, ByteStream, ...] }

每个成功下载的文件对应一个ByteStream，且该对象的meta中已写入两个键：meta["file_path"]（bucket 内原始路径）与meta["bucket_name"]（所属 bucket 名）。下游转换器可利用这些元数据溯源文件来源。

6.5 ByteStream 是什么

ByteStream是 Haystack 表示二进制数据的核心数据类，定义于 haystack/dataclasses/byte_stream.py，包含三个字段：

• data: bytes——二进制内容；
• meta: dict[str, Any]——附加元数据（下载器在此写入file_path与bucket_name）；
• mime_type: str | None——MIME 类型。

它提供to_file()、from_file_path()、from_string()、to_string()、to_dict()/from_dict()等便捷方法。例如将下载结果落盘：

for stream in streams: # 注意：meta 在写盘时不会保留 stream.to_file(f"downloaded_{stream.meta['file_path'].split('/')[-1]}")

TextFileToDocument、PyPDFToDocument等转换器均直接接受ByteStream列表作为输入，因此下载器可与它们无缝串联成索引管线。

七、端到端实战：从 Storage 下载到 RAG 查询

将上述组件组合成一条完整的索引 + 查询链路：SupabaseBucketDownloader拉取文件 → 转换器转文档 → 嵌入器生成向量 → 写入SupabasePgvectorDocumentStore→ 查询时用SupabasePgvectorEmbeddingRetriever检索 → 交给 LLM 生成答案。

from haystack import Document, Pipeline from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.document_stores.types.policy import DuplicatePolicy from haystack.utils import Secret from haystack_integrations.components.downloaders.supabase import SupabaseBucketDownloader from haystack_integrations.components.converters import TextFileToDocument from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder, ) from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore from haystack_integrations.components.retrievers.supabase import SupabasePgvectorEmbeddingRetriever # 1. 文档存储 document_store = SupabasePgvectorDocumentStore( embedding_dimension=768, vector_function="cosine_similarity", recreate_table=True, ) # 2. 索引管线：下载 -> 转换 -> 嵌入 -> 写入 indexing = Pipeline() indexing.add_component( "downloader", SupabaseBucketDownloader( supabase_url="https://<project-ref>.supabase.co", supabase_key=Secret.from_env_var("SUPABASE_SERVICE_KEY"), bucket_name="my-documents", file_extensions=[".pdf", ".txt"], ), ) indexing.add_component("converter", TextFileToDocument()) indexing.add_component("embedder", SentenceTransformersDocumentEmbedder()) indexing.add_component("writer", document_store.writer_script if False else None) # 占位，见下方说明 indexing.connect("downloader.streams", "converter.sources")

说明：上面writer一步示意性地标注了串联点。实际写库建议直接用document_store.write_documents(docs, policy=DuplicatePolicy.OVERWRITE)（与 supabasedocumentstore.mdx 中官方示例一致），把"下载 → 转换 → 嵌入"封装为索引管线，再统一写库，避免在管线中维护 writer 组件的额外配置。

查询侧与第四章示例一致；若希望检索结果更丰富，可将向量检索与关键词检索并行接入DocumentJoiner，实现混合检索（hybrid search）：

query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder()) query_pipeline.add_component( "embedding_retriever", SupabasePgvectorEmbeddingRetriever(document_store=document_store, top_k=5), ) query_pipeline.add_component( "keyword_retriever", SupabasePgvectorKeywordRetriever(document_store=document_store, top_k=5), ) query_pipeline.add_component( "joiner", DocumentJoiner(join_mode="concatenate"), # 来自 haystack.components.joiners ) query_pipeline.connect("text_embedder.embedding", "embedding_retriever.query_embedding") query_pipeline.connect("embedding_retriever.documents", "joiner.documents") query_pipeline.connect("keyword_retriever.documents", "joiner.documents") result = query_pipeline.run({"text_embedder": {"text": "How many languages are there?"}})

query_pipeline.run返回的result["retriever"]["documents"]中每个Document均包含content、embedding与meta（写入时保留的元数据），可直接进入PromptBuilder/ChatPromptBuilder构造上下文，参考 supabasedocumentstore.mdx 中的完整 RAG 示例。

八、序列化：to_dict 与 from_dict

三个数据组件与两个检索器均实现 Haystack 标准的序列化协议：

• to_dict() -> dict[str, Any]：将组件序列化为字典（Secret字段会以安全形式表示，不会明文导出密钥）；
• from_dict(data: dict[str, Any]) -> 组件类型：从字典反序列化重建组件实例。

序列化让组件可以安全地存入 YAML/JSON 管道定义（配合haystack.marshal的 YAML 支持），便于管道版本管理与跨环境迁移。例如保存与恢复文档存储配置：

import json data = document_store.to_dict() with open("document_store_config.json", "w") as f: json.dump(data, f) restored = SupabasePgvectorDocumentStore.from_dict(data)

需要注意的是：from_dict反序列化时，连接信息仍依赖SUPABASE_DB_URL环境变量（Secret默认指向该变量），因此迁移环境时需同步配置环境变量。

九、补充说明与最佳实践

1. 维度一致性：embedding_dimension必须与所用嵌入模型输出维度一致。以sentence-transformers的all-MiniLM-L6-v2（384 维）或部分模型（768 维）为参考，768是文档与代码示例中的常用值。
2. recreate_table=True慎用：它会在表存在时重建（清空既有数据），适合开发调试，生产环境应保持False并配合DuplicatePolicy控制写入行为。
3. 大表请用 HNSW：数据量增长后，exact_nearest_neighbor（精确最近邻）会退化为逐行扫描；切换search_strategy="hnsw"并用hnsw_ef_search调节查询精度/速度权衡时，务必保证检索器的vector_function与建索引时一致。
4. 私有桶密钥管理：SupabaseBucketDownloader访问私有 bucket 时必须使用 service role key，建议通过Secret.from_env_var("SUPABASE_SERVICE_KEY")注入，避免密钥出现在代码或管道文件中。
5. 会话模式连接：优先使用端口 5432（session mode）或直连，避免事务模式连接池影响 pgvector 操作。

十、参考资料

• 本文主体来源：integrations-api/supabase.md
• 官方使用指南：supabasedocumentstore.mdx
• 底层基类使用指南：pgvectordocumentstore.mdx
• ByteStream数据类实现：haystack/dataclasses/byte_stream.py
• FilterPolicy过滤器策略实现：haystack/document_stores/types/filter_policy.py

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack