在 Haystack 中使用 SupabaseSupabasePgvectorDocumentStore、双检索器与存储桶下载器实战指南【免费下载链接】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/haystackHaystack 的 Supabase 集成supabase-haystack让开发者可以直接把 Supabase 托管的 PostgreSQL 数据库作为生产级文档存储利用预装的 pgvector 扩展做向量相似度检索利用 PostgreSQL 全文检索做关键词检索并可通过SupabaseBucketDownloader从 Supabase Storage 拉取文件参与索引管线。本指南围绕 version-2.22 的 Supabase 集成 API 参考 展开覆盖SupabasePgvectorDocumentStore的全部初始化参数、SupabasePgvectorEmbeddingRetriever与SupabasePgvectorKeywordRetriever的调用方式、序列化接口以及存储桶下载器的完整用法并配合官方使用文档与仓库源码给出可复制、可运行的示例。一、集成概览与安装Supabase 是构建在 PostgreSQL 之上的开源后端平台。Haystack 的 Supabase 集成提供两类文档存储SupabasePgvectorDocumentStore——基于 pgvector 扩展的向量相似度检索pgvector 在 Supabase 上预装无需手动创建扩展SupabaseGroongaDocumentStore——基于 PGroonga 扩展的多语言全文检索不需要任何嵌入向量。本文聚焦 pgvector 路线即 API 参考文档覆盖的范围它由三组核心类组成组件所属模块作用SupabasePgvectorDocumentStorehaystack_integrations.document_stores.supabase.document_store文档存储封装连接、建表、写入与检索 SQLSupabasePgvectorEmbeddingRetrieverhaystack_integrations.components.retrievers.supabase.embedding_retriever基于稠密向量相似度召回文档SupabasePgvectorKeywordRetrieverhaystack_integrations.components.retrievers.supabase.keyword_retriever基于关键词全文检索召回文档SupabaseBucketDownloaderhaystack_integrations.components.downloaders.supabase.supabase_bucket_downloader从 Supabase Storage 下载文件为ByteStream安装集成与示例所需的 Sentence Transformers 嵌入器pip install supabase-haystack pip install sentence-transformers-haystack # 运行本文示例所需官方使用文档见 SupabaseDocumentStore 指南API 参考见 version-2.22 integrations-api/supabase.md。二、连接 Supabase 数据库SUPABASE_DB_URL 与端口选择所有 pgvector 组件都通过SUPABASE_DB_URL环境变量读取连接字符串因此第一步是配置该变量export SUPABASE_DB_URLpostgresql://postgres.[project-ref]:[password]aws-0-[region].pooler.supabase.com:5432/postgres连接要点官方文档明确提示Supabase 提供两个连接池端口——事务模式6543与会话模式5432。为了保证 pgvector 操作的兼容性建议使用会话模式5432或直连。连接字符串中的[project-ref]、[password]、[region]需替换为 Supabase 项目控制台提供的信息。若你的数据库密码包含、:、/等特殊字符URI 形式需要对特殊字符做百分号编码否则会出现类似psycopg.OperationalError的连接错误也可以改用 PostgreSQL 的 keyword/value 形式host... port... dbname... user... password...该形式无需百分号编码。这两种格式同样适用于自托管 pgvector 场景参见 PgvectorDocumentStore 指南。三、SupabasePgvectorDocumentStore初始化参数全解析SupabasePgvectorDocumentStore是PgvectorDocumentStore的轻量封装针对 Supabase 做了两个默认调整连接字符串默认从SUPABASE_DB_URL环境变量读取create_extension默认设为False因为 pgvector 在 Supabase 上已预装。3.1 构造函数签名与全部参数__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 ) - None3.2 参数详解参数类型默认值说明connection_stringSecretSecret.from_env_var(SUPABASE_DB_URL)数据库连接字符串以环境变量形式注入格式见上文create_extensionboolFalse是否在扩展不存在时创建 pgvector 扩展。Supabase 已预装默认关闭schema_namestrpublic建表所在的 schematable_namestrhaystack_documents存储 Haystack 文档的表名languagestrenglish关键词检索时解析查询与文档内容所用的语言embedding_dimensionint768嵌入向量的维度须与嵌入模型输出维度一致vector_typeLiteral[vector, halfvec]vector向量存储类型vectorfloat32或halfvecfloat16可节省存储空间vector_functionLiteral[cosine_similarity, inner_product, l2_distance]cosine_similarity向量相似度函数recreate_tableboolFalse若表已存在是否重建会丢失已有数据search_strategyLiteral[exact_nearest_neighbor, hnsw]exact_nearest_neighbor检索策略精确最近邻或 HNSW 近似最近邻hnsw_recreate_index_if_existsboolFalseHNSW 索引已存在时是否重建hnsw_index_creation_kwargsdict[str, int] \| NoneNone创建 HNSW 索引的额外关键字参数hnsw_index_namestrhaystack_hnsw_indexHNSW 索引名hnsw_ef_searchint \| NoneNone查询时 HNSW 的ef_search参数keyword_index_namestrhaystack_keyword_index关键词索引名3.3 最小初始化示例from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore document_store SupabasePgvectorDocumentStore( embedding_dimension768, vector_functioncosine_similarity, recreate_tableTrue, )注意recreate_tableTrue会删除并重建表开发阶段方便反复试验生产环境慎用。若改用DuplicatePolicy.SKIP写入则可以重复运行脚本而不报错。3.4 序列化接口to_dict() - dict[str, Any]将组件序列化为字典便于保存到 YAML/JSON 或存入数据库from_dict(data: dict[str, Any]) - SupabasePgvectorDocumentStore从字典反序列化重建组件。Haystack 的组件序列化机制保证了 Pipeline 可以被整体导出、版本化管理与跨环境复现详见 Haystack 序列化相关实现。四、写入文档与构建语义检索管线4.1 索引文档写入带嵌入的文档from haystack import Document from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) 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, )write_documents接收DuplicatePolicy控制重复文档处理方式如OVERWRITE覆盖、SKIP跳过这在索引侧保证幂等性。4.2 语义检索Embedding RetrieverSupabasePgvectorEmbeddingRetriever继承自PgvectorEmbeddingRetriever用于按查询向量的相似度召回文档。完整初始化签名__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实例filtersNone应用于检索结果的元数据过滤条件top_k10最多返回的文档数vector_functionNone检索用相似度函数默认沿用document_store上设置的函数filter_policyFilterPolicy.REPLACE运行时过滤器与初始化过滤器的合并策略关于vector_function需要特别注意cosine_similarity与inner_product属于相似度函数分数越高越相似l2_distance返回向量间的直线距离分数越小越相似重要若文档存储使用hnsw检索策略检索时使用的向量函数必须与建索引时一致才能充分利用 HNSW 索引。filter_policy的可选值在 Haystack 的 FilterPolicy 枚举 中定义REPLACE表示运行时过滤器替换初始化过滤器MERGE表示两者合并重叠键由运行时过滤器覆盖此外还有KEEP等取值。在 Pipeline 中使用from haystack import Document, Pipeline from haystack.document_stores.types.policy import DuplicatePolicy 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.管线中检索器通常位于Text Embedder 之后、PromptBuilder/阅读器之前先由嵌入器把查询文本编码为query_embedding再通过retriever.query_embedding输入口传入这是检索器唯一的必填运行参数。输出为documents列表。单独使用调试/离线场景也可以直接传入手工构造的向量retriever SupabasePgvectorEmbeddingRetriever(document_storedocument_store) retriever.run(query_embedding[0.1] * 768) # 与 embedding_dimension 等长的向量4.3 报错约定SupabasePgvectorEmbeddingRetriever.__init__在以下情况抛出ValueErrordocument_store不是SupabasePgvectorDocumentStore实例vector_function不属于cosine_similarity/inner_product/l2_distance三个合法取值。两个检索器同样提供to_dict()/from_dict()序列化接口可随 Pipeline 一起导出复现。五、关键词全文检索SupabasePgvectorKeywordRetrieverSupabasePgvectorKeywordRetriever继承自PgvectorKeywordRetriever不依赖任何嵌入向量直接以关键词匹配文档适合做零成本的基础检索。5.1 排序原理组件使用 PostgreSQL 全文检索的ts_rank_cd函数对文档排序。该函数的排序依据是查询词在文档中出现的频率查询词在文档中彼此相距的远近越近得分越高词出现在文档的哪个部分不同位置权重不同。5.2 初始化签名与参数__init__( *, document_store: SupabasePgvectorDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数默认值说明document_store必填SupabasePgvectorDocumentStore实例filtersNone检索结果的元数据过滤条件top_k10最多返回的文档数filter_policyFilterPolicy.REPLACE过滤器合并策略ValueError触发条件document_store不是SupabasePgvectorDocumentStore实例。5.3 单独使用from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore from haystack_integrations.components.retrievers.supabase import ( SupabasePgvectorKeywordRetriever, ) document_store SupabasePgvectorDocumentStore() retriever SupabasePgvectorKeywordRetriever(document_storedocument_store) retriever.run(querymy nice query)5.4 关键词检索 索引 结果验证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.注意关键词检索默认不做模糊匹配与ElasticsearchBM25Retriever不同查询词拼写稍有偏差就可能返回零结果因此需要精心构造查询。解析查询与文档所用的语言由文档存储的language参数控制默认english中文检索时建议确认 Supabase 侧 PostgreSQL 支持的语言配置。5.5 完整关键词 RAG 管线from haystack import Document, Pipeline from haystack.components.builders.answer_builder import AnswerBuilder from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore from haystack_integrations.components.retrievers.supabase import ( SupabasePgvectorKeywordRetriever, ) prompt_template [ ChatMessage.from_user( Given these documents, answer the question.\nDocuments:\n {% for doc in documents %}{{ doc.content }}{% endfor %}\n Question: {{question}}\nAnswer:, ), ] document_store SupabasePgvectorDocumentStore( languageenglish, 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 a high level of self-awareness, such as recognizing themselves in mirrors.), Document(contentIn certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.), ] document_store.write_documents(documentsdocuments, policyDuplicatePolicy.SKIP) retriever SupabasePgvectorKeywordRetriever(document_storedocument_store) rag_pipeline Pipeline() rag_pipeline.add_component(nameretriever, instanceretriever) rag_pipeline.add_component( instanceChatPromptBuilder(templateprompt_template, required_variables{question, documents}), nameprompt_builder, ) rag_pipeline.add_component(instanceOpenAIChatGenerator(), namellm) rag_pipeline.add_component(instanceAnswerBuilder(), nameanswer_builder) rag_pipeline.connect(retriever, prompt_builder.documents) rag_pipeline.connect(prompt_builder.prompt, llm.messages) rag_pipeline.connect(llm.replies, answer_builder.replies) rag_pipeline.connect(retriever, answer_builder.documents) question languages spoken around the world today result rag_pipeline.run( { retriever: {query: question}, prompt_builder: {question: question}, answer_builder: {query: question}, }, ) print(result[answer_builder])运行该管线需要提前设置两个环境变量OPENAI_API_KEYOpenAI API 密钥与SUPABASE_DB_URLSupabase 连接串。六、完整的向量 RAG 管线端到端示例将嵌入检索与生成环节串联起来构成一条可运行的问答管线from haystack import Document, Pipeline from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage 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 a high level of self-awareness.), Document(contentIn certain places, you can witness the phenomenon of bioluminescent waves.), ] embedder SentenceTransformersDocumentEmbedder() documents_with_embeddings embedder.run(documents) document_store.write_documents( documents_with_embeddings[documents], policyDuplicatePolicy.OVERWRITE, ) prompt_template [ ChatMessage.from_system(Answer the question based on the provided context.), ChatMessage.from_user( Query: {{query}}\nDocuments:\n{% for doc in documents %}{{ doc.content }}\n{% endfor %}\nAnswer:, ), ] query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder()) query_pipeline.add_component( retriever, SupabasePgvectorEmbeddingRetriever(document_storedocument_store), ) query_pipeline.add_component( prompt_builder, ChatPromptBuilder(templateprompt_template, required_variables[query, documents]), ) query_pipeline.add_component(generator, OpenAIChatGenerator(modelgpt-4o)) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) query_pipeline.connect(retriever.documents, prompt_builder.documents) query_pipeline.connect(prompt_builder.prompt, generator.messages) result query_pipeline.run( { text_embedder: {text: How many languages are there?}, prompt_builder: {query: How many languages are there?}, }, )该示例体现了三条核心连接text_embedder.embedding → retriever.query_embedding查询编码、retriever.documents → prompt_builder.documents召回结果注入提示词、prompt_builder.prompt → generator.messages生成答案。七、SupabaseBucketDownloader从 Supabase Storage 下载文件7.1 用途与数据流SupabaseBucketDownloader用于从 Supabase Storage 存储桶下载文件并将文件在内存中转换为ByteStream对象返回随后可直接交给DocumentConverter等组件进入索引管线。它不落盘适合流式/内存式处理。7.2 初始化签名与参数__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_url必填Supabase 项目 URL形如https://project-ref.supabase.cosupabase_keySecret.from_env_var(SUPABASE_SERVICE_KEY)认证用 Supabase API Key私有存储桶需使用service role key服务角色密钥bucket_name必填要下载文件的存储桶名称file_extensionsNone可选扩展名过滤列表如[.pdf, .txt]None表示下载全部文件扩展名大小写不敏感匹配7.3 使用方法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]7.4 run() 参数与返回值run(sources: list[str]) - dict[str, list[ByteStream]]sources存储桶内的文件路径列表如[folder/file.pdf, notes.txt]返回值字典键streams对应ByteStream列表每个成功下载的文件对应一个ByteStream且该对象的meta中携带file_path与bucket_name两个字段。ByteStream是 Haystack 表示二进制数据的核心数据类型定义于 haystack/dataclasses/byte_stream.py支持to_file()落盘、from_file_path()从文件读取、from_string()从字符串构造并自动携带meta与mime_type字段。利用meta[file_path]可以追溯每个流对应的原始文件便于下游TextFileToDocument、PDFToDocument等转换器记录来源。7.5 warm_up() 与生命周期warm_up() - Nonewarm_up()负责初始化 Supabase 客户端会在首次调用run()时自动执行也可以在 Pipeline 中显式提前调用。在 Pipeline 场景中Haystack 会在组件执行前统一调用各组件warm_up()Haystack 组件生命周期约定warm_up用于加载模型/初始化客户端等耗时操作run只做实际业务处理从而避免首次查询的冷启动延迟。7.6 组合示例下载 → 转换 → 写入文档存储把下载器接入索引管线即可实现存储桶文件 → 文本 → 文档的完整链路from haystack import Pipeline from haystack.components.converters import TextFileToDocument from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.components.downloaders.supabase import SupabaseBucketDownloader from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore downloader SupabaseBucketDownloader( supabase_urlhttps://project-ref.supabase.co, bucket_namemy-documents, ) converter TextFileToDocument() document_store SupabasePgvectorDocumentStore(embedding_dimension768) indexing Pipeline() indexing.add_component(downloader, downloader) indexing.add_component(converter, converter) indexing.add_component(store, document_store) indexing.connect(downloader.streams, converter.sources) indexing.connect(converter.documents, store.documents) indexing.run({downloader: {sources: [notes.txt, readme.md]}})八、搜索策略与向量函数选型建议结合上文参数选型时可以参考以下决策路径数据规模小数据集万级以内用默认的exact_nearest_neighbor即可获得精确结果大数据集建议search_strategyhnsw换取检索速度。向量函数一致性若启用 HNSWdocument_store初始化时的vector_function与检索时的vector_function必须一致否则索引无法生效见 4.2 节重要提示。embedding_dimension 对齐embedding_dimension必须与嵌入模型输出维度一致本文示例用 768 维对应sentence-transformers常用模型否则写入/检索时报维度错误。存储与成本向量规模大时可选用vector_typehalfvecfloat16降低存储开销。HNSW 调优通过hnsw_index_creation_kwargs如m、ef_construction、hnsw_ef_search查询时的候选集规模越大召回越准、越慢与hnsw_index_name控制索引行为。九、总结supabase-haystack集成为 Haystack 用户提供了托管 PostgreSQL 向量检索 关键词检索 对象存储的一站式后端能力SupabasePgvectorDocumentStore让 Supabase 数据库直接成为 Haystack 文档存储默认读取SUPABASE_DB_URL并针对 Supabase 预装的 pgvector 优化了默认配置SupabasePgvectorEmbeddingRetriever负责稠密向量语义检索SupabasePgvectorKeywordRetriever负责基于ts_rank_cd的关键词全文检索两者均可独立使用或嵌入 RAG PipelineSupabaseBucketDownloader支持从 Supabase Storage 内存化下载文件为ByteStream直接衔接文档转换与索引环节所有组件均实现to_dict()/from_dict()可随 Pipeline 序列化导出与复现。在此基础上还可以参考 SupabaseDocumentStore 官方指南 了解SupabaseGroongaDocumentStorePGroonga 多语言全文检索无需嵌入向量及其SupabaseGroongaBM25Retriever实现 pgvector 与 PGroonga 混合检索的进阶方案。【免费下载链接】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），仅供参考 在 Haystack 中使用 Supabase：SupabasePgvectorDocumentStore、双检索器与存储桶下载器实战指南
在 Haystack 中使用 SupabaseSupabasePgvectorDocumentStore、双检索器与存储桶下载器实战指南【免费下载链接】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/haystackHaystack 的 Supabase 集成supabase-haystack让开发者可以直接把 Supabase 托管的 PostgreSQL 数据库作为生产级文档存储利用预装的 pgvector 扩展做向量相似度检索利用 PostgreSQL 全文检索做关键词检索并可通过SupabaseBucketDownloader从 Supabase Storage 拉取文件参与索引管线。本指南围绕 version-2.22 的 Supabase 集成 API 参考 展开覆盖SupabasePgvectorDocumentStore的全部初始化参数、SupabasePgvectorEmbeddingRetriever与SupabasePgvectorKeywordRetriever的调用方式、序列化接口以及存储桶下载器的完整用法并配合官方使用文档与仓库源码给出可复制、可运行的示例。一、集成概览与安装Supabase 是构建在 PostgreSQL 之上的开源后端平台。Haystack 的 Supabase 集成提供两类文档存储SupabasePgvectorDocumentStore——基于 pgvector 扩展的向量相似度检索pgvector 在 Supabase 上预装无需手动创建扩展SupabaseGroongaDocumentStore——基于 PGroonga 扩展的多语言全文检索不需要任何嵌入向量。本文聚焦 pgvector 路线即 API 参考文档覆盖的范围它由三组核心类组成组件所属模块作用SupabasePgvectorDocumentStorehaystack_integrations.document_stores.supabase.document_store文档存储封装连接、建表、写入与检索 SQLSupabasePgvectorEmbeddingRetrieverhaystack_integrations.components.retrievers.supabase.embedding_retriever基于稠密向量相似度召回文档SupabasePgvectorKeywordRetrieverhaystack_integrations.components.retrievers.supabase.keyword_retriever基于关键词全文检索召回文档SupabaseBucketDownloaderhaystack_integrations.components.downloaders.supabase.supabase_bucket_downloader从 Supabase Storage 下载文件为ByteStream安装集成与示例所需的 Sentence Transformers 嵌入器pip install supabase-haystack pip install sentence-transformers-haystack # 运行本文示例所需官方使用文档见 SupabaseDocumentStore 指南API 参考见 version-2.22 integrations-api/supabase.md。二、连接 Supabase 数据库SUPABASE_DB_URL 与端口选择所有 pgvector 组件都通过SUPABASE_DB_URL环境变量读取连接字符串因此第一步是配置该变量export SUPABASE_DB_URLpostgresql://postgres.[project-ref]:[password]aws-0-[region].pooler.supabase.com:5432/postgres连接要点官方文档明确提示Supabase 提供两个连接池端口——事务模式6543与会话模式5432。为了保证 pgvector 操作的兼容性建议使用会话模式5432或直连。连接字符串中的[project-ref]、[password]、[region]需替换为 Supabase 项目控制台提供的信息。若你的数据库密码包含、:、/等特殊字符URI 形式需要对特殊字符做百分号编码否则会出现类似psycopg.OperationalError的连接错误也可以改用 PostgreSQL 的 keyword/value 形式host... port... dbname... user... password...该形式无需百分号编码。这两种格式同样适用于自托管 pgvector 场景参见 PgvectorDocumentStore 指南。三、SupabasePgvectorDocumentStore初始化参数全解析SupabasePgvectorDocumentStore是PgvectorDocumentStore的轻量封装针对 Supabase 做了两个默认调整连接字符串默认从SUPABASE_DB_URL环境变量读取create_extension默认设为False因为 pgvector 在 Supabase 上已预装。3.1 构造函数签名与全部参数__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 ) - None3.2 参数详解参数类型默认值说明connection_stringSecretSecret.from_env_var(SUPABASE_DB_URL)数据库连接字符串以环境变量形式注入格式见上文create_extensionboolFalse是否在扩展不存在时创建 pgvector 扩展。Supabase 已预装默认关闭schema_namestrpublic建表所在的 schematable_namestrhaystack_documents存储 Haystack 文档的表名languagestrenglish关键词检索时解析查询与文档内容所用的语言embedding_dimensionint768嵌入向量的维度须与嵌入模型输出维度一致vector_typeLiteral[vector, halfvec]vector向量存储类型vectorfloat32或halfvecfloat16可节省存储空间vector_functionLiteral[cosine_similarity, inner_product, l2_distance]cosine_similarity向量相似度函数recreate_tableboolFalse若表已存在是否重建会丢失已有数据search_strategyLiteral[exact_nearest_neighbor, hnsw]exact_nearest_neighbor检索策略精确最近邻或 HNSW 近似最近邻hnsw_recreate_index_if_existsboolFalseHNSW 索引已存在时是否重建hnsw_index_creation_kwargsdict[str, int] \| NoneNone创建 HNSW 索引的额外关键字参数hnsw_index_namestrhaystack_hnsw_indexHNSW 索引名hnsw_ef_searchint \| NoneNone查询时 HNSW 的ef_search参数keyword_index_namestrhaystack_keyword_index关键词索引名3.3 最小初始化示例from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore document_store SupabasePgvectorDocumentStore( embedding_dimension768, vector_functioncosine_similarity, recreate_tableTrue, )注意recreate_tableTrue会删除并重建表开发阶段方便反复试验生产环境慎用。若改用DuplicatePolicy.SKIP写入则可以重复运行脚本而不报错。3.4 序列化接口to_dict() - dict[str, Any]将组件序列化为字典便于保存到 YAML/JSON 或存入数据库from_dict(data: dict[str, Any]) - SupabasePgvectorDocumentStore从字典反序列化重建组件。Haystack 的组件序列化机制保证了 Pipeline 可以被整体导出、版本化管理与跨环境复现详见 Haystack 序列化相关实现。四、写入文档与构建语义检索管线4.1 索引文档写入带嵌入的文档from haystack import Document from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) 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, )write_documents接收DuplicatePolicy控制重复文档处理方式如OVERWRITE覆盖、SKIP跳过这在索引侧保证幂等性。4.2 语义检索Embedding RetrieverSupabasePgvectorEmbeddingRetriever继承自PgvectorEmbeddingRetriever用于按查询向量的相似度召回文档。完整初始化签名__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实例filtersNone应用于检索结果的元数据过滤条件top_k10最多返回的文档数vector_functionNone检索用相似度函数默认沿用document_store上设置的函数filter_policyFilterPolicy.REPLACE运行时过滤器与初始化过滤器的合并策略关于vector_function需要特别注意cosine_similarity与inner_product属于相似度函数分数越高越相似l2_distance返回向量间的直线距离分数越小越相似重要若文档存储使用hnsw检索策略检索时使用的向量函数必须与建索引时一致才能充分利用 HNSW 索引。filter_policy的可选值在 Haystack 的 FilterPolicy 枚举 中定义REPLACE表示运行时过滤器替换初始化过滤器MERGE表示两者合并重叠键由运行时过滤器覆盖此外还有KEEP等取值。在 Pipeline 中使用from haystack import Document, Pipeline from haystack.document_stores.types.policy import DuplicatePolicy 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.管线中检索器通常位于Text Embedder 之后、PromptBuilder/阅读器之前先由嵌入器把查询文本编码为query_embedding再通过retriever.query_embedding输入口传入这是检索器唯一的必填运行参数。输出为documents列表。单独使用调试/离线场景也可以直接传入手工构造的向量retriever SupabasePgvectorEmbeddingRetriever(document_storedocument_store) retriever.run(query_embedding[0.1] * 768) # 与 embedding_dimension 等长的向量4.3 报错约定SupabasePgvectorEmbeddingRetriever.__init__在以下情况抛出ValueErrordocument_store不是SupabasePgvectorDocumentStore实例vector_function不属于cosine_similarity/inner_product/l2_distance三个合法取值。两个检索器同样提供to_dict()/from_dict()序列化接口可随 Pipeline 一起导出复现。五、关键词全文检索SupabasePgvectorKeywordRetrieverSupabasePgvectorKeywordRetriever继承自PgvectorKeywordRetriever不依赖任何嵌入向量直接以关键词匹配文档适合做零成本的基础检索。5.1 排序原理组件使用 PostgreSQL 全文检索的ts_rank_cd函数对文档排序。该函数的排序依据是查询词在文档中出现的频率查询词在文档中彼此相距的远近越近得分越高词出现在文档的哪个部分不同位置权重不同。5.2 初始化签名与参数__init__( *, document_store: SupabasePgvectorDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数默认值说明document_store必填SupabasePgvectorDocumentStore实例filtersNone检索结果的元数据过滤条件top_k10最多返回的文档数filter_policyFilterPolicy.REPLACE过滤器合并策略ValueError触发条件document_store不是SupabasePgvectorDocumentStore实例。5.3 单独使用from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore from haystack_integrations.components.retrievers.supabase import ( SupabasePgvectorKeywordRetriever, ) document_store SupabasePgvectorDocumentStore() retriever SupabasePgvectorKeywordRetriever(document_storedocument_store) retriever.run(querymy nice query)5.4 关键词检索 索引 结果验证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.注意关键词检索默认不做模糊匹配与ElasticsearchBM25Retriever不同查询词拼写稍有偏差就可能返回零结果因此需要精心构造查询。解析查询与文档所用的语言由文档存储的language参数控制默认english中文检索时建议确认 Supabase 侧 PostgreSQL 支持的语言配置。5.5 完整关键词 RAG 管线from haystack import Document, Pipeline from haystack.components.builders.answer_builder import AnswerBuilder from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore from haystack_integrations.components.retrievers.supabase import ( SupabasePgvectorKeywordRetriever, ) prompt_template [ ChatMessage.from_user( Given these documents, answer the question.\nDocuments:\n {% for doc in documents %}{{ doc.content }}{% endfor %}\n Question: {{question}}\nAnswer:, ), ] document_store SupabasePgvectorDocumentStore( languageenglish, 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 a high level of self-awareness, such as recognizing themselves in mirrors.), Document(contentIn certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.), ] document_store.write_documents(documentsdocuments, policyDuplicatePolicy.SKIP) retriever SupabasePgvectorKeywordRetriever(document_storedocument_store) rag_pipeline Pipeline() rag_pipeline.add_component(nameretriever, instanceretriever) rag_pipeline.add_component( instanceChatPromptBuilder(templateprompt_template, required_variables{question, documents}), nameprompt_builder, ) rag_pipeline.add_component(instanceOpenAIChatGenerator(), namellm) rag_pipeline.add_component(instanceAnswerBuilder(), nameanswer_builder) rag_pipeline.connect(retriever, prompt_builder.documents) rag_pipeline.connect(prompt_builder.prompt, llm.messages) rag_pipeline.connect(llm.replies, answer_builder.replies) rag_pipeline.connect(retriever, answer_builder.documents) question languages spoken around the world today result rag_pipeline.run( { retriever: {query: question}, prompt_builder: {question: question}, answer_builder: {query: question}, }, ) print(result[answer_builder])运行该管线需要提前设置两个环境变量OPENAI_API_KEYOpenAI API 密钥与SUPABASE_DB_URLSupabase 连接串。六、完整的向量 RAG 管线端到端示例将嵌入检索与生成环节串联起来构成一条可运行的问答管线from haystack import Document, Pipeline from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack.components.builders import ChatPromptBuilder from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage 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 a high level of self-awareness.), Document(contentIn certain places, you can witness the phenomenon of bioluminescent waves.), ] embedder SentenceTransformersDocumentEmbedder() documents_with_embeddings embedder.run(documents) document_store.write_documents( documents_with_embeddings[documents], policyDuplicatePolicy.OVERWRITE, ) prompt_template [ ChatMessage.from_system(Answer the question based on the provided context.), ChatMessage.from_user( Query: {{query}}\nDocuments:\n{% for doc in documents %}{{ doc.content }}\n{% endfor %}\nAnswer:, ), ] query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder()) query_pipeline.add_component( retriever, SupabasePgvectorEmbeddingRetriever(document_storedocument_store), ) query_pipeline.add_component( prompt_builder, ChatPromptBuilder(templateprompt_template, required_variables[query, documents]), ) query_pipeline.add_component(generator, OpenAIChatGenerator(modelgpt-4o)) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) query_pipeline.connect(retriever.documents, prompt_builder.documents) query_pipeline.connect(prompt_builder.prompt, generator.messages) result query_pipeline.run( { text_embedder: {text: How many languages are there?}, prompt_builder: {query: How many languages are there?}, }, )该示例体现了三条核心连接text_embedder.embedding → retriever.query_embedding查询编码、retriever.documents → prompt_builder.documents召回结果注入提示词、prompt_builder.prompt → generator.messages生成答案。七、SupabaseBucketDownloader从 Supabase Storage 下载文件7.1 用途与数据流SupabaseBucketDownloader用于从 Supabase Storage 存储桶下载文件并将文件在内存中转换为ByteStream对象返回随后可直接交给DocumentConverter等组件进入索引管线。它不落盘适合流式/内存式处理。7.2 初始化签名与参数__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_url必填Supabase 项目 URL形如https://project-ref.supabase.cosupabase_keySecret.from_env_var(SUPABASE_SERVICE_KEY)认证用 Supabase API Key私有存储桶需使用service role key服务角色密钥bucket_name必填要下载文件的存储桶名称file_extensionsNone可选扩展名过滤列表如[.pdf, .txt]None表示下载全部文件扩展名大小写不敏感匹配7.3 使用方法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]7.4 run() 参数与返回值run(sources: list[str]) - dict[str, list[ByteStream]]sources存储桶内的文件路径列表如[folder/file.pdf, notes.txt]返回值字典键streams对应ByteStream列表每个成功下载的文件对应一个ByteStream且该对象的meta中携带file_path与bucket_name两个字段。ByteStream是 Haystack 表示二进制数据的核心数据类型定义于 haystack/dataclasses/byte_stream.py支持to_file()落盘、from_file_path()从文件读取、from_string()从字符串构造并自动携带meta与mime_type字段。利用meta[file_path]可以追溯每个流对应的原始文件便于下游TextFileToDocument、PDFToDocument等转换器记录来源。7.5 warm_up() 与生命周期warm_up() - Nonewarm_up()负责初始化 Supabase 客户端会在首次调用run()时自动执行也可以在 Pipeline 中显式提前调用。在 Pipeline 场景中Haystack 会在组件执行前统一调用各组件warm_up()Haystack 组件生命周期约定warm_up用于加载模型/初始化客户端等耗时操作run只做实际业务处理从而避免首次查询的冷启动延迟。7.6 组合示例下载 → 转换 → 写入文档存储把下载器接入索引管线即可实现存储桶文件 → 文本 → 文档的完整链路from haystack import Pipeline from haystack.components.converters import TextFileToDocument from haystack.document_stores.types.policy import DuplicatePolicy from haystack_integrations.components.downloaders.supabase import SupabaseBucketDownloader from haystack_integrations.document_stores.supabase import SupabasePgvectorDocumentStore downloader SupabaseBucketDownloader( supabase_urlhttps://project-ref.supabase.co, bucket_namemy-documents, ) converter TextFileToDocument() document_store SupabasePgvectorDocumentStore(embedding_dimension768) indexing Pipeline() indexing.add_component(downloader, downloader) indexing.add_component(converter, converter) indexing.add_component(store, document_store) indexing.connect(downloader.streams, converter.sources) indexing.connect(converter.documents, store.documents) indexing.run({downloader: {sources: [notes.txt, readme.md]}})八、搜索策略与向量函数选型建议结合上文参数选型时可以参考以下决策路径数据规模小数据集万级以内用默认的exact_nearest_neighbor即可获得精确结果大数据集建议search_strategyhnsw换取检索速度。向量函数一致性若启用 HNSWdocument_store初始化时的vector_function与检索时的vector_function必须一致否则索引无法生效见 4.2 节重要提示。embedding_dimension 对齐embedding_dimension必须与嵌入模型输出维度一致本文示例用 768 维对应sentence-transformers常用模型否则写入/检索时报维度错误。存储与成本向量规模大时可选用vector_typehalfvecfloat16降低存储开销。HNSW 调优通过hnsw_index_creation_kwargs如m、ef_construction、hnsw_ef_search查询时的候选集规模越大召回越准、越慢与hnsw_index_name控制索引行为。九、总结supabase-haystack集成为 Haystack 用户提供了托管 PostgreSQL 向量检索 关键词检索 对象存储的一站式后端能力SupabasePgvectorDocumentStore让 Supabase 数据库直接成为 Haystack 文档存储默认读取SUPABASE_DB_URL并针对 Supabase 预装的 pgvector 优化了默认配置SupabasePgvectorEmbeddingRetriever负责稠密向量语义检索SupabasePgvectorKeywordRetriever负责基于ts_rank_cd的关键词全文检索两者均可独立使用或嵌入 RAG PipelineSupabaseBucketDownloader支持从 Supabase Storage 内存化下载文件为ByteStream直接衔接文档转换与索引环节所有组件均实现to_dict()/from_dict()可随 Pipeline 序列化导出与复现。在此基础上还可以参考 SupabaseDocumentStore 官方指南 了解SupabaseGroongaDocumentStorePGroonga 多语言全文检索无需嵌入向量及其SupabaseGroongaBM25Retriever实现 pgvector 与 PGroonga 混合检索的进阶方案。【免费下载链接】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），仅供参考