基于 Google Cloud AlloyDB 与 pgvector 构建 Haystack 向量检索与关键词检索系统

基于 Google Cloud AlloyDB 与 pgvector 构建 Haystack 向量检索与关键词检索系统

【免费下载链接】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

AlloyDB 是 Google Cloud 提供的全托管、兼容 PostgreSQL 的数据库服务，本篇文章围绕 Haystack 生态中的alloydb-haystack集成展开，系统讲解AlloyDBDocumentStore、AlloyDBEmbeddingRetriever与AlloyDBKeywordRetriever的安装、初始化、参数配置、检索策略与元数据过滤能力。读完本文，你将能够把 AlloyDB + pgvector 作为 Haystack 的向量数据库后端，搭建语义检索、关键词检索与 RAG 问答管线。

本文以官方 API 参考文档 integrations-api/alloydb.md 为主体骨架，并补充使用指南与核心源码佐证。

一、集成概览：AlloyDB 在 Haystack 中的定位

AlloyDBDocumentStore是一个基于 Google Cloud AlloyDB 的 Document Store 实现，它利用pgvector 扩展执行向量相似度搜索。AlloyDB 本身是 Google Cloud 上全托管的 PostgreSQL 兼容数据库服务，因此该集成天然继承了 PostgreSQL 成熟的关系查询、全文检索与 JSONB 元数据处理能力。

连接层面，集成通过AlloyDB Python Connector安全地建立连接，提供 TLS 加密与基于 IAM 的授权，无需手工管理 SSL 证书、防火墙规则或 IP 白名单。连接采用惰性建立机制——首次使用时才真正与数据库建立连接。

从 Haystack 核心架构看，AlloyDBDocumentStore需要满足核心仓库 protocol.py 中定义的DocumentStore协议，协议要求实现to_dict/from_dict（序列化）、count_documents、filter_documents、write_documents等方法；同时它支持三种检索能力：

• 嵌入向量检索（AlloyDBEmbeddingRetriever）
• 关键词检索（AlloyDBKeywordRetriever，基于 PostgreSQL 全文检索）
• 元数据过滤（过滤语法遵循 Haystack 标准）

集成对应的 Python 包名为alloydb-haystack，API 参考见 alloydb.md。

二、安装与环境准备

安装集成包：

pip install alloydb-haystack

要创建 AlloyDB 集群与实例，请遵循 AlloyDB 官方快速入门指南。文中示例还会用到 Sentence Transformers 嵌入器（来自sentence-transformers-haystack包）：

pip install sentence-transformers-haystack

认证与连接信息

AlloyDBDocumentStore使用 Secrets 机制，默认从环境变量读取连接信息：

• ALLOYDB_INSTANCE_URI：AlloyDB 实例 URI，格式为projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCE；
• ALLOYDB_USER：数据库用户；使用 IAM 数据库认证时，需使用服务账号邮箱（去掉.gserviceaccount.com后缀）或完整 IAM 用户邮箱；
• ALLOYDB_PASSWORD：数据库密码；当enable_iam_auth=True时无需提供。

export ALLOYDB_INSTANCE_URI="projects/MY_PROJECT/locations/MY_REGION/clusters/MY_CLUSTER/instances/MY_INSTANCE" export ALLOYDB_USER="my-db-user" export ALLOYDB_PASSWORD="my-db-password"

若改用 IAM 认证，设置enable_iam_auth=True并授予 IAM 主体 AlloyDB Client 角色即可（详见 AlloyDB 的 IAM 认证文档）。用户名中带.gserviceaccount.com前缀的细节遵循上文规则。

三、AlloyDBDocumentStore 深度解析

3.1 快速上手

from haystack import Document from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore document_store = AlloyDBDocumentStore( db="my-database", embedding_dimension=768, vector_function="cosine_similarity", recreate_table=True, ) document_store.write_documents( [ Document(content="This is first", embedding=[0.1] * 768), Document(content="This is second", embedding=[0.3] * 768), ], ) print(document_store.count_documents())

关键行为说明：

• 连接在首次使用时惰性建立；
• 若用于存储 Haystack 文档的表不存在，会自动创建；
• recreate_table=True表示表已存在时重建，适合开发调试阶段。

3.2 构造参数全解

构造签名（__init__）如下，默认值均来自 API 参考：

__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", strict=False), 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

各参数含义与使用要点：

表结构、schema 与表名由schema_name/table_name决定；调用delete_table()可删除该表，删除范围即由这两个参数限定。

3.3 向量函数（vector_function）

vector_function支持三种取值，含义与评分方向密切相关：

• "cosine_similarity"：余弦相似度，得分越高表示文档与查询越相似；
• "inner_product"：内积，同样是相似度函数，得分越高越相似；
• "l2_distance"：L2 距离，返回向量间的直线距离，得分越小越相似。

重要注意：当使用"hnsw"检索策略时，创建的索引与所选的vector_function强绑定，后续查询必须继续使用相同的向量函数，才能充分利用索引加速；否则索引无法被命中。

3.4 检索策略（search_strategy）

• "exact_nearest_neighbor"（默认）：精确最近邻，召回完美，但在文档数量很大时可能较慢；
• "hnsw"：近似最近邻，以少量精度换取速度，适合大规模文档场景。

使用"hnsw"时，可通过以下参数调优：

• hnsw_index_creation_kwargs：传给索引创建的参数，合法键为m与ef_construction（细节见 pgvector 文档）；
• hnsw_ef_search：查询时控制搜索范围/质量的参数；
• hnsw_recreate_index_if_exists：索引已存在时是否重建；
• hnsw_index_name：索引名称。

3.5 元数据过滤

AlloyDBDocumentStore对标准 Haystack 过滤语法（详见 metadata-filtering.mdx）支持如下：

• 比较运算符：==、!=、>、>=、<、<=、in、not in、like、not like；
• 逻辑运算符：AND、OR。

其中like/not like是 PostgreSQL 对标准 Haystack 过滤语法的扩展，映射到 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再用反转后的比较运算表达。

3.6 文档读写与元数据能力

文档存储相关方法：

DuplicatePolicy是核心仓库定义的枚举，见 policy.py，取值为NONE、SKIP、OVERWRITE、FAIL。

元数据内省能力（因为元数据存储在 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"collation 返回字典序 min/max；字段无值或存储为空时返回{"min": None, "max": None}；
• get_metadata_field_unique_values(metadata_field, search_term=None, from_=0, size=10, filters=None) -> tuple[list[Any], int]：分页返回字段的唯一值列表与唯一值总数；search_term按不区分大小写的子串对字段值做过滤，from_/size控制分页，filters收窄考察范围。

3.7 序列化

与 Haystack 所有组件一致，AlloyDBDocumentStore实现了to_dict()（序列化为字典）与from_dict(data)（从字典反序列化），便于将配置写入 YAML/JSON 并在管线中复用。

四、AlloyDBEmbeddingRetriever：向量语义检索

4.1 定位与核心参数

AlloyDBEmbeddingRetriever是嵌入向量检索器，通过比较查询向量与文档向量的相似度，从AlloyDBDocumentStore中取回最相关的文档，必须连接到AlloyDBDocumentStore。

构造签名：

__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实例，非该类型时抛ValueError；
• filters：应用于检索结果的过滤条件；
• top_k：最多返回的文档数（默认 10）；
• vector_function：检索时使用的相似度函数，覆盖Document Store 中设置的值；不指定时使用AlloyDBDocumentStore的vector_function；
• filter_policy：查询时过滤策略，FilterPolicy.REPLACE（默认）用运行时过滤条件替换初始化时的过滤条件，FilterPolicy.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：覆盖初始化时设置的top_k；
• vector_function：覆盖初始化时设置的相似度函数；
• 返回：{"documents": [...]}字典。

4.2 FilterPolicy 的源码行为

FilterPolicy定义于核心仓库 filter_policy.py，取值REPLACE与MERGE。apply_filter_policy（见同文件 filter_policy.py）实现了合并逻辑：比较过滤与逻辑过滤两两组合（combine_two_comparison_filters、combine_init_comparison_and_runtime_logical_filters、combine_runtime_comparison_and_init_logical_filters、combine_two_logical_filters），同字段时运行时值覆盖初始值；其余情况直接采用运行时过滤条件或初始过滤条件。

4.3 独立使用示例

需要先设置ALLOYDB_INSTANCE_URI、ALLOYDB_USER、ALLOYDB_PASSWORD环境变量，并确保文档已写入 Document Store：

from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import ( AlloyDBEmbeddingRetriever, ) document_store = AlloyDBDocumentStore() retriever = AlloyDBEmbeddingRetriever(document_store=document_store) # 使用假向量保持示例简单 retriever.run(query_embedding=[0.1] * 768)

4.4 在语义检索 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_dimension=768, vector_function="cosine_similarity", recreate_table=True, ) documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document( content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.", ), Document( content="In 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"), policy=DuplicatePolicy.OVERWRITE, ) query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder()) query_pipeline.add_component( "retriever", AlloyDBEmbeddingRetriever(document_store=document_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])

要点：索引侧用SentenceTransformersDocumentEmbedder生成文档向量（详见 sentencetransformersdocumentembedder.mdx），查询侧用SentenceTransformersTextEmbedder生成查询向量；AlloyDBEmbeddingRetriever位于 Text Embedder 之后、PromptBuilder之前（RAG 场景），或作为语义搜索管线的最后一个组件，也可放在TransformersExtractiveReader之前构建抽取式问答（详见 transformersextractivereader.mdx）。

五、AlloyDBKeywordRetriever：PostgreSQL 全文关键词检索

5.1 工作原理

AlloyDBKeywordRetriever基于 PostgreSQL 全文检索（to_tsvector/plainto_tsquery）查找文档，并用ts_rank_cd排序。排序综合考虑：查询词在文档中出现的频率、词项之间的接近程度，以及出现位置在文档中的重要性权重。其实现位于haystack_integrations.components.retrievers.alloydb.keyword_retriever。

需要注意，与ElasticsearchBM25Retriever等组件不同，它默认不做模糊搜索，因此查询措辞需要精心组织，否则可能得到零结果。

关键词检索所用语言由AlloyDBDocumentStore的language参数决定（默认"english"）。可在数据库中查看支持的语言列表：

SELECT cfgname FROM pg_ts_config;

5.2 构造与运行

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

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": [...]}。

独立使用：

from haystack_integrations.document_stores.alloydb import AlloyDBDocumentStore from haystack_integrations.components.retrievers.alloydb import ( AlloyDBKeywordRetriever, ) document_store = AlloyDBDocumentStore() retriever = AlloyDBKeywordRetriever(document_store=document_store) retriever.run(query="my nice query")

5.3 在 RAG Pipeline 中结合 LLM 使用

前置条件：设置OPENAI_API_KEY，以及ALLOYDB_INSTANCE_URI、ALLOYDB_USER、ALLOYDB_PASSWORD环境变量。

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, ) # 创建 RAG 查询管线 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( language="english", # 该参数影响关键词检索的文本解析 recreate_table=True, ) documents = [ Document(content="There are over 7,000 languages spoken around the world today."), Document( content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.", ), Document( content="In 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(documents=documents, policy=DuplicatePolicy.SKIP) retriever = AlloyDBKeywordRetriever(document_store=document_store) rag_pipeline = Pipeline() rag_pipeline.add_component(name="retriever", instance=retriever) rag_pipeline.add_component( instance=ChatPromptBuilder( template=prompt_template, required_variables={"question", "documents"}, ), name="prompt_builder", ) rag_pipeline.add_component(instance=OpenAIChatGenerator(), name="llm") rag_pipeline.add_component(instance=AnswerBuilder(), name="answer_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 核心机制的契合点

• DocumentStore 协议：AlloyDBDocumentStore遵循核心仓库 protocol.py 定义的DocumentStore协议，因此可无缝接入 Haystack Pipeline，配合 Retriever、Writer 等组件工作；
• DuplicatePolicy：写入策略来自核心仓库 policy.py 的DuplicatePolicy枚举（NONE/SKIP/OVERWRITE/FAIL）；
• FilterPolicy：检索器的过滤合并策略复用核心仓库 filter_policy.py 的FilterPolicy，与 InMemory、Elasticsearch 等其他检索器行为保持一致；
• Secret 管理：连接凭据通过 Haystack 的 Secret 机制从环境变量注入（见 secret-management.mdx）。

七、常见问题与使用建议

1. HNSW 索引失效：使用"hnsw"策略时，查询端（Retriever 的vector_function或 Document Store 的vector_function）必须与建索引时保持一致，否则无法利用索引。
2. NOT操作符不可用：AlloyDB 过滤不支持NOT逻辑操作符，务必使用!=、not in等否定比较运算符，或用德摩根定律改写嵌套否定。
3. 关键词检索无结果：该检索器默认无模糊匹配，需谨慎组织查询词；必要时通过language参数选择合适的全文检索语言配置。
4. 扩展权限：create_extension=True时自动创建 pgvector 扩展可能需要超级用户权限；若权限不足，可预先在数据库手工安装扩展并将create_extension设为False。
5. 维度一致性：embedding_dimension必须与实际嵌入模型的输出维度一致（如示例中的 768），否则写入与检索会因向量维度不匹配而失败。

八、延伸阅读

• 完整 API 参考：integrations-api/alloydb.md
• 文档存储使用指南：alloydbdocumentstore.mdx
• 嵌入检索器指南：alloydbembeddingretriever.mdx
• 关键词检索器指南：alloydbkeywordretriever.mdx
• 元数据过滤语法：metadata-filtering.mdx
• Prompt 构建：promptbuilder.mdx

【免费下载链接】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