文档API 参考📓 教程🧑‍🍳 食谱🤝 集成💜 Discord🎨 Studio
API 参考

Pgvector

Haystack 的 Pgvector 集成

模块 haystack_integrations.components.retrievers.pgvector.embedding_retriever

PgvectorEmbeddingRetriever

从PgvectorDocumentStore 中检索文档，依据它们的密集嵌入。

示例用法

from haystack.document_stores import DuplicatePolicy
from haystack import Document, Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder

from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore
from haystack_integrations.components.retrievers.pgvector import PgvectorEmbeddingRetriever

# Set an environment variable `PG_CONN_STR` with the connection string to your PostgreSQL database.
# e.g., "postgresql://USER:PASSWORD@HOST:PORT/DB_NAME"

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..."),
             Document(content="In certain places, you can witness the phenomenon of bioluminescent waves.")]

document_embedder = SentenceTransformersDocumentEmbedder()
document_embedder.warm_up()
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?"

res = query_pipeline.run({"text_embedder": {"text": query}})

assert res['retriever']['documents'][0].content == "There are over 7,000 languages spoken around the world today."

PgvectorEmbeddingRetriever.__init__

def __init__(*,
             document_store: PgvectorDocumentStore,
             filters: Optional[Dict[str, Any]] = None,
             top_k: int = 10,
             vector_function: Optional[Literal["cosine_similarity",
                                               "inner_product",
                                               "l2_distance"]] = None,
             filter_policy: Union[str, FilterPolicy] = FilterPolicy.REPLACE)

参数:

  • document_store: 一个PgvectorDocumentStore.
  • filters: 应用于检索到的文档的过滤器。
  • top_k: 要返回的最大文档数量。
  • vector_function: 在搜索相似嵌入时使用的相似度函数。默认为在document_store 实例中设置的函数。"cosine_similarity" 和"inner_product" 是相似度函数，分数越高表示文档之间相似度越高。"l2_distance" 返回向量之间的直线距离，得分最低的文档最相似。**重要提示**：如果文档存储使用"hnsw" 搜索策略，则向量函数应与索引创建时使用的函数匹配，以便利用索引。
  • filter_policy: 确定如何应用过滤器的策略。

引发:

  • ValueError: 如果document_store 不是一个实例PgvectorDocumentStore 或如果vector_function 不是有效选项之一。

PgvectorEmbeddingRetriever.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

PgvectorEmbeddingRetriever.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "PgvectorEmbeddingRetriever"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

PgvectorEmbeddingRetriever.run

@component.output_types(documents=List[Document])
def run(
    query_embedding: List[float],
    filters: Optional[Dict[str, Any]] = None,
    top_k: Optional[int] = None,
    vector_function: Optional[Literal["cosine_similarity", "inner_product",
                                      "l2_distance"]] = None
) -> Dict[str, List[Document]]

从PgvectorDocumentStore 中检索文档，依据它们的嵌入。

参数:

  • query_embedding: 查询的嵌入。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要返回的最大文档数量。
  • vector_function: 在搜索相似嵌入时使用的相似度函数。

返回值:

包含以下键的字典

  • documents: 匹配查询的Document列表。Documents 与query_embedding.

PgvectorEmbeddingRetriever.run_async

@component.output_types(documents=List[Document])
async def run_async(
    query_embedding: List[float],
    filters: Optional[Dict[str, Any]] = None,
    top_k: Optional[int] = None,
    vector_function: Optional[Literal["cosine_similarity", "inner_product",
                                      "l2_distance"]] = None
) -> Dict[str, List[Document]]

异步检索文档。PgvectorDocumentStore 中检索文档，依据它们的嵌入。

参数:

  • query_embedding: 查询的嵌入。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要返回的最大文档数量。
  • vector_function: 在搜索相似嵌入时使用的相似度函数。

返回值:

包含以下键的字典

  • documents: 匹配查询的Document列表。Documents 与query_embedding.

模块 haystack_integrations.components.retrievers.pgvector.keyword_retriever

PgvectorKeywordRetriever

从PgvectorDocumentStore，依据关键词。

为了对文档进行排序，使用了 PostgreSQL 的ts_rank_cd 函数。它考虑了查询词在文档中出现的频率、词在文档中的接近程度以及词所在文档部分的重要性。更多细节，请参阅 Postgres 文档。

使用示例

from haystack.document_stores import DuplicatePolicy
from haystack import Document

from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore
from haystack_integrations.components.retrievers.pgvector import PgvectorKeywordRetriever

# Set an environment variable `PG_CONN_STR` with the connection string to your PostgreSQL database.
# e.g., "postgresql://USER:PASSWORD@HOST:PORT/DB_NAME"

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..."),
    Document(content="In certain places, you can witness the phenomenon of bioluminescent waves.")]

document_store.write_documents(documents_with_embeddings.get("documents"), policy=DuplicatePolicy.OVERWRITE)

retriever = PgvectorKeywordRetriever(document_store=document_store)

result = retriever.run(query="languages")

assert res['retriever']['documents'][0].content == "There are over 7,000 languages spoken around the world today."

<a id="haystack_integrations.components.retrievers.pgvector.keyword_retriever.PgvectorKeywordRetriever.__init__"></a>

#### PgvectorKeywordRetriever.\_\_init\_\_

```python
def __init__(*,
             document_store: PgvectorDocumentStore,
             filters: Optional[Dict[str, Any]] = None,
             top_k: int = 10,
             filter_policy: Union[str, FilterPolicy] = FilterPolicy.REPLACE)

参数:

  • document_store: 一个PgvectorDocumentStore.
  • filters: 应用于检索到的文档的过滤器。
  • top_k: 要返回的最大文档数量。
  • filter_policy: 确定如何应用过滤器的策略。

引发:

  • ValueError: 如果document_store 不是一个实例PgvectorDocumentStore.

PgvectorKeywordRetriever.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

PgvectorKeywordRetriever.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "PgvectorKeywordRetriever"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

PgvectorKeywordRetriever.run

@component.output_types(documents=List[Document])
def run(query: str,
        filters: Optional[Dict[str, Any]] = None,
        top_k: Optional[int] = None) -> Dict[str, List[Document]]

从PgvectorDocumentStore，依据关键词。

参数:

  • query: 要在Documents 内容中搜索的字符串。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要返回的最大文档数量。

返回值:

包含以下键的字典

  • documents: 匹配查询的Document列表。匹配查询的 Documents。

PgvectorKeywordRetriever.run_async

@component.output_types(documents=List[Document])
async def run_async(query: str,
                    filters: Optional[Dict[str, Any]] = None,
                    top_k: Optional[int] = None) -> Dict[str, List[Document]]

异步检索文档。PgvectorDocumentStore，依据关键词。

参数:

  • query: 要在Documents 内容中搜索的字符串。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要返回的最大文档数量。

返回值:

包含以下键的字典

  • documents: 匹配查询的Document列表。匹配查询的 Documents。

模块 haystack_integrations.document_stores.pgvector.document_store

PgvectorDocumentStore

一个使用 PostgreSQL 和 pgvector 扩展的 Document Store。

PgvectorDocumentStore.__init__

def __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: Optional[Dict[str, int]] = None,
             hnsw_index_name: str = "haystack_hnsw_index",
             hnsw_ef_search: Optional[int] = None,
             keyword_index_name: str = "haystack_keyword_index")

创建一个新的 PgvectorDocumentStore 实例。

它旨在连接到安装了 pgvector 扩展的 PostgreSQL 数据库。如果不存在，将创建一个用于存储 Haystack 文档的特定表。

参数:

  • connection_string: 用于连接 PostgreSQL 数据库的连接字符串，定义为环境变量。它可以是 URI 格式，例如：PG_CONN_STR="postgresql://USER:PASSWORD@HOST:PORT/DB_NAME"，或关键字/值格式，例如：PG_CONN_STR="host=HOST port=PORT dbname=DBNAME user=USER password=PASSWORD"。有关更多详细信息，请参阅 PostgreSQL 文档。
  • create_extension: 如果 pgvector 扩展不存在，是否创建它。将其设置为True (默认) 以在扩展缺失时自动创建。创建扩展可能需要超级用户权限。如果设置为False，请确保扩展已安装；否则，将引发错误。
  • schema_name: 创建表的模式名称。该模式必须已存在。
  • table_name: 用于存储 Haystack 文档的表的名称。
  • language: 在关键词检索中用于解析查询和文档内容的语言。要查看可用语言列表，您可以在 PostgreSQL 数据库中运行以下 SQL 查询：SELECT cfgname FROM pg_ts_config;。您可以在此 StackOverflow 回答 中找到更多信息。
  • embedding_dimension: 嵌入的维度。
  • vector_type: 用于嵌入存储的向量类型。"vector" 是默认值。"halfvec" 以半精度存储嵌入，这对于高维嵌入（维度大于 2000 且高达 4000）特别有用。需要 pgvector 版本 0.7.0 或更高版本。有关更多信息，请参阅 pgvector 文档。
  • vector_function: 在搜索相似嵌入时使用的相似度函数。"cosine_similarity" 和"inner_product" 是相似度函数，分数越高表示文档之间相似度越高。"l2_distance" 返回向量之间的直线距离，得分最低的文档最相似。**重要提示**：当使用"hnsw" 搜索策略时，将创建一个依赖于此处传递的vector_function 的索引。确保后续查询继续使用相同的向量相似度函数，以便利用索引。
  • recreate_table: 如果表已存在，是否重新创建它。
  • search_strategy: 在搜索相似嵌入时使用的搜索策略。"exact_nearest_neighbor" 提供完美的召回率，但对于大量文档可能速度较慢。"hnsw" 是一种近似最近邻搜索策略，它在精度上有所牺牲以换取速度；建议用于大量文档。**重要提示**：当使用"hnsw" 搜索策略时，将创建一个依赖于此处传递的vector_function 的索引。确保后续查询继续使用相同的向量相似度函数，以便利用索引。
  • hnsw_recreate_index_if_exists: 如果 HNSW 索引已存在，是否重新创建它。仅在 search_strategy 设置为"hnsw" 时使用。.
  • hnsw_index_creation_kwargs: 传递给 HNSW 索引创建的其他关键字参数。仅在 search_strategy 设置为"hnsw" 时使用。您可以在 pgvector 文档 中找到有效参数列表。
  • hnsw_index_name: HNSW 索引的索引名称。
  • hnsw_ef_search: 在查询时使用的ef_search 参数。仅在 search_strategy 设置为"hnsw" 时使用。您可以在 pgvector 文档 中找到有关此参数的更多信息。
  • keyword_index_name: 关键词索引的索引名称。

PgvectorDocumentStore.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

PgvectorDocumentStore.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "PgvectorDocumentStore"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

PgvectorDocumentStore.delete_table

def delete_table()

删除用于存储 Haystack 文档的表。模式名称（schema_name）和表名称（table_name）在初始化PgvectorDocumentStore.

PgvectorDocumentStore.delete_table_async

async def delete_table_async()

异步删除用于存储 Haystack 文档的表的方法。

PgvectorDocumentStore.count_documents

def count_documents() -> int

返回文档存储中存在的文档数量。

返回值:

文档存储中的文档数量。

PgvectorDocumentStore.count_documents_async

async def count_documents_async() -> int

返回文档存储中存在的文档数量。

返回值:

文档存储中的文档数量。

PgvectorDocumentStore.filter_documents

def filter_documents(
        filters: Optional[Dict[str, Any]] = None) -> List[Document]

返回与提供的过滤器匹配的文档。

有关过滤器的详细规范，请参阅 文档。

参数:

  • filters: 要应用于文档列表的过滤器。

引发:

  • TypeError: 如果filters 不是字典。
  • ValueError: 如果过滤器的语法无效。

返回值:

与给定过滤器匹配的文档列表。

PgvectorDocumentStore.filter_documents_async

async def filter_documents_async(
        filters: Optional[Dict[str, Any]] = None) -> List[Document]

异步返回与提供的过滤器匹配的文档。

有关过滤器的详细规范，请参阅 文档。

参数:

  • filters: 要应用于文档列表的过滤器。

引发:

  • TypeError: 如果filters 不是字典。
  • ValueError: 如果过滤器的语法无效。

返回值:

与给定过滤器匹配的文档列表。

PgvectorDocumentStore.write_documents

def write_documents(documents: List[Document],
                    policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int

将文档写入文档存储。

参数:

  • documents: 要写入文档存储的文档列表。
  • policy: 写入文档时使用的重复策略。

引发:

  • ValueError: 如果文档 documents 包含的不是类型Document.
  • DuplicateDocumentError: 如果具有相同 ID 的文档已存在于文档存储中，并且策略设置为DuplicatePolicy.FAIL (或未指定)。
  • DocumentStoreError: 如果写入操作因任何其他原因失败。

返回值:

写入文档存储的文档数量。

PgvectorDocumentStore.write_documents_async

async def write_documents_async(
        documents: List[Document],
        policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int

异步将文档写入文档存储。

参数:

  • documents: 要写入文档存储的文档列表。
  • policy: 写入文档时使用的重复策略。

引发:

  • ValueError: 如果文档 documents 包含的不是类型Document.
  • DuplicateDocumentError: 如果具有相同 ID 的文档已存在于文档存储中，并且策略设置为DuplicatePolicy.FAIL (或未指定)。
  • DocumentStoreError: 如果写入操作因任何其他原因失败。

返回值:

写入文档存储的文档数量。

PgvectorDocumentStore.delete_documents

def delete_documents(document_ids: List[str]) -> None

删除与提供的document_ids 匹配的文档。

参数:

  • document_ids: 要删除的文档 ID

PgvectorDocumentStore.delete_documents_async

async def delete_documents_async(document_ids: List[str]) -> None

异步删除与提供的document_ids 匹配的文档。

参数:

  • document_ids: 要删除的文档 ID

PgvectorDocumentStore.delete_all_documents

def delete_all_documents() -> None

删除文档存储中的所有文档。

PgvectorDocumentStore.delete_all_documents_async

async def delete_all_documents_async() -> None

异步删除文档存储中的所有文档。