企业官网建设流程全解析

Haystack 集成指南：使用 PgvectorDocumentStore 构建 PostgreSQL 向量检索与 RAG 管线

【免费下载链接】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 生态中的Pgvector 集成（pgvector-haystack包），系统讲解如何在 PostgreSQL + pgvector 扩展之上构建完整的向量检索能力：从环境搭建、PgvectorDocumentStore的初始化参数与文档管理 API，到PgvectorEmbeddingRetriever的语义检索与PgvectorKeywordRetriever的关键词检索，再到过滤器策略的底层实现。读完本文，你将能够把 PostgreSQL 作为 Haystack 的生产级 Document Store，搭建可复制的语义搜索、关键词搜索与 RAG 管线。文章以 version-2.21 的 Pgvector 集成 API 参考 为核心骨架，并辅以仓库内文档站点与核心源码佐证。

Pgvector 集成概览

Pgvector 是 PostgreSQL 的扩展，它在保留 PostgreSQL 经典能力（如 ACID 事务、point-in-time recovery）的同时，引入了向量相似度搜索：既支持精确最近邻（exact nearest neighbor），也支持基于 HNSW 的近似最近邻（approximate nearest neighbor）。因此，把 PostgreSQL 用作 Haystack 的 Document Store，可以在同一套数据库基础设施上同时获得关系查询、事务保证与向量检索能力。

在 Haystack 中，这一能力通过pgvector-haystack集成包提供，核心组件有三个（定义见 API 参考）：

• PgvectorDocumentStore：基于 pgvector 扩展的 Document Store，负责建表、写入、过滤、删除、统计等文档管理操作；
• PgvectorEmbeddingRetriever：基于稠密向量（dense embedding）的检索器，从 Document Store 中取回与查询向量最相似的文档；
• PgvectorKeywordRetriever：基于关键词的检索器，使用 PostgreSQL 全文检索的ts_rank_cd函数对文档排序。

从仓库文档站点的 Document Store 选择指南 可以看到，Pgvector 属于"关系型数据库 + 向量扩展"这一类 Document Store，同时支持 Embedding 检索与关键词（Keyword）检索，适合希望复用既有 PostgreSQL 集群、又要向量能力的场景。

环境准备与安装

使用 Docker 快速启动 PostgreSQL + pgvector

文档站点推荐直接用官方镜像启动一个带 pgvector 的 PostgreSQL 实例（详见 PgvectorDocumentStore 指南）：

docker run -d -p 5432:5432 \ -e POSTGRES_USER=postgres \ -e POSTGRES_PASSWORD=postgres \ -e POSTGRES_DB=postgres \ pgvector/pgvector:pg17

该命令会在本机 5432 端口启动一个 PostgreSQL 17 容器，并预置用户、密码与数据库，均为postgres。

安装 Python 集成包

pip install pgvector-haystack

如果还要运行文档中的嵌入示例（使用 Sentence Transformers 生成向量），需要额外安装：

pip install sentence-transformers-haystack

配置连接字符串

PgvectorDocumentStore通过PG_CONN_STR环境变量读取连接字符串（默认值即Secret.from_env_var("PG_CONN_STR")，见 API 参考）。支持两种格式：

URI 格式：

export PG_CONN_STR="postgresql://USER:PASSWORD@HOST:PORT/DB_NAME"

关键字/值（keyword/value）格式：

export PG_CONN_STR="host=HOST port=PORT dbname=DB_NAME user=USER password=PASSWORD"

需要注意一个常见的坑（文档站点有专门提醒，见 PgvectorDocumentStore 指南）：URI 格式中密码等特殊字符必须做percent-encoding。例如密码为p=ssword，则应写成：

export PG_CONN_STR="postgresql://postgres:p%3Dssword@localhost:5432/postgres"

否则可能触发类似psycopg.OperationalError: [Errno -2] Name or service not known的连接错误。若不想处理转义，可直接改用关键字/值格式：

export PG_CONN_STR="host=localhost port=5432 dbname=postgres user=postgres password=p=ssword"

PgvectorDocumentStore 详解

初始化签名与参数

PgvectorDocumentStore的完整初始化签名（来自 API 参考）：

__init__( *, connection_string: Secret = Secret.from_env_var("PG_CONN_STR"), create_extension: bool = True, 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" ) -> None

各参数含义如下：

初始化与写入文档

文档站点给出了最简初始化与写入示例（见 PgvectorDocumentStore 指南）：

from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore from haystack import Document document_store = PgvectorDocumentStore( embedding_dimension=768, vector_function="cosine_similarity", recreate_table=True, search_strategy="hnsw", ) 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())

要点：

• embedding_dimension=768需与嵌入模型输出维度一致；
• recreate_table=True适合开发调试，会丢弃已有表结构；生产环境请谨慎；
• 写入带embedding字段的Document即可被向量检索命中；不带向量的文档仍可参与关键词检索与元数据过滤。

关于向量维度的上限："vector"类型支持最高约 2000 维（受 PostgreSQL 行大小限制），超过 2000 且不超过 4000 的维度应使用vector_type="halfvec"（半精度存储），这要求 pgvector 0.7.0 及以上。这一点从vector_type参数文档可以直接确认。

向量函数语义

三个可选向量函数中，"cosine_similarity"与"inner_product"是相似度函数，分数越高代表越相似；而"l2_distance"返回向量间的直线距离，分数越小越相似。若使用 HNSW 搜索策略，索引的构建依赖初始化时传入的vector_function，因此后续查询必须沿用同一个向量函数，否则无法利用该索引。

文档写入与去重策略（DuplicatePolicy）

write_documents(documents, policy=DuplicatePolicy.NONE)的policy参数控制文档去重行为。仓库核心库中的DuplicatePolicy定义于 haystack/document_stores/types/policy.py：

• SKIP：同 id 文档已存在时跳过，不写入；
• OVERWRITE：同 id 文档已存在时覆盖；
• FAIL：同 id 文档已存在时抛错；
• NONE：不指定，实际默认回落为FAIL语义（这一点可从核心库内存存储的实现注释确认，见 haystack/document_stores/in_memory/document_store.py）。

Pgvector 集成文档中的示例普遍使用DuplicatePolicy.OVERWRITE，方便反复运行脚本；在 RAG 示例中也可见到用DuplicatePolicy.SKIP实现"可重复运行不报错"的写法。

Raises：

• 若documents含非Document对象，抛出ValueError；
• 若文档 id 已存在且策略为FAIL（或未指定），抛出DuplicateDocumentError；
• 写入失败的其他情况抛出DocumentStoreError。

文档管理 API 一览

PgvectorDocumentStore除了写入，还提供一整套文档管理方法，且每个方法都有对应的异步版本（*_async）：

关于get_metadata_field_unique_values，有一个值得注意的细节（API 参考中有明确说明）：不同 JSON 类型类别的值会保持独立——字符串"1"与数字1不会合并；唯一例外是 JSONB 的相等语义会把整数值浮点1.0与整数1视为相等从而合并，而带小数部分的浮点（如1.5）不受影响。

get_metadata_fields_info的返回示例：

{ 'category': {'type': 'text'}, 'status': {'type': 'text'}, 'priority': {'type': 'integer'}, }

序列化与资源释放

PgvectorDocumentStore、PgvectorEmbeddingRetriever、PgvectorKeywordRetriever三个组件均实现 Haystack 标准的序列化协议：

• to_dict() -> dict[str, Any]：将组件序列化为字典（用于 YAML/JSON 流水线描述）；
• from_dict(data) -> 组件类型：从字典反序列化重建组件；
• close()/close_async()：释放底层 Document Store 的同步/异步资源。

PgvectorEmbeddingRetriever：基于稠密向量的语义检索

初始化参数

__init__( *, document_store: PgvectorDocumentStore, 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（必填）：PgvectorDocumentStore实例。若非该类型实例，抛出ValueError；
• filters：作用于被检索文档的过滤器；
• top_k：最多返回的文档数，默认 10；
• vector_function：本次检索使用的相似度函数。默认为None，此时使用 Document Store 初始化时设定的函数；若传入非法值，抛出ValueError。重要：若 Document Store 使用"hnsw"搜索策略，此处传入的函数应与建索引时所用函数一致，才能命中索引；
• filter_policy：过滤器应用策略（详见"过滤机制深入"一节）。

单独使用

检索器本身不负责生成查询向量，需要先把query_embedding准备好：

import os from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore from haystack_integrations.components.retrievers.pgvector import ( PgvectorEmbeddingRetriever, ) os.environ["PG_CONN_STR"] = "postgresql://postgres:postgres@localhost:5432/postgres" document_store = PgvectorDocumentStore() retriever = PgvectorEmbeddingRetriever(document_store=document_store) # 这里用假向量保持示例简洁，实际应使用 Text Embedder 生成 retriever.run(query_embedding=[0.1] * 768)

在 Pipeline 中使用（语义搜索）

将PgvectorEmbeddingRetriever与文本嵌入器接入 Haystack Pipeline，是文档推荐的标准用法（完整示例见 PgvectorEmbeddingRetriever 指南）：

import os from haystack.document_stores.types import DuplicatePolicy from haystack import Document, Pipeline from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore from haystack_integrations.components.retrievers.pgvector import ( PgvectorEmbeddingRetriever, ) os.environ["PG_CONN_STR"] = "postgresql://postgres:postgres@localhost:5432/postgres" document_store = PgvectorDocumentStore( 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", PgvectorEmbeddingRetriever(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])

这条管线的数据流是：SentenceTransformersTextEmbedder把查询文本编码为向量 → 通过text_embedder.embedding → retriever.query_embedding连接传入检索器 → 检索器在 PostgreSQL 中执行向量相似度查询 → 返回documents列表。最终结果从result["retriever"]["documents"]取回。

run / run_async 签名

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]]

运行时参数与初始化参数同名者会覆盖初始化值。返回值是字典，键为documents，值为与query_embedding最相似的Document列表。run_async提供等价异步实现。

PgvectorKeywordRetriever：基于关键词的全文检索

初始化参数

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

参数含义与嵌入检索器一致：document_store必填（非PgvectorDocumentStore实例抛ValueError），filters限定检索范围，top_k控制返回数量，filter_policy控制过滤器应用策略。

排序原理：ts_rank_cd

与嵌入检索不同，关键词检索直接对文档文本内容做全文检索，排序依据是 PostgreSQL 的ts_rank_cd函数。它综合考虑：

• 查询词在文档中出现的频率；
• 查询词在文档中彼此相距的远近（越近分越高）；
• 查询词出现的位置重要性（出现在标题等更重要的位置得分更高）。

因此在配置PgvectorDocumentStore时，language参数直接影响全文检索的效果——它决定了查询与文档内容按何种语言的文本搜索配置（ts_config）进行解析。可用语言可以通过 SQL 在数据库中查询：

SELECT cfgname FROM pg_ts_config;

易错点：文档明确指出，与ElasticsearchBM25Retriever等组件不同，PgvectorKeywordRetriever默认不提供模糊搜索，查询词必须与文档内容精确匹配，否则可能得到零结果。构造查询时需要注意措辞。

单独使用

from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore from haystack_integrations.components.retrievers.pgvector import ( PgvectorKeywordRetriever, ) document_store = PgvectorDocumentStore() retriever = PgvectorKeywordRetriever(document_store=document_store) retriever.run(query="my nice query")

在 RAG 管线中使用

文档给出了一个完整的"关键词检索 + LLM 生成"RAG 示例（见 PgvectorKeywordRetriever 指南），前提是设置OPENAI_API_KEY与PG_CONN_STR两个环境变量：

from haystack import Document from haystack import 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.pgvector import PgvectorDocumentStore from haystack_integrations.components.retrievers.pgvector import ( PgvectorKeywordRetriever, ) prompt_template = [ ChatMessage.from_user( """ Given these documents, answer the question.\nDocuments: {% for doc in documents %} {{ doc.content }} {% endfor %} \nQuestion: {{question}} \nAnswer: """, ), ] document_store = PgvectorDocumentStore( 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.", ), ] # DuplicatePolicy.SKIP 可选，便于脚本重复运行而不抛错 document_store.write_documents(documents=documents, policy=DuplicatePolicy.SKIP) retriever = PgvectorKeywordRetriever(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="*"), 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"])

这段示例展示了关键词检索 RAG 的完整链路：retriever取回相关文档 → 文档注入prompt_builder的提示模板 →llm生成回答 →answer_builder组装带引用的答案。

run / run_async 签名

run( query: str, filters: dict[str, Any] | None = None, top_k: int | None = None ) -> dict[str, list[Document]]

query是在文档内容中搜索的字符串；返回documents键下的匹配文档列表。run_async提供等价异步实现。

过滤机制深入：FilterPolicy 的 REPLACE 与 MERGE

两个检索器初始化时都接受filter_policy参数，默认FilterPolicy.REPLACE。它的语义定义在仓库核心库的 haystack/document_stores/types/filter_policy.py 中：

• REPLACE（默认）：run()时传入的运行时过滤器整体替换初始化时设置的过滤器；
• MERGE：运行时过滤器与初始化过滤器合并，运行时值优先覆盖初始化值中重叠的字段。

底层实现（apply_filter_policy函数）会根据过滤器形态组合出不同的合并策略，例如：

• 两个比较过滤器（comparison filter，含field/operator/value）合并为一个AND逻辑过滤器；
• 逻辑过滤器与比较过滤器合并时，若运算符一致（如都是AND），则把比较条件并入逻辑条件；若同名字段冲突，运行时条件优先；
• 两个同运算符的逻辑过滤器合并时，条件列表直接拼接；运算符不一致时，初始化过滤器被忽略并给出警告日志。

一个直观的合并效果（源码 docstring 中的示例）：

init_filters = { "operator": "AND", "conditions": [ {"field": "meta.type", "operator": "==", "value": "article"}, {"field": "meta.rating", "operator": ">=", "value": 3}, ] } runtime_filters = { "operator": "AND", "conditions": [ {"field": "meta.genre", "operator": "IN", "value": ["economy", "politics"]}, {"field": "meta.publisher", "operator": "==", "value": "nytimes"}, ] } # MERGE 策略下得到： { "operator": "AND", "conditions": [ ...四个条件合并... ] }

因此，若希望"初始化时固定一部分过滤条件（如租户隔离、数据范围），运行时追加动态条件"，应使用MERGE；若每次运行时都想完全重新指定过滤条件，则保持默认REPLACE即可。

同步 / 异步 API 对照

Pgvector 集成紧跟 Haystack 的异步支持。所有涉及数据库操作的方法都提供_async变体，便于在异步 Pipeline（run_async）或async上下文中使用：

总结与选型建议

围绕 Pgvector 集成 API 参考，本文完整覆盖了 Pgvector 集成的三个核心组件：

1. PgvectorDocumentStore：以 PostgreSQL + pgvector 为底座，提供建表、写入（含去重策略）、过滤、删除、更新、统计、元数据探索等完整文档管理能力，并支持vector/halfvec两种向量类型、精确搜索与 HNSW 近似搜索两种策略；
2. PgvectorEmbeddingRetriever：基于稠密向量的语义检索，可与任意 Text Embedder 接入 Haystack Pipeline；
3. PgvectorKeywordRetriever：基于 PostgreSQL 全文检索（ts_rank_cd）的关键词检索，适合精确词匹配场景。

从选型角度（参考 Document Store 选择指南），Pgvector 适合以下场景：已经在使用或希望继续使用 PostgreSQL、看重 ACID 与点时间恢复能力、同时需要 Embedding 与关键词两种检索方式，且文档规模可接受精确最近邻（小规模）或愿意用 HNSW 换取检索速度（大规模）的情况。

实际使用时建议遵循以下几条经验：

• 生产环境不要设置recreate_table=True，避免误删表数据；
• 向量维度与嵌入模型对齐，高维（>2000）使用halfvec并确保 pgvector ≥ 0.7.0；
• 使用 HNSW 策略时，建索引、查询、检索器三处的vector_function必须保持一致，否则索引无法生效；
• 关键词检索无模糊匹配，查询构造需精确，必要时可结合嵌入检索互为补充；
• 连接串中特殊字符做好 percent-encoding，或直接使用关键字/值格式规避。

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