Elasticsearch

ElasticsearchKVStore

Bases: BaseKVStore

Elasticsearch键值存储。

Parameters:

Name	Type	Description	Default
index_name	str	Elasticsearch索引的名称。	required
es_client	Optional[Any]	可选。预先存在的AsyncElasticsearch客户端。	required
es_url	Optional[str]	可选。Elasticsearch URL。	None
es_cloud_id	Optional[str]	可选。Elasticsearch云ID。	None
es_api_key	Optional[str]	可选。Elasticsearch API密钥。	None
es_user	Optional[str]	可选。Elasticsearch用户名。	None
es_password	Optional[str]	可选。Elasticsearch密码。	None

Raises:

Type	Description
ConnectionError	如果AsyncElasticsearch客户端无法连接到Elasticsearch。
ValueError	如果未提供es_client、es_url或es_cloud_id。

Source code in llama_index/storage/kvstore/elasticsearch/base.py

class ElasticsearchKVStore(BaseKVStore):
    """Elasticsearch键值存储。

    Args:
        index_name: Elasticsearch索引的名称。
        es_client: 可选。预先存在的AsyncElasticsearch客户端。
        es_url: 可选。Elasticsearch URL。
        es_cloud_id: 可选。Elasticsearch云ID。
        es_api_key: 可选。Elasticsearch API密钥。
        es_user: 可选。Elasticsearch用户名。
        es_password: 可选。Elasticsearch密码。


    Raises:
        ConnectionError: 如果AsyncElasticsearch客户端无法连接到Elasticsearch。
        ValueError: 如果未提供es_client、es_url或es_cloud_id。"""

    es_client: Optional[Any]
    es_url: Optional[str]
    es_cloud_id: Optional[str]
    es_api_key: Optional[str]
    es_user: Optional[str]
    es_password: Optional[str]

    def __init__(
        self,
        index_name: str,
        es_client: Optional[Any],
        es_url: Optional[str] = None,
        es_cloud_id: Optional[str] = None,
        es_api_key: Optional[str] = None,
        es_user: Optional[str] = None,
        es_password: Optional[str] = None,
    ) -> None:
        nest_asyncio.apply()

        """Init a ElasticsearchKVStore."""
        try:
            from elasticsearch import AsyncElasticsearch
        except ImportError:
            raise ImportError(IMPORT_ERROR_MSG)

        if es_client is not None:
            self._client = es_client.options(
                headers={"user-agent": self.get_user_agent()}
            )
        elif es_url is not None or es_cloud_id is not None:
            self._client: AsyncElasticsearch = _get_elasticsearch_client(
                es_url=es_url,
                username=es_user,
                password=es_password,
                cloud_id=es_cloud_id,
                api_key=es_api_key,
            )
        else:
            raise ValueError(
                """Either provide a pre-existing AsyncElasticsearch or valid \
                credentials for creating a new connection."""
            )

    @property
    def client(self) -> Any:
        """获取异步的elasticsearch客户端。"""
        return self._client

    @staticmethod
    def get_user_agent() -> str:
        """获取用于elasticsearch客户端的用户代理。"""
        return "llama_index-py-vs"

    async def _create_index_if_not_exists(self, index_name: str) -> None:
        """创建AsyncElasticsearch索引（如果尚不存在）。

Args:
    index_name：要创建的AsyncElasticsearch索引的名称。
"""
        if await self.client.indices.exists(index=index_name):
            logger.debug(f"Index {index_name} already exists. Skipping creation.")

        else:
            index_settings = {"mappings": {"_source": {"enabled": True}}}

            logger.debug(
                f"Creating index {index_name} with mappings {index_settings['mappings']}"
            )
            await self.client.indices.create(index=index_name, **index_settings)

    def put(
        self,
        key: str,
        val: dict,
        collection: str = DEFAULT_COLLECTION,
    ) -> None:
        """将一个键值对放入存储中。

Args:
    key（str）：键
    val（dict）：值
    collection（str）：集合名称
"""
        self.put_all([(key, val)], collection=collection)

    async def aput(
        self,
        key: str,
        val: dict,
        collection: str = DEFAULT_COLLECTION,
    ) -> None:
        """将一个键值对放入存储中。

Args:
    key（str）：键
    val（dict）：值
    collection（str）：集合名称
"""
        await self.aput_all([(key, val)], collection=collection)

    def put_all(
        self,
        kv_pairs: List[Tuple[str, dict]],
        collection: str = DEFAULT_COLLECTION,
        batch_size: int = DEFAULT_BATCH_SIZE,
    ) -> None:
        return asyncio.get_event_loop().run_until_complete(
            self.aput_all(kv_pairs, collection, batch_size)
        )

    async def aput_all(
        self,
        kv_pairs: List[Tuple[str, dict]],
        collection: str = DEFAULT_COLLECTION,
        batch_size: int = DEFAULT_BATCH_SIZE,
    ) -> None:
        await self._create_index_if_not_exists(collection)

        # Prepare documents with '_id' set to the key for batch insertion
        docs = [{"_id": key, **value} for key, value in kv_pairs]

        # Insert documents in batches
        for batch in (
            docs[i : i + batch_size] for i in range(0, len(docs), batch_size)
        ):
            requests = []
            for doc in batch:
                doc_id = doc["_id"]
                doc.pop("_id")
                logger.debug(doc)
                request = {
                    "_op_type": "index",
                    "_index": collection,
                    **doc,
                    "_id": doc_id,
                }
                requests.append(request)
            await async_bulk(self.client, requests, chunk_size=batch_size, refresh=True)

    def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> Optional[dict]:
        """从存储中获取一个值。

Args:
    key（str）：键
    collection（str）：集合名称
"""
        return asyncio.get_event_loop().run_until_complete(self.aget(key, collection))

    async def aget(
        self, key: str, collection: str = DEFAULT_COLLECTION
    ) -> Optional[dict]:
        """Get a value from the store.

        Args:
            key (str): key
            collection (str): collection name

        """
        await self._create_index_if_not_exists(collection)

        try:
            response = await self._client.get(index=collection, id=key, source=True)
            return response.body["_source"]
        except elasticsearch.NotFoundError:
            return None

    def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """从商店中获取所有的数值。

Args:
    collection (str): 集合名称
"""
        return asyncio.get_event_loop().run_until_complete(self.aget_all(collection))

    async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """从商店中获取所有的数值。

Args:
    collection (str): 集合名称
"""
        await self._create_index_if_not_exists(collection)

        q = {"query": {"match_all": {}}}
        response = await self._client.search(index=collection, body=q, source=True)
        result = {}
        for r in response["hits"]["hits"]:
            doc_id = r["_id"]
            content = r["_source"]
            result[doc_id] = content
        return result

    def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """从存储中删除一个值。

Args:
    key (str): 键
    collection (str): 集合名称
"""
        return asyncio.get_event_loop().run_until_complete(
            self.adelete(key, collection)
        )

    async def adelete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """从存储中删除一个值。

Args:
    key (str): 键
    collection (str): 集合名称
"""
        await self._create_index_if_not_exists(collection)

        try:
            response = await self._client.delete(index=collection, id=key)
            return response.body["result"] == "deleted"
        except elasticsearch.NotFoundError:
            return False

client property

client: Any

获取异步的elasticsearch客户端。

get_user_agent staticmethod

get_user_agent() -> str

获取用于elasticsearch客户端的用户代理。

Source code in llama_index/storage/kvstore/elasticsearch/base.py

@staticmethod
def get_user_agent() -> str:
    """获取用于elasticsearch客户端的用户代理。"""
    return "llama_index-py-vs"

put

put(
    key: str,
    val: dict,
    collection: str = DEFAULT_COLLECTION,
) -> None

将一个键值对放入存储中。

Source code in llama_index/storage/kvstore/elasticsearch/base.py

    def put(
        self,
        key: str,
        val: dict,
        collection: str = DEFAULT_COLLECTION,
    ) -> None:
        """将一个键值对放入存储中。

Args:
    key（str）：键
    val（dict）：值
    collection（str）：集合名称
"""
        self.put_all([(key, val)], collection=collection)

aput async

aput(
    key: str,
    val: dict,
    collection: str = DEFAULT_COLLECTION,
) -> None

将一个键值对放入存储中。

Source code in llama_index/storage/kvstore/elasticsearch/base.py

    async def aput(
        self,
        key: str,
        val: dict,
        collection: str = DEFAULT_COLLECTION,
    ) -> None:
        """将一个键值对放入存储中。

Args:
    key（str）：键
    val（dict）：值
    collection（str）：集合名称
"""
        await self.aput_all([(key, val)], collection=collection)

get

get(
    key: str, collection: str = DEFAULT_COLLECTION
) -> Optional[dict]

从存储中获取一个值。

Source code in llama_index/storage/kvstore/elasticsearch/base.py

    def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> Optional[dict]:
        """从存储中获取一个值。

Args:
    key（str）：键
    collection（str）：集合名称
"""
        return asyncio.get_event_loop().run_until_complete(self.aget(key, collection))

aget async

aget(
    key: str, collection: str = DEFAULT_COLLECTION
) -> Optional[dict]

Get a value from the store.

Parameters:

Name	Type	Description	Default
key	str	key	required
collection	str	collection name	DEFAULT_COLLECTION

Source code in llama_index/storage/kvstore/elasticsearch/base.py

async def aget(
    self, key: str, collection: str = DEFAULT_COLLECTION
) -> Optional[dict]:
    """Get a value from the store.

    Args:
        key (str): key
        collection (str): collection name

    """
    await self._create_index_if_not_exists(collection)

    try:
        response = await self._client.get(index=collection, id=key, source=True)
        return response.body["_source"]
    except elasticsearch.NotFoundError:
        return None

get_all

get_all(
    collection: str = DEFAULT_COLLECTION,
) -> Dict[str, dict]

从商店中获取所有的数值。

Parameters:

Name	Type	Description	Default
collection	str	集合名称	DEFAULT_COLLECTION

Source code in llama_index/storage/kvstore/elasticsearch/base.py

    def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """从商店中获取所有的数值。

Args:
    collection (str): 集合名称
"""
        return asyncio.get_event_loop().run_until_complete(self.aget_all(collection))

aget_all async

aget_all(
    collection: str = DEFAULT_COLLECTION,
) -> Dict[str, dict]

从商店中获取所有的数值。

Parameters:

Name	Type	Description	Default
collection	str	集合名称	DEFAULT_COLLECTION

Source code in llama_index/storage/kvstore/elasticsearch/base.py

    async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """从商店中获取所有的数值。

Args:
    collection (str): 集合名称
"""
        await self._create_index_if_not_exists(collection)

        q = {"query": {"match_all": {}}}
        response = await self._client.search(index=collection, body=q, source=True)
        result = {}
        for r in response["hits"]["hits"]:
            doc_id = r["_id"]
            content = r["_source"]
            result[doc_id] = content
        return result

delete

delete(
    key: str, collection: str = DEFAULT_COLLECTION
) -> bool

从存储中删除一个值。

Parameters:

Name	Type	Description	Default
key	str	键	required
collection	str	集合名称	DEFAULT_COLLECTION

Source code in llama_index/storage/kvstore/elasticsearch/base.py

    def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """从存储中删除一个值。

Args:
    key (str): 键
    collection (str): 集合名称
"""
        return asyncio.get_event_loop().run_until_complete(
            self.adelete(key, collection)
        )

adelete async

adelete(
    key: str, collection: str = DEFAULT_COLLECTION
) -> bool

从存储中删除一个值。

Parameters:

Name	Type	Description	Default
key	str	键	required
collection	str	集合名称	DEFAULT_COLLECTION

Source code in llama_index/storage/kvstore/elasticsearch/base.py

    async def adelete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """从存储中删除一个值。

Args:
    key (str): 键
    collection (str): 集合名称
"""
        await self._create_index_if_not_exists(collection)

        try:
            response = await self._client.delete(index=collection, id=key)
            return response.body["result"] == "deleted"
        except elasticsearch.NotFoundError:
            return False