类class agentscope.rag

MilvusLiteStore

class MilvusLiteStore(VectorStoreBase)

源码Source agentscope/rag/_vdb/_milvus_lite.py:26–566

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

Each knowledge base maps to one Milvus collection. Every entity stores the owning document_id, the serialized :class:~agentscope.rag.Chunk, and the chunk metadata in a JSON field used for flat equality filtering.

示例Example

store = MilvusLiteStore(uri="./rag_demo.db")

async with store:
    await store.create_collection("kb_1", dimensions=768)

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(uri: str = './agentscope_milvus_lite.db', metric_type: Literal['COSINE', 'IP', 'L2'] = 'COSINE', index_typ…`	初始化 `MilvusLiteStore` 实例。 Initialize the Milvus Lite vector store.

方法摘要Method Summary

方法Method	说明Description
`def get_client() -> 'MilvusClient'`	获取客户端。 Lazily create and cache the Milvus client.
`async def __aexit__(exc_type: type[BaseException] \| None, exc_val: BaseException \| None, exc_tb: Any) -> None`async	退出异步上下文并清理资源。 Exit the async context and close the underlying client.
`async def create_collection(name: str, dimensions: int) -> None`async	创建集合。 Create a new Milvus collection.
`async def delete_collection(name: str) -> None`async	删除集合。 Delete a collection and all its data.
`async def has_collection(name: str) -> bool`async	判断是否具有集合。 Check whether a collection exists.
`async def insert(collection: str, records: list[VectorRecord]) -> None`async	执行插入相关操作。 Insert records into a collection.
`async def delete(collection: str, document_id: str, metadata_filter: dict[str, Any] \| None = None) -> None`async	删除数据。 Delete all records 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	执行搜索相关操作。 Find the most similar records to a query vector.
`async def list_documents(collection: str, metadata_filter: dict[str, Any] \| None = None) -> list[DocumentSummary]`async	列出文档。 List all distinct source documents indexed in a collection.
`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

MilvusLiteStore

def __init__(uri: str = './agentscope_milvus_lite.db', metric_type: Literal['COSINE', 'IP', 'L2'] = 'COSINE', index_type: str = 'AUTOINDEX', client_kwargs: dict[str, Any] | None = None, batch_size: int = 256) -> None

初始化 MilvusLiteStore 实例。

Initialize the Milvus Lite vector store.

参数Parameters

名称Name	类型Type	说明Description
`uri`默认default `'./agentscope_milvus_lite.db'`	str	URI（类型 `str`）。Local `.db` path for Milvus Lite, or a Milvus-compatible endpoint URI.
`metric_type`默认default `'COSINE'`	Literal['COSINE', 'IP', 'L2']	指标类型（类型 `Literal['COSINE', 'IP', 'L2']`）。The metric used for the vector index and search.
`index_type`默认default `'AUTOINDEX'`	str	索引类型（类型 `str`）。The Milvus index type used for newly created collections.
`client_kwargs`默认default `None`	dict[str, Any] \| None	客户端关键字参数（类型 `dict[str, Any] \| None`）。Extra keyword arguments forwarded to :class:`pymilvus.MilvusClient`.
`batch_size`默认default `256`	int	batch 大小（类型 `int`）。Batch size used by inserts and document listing.

源码Source agentscope/rag/_vdb/_milvus_lite.py:45–76

方法详情Method Detail

get_client

def get_client() -> 'MilvusClient'

获取客户端。

Lazily create and cache the Milvus client.

返回Returns 'MilvusClient'

返回'MilvusClient'。Returns Milvus client.

源码Source agentscope/rag/_vdb/_milvus_lite.py:78–84

aexit

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

async

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

Exit the async context and close the underlying client.

参数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/_milvus_lite.py:86–102

create_collection

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

async

创建集合。

No-op if the collection already exists.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	名称（类型 `str`）。The collection name. Typically, the knowledge base ID.
`dimensions`	int	向量维度（类型 `int`）。The fixed vector dimensionality for this collection. All vectors inserted later must have this many elements.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_milvus_lite.py:108–175

delete_collection

async def delete_collection(name: str) -> None

async

删除集合。

Delete a collection and all its data.

参数Parameters

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

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_milvus_lite.py:177–182

has_collection

async def has_collection(name: str) -> bool

async

判断是否具有集合。

Check whether a collection exists.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/rag/_vdb/_milvus_lite.py:184–189

insert

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

async

执行插入相关操作。

Each entity stores the :attr:VectorRecord.document_id, the serialized :class:Chunk, and a copy of chunk.metadata in a JSON field used by :meth:search and :meth:list_documents for flat equality filtering.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	目标集合名称。The target collection name.
`records`	list[VectorRecord]	records（类型 `list[VectorRecord]`）。The records to insert, each carrying a :class:`Chunk` and its embedding vector.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_milvus_lite.py:195–237

delete

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

async

删除数据。

Delete all records 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/_milvus_lite.py:259–284

search

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

async

执行搜索相关操作。

Find the most similar records to a query vector.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	集合到搜索。The collection to search.
`query_vector`	list[float]	查询向量化向量。The query embedding vector.
`top_k`默认default `5`	int	返回条数（类型 `int`）。Maximum number of results to return.
`metadata_filter`默认default `None`	dict[str, Any] \| None	元数据过滤器（类型 `dict[str, Any] \| None`）。If provided, restrict the search to records whose `chunk.metadata` matches every `key == value` pair in this dict. Milvus Lite stores this metadata in a JSON field and supports flat scalar equality filters.

返回Returns list[VectorSearchResult]

返回VectorSearchResult列表。`list[VectorSearchResult]`:

源码Source agentscope/rag/_vdb/_milvus_lite.py:290–337

list_documents

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

async

列出文档。

Scans stored entities and aggregates them by document_id.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	目标集合名称。The target collection name.
`metadata_filter`默认default `None`	dict[str, Any] \| None	元数据过滤器（类型 `dict[str, Any] \| None`）。If provided, restrict aggregation to records whose `chunk.metadata` matches every `key == value` pair in this dict.

返回Returns list[DocumentSummary]

返回DocumentSummary列表。`list[DocumentSummary]`:

源码Source agentscope/rag/_vdb/_milvus_lite.py:343–396

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。

Selects the page with a JSON-path range expression offset <= chunk["chunk_index"] < offset + limit instead of sort-and-skip — Milvus query has no ORDER BY, but because chunk_index is dense the range filter yields exactly the requested page, which is then sorted in Python (…

参数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/_milvus_lite.py:450–511

继承成员Inherited Members

继承自Inherited from VectorStoreBase

__aenter__