Haystack AlloyDB 集成完全指南AlloyDBDocumentStore、Embedding/Keyword Retriever 实战与原理【免费下载链接】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 参考文档docs-website/reference_versioned_docs/version-2.20/integrations-api/alloydb.md为骨架系统讲解基于 Google Cloud AlloyDB 的AlloyDBDocumentStore及其配套检索组件AlloyDBEmbeddingRetriever、AlloyDBKeywordRetriever的完整用法。读完本文你将掌握如何安全连接 AlloyDB 实例、如何配置 pgvector 向量检索精确近邻与 HNSW 两种策略、如何用关键词全文检索与元数据过滤以及如何在 Haystack Pipeline 中组合这些组件搭建 RAG 应用。一、集成概览在 Haystack 中使用 Google Cloud AlloyDBalloydb-haystack是 Haystack 生态的官方集成将 Google Cloud AlloyDB——一个完全托管的、PostgreSQL 兼容的云数据库服务——作为 Haystack 的 Document Store 后端。它依赖 pgvector 扩展 完成向量相似度检索支持向量检索Embedding Retrieval基于查询向量与文档向量的相似度召回文档关键词检索Keyword Retrieval基于 PostgreSQL 全文检索to_tsvector/plainto_tsquery召回文档元数据过滤Metadata Filtering通过结构化条件精确收窄检索范围。集成组件分布在haystack_integrations命名空间下参见参考文档中的模块路径haystack_integrations.components.retrievers.alloydb与haystack_integrations.document_stores.alloydb包括组件所属模块核心能力AlloyDBDocumentStorehaystack_integrations.document_stores.alloydb文档存储、写入/删除/过滤、向量与关键词索引管理AlloyDBEmbeddingRetrieverhaystack_integrations.components.retrievers.alloydb按嵌入相似度检索AlloyDBKeywordRetrieverhaystack_integrations.components.retrievers.alloydb按 PostgreSQL 全文检索检索这些组件实现了 Haystack 的DocumentStore协议与检索器约定。核心协议定义在 haystack/document_stores/types/protocol.py 中该协议规定了文档存储需提供的write_documents、filter_documents、count_documents等标准接口因此AlloyDBDocumentStore可以无缝接入 Haystack Pipeline。二、安装与环境准备2.1 安装集成包pip install alloydb-haystack如需运行本文的嵌入向量示例还需安装 Sentence Transformers 集成用于生成文本/文档嵌入pip install sentence-transformers-haystack2.2 前置条件一个已创建好的 AlloyDB 集群与实例。可按 AlloyDB 快速入门 完成环境搭建目标数据库默认连接postgres库需支持安装pgvector扩展create_extensionTrue时集成会自动创建。2.3 连接与认证方式AlloyDBDocumentStore通过 AlloyDB Python Connector 建立连接该连接器提供TLS 加密与IAM 授权能力无需手动管理 SSL 证书、防火墙规则或 IP 白名单参考文档在AlloyDBDocumentStore类说明中明确这一点。默认通过环境变量读取连接信息参考文档中的Usage exampleexport ALLOYDB_INSTANCE_URIprojects/MY_PROJECT/locations/MY_REGION/clusters/MY_CLUSTER/instances/MY_INSTANCE export ALLOYDB_USERmy-db-user export ALLOYDB_PASSWORDmy-db-passwordALLOYDB_INSTANCE_URI实例 URI格式为projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCEALLOYDB_USER数据库用户。若使用 IAM 数据库认证应填服务账号邮箱省略.gserviceaccount.com后缀或完整的 IAM 用户邮箱ALLOYDB_PASSWORD数据库密码enable_iam_authTrue时无需提供。也可以选择IAM 数据库认证设置enable_iam_authTrue授予 IAM 主体 AlloyDB Client 角色并创建对应的 IAM 数据库用户详见 AlloyDB IAM 认证文档。启用后password参数将被忽略。三、AlloyDBDocumentStore核心参数与初始化3.1 最小可用示例连接是惰性建立的——首次使用时才真正连库存储 Haystack 文档的专用表若不存在会自动创建import os 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())3.2 初始化参数全解析AlloyDBDocumentStore.__init__的完整签名来自参考文档__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_uriSecret环境变量ALLOYDB_INSTANCE_URIAlloyDB 实例 URI格式projects/.../locations/.../clusters/.../instances/...userSecret环境变量ALLOYDB_USER数据库用户IAM 认证时使用服务账号/IAM 用户邮箱passwordSecret环境变量ALLOYDB_PASSWORD数据库密码enable_iam_authTrue时不需要dbstrpostgres要连接的数据库名enable_iam_authboolFalse是否使用 IAM 数据库认证为True时忽略passwordip_typeLiteral[PRIVATE,PUBLIC,PSC]PRIVATE连接 IP 类型PRIVATE走私有 VPC IPPUBLIC走公网 IPPSC走 Private Service Connect表结构与扩展类参数类型默认值说明create_extensionboolTrue是否自动创建 pgvector 扩展若不存在。创建扩展可能需要超级用户权限若设为False需保证扩展已安装否则报错schema_namestrpublic建表所用的 schema该 schema 必须已存在table_namestrhaystack_documents存储 Haystack 文档的表名recreate_tableboolFalse若表已存在是否重建会清空原表向量检索类参数类型默认值说明embedding_dimensionint768嵌入向量的维度须与所用嵌入模型输出维度一致vector_functionLiteral[cosine_similarity,inner_product,l2_distance]cosine_similarity相似度函数cosine_similarity与inner_product分数越高越相似l2_distance返回向量间直线距离分数越小越相似search_strategyLiteral[exact_nearest_neighbor,hnsw]exact_nearest_neighbor检索策略精确近邻召回完美但大文档集下较慢hnsw为近似近邻牺牲少量精度换取速度适合大规模文档hnsw_recreate_index_if_existsboolFalseHNSW 索引已存在时是否重建仅search_strategyhnsw时生效hnsw_index_creation_kwargsdict[str, int] \| NoneNoneHNSW 建索引进阶参数合法键为m与ef_construction仅 HNSW 策略生效hnsw_index_namestrhaystack_hnsw_indexHNSW 索引名hnsw_ef_searchint \| NoneNone查询时的ef_search参数仅 HNSW 策略生效关键词检索类参数类型默认值说明languagestrenglish关键词检索解析查询与文档内容所用的全文检索语言keyword_index_namestrhaystack_keyword_index关键词检索使用的 GIN 索引名关键注意事项HNSW 索引与vector_function强绑定当使用hnsw检索策略时索引的构建依赖于初始化时传入的vector_function。后续查询必须持续使用同一相似度函数才能充分利用该索引否则索引失效甚至产生错误结果。可用的全文检索语言可通过在 PostgreSQL 中执行以下 SQL 查询当前库支持的语言配置SELECT cfgname FROM pg_ts_config;3.3 文档写入与重复策略write_documents(documents, policyDuplicatePolicy.FAIL)将文档写入存储。DuplicatePolicy定义在 haystack/document_stores/types/policy.py取值包括NONE默认行为具体取决于存储实现SKIP同 id 文档已存在时跳过不写入OVERWRITE同 id 文档已存在时覆盖FAIL同 id 文档已存在时报DuplicateDocumentError。当policy为FAIL或未指定且同 id 文档已存在时会抛出DuplicateDocumentError文档列表含非Document对象时抛出ValueError其他写入失败场景抛出DocumentStoreError。四、AlloyDBEmbeddingRetriever基于嵌入相似度的检索4.1 组件定位AlloyDBEmbeddingRetriever是一个基于嵌入相似度的检索器必须与AlloyDBDocumentStore搭配使用。在 Pipeline 中的典型位置参考 alloydbembeddingretriever.mdxRAG 流水线中位于 Text Embedder 之后、PromptBuilder 之前语义搜索流水线的最后一个组件抽取式问答中位于 Text Embedder 之后、抽取式 Reader 之前。使用时需保证查询与文档嵌入可用索引流水线中用Document Embedder生成文档嵌入查询流水线中用Text Embedder生成查询嵌入。4.2 初始化参数__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_storeAlloyDBDocumentStore必填绑定的文档存储实例filtersdict[str, Any] \| NoneNone应用于检索结果的元数据过滤条件top_kint10返回的最大文档数vector_functionLiteral[...] \| NoneNone覆盖 Document Store 上的相似度函数未指定时使用AlloyDBDocumentStore中设置的函数filter_policystr \| FilterPolicyFilterPolicy.REPLACE运行时过滤策略REPLACE用运行时过滤条件替换初始化时的过滤条件MERGE将两者合并若document_store不是AlloyDBDocumentStore实例初始化抛出ValueError。4.3 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查询的向量表示list[float]必填filters运行时过滤条件与初始化过滤条件的组合方式由filter_policy决定top_k覆盖初始化时的top_kvector_function覆盖初始化时的相似度函数。返回值{documents: [...]}即从 Document Store 检索到的文档列表。4.4 独立使用示例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)4.5 在 Pipeline 中构建语义搜索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, 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_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) query How many languages are there? result query_pipeline.run({text_embedder: {text: query}}) print(result[retriever][documents][0])五、AlloyDBKeywordRetrieverPostgreSQL 全文检索5.1 检索原理AlloyDBKeywordRetriever是基于关键词的检索器使用 PostgreSQL 全文检索to_tsvector/plainto_tsquery查找文档并用ts_rank_cd排序。排序综合考虑查询词在文档中出现的频率、词与词之间的距离以及出现位置的权重详见 PostgreSQL 文本搜索排序文档。重要差异与ElasticsearchBM25Retriever等组件不同该检索器默认不提供模糊匹配fuzzy search查询词需精确匹配分词结果否则可能得到零结果因此需谨慎构造查询语句。语言配置解析查询与文档内容所用的语言由AlloyDBDocumentStore的language参数决定默认english可通过SELECT cfgname FROM pg_ts_config;查看库内可用的语言配置。5.2 初始化与 run 方法__init__( *, document_store: AlloyDBDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - Nonerun( query: str, filters: dict[str, Any] | None None, top_k: int | None None ) - dict[str, list[Document]]query必填关键词查询字符串filters运行时过滤条件组合方式由filter_policy决定top_k覆盖初始化时的top_k。返回{documents: [...]}。document_store非AlloyDBDocumentStore实例时初始化抛出ValueError。5.3 独立使用示例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)5.4 在 RAG Pipeline 中使用以下示例构建了一个「关键词检索 → 提示构建 → OpenAI 生成 → 答案组装」的 RAG 流水线。运行前提设置OPENAI_API_KEY环境变量以及前述三个ALLOYDB_*环境变量。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, ) ## Create a RAG query pipeline 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, # this parameter influences text parsing for keyword retrieval 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 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])六、元数据过滤与 FilterPolicy 机制6.1 支持的过滤算子AlloyDBDocumentStore.filter_documents及检索器支持比较算子与逻辑算子比较算子、!、、、、、in、not in、like、not like其中like/not like是 AlloyDB 集成对标准 Haystack 过滤语法的 PostgreSQL 扩展映射到 SQL 的LIKE/NOT LIKE模式匹配逻辑算子AND、OR完整支持NOT不支持。由于每个比较算子都有否定对应物/!、in/not in、like/not like任何对单个条件的NOT都可以通过反转比较算子表达。对嵌套的AND/OR组合取反可应用De Morgan 定律NOT (A AND B)等价于(NOT A) OR (NOT B)其中每个NOT A/NOT B再用反转的比较算子表达。Haystack 标准的过滤语法比较字典包含field/operator/value三个键逻辑字典包含operator/conditions定义在 haystack/document_stores/types/protocol.py更完整的过滤规则可参考 metadata-filtering 概念文档。6.2 FilterPolicy初始化过滤与运行时过滤的组合FilterPolicy枚举定义在 haystack/document_stores/types/filter_policy.pyREPLACE默认运行时过滤条件替换初始化时的过滤条件便于对每次查询动态更换过滤逻辑MERGE运行时过滤条件与初始化过滤条件合并进一步收窄搜索范围同一字段冲突时运行时值覆盖初始化值。apply_filter_policyfilter_policy.py实现了合并逻辑两个比较条件用AND组合比较条件与逻辑条件、或两个逻辑条件则按条件形状递归合并同操作符的逻辑条件合并conditions列表不同操作符时以运行时条件为准并记录警告日志。MERGE只在一个非空条件存在时返回该条件REPLACE则直接返回runtime_filters or init_filters。七、文档存储的完整操作接口AlloyDBDocumentStore除读写外还提供以下运维与统计方法方法签名说明count_documents() - int返回存储中的文档总数filter_documents(filtersNone) - list[Document]返回匹配过滤条件的文档filters非字典抛TypeError语法非法抛ValueErrordelete_documents(document_ids: list[str]) - None按文档 id 批量删除delete_all_documents() - None清空所有文档delete_by_filter(filters) - int删除匹配过滤条件的文档返回删除数量update_by_filter(filters, meta) - int更新匹配文档的元数据字段返回更新数量count_documents_by_filter(filters) - int统计匹配过滤条件的文档数count_unique_metadata_by_filter(filters, metadata_fields) - dict[str, int]统计指定元数据字段的唯一值数量字段名可带或不带meta.前缀get_metadata_fields_info() - dict[str, dict[str, str]]返回元数据字段的类型信息因元数据存于 JSONB 字段通过分析实际数据推断类型例如{category: {type: text}, priority: {type: integer}}get_metadata_field_min_max(field) - dict[str, Any]返回某字段的min/max数值字段返回数值极值文本等非数值字段使用C排序规则返回字典序极值字段无值或存储为空时返回{min: None, max: None}get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]分页返回某字段的唯一值可选search_term按大小写不敏感子串过滤、filters收窄范围返回(唯一值列表, 唯一值总数)delete_table() - None删除用于存储 Haystack 文档的表表名由schema_name与table_name决定close() - None释放底层同步资源to_dict/from_dict—序列化 / 反序列化组件便于 Pipeline YAML 与序列化场景使用八、搜索策略深度解读精确近邻与 HNSW8.1 两种策略的取舍exact_nearest_neighbor默认对全部向量做精确扫描召回完美但在文档规模很大时较慢hnsw近似最近邻搜索基于 pgvector 的 HNSW 索引以少量精度换取速度官方推荐用于大规模文档集。8.2 HNSW 使用要点向量函数一致性HNSW 索引在创建时依赖vector_functioncosine_similarity/inner_product/l2_distance后续查询必须使用相同函数否则无法利用索引。索引创建调参通过hnsw_index_creation_kwargs传递m与ef_construction两个参数详见 pgvector 的 HNSW 文档m每个节点的最大连接数影响索引质量与内存占用ef_construction构建索引时的搜索范围越大索引质量越高但构建越慢。查询期调参通过hnsw_ef_search控制查询时的搜索广度——值越大召回越准但查询越慢。索引重建若 HNSW 索引已存在设置hnsw_recreate_index_if_existsTrue可强制重建。关键词索引关键词检索使用 GIN 索引索引名由keyword_index_name指定默认haystack_keyword_index。九、序列化与 Pipeline 集成三个组件均实现to_dict()与from_dict()to_dict() - dict[str, Any]将组件序列化为字典from_dict(data: dict[str, Any])从字典反序列化还原组件AlloyDBDocumentStore返回AlloyDBDocumentStore检索器返回对应检索器实例。这使它们能够无缝接入 Haystack 的 Pipeline 序列化YAML体系便于流水线版本管理与部署。在 Pipeline 中使用时务必用Pipeline.connect()将 Text Embedder 的embedding输出接到检索器的query_embedding输入见 4.5 节或将检索器的documents输出接到PromptBuilder、AnswerBuilder等下游组件。十、官方资料索引AlloyDB 集成 API 参考当前版本 docs-website/reference_versioned_docs/version-2.20/integrations-api/alloydb.md最新版见 docs-website/reference/integrations-api/alloydb.md使用指南AlloyDBDocumentStore见 docs-website/docs/document-stores/alloydbdocumentstore.mdxAlloyDBEmbeddingRetriever见 docs-website/docs/pipeline-components/retrievers/alloydbembeddingretriever.mdxAlloyDBKeywordRetriever见 docs-website/docs/pipeline-components/retrievers/alloydbkeywordretriever.mdx底层协议与策略实现haystack/document_stores/types/protocol.py、haystack/document_stores/types/filter_policy.py、haystack/document_stores/types/policy.py。说明本文基于 Haystack 2.20 版本的 AlloyDB 集成 API 参考撰写所述接口与参数以 docs-website/reference_versioned_docs/version-2.20/integrations-api/alloydb.md 为准。实际使用时请确认所安装的alloydb-haystack包版本与 API 签名一致。【免费下载链接】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），仅供参考