类class agentscope.rag

ElasticsearchStore

class ElasticsearchStore(VectorStoreBase)

源码Source agentscope/rag/_vdb/_elasticsearch.py:20–415

Elasticsearch 存储，负责持久化与读取数据。位于 agentscope.rag 包，继承自 VectorStoreBase。核心成员包括 get_client、create_collection、delete_collection、has_collection、insert、delete。

Each knowledge base maps to one Elasticsearch index. Vectors use cosine similarity and approximate k-nearest-neighbour search. Chunk payloads are retained in _source while metadata is duplicated into a flattened field so exact-match filters do not create unbounded dynamic mappings.

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(hosts: str \| list[str], *, num_candidates: int = 100, refresh: bool \| Literal['wait_for'] = 'wait_for', cl…`	初始化 `ElasticsearchStore` 实例。 Initialize the Elasticsearch vector store.

方法摘要Method Summary

方法Method	说明Description
`def get_client() -> 'AsyncElasticsearch'`	获取客户端。 Lazily create and cache the shared async client.
`async def __aexit__(exc_type: type[BaseException] \| None, exc_val: BaseException \| None, exc_tb: Any) -> None`async	退出异步上下文并清理资源。 Close the underlying client if this store created it.
`async def create_collection(name: str, dimensions: int) -> None`async	创建集合。 Create an Elasticsearch index for one knowledge base.
`async def delete_collection(name: str) -> None`async	删除集合。 Delete an Elasticsearch index and all its records.
`async def has_collection(name: str) -> bool`async	判断是否具有集合。 Return whether an Elasticsearch index exists.
`async def insert(collection: str, records: list[VectorRecord]) -> None`async	执行插入相关操作。 Bulk-index records using deterministic, retry-safe IDs.
`async def delete(collection: str, document_id: str, metadata_filter: dict[str, Any] \| None = None) -> None`async	删除数据。 Delete every chunk belonging to one source document.
`async def search(collection: str, query_vector: list[float], top_k: int = 5, metadata_filter: dict[str, Any] \| None = N…`async	执行搜索相关操作。 Run an approximate cosine kNN search.
`async def list_documents(collection: str, metadata_filter: dict[str, Any] \| None = None) -> list[DocumentSummary]`async	列出文档。 Aggregate documents with paginated composite buckets.
`async def list_chunks(collection: str, document_id: str, *, offset: int = 0, limit: int = 30, metadata_filter: dict[str…`async	列出 `list_chunks`。 List one document's chunks ordered by ``chunk_index``.

构造器详情Constructor Detail

ElasticsearchStore

def __init__(hosts: str | list[str], *, num_candidates: int = 100, refresh: bool | Literal['wait_for'] = 'wait_for', client_kwargs: dict[str, Any] | None = None) -> None

初始化 ElasticsearchStore 实例。

Initialize the Elasticsearch vector store.

参数Parameters

名称Name	类型Type	说明Description
`hosts`	str \| list[str]	hosts（类型 `str \| list[str]`）。Elasticsearch URL or list of URLs.
`num_candidates`kw默认default `100`	int	num candidates（类型 `int`）。Minimum HNSW candidates considered per shard. The effective value is raised to `top_k` when necessary.
`refresh`kw默认default `'wait_for'`	bool \| Literal['wait_for']	刷新（类型 `bool \| Literal['wait_for']`）。Refresh policy for writes. Set to `False` for higher indexing throughput when immediate search visibility is not required. Elasticsearch's delete-by-query API only accepts a boolean, so `"wait_for"` maps to `True` for deletes.
`client_kwargs`kw默认default `None`	dict[str, Any] \| None	客户端关键字参数（类型 `dict[str, Any] \| None`）。Extra arguments forwarded to `AsyncElasticsearch` such as `api_key`, `basic_auth` or `ca_certs`.

源码Source agentscope/rag/_vdb/_elasticsearch.py:33–65

方法详情Method Detail

get_client

def get_client() -> 'AsyncElasticsearch'

获取客户端。

Lazily create and cache the shared async client.

返回Returns 'AsyncElasticsearch'

返回'AsyncElasticsearch'。Returns async elasticsearch.

源码Source agentscope/rag/_vdb/_elasticsearch.py:67–76

aexit

async def __aexit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None

async

退出异步上下文并清理资源。

Close the underlying client if this store created it.

参数Parameters

名称Name	类型Type	说明Description
`exc_type`	type[BaseException] \| None	exc 类型（类型 `type[BaseException] \| None`）。The exc type.
`exc_val`	BaseException \| None	exc val（类型 `BaseException \| None`）。The exc val.
`exc_tb`	Any	exc tb（类型 `Any`）。The exc tb.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_elasticsearch.py:78–87

create_collection

async def create_collection(name: str, dimensions: int) -> None

async

创建集合。

Create an Elasticsearch index for one knowledge base.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	名称（类型 `str`）。The name.
`dimensions`	int	向量维度（类型 `int`）。The dimensions.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_elasticsearch.py:89–109

delete_collection

async def delete_collection(name: str) -> None

async

删除集合。

Delete an Elasticsearch index and all its records.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	名称（类型 `str`）。The name.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_elasticsearch.py:111–113

has_collection

async def has_collection(name: str) -> bool

async

判断是否具有集合。

Return whether an Elasticsearch index exists.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	名称（类型 `str`）。The name.

返回Returns bool

返回布尔值，表示判断结果。Returns a boolean result.

源码Source agentscope/rag/_vdb/_elasticsearch.py:115–117

insert

async def insert(collection: str, records: list[VectorRecord]) -> None

async

执行插入相关操作。

Bulk-index records using deterministic, retry-safe IDs.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	集合名称（类型 `str`）。The collection.
`records`	list[VectorRecord]	records（类型 `list[VectorRecord]`）。The records.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_elasticsearch.py:119–160

delete

async def delete(collection: str, document_id: str, metadata_filter: dict[str, Any] | None = None) -> None

async

删除数据。

Delete every chunk belonging to one source document.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	目标集合名称。The target collection name.
`document_id`	str	文档标识（类型 `str`）。The source document ID whose records should be removed.
`metadata_filter`默认default `None`	dict[str, Any] \| None	元数据过滤器（类型 `dict[str, Any] \| None`）。Restrict deletion to records matching every metadata pair in addition to the document ID. `None` disables filtering.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_elasticsearch.py:162–188

search

async def search(collection: str, query_vector: list[float], top_k: int = 5, metadata_filter: dict[str, Any] | None = None) -> list[VectorSearchResult]

async

执行搜索相关操作。

Run an approximate cosine kNN search.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	集合名称（类型 `str`）。The collection.
`query_vector`	list[float]	查询向量（类型 `list[float]`）。The query vector.
`top_k`默认default `5`	int	返回条数（类型 `int`）。The top k.
`metadata_filter`默认default `None`	dict[str, Any] \| None	元数据过滤器（类型 `dict[str, Any] \| None`）。The metadata filter.

返回Returns list[VectorSearchResult]

返回VectorSearchResult列表。Returns a list of vector search result.

源码Source agentscope/rag/_vdb/_elasticsearch.py:190–229

list_documents

async def list_documents(collection: str, metadata_filter: dict[str, Any] | None = None) -> list[DocumentSummary]

async

列出文档。

Aggregate documents with paginated composite buckets.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	集合名称（类型 `str`）。The collection.
`metadata_filter`默认default `None`	dict[str, Any] \| None	元数据过滤器（类型 `dict[str, Any] \| None`）。The metadata filter.

返回Returns list[DocumentSummary]

返回DocumentSummary列表。Returns a list of document summary.

源码Source agentscope/rag/_vdb/_elasticsearch.py:231–289

list_chunks

async def list_chunks(collection: str, document_id: str, *, offset: int = 0, limit: int = 30, metadata_filter: dict[str, Any] | None = None) -> list[Chunk]

async

列出 list_chunks。

The index mapping stores chunk with enabled: false (it is a verbatim payload, neither indexed nor sortable), so unlike the other backends this implementation cannot push the chunk_index range down to the server.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	目标集合名称。The target collection name.
`document_id`	str	文档标识（类型 `str`）。The source document whose chunks should be listed.
`offset`kw默认default `0`	int	偏移量（类型 `int`）。Number of leading chunks to skip.
`limit`kw默认default `30`	int	数量上限（类型 `int`）。Maximum number of chunks to return.
`metadata_filter`kw默认default `None`	dict[str, Any] \| None	元数据过滤器（类型 `dict[str, Any] \| None`）。Extra `chunk.metadata` equality constraints.

返回Returns list[Chunk]

返回分块列表。`list[Chunk]`:

源码Source agentscope/rag/_vdb/_elasticsearch.py:295–388

继承成员Inherited Members

继承自Inherited from VectorStoreBase

__aenter__