Astra db

AstraDBReader

Bases: BaseReader

Astra DB 读取器。

从 Astra DB 实例中检索文档。

Parameters:

Name	Type	Description	Default
collection_name	str	要使用的集合名称。如果不存在，将被创建。	required
token	str	要使用的 Astra DB 应用程序令牌。	required
api_endpoint	str	数据库的 Astra DB JSON API 端点。	required
embedding_dimension	int	使用的嵌入向量的长度。	required
namespace	Optional[str]	要使用的命名空间。如果未提供，则为 'default_keyspace'。	None
client	Optional[Any]	要使用的 Astra DB 客户端。如果未提供，将创建一个。	None

Source code in llama_index/readers/astra_db/base.py

class AstraDBReader(BaseReader):
    """Astra DB 读取器。

    从 Astra DB 实例中检索文档。

    Args:
        collection_name (str): 要使用的集合名称。如果不存在，将被创建。
        token (str): 要使用的 Astra DB 应用程序令牌。
        api_endpoint (str): 数据库的 Astra DB JSON API 端点。
        embedding_dimension (int): 使用的嵌入向量的长度。
        namespace (Optional[str]): 要使用的命名空间。如果未提供，则为 'default_keyspace'。
        client (Optional[Any]): 要使用的 Astra DB 客户端。如果未提供，将创建一个。"""

    def __init__(
        self,
        *,
        collection_name: str,
        token: str,
        api_endpoint: str,
        embedding_dimension: int,
        namespace: Optional[str] = None,
        client: Optional[Any] = None,
    ) -> None:
        """使用参数进行初始化。"""
        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:
        """从Astra DB加载数据。

Args:
    vector (Any): 查询
    limit (int): 要返回的结果数量。
    kwargs (Any): 传递给Astra DB查询的其他参数。

Returns:
    List[Document]: 文档列表。
"""
        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加载数据。

Parameters:

Name	Type	Description	Default
vector	Any	查询	required
limit	int	要返回的结果数量。	10
kwargs	Any	传递给Astra DB查询的其他参数。	{}

Returns:

Type	Description
Any	List[Document]: 文档列表。

Source code in llama_index/readers/astra_db/base.py

    def load_data(self, vector: List[float], limit: int = 10, **kwargs: Any) -> Any:
        """从Astra DB加载数据。

Args:
    vector (Any): 查询
    limit (int): 要返回的结果数量。
    kwargs (Any): 传递给Astra DB查询的其他参数。

Returns:
    List[Document]: 文档列表。
"""
        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