Astra DB

AstraDBReader

基类：BaseReader

Astra DB 阅读器。

从 Astra DB 实例检索文档。

参数

名称	类型	描述	默认
collection_name	str	要使用的集合名称。如果不存在，则会自动创建。	必需
token	str	要使用的 Astra DB 应用程序令牌。	必需
api_endpoint	str	您的数据库的 Astra DB JSON API 端点。	必需
embedding_dimension	int	正在使用的嵌入向量的长度。	必需
namespace	可选[str]	要使用的命名空间。如果未提供，则使用 'default_keyspace'	无
client	可选[Any]	要使用的 Astra DB 客户端。如果未提供，则会创建一个。	无

源代码位于 llama-index-integrations/readers/llama-index-readers-astra-db/llama_index/readers/astra_db/base.py

class AstraDBReader(BaseReader):
    """
    Astra DB reader.

    Retrieve documents from an Astra DB Instance.

    Args:
        collection_name (str): collection name to use. If not existing, it will be created.
        token (str): The Astra DB Application Token to use.
        api_endpoint (str): The Astra DB JSON API endpoint for your database.
        embedding_dimension (int): Length of the embedding vectors in use.
        namespace (Optional[str]): The namespace to use. If not provided, 'default_keyspace'
        client (Optional[Any]): Astra DB client to use. If not provided, one will be created.

    """

    def __init__(
        self,
        *,
        collection_name: str,
        token: str,
        api_endpoint: str,
        embedding_dimension: int,
        namespace: Optional[str] = None,
        client: Optional[Any] = None,
    ) -> None:
        """Initialize with parameters."""
        import_err_msg = (
            "`astrapy` package not found, please run `pip install --upgrade astrapy`"
        )

        # Try to import astrapy for use
        try:
            from astrapy.db import AstraDB
        except ImportError:
            raise ImportError(import_err_msg)

        if client is not None:
            self._client = client.copy()
            self._client.set_caller(
                caller_name=getattr(llama_index, "__name__", "llama_index"),
                caller_version=getattr(llama_index.core, "__version__", None),
            )
        else:
            # Build the Astra DB object
            self._client = AstraDB(
                api_endpoint=api_endpoint,
                token=token,
                namespace=namespace,
                caller_name=getattr(llama_index, "__name__", "llama_index"),
                caller_version=getattr(llama_index.core, "__version__", None),
            )

        self._collection = self._client.create_collection(
            collection_name=collection_name, dimension=embedding_dimension
        )

    def load_data(self, vector: List[float], limit: int = 10, **kwargs: Any) -> Any:
        """
        Load data from Astra DB.

        Args:
            vector (Any): Query
            limit (int): Number of results to return.
            kwargs (Any): Additional arguments to pass to the Astra DB query.

        Returns:
            List[Document]: A list of documents.

        """
        results = self._collection.vector_find(
            vector,
            limit=limit,
            fields=["*"],
            **kwargs,
        )

        documents: List[Document] = []
        for result in results:
            document = Document(
                doc_id=result["_id"],
                text=result["content"],
                embedding=result["$vector"],
            )

            documents.append(document)

        return documents

load_data

load_data(vector: List[float], limit: int = 10, **kwargs: Any) -> Any

从 Astra DB 加载数据。

参数

名称	类型	描述	默认
vector	Any	查询	必需
limit	int	要返回的结果数量。	10
kwargs	Any	要传递给 Astra DB 查询的附加参数。	{}

返回值

类型	描述
Any	List[Document]：文档列表。

源代码位于 llama-index-integrations/readers/llama-index-readers-astra-db/llama_index/readers/astra_db/base.py

def load_data(self, vector: List[float], limit: int = 10, **kwargs: Any) -> Any:
    """
    Load data from Astra DB.

    Args:
        vector (Any): Query
        limit (int): Number of results to return.
        kwargs (Any): Additional arguments to pass to the Astra DB query.

    Returns:
        List[Document]: A list of documents.

    """
    results = self._collection.vector_find(
        vector,
        limit=limit,
        fields=["*"],
        **kwargs,
    )

    documents: List[Document] = []
    for result in results:
        document = Document(
            doc_id=result["_id"],
            text=result["content"],
            embedding=result["$vector"],
        )

        documents.append(document)

    return documents