Module haystack_integrations.components.retrievers.astra.retriever

AstraEmbeddingRetriever

一个用于从 AstraDocumentStore 中检索文档的组件。

使用示例

from haystack_integrations.document_stores.astra import AstraDocumentStore
from haystack_integrations.components.retrievers.astra import AstraEmbeddingRetriever

document_store = AstraDocumentStore(
    api_endpoint=api_endpoint,
    token=token,
    collection_name=collection_name,
    duplicates_policy=DuplicatePolicy.SKIP,
    embedding_dim=384,
)

retriever = AstraEmbeddingRetriever(document_store=document_store)

AstraEmbeddingRetriever.__init__

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

参数:

• document_store: AstraDocumentStore 的一个实例。
• filters: 一个包含用于缩小搜索范围的过滤器的字典。
• top_k: 要检索的最大文档数。
• filter_policy: 确定如何应用过滤器的策略。

AstraEmbeddingRetriever.run

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

从 AstraDocumentStore 检索文档。

参数:

• query_embedding: 表示查询嵌入的浮点数
• filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
• top_k: 要检索的最大文档数。

返回值:

一个包含以下键的字典

• documents: 从 AstraDocumentStore 检索到的文档列表。

AstraEmbeddingRetriever.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AstraEmbeddingRetriever.from_dict

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

从字典反序列化组件。

参数:

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

返回值:

反序列化后的组件。

Module haystack_integrations.document_stores.astra.document_store

AstraDocumentStore

Haystack 的 AstraDocumentStore 文档存储。

示例用法

from haystack_integrations.document_stores.astra import AstraDocumentStore

document_store = AstraDocumentStore(
    api_endpoint=api_endpoint,
    token=token,
    collection_name=collection_name,
    duplicates_policy=DuplicatePolicy.SKIP,
    embedding_dim=384,
)

AstraDocumentStore.__init__

def __init__(
        api_endpoint: Secret = Secret.from_env_var("ASTRA_DB_API_ENDPOINT"),
        token: Secret = Secret.from_env_var("ASTRA_DB_APPLICATION_TOKEN"),
        collection_name: str = "documents",
        embedding_dimension: int = 768,
        duplicates_policy: DuplicatePolicy = DuplicatePolicy.NONE,
        similarity: str = "cosine",
        namespace: Optional[str] = None)

与 Astra DB 的连接通过 JSON API 进行建立和管理。

所需的凭据（API 端点和应用程序令牌）可以通过 UI 生成，方法是单击“连接”选项卡，然后选择 JSON API 并生成配置。

参数:

• api_endpoint: Astra DB API 端点。
• token: Astra DB 应用程序令牌。
• collection_name: Astra DB 中当前键空间中的当前集合。
• embedding_dimension: 嵌入向量的维度。
• duplicates_policy: 基于 DuplicatePolicy 参数选项处理重复文档。参数选项：（SKIP, OVERWRITE, FAIL, NONE)
• DuplicatePolicy.NONE: 默认策略，如果具有相同 ID 的文档已存在，则跳过写入。
• DuplicatePolicy.SKIP: 如果具有相同 ID 的文档已存在，则跳过写入。
• DuplicatePolicy.OVERWRITE: 如果具有相同 ID 的文档已存在，则覆盖。
• DuplicatePolicy.FAIL: 如果具有相同 ID 的文档已存在，则引发错误。
• similarity: 用于比较文档向量的相似度函数。

引发:

• ValueError: 如果未设置 API 端点或令牌。

AstraDocumentStore.from_dict

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

从字典反序列化组件。

参数:

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

返回值:

反序列化后的组件。

AstraDocumentStore.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AstraDocumentStore.write_documents

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

为后续查询索引文档。

参数:

• documents: Haystack Document 对象列表。
• policy: 基于 DuplicatePolicy 参数选项处理重复文档。参数选项：（SKIP, OVERWRITE, FAIL, NONE)
• DuplicatePolicy.NONE: 默认策略，如果具有相同 ID 的文档已存在，则跳过写入。
• DuplicatePolicy.SKIP: 如果具有相同 ID 的文档已存在，则跳过写入。
• DuplicatePolicy.OVERWRITE: 如果具有相同 ID 的文档已存在，则覆盖。
• DuplicatePolicy.FAIL: 如果具有相同 ID 的文档已存在，则引发错误。

引发:

• ValueError: 如果文档不是 Document 或 dict 类型。
• DuplicateDocumentError: 如果具有相同 ID 的文档已存在且策略设置为 FAIL。
• Exception: 如果文档 ID 不是字符串或如果id 和_id 同时存在于文档中。

返回值:

写入的文档数量。

AstraDocumentStore.count_documents

def count_documents() -> int

计算文档存储中的文档数量。

返回值:

文档存储中的文档数量。

AstraDocumentStore.filter_documents

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

返回最多 1000 个匹配过滤器的文档。

参数:

• filters: 要应用的过滤器。

引发:

• AstraDocumentStoreFilterError: 如果过滤器无效或不被此类支持。

返回值:

匹配的文档。

AstraDocumentStore.get_documents_by_id

def get_documents_by_id(ids: List[str]) -> List[Document]

根据 ID 获取文档。

参数:

• ids: 要检索的文档的 ID。

返回值:

匹配的文档。

AstraDocumentStore.get_document_by_id

def get_document_by_id(document_id: str) -> Document

根据 ID 获取文档。

参数:

• document_id: 要过滤的 ID

引发:

• MissingDocumentError: 如果找不到文档

返回值:

找到的文档

AstraDocumentStore.search

def search(query_embedding: List[float],
           top_k: int,
           filters: Optional[Dict[str, Any]] = None) -> List[Document]

对查询列表执行搜索。

参数:

• query_embedding: 查询嵌入列表。
• top_k: 要返回的结果数量。
• filters: 搜索过程中要应用的过滤器。

返回值:

匹配的文档。

AstraDocumentStore.delete_documents

def delete_documents(document_ids: Optional[List[str]] = None,
                     *,
                     delete_all: Optional[bool] = None) -> None

从文档存储中删除文档。

参数:

• document_ids: 要删除的文档的 ID。
• delete_all: 如果True，则删除所有文档。

引发:

• MissingDocumentError: 如果未删除任何文档但提供了文档 ID。

Module haystack_integrations.document_stores.astra.errors

AstraDocumentStoreError

所有 AstraDocumentStore 错误的父类。

AstraDocumentStoreFilterError

当向 AstraDocumentStore 传递无效过滤器时引发。

AstraDocumentStoreConfigError

当向 AstraDocumentStore 传递无效配置时引发。