Haystack 与 Google Cloud AlloyDB 集成实战AlloyDBDocumentStore 与嵌入/关键词检索器完全指南【免费下载链接】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/haystackGoogle Cloud AlloyDB 是 Google Cloud 上完全托管的 PostgreSQL 兼容数据库服务而本仓库Haystack 开源 AI 编排框架通过alloydb-haystack集成包将其纳入 LLM 应用基础设施提供基于 pgvector 的向量相似度检索、基于 PostgreSQL 全文检索的关键词检索以及元数据过滤能力。阅读本文后你将掌握AlloyDBDocumentStore、AlloyDBEmbeddingRetriever、AlloyDBKeywordRetriever三个核心组件的完整 API、配置参数语义、底层实现原理并能在 RAG 与语义搜索流水线中直接落地使用。集成概览AlloyDB 在 Haystack 中的定位AlloyDBDocumentStore是一个由 Google Cloud AlloyDB 支撑的文档存储实现类继承自 Haystack 的DocumentStore基类见 docs-website/reference_versioned_docs/version-2.23/integrations-api/alloydb.md。它使用 pgvector 扩展执行向量搜索支持嵌入检索、关键词检索和元数据过滤三类能力。连接层由 AlloyDB Python Connector 负责它通过 TLS 加密与 IAM 授权保障连接安全无需手动管理 SSL 证书、配置防火墙规则或维护 IP 白名单。从源码结构看该集成位于haystack_integrations.document_stores.alloydb与haystack_integrations.components.retrievers.alloydb命名空间下属于 Haystack 生态的外部集成组件deepset-ai/haystack-core-integrations 中的 alloydb 集成。安装与前置条件pip install alloydb-haystack运行示例前还需准备一个已创建的 AlloyDB 集群与实例参照 AlloyDB quickstart 完成 GCP 侧配置三个连接环境变量ALLOYDB_INSTANCE_URI、ALLOYDB_USER、ALLOYDB_PASSWORD若运行嵌入检索示例还需要sentence-transformers-haystack包提供 Sentence Transformers 嵌入器。环境变量格式如下export ALLOYDB_INSTANCE_URIprojects/MY_PROJECT/locations/MY_REGION/clusters/MY_CLUSTER/instances/MY_INSTANCE export ALLOYDB_USERmy-db-user export ALLOYDB_PASSWORDmy-db-password其中实例 URI 遵循projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCE的规范格式。AlloyDBDocumentStore使用 Haystack 的 Secret 机制读取这些变量详见 docs-website/docs/document-stores/alloydbdocumentstore.mdx。AlloyDBDocumentStore核心文档存储初始化与全部参数语义连接在首次使用时才建立lazy 建立存放 Haystack 文档的表若不存在会被自动创建。构造函数签名如下__init__( *, instance_uri: Secret Secret.from_env_var(ALLOYDB_INSTANCE_URI), user: Secret Secret.from_env_var(ALLOYDB_USER), password: Secret Secret.from_env_var(ALLOYDB_PASSWORD, strictFalse), db: str postgres, enable_iam_auth: bool False, ip_type: Literal[PRIVATE, PUBLIC, PSC] PRIVATE, create_extension: bool True, schema_name: str public, table_name: str haystack_documents, language: str english, embedding_dimension: int 768, 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各参数的完整语义如下表参数默认值说明instance_uriALLOYDB_INSTANCE_URI环境变量AlloyDB 实例 URI格式为projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCEuserALLOYDB_USER环境变量数据库用户。使用 IAM 数据库认证时填写去掉.gserviceaccount.com后缀的服务账号邮箱或完整 IAM 用户邮箱passwordALLOYDB_PASSWORD环境变量数据库密码enable_iam_authTrue时不需要dbpostgres要连接的数据库名称enable_iam_authFalse是否用 IAM 数据库认证替代密码认证。为True时password被忽略IAM 主体需被授予 AlloyDB Client 角色并创建 IAM 数据库用户ip_typePRIVATE连接使用的 IP 类型PRIVATE走私有 VPC IPPUBLIC走公网 IPPSC走 Private Service Connectcreate_extensionTrue是否在 pgvector 扩展缺失时自动创建。创建扩展可能需要超级用户权限设为False时需保证扩展已安装否则报错schema_namepublic建表所在的 schema该 schema 必须已存在table_namehaystack_documents存储 Haystack 文档的表名languageenglish关键词检索中解析查询与文档内容所用语言可通过SELECT cfgname FROM pg_ts_config;查询数据库支持的配置名embedding_dimension768嵌入向量的维度vector_functioncosine_similarity向量相似度函数详见下文recreate_tableFalse表已存在时是否重建search_strategyexact_nearest_neighbor嵌入检索策略exact_nearest_neighbor或hnswhnsw_recreate_index_if_existsFalse仅hnsw策略下生效HNSW 索引已存在时是否重建hnsw_index_creation_kwargsNone仅hnsw策略下生效传给 HNSW 索引创建的额外参数合法键为m与ef_constructionhnsw_index_namehaystack_hnsw_indexHNSW 索引名称hnsw_ef_searchNone仅hnsw策略下生效查询时的ef_search参数keyword_index_namehaystack_keyword_index关键词 GIN 索引名称vector_function三种取值的得分语义差异很大务必区分cosine_similarity与inner_product是相似度函数得分越高越相似l2_distance返回向量间的直线距离得分越小越相似。一个关键约束使用hnsw检索策略时HNSW 索引依赖创建时传入的vector_function后续查询必须持续使用相同的向量相似度函数才能命中索引、获得加速效果。最小可用示例from haystack import Document from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore document_store AlloyDBDocumentStore( dbmy-database, embedding_dimension768, vector_functioncosine_similarity, recreate_tableTrue, ) document_store.write_documents( [ Document(contentThis is first, embedding[0.1] * 768), Document(contentThis is second, embedding[0.3] * 768), ], ) print(document_store.count_documents())文档的嵌入向量可以由 Haystack 的文档嵌入器生成例如SentenceTransformersDocumentEmbedder见 docs-website/docs/pipeline-components/retrievers/alloydbembeddingretriever.mdx。检索策略选择AlloyDBDocumentStore为嵌入检索提供两种策略exact_nearest_neighbor默认精确最近邻召回率完美但在文档量大时速度偏慢hnsw近似最近邻用少量精度换取速度推荐用于大规模文档集。HNSW 索引的构建可通过hnsw_index_creation_kwargsm、ef_construction调节查询时可另设hnsw_ef_search平衡召回与延迟。索引创建与查询使用的向量函数必须一致。元数据过滤能力与已知限制AlloyDBDocumentStore完整支持比较运算符、!、、、、、in、not in、like、not like以及逻辑运算符AND、OR。其中like/not like是 AlloyDB 对标准 Haystack 过滤语法的 PostgreSQL 特有扩展映射为 SQL 的LIKE/NOT LIKE模式匹配。已知限制NOT逻辑运算符不受支持。由于每个比较运算符都有对应的否定形式/!、in/not in、like/not like对单个条件的NOT都可以通过反转比较运算符来表达对嵌套AND/OR组的否定可依据德摩根定律改写例如NOT (A AND B)等价于(NOT A) OR (NOT B)其中每个NOT A/NOT B用反转后的比较运算符表达。文档写入、删除与批量管理方法AlloyDBDocumentStore提供完整的文档生命周期管理 APIwrite_documents(documents, policyDuplicatePolicy.FAIL) - int写入文档列表并返回写入数量。policy控制重复文档的处理策略DuplicatePolicy枚举定义于 haystack/document_stores/types/policy.py。异常约定ValueErrordocuments含非Document对象DuplicateDocumentError文档 id 已存在且策略为DuplicatePolicy.FAIL或未指定DocumentStoreError写入因其他原因失败。filter_documents(filtersNone) - list[Document]按过滤条件返回匹配文档。操作符支持范围同上文元数据过滤filters非字典抛TypeError语法非法抛ValueError。delete_documents(document_ids) / delete_all_documents()按 id 列表删除文档 / 清空全部文档。delete_by_filter(filters) - int删除匹配过滤条件的文档并返回删除数量。update_by_filter(filters, meta) - int批量更新匹配文档的元数据字段返回更新数量。count_documents() - int返回文档总数。count_documents_by_filter(filters) - int返回匹配过滤条件的文档数量。count_unique_metadata_by_filter(filters, metadata_fields) - dict[str, int]统计指定元数据字段的唯一值数量字段名可带或不带meta.前缀。元数据内省类型推断与取值探查由于元数据存储在 JSONB 字段中该存储提供了一组分析真实数据来推断结构的方法get_metadata_fields_info() - dict[str, dict[str, str]]分析实际数据推断各元数据字段类型返回形如{ category: {type: text}, priority: {type: integer}, }get_metadata_field_min_max(field) - dict[str, Any]返回某元数据字段的最小/最大值数值字段integer、real返回数值 min/max文本等非数值字段使用C排序规则返回字典序 min/max。字段为空或存储为空时返回{min: None, max: None}。get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]返回某字段的唯一取值列表及其总数支持search_term对字段取值做大小写不敏感的子串匹配过滤from_/size基于 0 的偏移量与返回条数用于分页filters先按过滤条件圈定考虑范围内的文档。生命周期与序列化方法与 Haystack 组件体系保持一致该存储实现了标准的序列化与资源管理接口to_dict() - dict[str, Any]将组件序列化为字典便于写入 YAML 等管线配置from_dict(data) - AlloyDBDocumentStore从字典反序列化还原组件实例close()释放底层关联的同步资源delete_table()删除用于存储 Haystack 文档的表表名由初始化时的schema_name与table_name决定。FilterPolicy与DuplicatePolicy等策略枚举同样定义于 Haystack 核心的 haystack/document_stores/types/filter_policy.py 与 haystack/document_stores/types/policy.py供存储与检索器共同使用。AlloyDBEmbeddingRetriever基于嵌入相似度的检索器AlloyDBEmbeddingRetriever通过嵌入相似度从AlloyDBDocumentStore检索文档必须与AlloyDBDocumentStore实例连接使用构造时传入的document_store若非该类型会抛出ValueError。初始化参数__init__( *, document_store: AlloyDBDocumentStore, 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必填AlloyDBDocumentStore实例filtersNone应用于检索结果的元数据过滤条件top_k10最多返回的文档数vector_functionNone检索时使用的相似度函数覆盖文档存储初始化时设定的值未指定时沿用存储的设定filter_policyFilterPolicy.REPLACE运行期过滤条件与初始化过滤条件的组合策略REPLACE用运行期条件替换初始化条件MERGE将两者合并run 方法run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, vector_function: Literal[cosine_similarity, inner_product, l2_distance] | None None, ) - dict[str, list[Document]]query_embedding必填查询的向量表示filters运行期过滤条件与初始化条件的组合方式由filter_policy决定top_k覆盖初始化时设定的返回上限vector_function覆盖初始化时设定的相似度函数。返回值为包含documents键的字典值为检索到的Document列表。注意vector_function语义cosine_similarity/inner_product得分越高越相似l2_distance得分越小越相似使用hnsw检索策略时必须与建索引时的向量函数保持一致。独立使用from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBEmbeddingRetriever document_store AlloyDBDocumentStore() retriever AlloyDBEmbeddingRetriever(document_storedocument_store) # 用假向量简化示例 retriever.run(query_embedding[0.1] * 768)在流水线中使用嵌入检索在管线中的典型位置是RAG 管线中位于 Text Embedder 之后、PromptBuilder 之前语义搜索管线中作为末组件抽取式 QA 管线中位于 Text Embedder 之后、Extractive Reader 之前。from haystack import Document, Pipeline from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBEmbeddingRetriever document_store AlloyDBDocumentStore( 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 parts of the world, you can witness 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, AlloyDBEmbeddingRetriever(document_storedocument_store)) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) result query_pipeline.run({text_embedder: {text: How many languages are there?}}) print(result[retriever][documents][0])AlloyDBKeywordRetriever基于 PostgreSQL 全文检索的关键词检索器AlloyDBKeywordRetriever通过关键词从AlloyDBDocumentStore检索文档底层使用 PostgreSQL 全文检索的to_tsvector/plainto_tsquery构建查询并用ts_rank_cd排序。排序综合考虑查询词在文档中出现的频率、词项之间的紧凑程度以及出现位置在文档中的重要性。它同样必须连接AlloyDBDocumentStore使用。需要注意与ElasticsearchBM25Retriever等组件不同该检索器默认不提供模糊搜索因此查询措辞需要仔细斟酌否则可能返回零结果。初始化参数__init__( *, document_store: AlloyDBDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数含义与AlloyDBEmbeddingRetriever一致document_store为必填的AlloyDBDocumentStore实例类型不符抛ValueErrorfilters为初始化过滤条件top_k默认10filter_policy默认REPLACE。run 方法run( query: str, filters: dict[str, Any] | None None, top_k: int | None None ) - dict[str, list[Document]]query必填关键词查询字符串filters运行期过滤条件组合方式由filter_policy决定top_k覆盖初始化时的返回上限。返回包含documents键的字典。语言配置解析查询与文档内容所用的语言由AlloyDBDocumentStore的language参数决定默认english。要查看数据库支持的全文检索语言配置执行SELECT cfgname FROM pg_ts_config;独立使用与 RAG 管线独立使用from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBKeywordRetriever document_store AlloyDBDocumentStore() retriever AlloyDBKeywordRetriever(document_storedocument_store) retriever.run(querymy nice query)完整的 RAG 查询管线示例需要OPENAI_API_KEY环境变量from haystack import Document, Pipeline from haystack.components.builders.answer_builder import AnswerBuilder from haystack.components.builders.chat_prompt_builder 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.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBKeywordRetriever prompt_template [ ChatMessage.from_system(You are a helpful assistant.), ChatMessage.from_user( Given these documents, answer the question.\nDocuments:\n {% for doc in documents %}{{ doc.content }}{% endfor %}\n Question: {{question}}\nAnswer: ), ] document_store AlloyDBDocumentStore( 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.), Document(contentIn certain parts of the world, you can witness bioluminescent waves.), ] document_store.write_documents(documentsdocuments, policyDuplicatePolicy.SKIP) retriever AlloyDBKeywordRetriever(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])检索器与存储的序列化与资源管理两个检索器均实现 Haystack 组件的标准接口to_dict() - dict[str, Any]序列化为字典含document_store的序列化表示from_dict(data) - 对应检索器类型从字典还原组件close()释放底层 Document Store 的同步资源。这使得AlloyDBDocumentStore、AlloyDBEmbeddingRetriever、AlloyDBKeywordRetriever可以无缝嵌入 Haystack 的 YAML 管线定义与运行时序列化体系保持与框架内其他组件一致的生命周期管理。结语把 AlloyDB 变成 Haystack 的生产级检索后端从 API 参考与配套指南docs-website/docs/document-stores/alloydbdocumentstore.mdx、docs-website/docs/pipeline-components/retrievers/alloydbembeddingretriever.mdx、docs-website/docs/pipeline-components/retrievers/alloydbkeywordretriever.mdx可以看到这套集成在统一 API 之下同时覆盖了向量检索与全文检索两条路径面向语义相似度场景选择AlloyDBEmbeddingRetriever pgvector按数据规模在精确最近邻与 HNSW 之间权衡并严格保证向量函数在索引构建与查询阶段一致面向精确关键词匹配场景选择AlloyDBKeywordRetriever PostgreSQL 全文检索通过language参数与ts_rank_cd排序获得可控的关键词召回无论哪条路径元数据过滤都遵循同一套操作符语法仅需规避NOT逻辑运算符并善用!/not in/like/not like等价改写。在 GCP 上运行生产级 RAG 或语义搜索服务时这套组件可以在不引入额外中间件的前提下借助 AlloyDB 的托管 PostgreSQL 与 pgvector 生态获得向量检索、全文检索、元数据过滤三位一体的检索后端。【免费下载链接】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 与 Google Cloud AlloyDB 集成实战：AlloyDBDocumentStore 与嵌入/关键词检索器完全指南
Haystack 与 Google Cloud AlloyDB 集成实战AlloyDBDocumentStore 与嵌入/关键词检索器完全指南【免费下载链接】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/haystackGoogle Cloud AlloyDB 是 Google Cloud 上完全托管的 PostgreSQL 兼容数据库服务而本仓库Haystack 开源 AI 编排框架通过alloydb-haystack集成包将其纳入 LLM 应用基础设施提供基于 pgvector 的向量相似度检索、基于 PostgreSQL 全文检索的关键词检索以及元数据过滤能力。阅读本文后你将掌握AlloyDBDocumentStore、AlloyDBEmbeddingRetriever、AlloyDBKeywordRetriever三个核心组件的完整 API、配置参数语义、底层实现原理并能在 RAG 与语义搜索流水线中直接落地使用。集成概览AlloyDB 在 Haystack 中的定位AlloyDBDocumentStore是一个由 Google Cloud AlloyDB 支撑的文档存储实现类继承自 Haystack 的DocumentStore基类见 docs-website/reference_versioned_docs/version-2.23/integrations-api/alloydb.md。它使用 pgvector 扩展执行向量搜索支持嵌入检索、关键词检索和元数据过滤三类能力。连接层由 AlloyDB Python Connector 负责它通过 TLS 加密与 IAM 授权保障连接安全无需手动管理 SSL 证书、配置防火墙规则或维护 IP 白名单。从源码结构看该集成位于haystack_integrations.document_stores.alloydb与haystack_integrations.components.retrievers.alloydb命名空间下属于 Haystack 生态的外部集成组件deepset-ai/haystack-core-integrations 中的 alloydb 集成。安装与前置条件pip install alloydb-haystack运行示例前还需准备一个已创建的 AlloyDB 集群与实例参照 AlloyDB quickstart 完成 GCP 侧配置三个连接环境变量ALLOYDB_INSTANCE_URI、ALLOYDB_USER、ALLOYDB_PASSWORD若运行嵌入检索示例还需要sentence-transformers-haystack包提供 Sentence Transformers 嵌入器。环境变量格式如下export ALLOYDB_INSTANCE_URIprojects/MY_PROJECT/locations/MY_REGION/clusters/MY_CLUSTER/instances/MY_INSTANCE export ALLOYDB_USERmy-db-user export ALLOYDB_PASSWORDmy-db-password其中实例 URI 遵循projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCE的规范格式。AlloyDBDocumentStore使用 Haystack 的 Secret 机制读取这些变量详见 docs-website/docs/document-stores/alloydbdocumentstore.mdx。AlloyDBDocumentStore核心文档存储初始化与全部参数语义连接在首次使用时才建立lazy 建立存放 Haystack 文档的表若不存在会被自动创建。构造函数签名如下__init__( *, instance_uri: Secret Secret.from_env_var(ALLOYDB_INSTANCE_URI), user: Secret Secret.from_env_var(ALLOYDB_USER), password: Secret Secret.from_env_var(ALLOYDB_PASSWORD, strictFalse), db: str postgres, enable_iam_auth: bool False, ip_type: Literal[PRIVATE, PUBLIC, PSC] PRIVATE, create_extension: bool True, schema_name: str public, table_name: str haystack_documents, language: str english, embedding_dimension: int 768, 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各参数的完整语义如下表参数默认值说明instance_uriALLOYDB_INSTANCE_URI环境变量AlloyDB 实例 URI格式为projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCEuserALLOYDB_USER环境变量数据库用户。使用 IAM 数据库认证时填写去掉.gserviceaccount.com后缀的服务账号邮箱或完整 IAM 用户邮箱passwordALLOYDB_PASSWORD环境变量数据库密码enable_iam_authTrue时不需要dbpostgres要连接的数据库名称enable_iam_authFalse是否用 IAM 数据库认证替代密码认证。为True时password被忽略IAM 主体需被授予 AlloyDB Client 角色并创建 IAM 数据库用户ip_typePRIVATE连接使用的 IP 类型PRIVATE走私有 VPC IPPUBLIC走公网 IPPSC走 Private Service Connectcreate_extensionTrue是否在 pgvector 扩展缺失时自动创建。创建扩展可能需要超级用户权限设为False时需保证扩展已安装否则报错schema_namepublic建表所在的 schema该 schema 必须已存在table_namehaystack_documents存储 Haystack 文档的表名languageenglish关键词检索中解析查询与文档内容所用语言可通过SELECT cfgname FROM pg_ts_config;查询数据库支持的配置名embedding_dimension768嵌入向量的维度vector_functioncosine_similarity向量相似度函数详见下文recreate_tableFalse表已存在时是否重建search_strategyexact_nearest_neighbor嵌入检索策略exact_nearest_neighbor或hnswhnsw_recreate_index_if_existsFalse仅hnsw策略下生效HNSW 索引已存在时是否重建hnsw_index_creation_kwargsNone仅hnsw策略下生效传给 HNSW 索引创建的额外参数合法键为m与ef_constructionhnsw_index_namehaystack_hnsw_indexHNSW 索引名称hnsw_ef_searchNone仅hnsw策略下生效查询时的ef_search参数keyword_index_namehaystack_keyword_index关键词 GIN 索引名称vector_function三种取值的得分语义差异很大务必区分cosine_similarity与inner_product是相似度函数得分越高越相似l2_distance返回向量间的直线距离得分越小越相似。一个关键约束使用hnsw检索策略时HNSW 索引依赖创建时传入的vector_function后续查询必须持续使用相同的向量相似度函数才能命中索引、获得加速效果。最小可用示例from haystack import Document from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore document_store AlloyDBDocumentStore( dbmy-database, embedding_dimension768, vector_functioncosine_similarity, recreate_tableTrue, ) document_store.write_documents( [ Document(contentThis is first, embedding[0.1] * 768), Document(contentThis is second, embedding[0.3] * 768), ], ) print(document_store.count_documents())文档的嵌入向量可以由 Haystack 的文档嵌入器生成例如SentenceTransformersDocumentEmbedder见 docs-website/docs/pipeline-components/retrievers/alloydbembeddingretriever.mdx。检索策略选择AlloyDBDocumentStore为嵌入检索提供两种策略exact_nearest_neighbor默认精确最近邻召回率完美但在文档量大时速度偏慢hnsw近似最近邻用少量精度换取速度推荐用于大规模文档集。HNSW 索引的构建可通过hnsw_index_creation_kwargsm、ef_construction调节查询时可另设hnsw_ef_search平衡召回与延迟。索引创建与查询使用的向量函数必须一致。元数据过滤能力与已知限制AlloyDBDocumentStore完整支持比较运算符、!、、、、、in、not in、like、not like以及逻辑运算符AND、OR。其中like/not like是 AlloyDB 对标准 Haystack 过滤语法的 PostgreSQL 特有扩展映射为 SQL 的LIKE/NOT LIKE模式匹配。已知限制NOT逻辑运算符不受支持。由于每个比较运算符都有对应的否定形式/!、in/not in、like/not like对单个条件的NOT都可以通过反转比较运算符来表达对嵌套AND/OR组的否定可依据德摩根定律改写例如NOT (A AND B)等价于(NOT A) OR (NOT B)其中每个NOT A/NOT B用反转后的比较运算符表达。文档写入、删除与批量管理方法AlloyDBDocumentStore提供完整的文档生命周期管理 APIwrite_documents(documents, policyDuplicatePolicy.FAIL) - int写入文档列表并返回写入数量。policy控制重复文档的处理策略DuplicatePolicy枚举定义于 haystack/document_stores/types/policy.py。异常约定ValueErrordocuments含非Document对象DuplicateDocumentError文档 id 已存在且策略为DuplicatePolicy.FAIL或未指定DocumentStoreError写入因其他原因失败。filter_documents(filtersNone) - list[Document]按过滤条件返回匹配文档。操作符支持范围同上文元数据过滤filters非字典抛TypeError语法非法抛ValueError。delete_documents(document_ids) / delete_all_documents()按 id 列表删除文档 / 清空全部文档。delete_by_filter(filters) - int删除匹配过滤条件的文档并返回删除数量。update_by_filter(filters, meta) - int批量更新匹配文档的元数据字段返回更新数量。count_documents() - int返回文档总数。count_documents_by_filter(filters) - int返回匹配过滤条件的文档数量。count_unique_metadata_by_filter(filters, metadata_fields) - dict[str, int]统计指定元数据字段的唯一值数量字段名可带或不带meta.前缀。元数据内省类型推断与取值探查由于元数据存储在 JSONB 字段中该存储提供了一组分析真实数据来推断结构的方法get_metadata_fields_info() - dict[str, dict[str, str]]分析实际数据推断各元数据字段类型返回形如{ category: {type: text}, priority: {type: integer}, }get_metadata_field_min_max(field) - dict[str, Any]返回某元数据字段的最小/最大值数值字段integer、real返回数值 min/max文本等非数值字段使用C排序规则返回字典序 min/max。字段为空或存储为空时返回{min: None, max: None}。get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]返回某字段的唯一取值列表及其总数支持search_term对字段取值做大小写不敏感的子串匹配过滤from_/size基于 0 的偏移量与返回条数用于分页filters先按过滤条件圈定考虑范围内的文档。生命周期与序列化方法与 Haystack 组件体系保持一致该存储实现了标准的序列化与资源管理接口to_dict() - dict[str, Any]将组件序列化为字典便于写入 YAML 等管线配置from_dict(data) - AlloyDBDocumentStore从字典反序列化还原组件实例close()释放底层关联的同步资源delete_table()删除用于存储 Haystack 文档的表表名由初始化时的schema_name与table_name决定。FilterPolicy与DuplicatePolicy等策略枚举同样定义于 Haystack 核心的 haystack/document_stores/types/filter_policy.py 与 haystack/document_stores/types/policy.py供存储与检索器共同使用。AlloyDBEmbeddingRetriever基于嵌入相似度的检索器AlloyDBEmbeddingRetriever通过嵌入相似度从AlloyDBDocumentStore检索文档必须与AlloyDBDocumentStore实例连接使用构造时传入的document_store若非该类型会抛出ValueError。初始化参数__init__( *, document_store: AlloyDBDocumentStore, 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必填AlloyDBDocumentStore实例filtersNone应用于检索结果的元数据过滤条件top_k10最多返回的文档数vector_functionNone检索时使用的相似度函数覆盖文档存储初始化时设定的值未指定时沿用存储的设定filter_policyFilterPolicy.REPLACE运行期过滤条件与初始化过滤条件的组合策略REPLACE用运行期条件替换初始化条件MERGE将两者合并run 方法run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, vector_function: Literal[cosine_similarity, inner_product, l2_distance] | None None, ) - dict[str, list[Document]]query_embedding必填查询的向量表示filters运行期过滤条件与初始化条件的组合方式由filter_policy决定top_k覆盖初始化时设定的返回上限vector_function覆盖初始化时设定的相似度函数。返回值为包含documents键的字典值为检索到的Document列表。注意vector_function语义cosine_similarity/inner_product得分越高越相似l2_distance得分越小越相似使用hnsw检索策略时必须与建索引时的向量函数保持一致。独立使用from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBEmbeddingRetriever document_store AlloyDBDocumentStore() retriever AlloyDBEmbeddingRetriever(document_storedocument_store) # 用假向量简化示例 retriever.run(query_embedding[0.1] * 768)在流水线中使用嵌入检索在管线中的典型位置是RAG 管线中位于 Text Embedder 之后、PromptBuilder 之前语义搜索管线中作为末组件抽取式 QA 管线中位于 Text Embedder 之后、Extractive Reader 之前。from haystack import Document, Pipeline from haystack.document_stores.types import DuplicatePolicy from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBEmbeddingRetriever document_store AlloyDBDocumentStore( 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 parts of the world, you can witness 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, AlloyDBEmbeddingRetriever(document_storedocument_store)) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) result query_pipeline.run({text_embedder: {text: How many languages are there?}}) print(result[retriever][documents][0])AlloyDBKeywordRetriever基于 PostgreSQL 全文检索的关键词检索器AlloyDBKeywordRetriever通过关键词从AlloyDBDocumentStore检索文档底层使用 PostgreSQL 全文检索的to_tsvector/plainto_tsquery构建查询并用ts_rank_cd排序。排序综合考虑查询词在文档中出现的频率、词项之间的紧凑程度以及出现位置在文档中的重要性。它同样必须连接AlloyDBDocumentStore使用。需要注意与ElasticsearchBM25Retriever等组件不同该检索器默认不提供模糊搜索因此查询措辞需要仔细斟酌否则可能返回零结果。初始化参数__init__( *, document_store: AlloyDBDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数含义与AlloyDBEmbeddingRetriever一致document_store为必填的AlloyDBDocumentStore实例类型不符抛ValueErrorfilters为初始化过滤条件top_k默认10filter_policy默认REPLACE。run 方法run( query: str, filters: dict[str, Any] | None None, top_k: int | None None ) - dict[str, list[Document]]query必填关键词查询字符串filters运行期过滤条件组合方式由filter_policy决定top_k覆盖初始化时的返回上限。返回包含documents键的字典。语言配置解析查询与文档内容所用的语言由AlloyDBDocumentStore的language参数决定默认english。要查看数据库支持的全文检索语言配置执行SELECT cfgname FROM pg_ts_config;独立使用与 RAG 管线独立使用from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBKeywordRetriever document_store AlloyDBDocumentStore() retriever AlloyDBKeywordRetriever(document_storedocument_store) retriever.run(querymy nice query)完整的 RAG 查询管线示例需要OPENAI_API_KEY环境变量from haystack import Document, Pipeline from haystack.components.builders.answer_builder import AnswerBuilder from haystack.components.builders.chat_prompt_builder 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.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import AlloyDBKeywordRetriever prompt_template [ ChatMessage.from_system(You are a helpful assistant.), ChatMessage.from_user( Given these documents, answer the question.\nDocuments:\n {% for doc in documents %}{{ doc.content }}{% endfor %}\n Question: {{question}}\nAnswer: ), ] document_store AlloyDBDocumentStore( 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.), Document(contentIn certain parts of the world, you can witness bioluminescent waves.), ] document_store.write_documents(documentsdocuments, policyDuplicatePolicy.SKIP) retriever AlloyDBKeywordRetriever(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])检索器与存储的序列化与资源管理两个检索器均实现 Haystack 组件的标准接口to_dict() - dict[str, Any]序列化为字典含document_store的序列化表示from_dict(data) - 对应检索器类型从字典还原组件close()释放底层 Document Store 的同步资源。这使得AlloyDBDocumentStore、AlloyDBEmbeddingRetriever、AlloyDBKeywordRetriever可以无缝嵌入 Haystack 的 YAML 管线定义与运行时序列化体系保持与框架内其他组件一致的生命周期管理。结语把 AlloyDB 变成 Haystack 的生产级检索后端从 API 参考与配套指南docs-website/docs/document-stores/alloydbdocumentstore.mdx、docs-website/docs/pipeline-components/retrievers/alloydbembeddingretriever.mdx、docs-website/docs/pipeline-components/retrievers/alloydbkeywordretriever.mdx可以看到这套集成在统一 API 之下同时覆盖了向量检索与全文检索两条路径面向语义相似度场景选择AlloyDBEmbeddingRetriever pgvector按数据规模在精确最近邻与 HNSW 之间权衡并严格保证向量函数在索引构建与查询阶段一致面向精确关键词匹配场景选择AlloyDBKeywordRetriever PostgreSQL 全文检索通过language参数与ts_rank_cd排序获得可控的关键词召回无论哪条路径元数据过滤都遵循同一套操作符语法仅需规避NOT逻辑运算符并善用!/not in/like/not like等价改写。在 GCP 上运行生产级 RAG 或语义搜索服务时这套组件可以在不引入额外中间件的前提下借助 AlloyDB 的托管 PostgreSQL 与 pgvector 生态获得向量检索、全文检索、元数据过滤三位一体的检索后端。【免费下载链接】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），仅供参考