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/haystack

Google 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_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"

其中实例 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", 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

各参数的完整语义如下表：

参数	默认值	说明
`instance_uri`	`ALLOYDB_INSTANCE_URI`环境变量	AlloyDB 实例 URI，格式为`projects/PROJECT/locations/REGION/clusters/CLUSTER/instances/INSTANCE`
`user`	`ALLOYDB_USER`环境变量	数据库用户。使用 IAM 数据库认证时，填写去掉`.gserviceaccount.com`后缀的服务账号邮箱或完整 IAM 用户邮箱
`password`	`ALLOYDB_PASSWORD`环境变量	数据库密码；`enable_iam_auth=True`时不需要
`db`	`"postgres"`	要连接的数据库名称
`enable_iam_auth`	`False`	是否用 IAM 数据库认证替代密码认证。为`True`时`password`被忽略，IAM 主体需被授予 AlloyDB Client 角色并创建 IAM 数据库用户
`ip_type`	`"PRIVATE"`	连接使用的 IP 类型：`"PRIVATE"`走私有 VPC IP，`"PUBLIC"`走公网 IP，`"PSC"`走 Private Service Connect
`create_extension`	`True`	是否在 pgvector 扩展缺失时自动创建。创建扩展可能需要超级用户权限；设为`False`时需保证扩展已安装，否则报错
`schema_name`	`"public"`	建表所在的 schema，该 schema 必须已存在
`table_name`	`"haystack_documents"`	存储 Haystack 文档的表名
`language`	`"english"`	关键词检索中解析查询与文档内容所用语言，可通过`SELECT cfgname FROM pg_ts_config;`查询数据库支持的配置名
`embedding_dimension`	`768`	嵌入向量的维度
`vector_function`	`"cosine_similarity"`	向量相似度函数，详见下文
`recreate_table`	`False`	表已存在时是否重建
`search_strategy`	`"exact_nearest_neighbor"`	嵌入检索策略，`"exact_nearest_neighbor"`或`"hnsw"`
`hnsw_recreate_index_if_exists`	`False`	仅`"hnsw"`策略下生效，HNSW 索引已存在时是否重建
`hnsw_index_creation_kwargs`	`None`	仅`"hnsw"`策略下生效，传给 HNSW 索引创建的额外参数，合法键为`m`与`ef_construction`
`hnsw_index_name`	`"haystack_hnsw_index"`	HNSW 索引名称
`hnsw_ef_search`	`None`	仅`"hnsw"`策略下生效，查询时的`ef_search`参数
`keyword_index_name`	`"haystack_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( 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 的文档嵌入器生成，例如SentenceTransformersDocumentEmbedder（见 docs-website/docs/pipeline-components/retrievers/alloydbembeddingretriever.mdx）。

检索策略选择

AlloyDBDocumentStore为嵌入检索提供两种策略：

"exact_nearest_neighbor"（默认）：精确最近邻，召回率完美，但在文档量大时速度偏慢；
"hnsw"：近似最近邻，用少量精度换取速度，推荐用于大规模文档集。

HNSW 索引的构建可通过hnsw_index_creation_kwargs（m、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提供完整的文档生命周期管理 API：

write_documents(documents, policy=DuplicatePolicy.FAIL) -> int

写入文档列表并返回写入数量。policy控制重复文档的处理策略，DuplicatePolicy枚举定义于 haystack/document_stores/types/policy.py。异常约定：

ValueError：documents含非Document对象；
DuplicateDocumentError：文档 id 已存在且策略为DuplicatePolicy.FAIL（或未指定）；
DocumentStoreError：写入因其他原因失败。

filter_documents(filters=None) -> 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_term=None, from_=0, size=10, filters=None) -> 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`实例
`filters`	`None`	应用于检索结果的元数据过滤条件
`top_k`	`10`	最多返回的文档数
`vector_function`	`None`	检索时使用的相似度函数，覆盖文档存储初始化时设定的值；未指定时沿用存储的设定
`filter_policy`	`FilterPolicy.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_store=document_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_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."), Document(content="In 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"), 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") 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实例（类型不符抛ValueError），filters为初始化过滤条件，top_k默认10，filter_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_store=document_store) retriever.run(query="my 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( 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."), Document(content="In certain parts of the world, you can witness 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 组件的标准接口：

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 生态获得向量检索、全文检索、元数据过滤三位一体的检索后端。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考