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稠密向量检索器、SupabasePgvectorKeywordRetrieverPostgreSQL 全文关键词检索器以及SupabaseBucketDownloader从 Supabase Storage 下载文件为ByteStream。读完本文你将能够完成从环境配置、文档索引、双路检索到文件下载接入索引管线的完整 Supabase RAG 方案搭建并理解每个参数在源码层面的实际作用。一、组件全景Haystack 为 Supabase 提供什么Supabase 是一个基于 PostgreSQL 的开源后端平台。Haystack 的 Supabase 集成围绕 Supabase 的两项核心能力展开pgvector 扩展Supabase 预装支撑向量相似度搜索Supabase Storage支撑对象存储文件的批量下载与索引。集成共提供四个组件全部位于haystack_integrations命名空间下组件所属模块职责SupabasePgvectorDocumentStorehaystack_integrations.document_stores.supabase基于 PostgreSQL pgvector 的文档存储SupabasePgvectorEmbeddingRetrieverhaystack_integrations.components.retrievers.supabase按稠密向量检索文档SupabasePgvectorKeywordRetrieverhaystack_integrations.components.retrievers.supabase按关键词全文检索文档SupabaseBucketDownloaderhaystack_integrations.components.downloaders.supabase从 Supabase Storage 下载文件并转为ByteStream其中前三个组件是对 pgvector 集成PgvectorDocumentStore及其检索器的薄封装thin wrapper仅调整了 Supabase 专属默认值第四个组件则是独立的文件下载器。更多官方用法可参考仓库中的 supabasedocumentstore.mdx。二、安装与环境准备2.1 安装集成包pip install supabase-haystack官方示例使用 Sentence Transformers 嵌入器它们已迁移至独立包sentence-transformers-haystack如需运行下文示例请一并安装pip install sentence-transformers-haystack2.2 配置两个环境变量Supabase 集成依赖两个环境变量# 1. 数据库连接串文档存储与检索器使用 export SUPABASE_DB_URLpostgresql://postgres.[project-ref]:[password]aws-0-[region].pooler.supabase.com:5432/postgres # 2. Service Role KeyStorage 下载器使用私有桶必须使用服务角色密钥 export SUPABASE_SERVICE_KEYyour-service-role-keySUPABASE_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 格式连接串时密码中的特殊字符需要进行百分号编码例如pssword应写为p%3Dssword否则可能触发psycopg.OperationalError连接错误。三、SupabasePgvectorDocumentStoreSupabase 上的向量文档存储3.1 设计定位SupabasePgvectorDocumentStore继承自PgvectorDocumentStore是一个薄封装只改变两处 Supabase 专属默认值连接串默认从SUPABASE_DB_URL环境变量读取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_dimension768, vector_functioncosine_similarity, recreate_tableTrue, )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各参数含义与默认值一览参数类型默认值说明connection_stringSecret环境变量SUPABASE_DB_URLSupabase PostgreSQL 连接串格式见上文create_extensionboolFalse是否在 pgvector 不存在时创建扩展Supabase 已预装故默认关闭schema_namestrpublic建表所在的 schematable_namestrhaystack_documents存放 Haystack 文档的表名languagestrenglish关键词检索时解析查询与文档内容所用的语言embedding_dimensionint768嵌入向量维度须与所用嵌入模型输出维度一致vector_typevector/halfvecvector向量存储类型halfvec为半精度向量省一半存储vector_functioncosine_similarity/inner_product/l2_distancecosine_similarity相似度函数recreate_tableboolFalse表已存在时是否重建会清空数据慎用search_strategyexact_nearest_neighbor/hnswexact_nearest_neighbor精确最近邻或 HNSW 近似最近邻hnsw_recreate_index_if_existsboolFalseHNSW 索引已存在时是否重建hnsw_index_creation_kwargsdict[str, int] \| NoneNoneHNSW 建索引的额外参数如m、ef_constructionhnsw_index_namestrhaystack_hnsw_indexHNSW 索引名hnsw_ef_searchint \| NoneNone查询时 HNSW 的ef_search参数keyword_index_namestrhaystack_keyword_index关键词全文索引名3.3 三种相似度函数的选择要点vector_function决定向量相似度度量直接影响检索排序语义cosine_similarity余弦相似度适合文本嵌入场景分值越高越相似inner_product内积分值越高越相似l2_distance欧氏距离返回向量间直线距离分值越小越相似与其他两者方向相反。重要当search_strategyhnsw时检索时使用的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(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates...), Document(contentIn 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), policyDuplicatePolicy.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_dimension768, vector_functioncosine_similarity, recreate_tableTrue, ) # —— 索引阶段 —— documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates...), Document(contentIn 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), policyDuplicatePolicy.OVERWRITE, ) # —— 查询阶段 —— query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder()) query_pipeline.add_component(retriever, SupabasePgvectorEmbeddingRetriever(document_storedocument_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_storeSupabasePgvectorDocumentStore—必填检索的文档存储实例filtersdict[str, Any] \| NoneNone作用于检索结果元数据meta的过滤器top_kint10返回的最大文档数vector_function三选一 |NoneNone检索时使用的相似度函数默认为document_store实例中设置的值。若 store 使用hnsw搜索策略此处应与建索引时的函数一致filter_policystr \| FilterPolicyFilterPolicy.REPLACE过滤器应用策略见 4.3异常当document_store不是SupabasePgvectorDocumentStore实例或vector_function不在合法选项中时抛出ValueError。4.3 filter_policy过滤器如何生效FilterPolicy定义于 haystack/document_stores/types/filter_policy.py是包含两个成员的枚举REPLACEreplace运行时传入的过滤器替换初始化时设置的过滤器MERGEmerge运行时过滤器与初始化过滤器合并字段冲突时运行时值覆盖初始化值。合并逻辑在源码的apply_filter_policy中实现会依据比较过滤器field/operator/value与逻辑过滤器operator/conditions支持AND/OR/NOT的不同组合执行相应的合并函数。实际使用时注意filters仅作用于文档meta字段例如retriever SupabasePgvectorEmbeddingRetriever( document_storedocument_store, filters{field: meta.category, operator: , value: article}, top_k5, )五、SupabasePgvectorKeywordRetrieverPostgreSQL 关键词检索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_dimension768, recreate_tableTrue, ) documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document(contentElephants have been observed to behave in a way that indicates...), Document(contentIn certain places, you can witness the phenomenon of bioluminescent waves.), ] document_store.write_documents(documents, policyDuplicatePolicy.OVERWRITE) retriever SupabasePgvectorKeywordRetriever(document_storedocument_store) result retriever.run(querylanguages) 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_storeSupabasePgvectorDocumentStore—必填检索的文档存储实例filtersdict[str, Any] \| NoneNone作用于文档meta的过滤器top_kint10返回的最大文档数filter_policystr \| FilterPolicyFilterPolicy.REPLACE过滤器应用策略同 4.3 节注意关键词检索的语言由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_urlhttps://project-ref.supabase.co, supabase_keySecret.from_env_var(SUPABASE_SERVICE_KEY), bucket_namemy-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参数类型默认值说明supabase_urlstr—必填Supabase 项目 URL如https://project-ref.supabase.cosupabase_keySecret环境变量SUPABASE_SERVICE_KEY认证 API 密钥私有 bucket 必须使用 service role keybucket_namestr—必填要下载文件的 Storage bucket 名称file_extensionslist[str] \| NoneNone可选的文件扩展名过滤列表如[.pdf, .txt]为None时下载全部文件扩展名匹配不区分大小写6.4 warm_up 与 runwarm_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_namemime_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(fdownloaded_{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_dimension768, vector_functioncosine_similarity, recreate_tableTrue, ) # 2. 索引管线下载 - 转换 - 嵌入 - 写入 indexing Pipeline() indexing.add_component( downloader, SupabaseBucketDownloader( supabase_urlhttps://project-ref.supabase.co, supabase_keySecret.from_env_var(SUPABASE_SERVICE_KEY), bucket_namemy-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, policyDuplicatePolicy.OVERWRITE)与 supabasedocumentstore.mdx 中官方示例一致把下载 → 转换 → 嵌入封装为索引管线再统一写库避免在管线中维护 writer 组件的额外配置。查询侧与第四章示例一致若希望检索结果更丰富可将向量检索与关键词检索并行接入DocumentJoiner实现混合检索hybrid searchquery_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder()) query_pipeline.add_component( embedding_retriever, SupabasePgvectorEmbeddingRetriever(document_storedocument_store, top_k5), ) query_pipeline.add_component( keyword_retriever, SupabasePgvectorKeywordRetriever(document_storedocument_store, top_k5), ) query_pipeline.add_component( joiner, DocumentJoiner(join_modeconcatenate), # 来自 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默认指向该变量因此迁移环境时需同步配置环境变量。九、补充说明与最佳实践维度一致性embedding_dimension必须与所用嵌入模型输出维度一致。以sentence-transformers的all-MiniLM-L6-v2384 维或部分模型768 维为参考768是文档与代码示例中的常用值。recreate_tableTrue慎用它会在表存在时重建清空既有数据适合开发调试生产环境应保持False并配合DuplicatePolicy控制写入行为。大表请用 HNSW数据量增长后exact_nearest_neighbor精确最近邻会退化为逐行扫描切换search_strategyhnsw并用hnsw_ef_search调节查询精度/速度权衡时务必保证检索器的vector_function与建索引时一致。私有桶密钥管理SupabaseBucketDownloader访问私有 bucket 时必须使用 service role key建议通过Secret.from_env_var(SUPABASE_SERVICE_KEY)注入避免密钥出现在代码或管道文件中。会话模式连接优先使用端口 5432session mode或直连避免事务模式连接池影响 pgvector 操作。十、参考资料本文主体来源integrations-api/supabase.md官方使用指南supabasedocumentstore.mdx底层基类使用指南pgvectordocumentstore.mdxByteStream数据类实现haystack/dataclasses/byte_stream.pyFilterPolicy过滤器策略实现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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考