类class agentscope.rag

VectorStoreBase

class VectorStoreBase(ABC)

源码Source agentscope/rag/_vdb/_vector_store.py:104–365

向量存储基类，把私有文档转化为可检索的知识。位于 agentscope.rag 包，继承自 ABC。核心成员包括 create_collection、delete_collection、has_collection、insert、delete、search。

Subclasses implement the concrete connection and query logic for a specific vector database (Chroma, Milvus, Qdrant, FAISS, etc.).

A single instance is shared across the entire application. The underlying client SDK is expected to handle connection pooling and thread safety internally.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 Enter the async context — open connections if needed.
`async def __aexit__(exc_type: type[BaseException] \| None, exc_val: BaseException \| None, exc_tb: Any) -> None`async	退出异步上下文并清理资源。 Exit the async context — close connections if needed.
`async def create_collection(name: str, dimensions: int) -> None`asyncabstractabstractmethod	创建新的集合(向量索引)。 Create a new collection (vector index).
`async def delete_collection(name: str) -> None`asyncabstractabstractmethod	删除集合。 Delete a collection and all its data.
`async def has_collection(name: str) -> bool`asyncabstractabstractmethod	判断是否具有集合。 Check whether a collection exists.
`async def insert(collection: str, records: list[VectorRecord]) -> None`asyncabstractabstractmethod	执行插入相关操作。 Insert records into a collection.
`async def delete(collection: str, document_id: str, metadata_filter: dict[str, Any] \| None = None) -> None`asyncabstractabstractmethod	删除数据。 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…`asyncabstractabstractmethod	执行搜索相关操作。 Find the most similar records to a query vector.
`async def list_documents(collection: str, metadata_filter: dict[str, Any] \| None = None) -> list[DocumentSummary]`asyncabstractabstractmethod	列出文档。 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 the chunks of one source document in ``chunk_index`` order.

方法详情Method Detail

aenter

async def __aenter__() -> Self

async

进入异步上下文并返回自身。

The default implementation is a no-op.

返回Returns Self

返回自身。`VectorStoreBase`: `self`.

源码Source agentscope/rag/_vdb/_vector_store.py:119–128

aexit

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

async

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

The default implementation is a no-op.

参数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/_vector_store.py:130–139

create_collection

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

asyncabstractabstractmethod

创建新的集合(向量索引)。

If the collection already exists, implementations should raise or silently no-op depending on the backend's semantics.

参数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/_vector_store.py:146–162

delete_collection

async def delete_collection(name: str) -> None

asyncabstractabstractmethod

删除集合。

Delete a collection and all its data.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	集合名称到删除。The collection name to delete.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_vector_store.py:165–171

has_collection

async def has_collection(name: str) -> bool

asyncabstractabstractmethod

判断是否具有集合。

Check whether a collection exists.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	集合名称到检查。The collection name to check.

返回Returns bool

返回布尔值，表示判断结果。`bool`: `True` if the collection exists.

源码Source agentscope/rag/_vdb/_vector_store.py:174–183

insert

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

asyncabstractabstractmethod

执行插入相关操作。

Insert records into a collection.

参数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/_vector_store.py:190–203

delete

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

asyncabstractabstractmethod

删除数据。

Identifies records by the :attr:VectorRecord.document_id field that backends persist at insertion time.

参数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 whose chunk metadata matches every `key == value` pair in addition to the document ID. `None` or an empty dict deletes all records of the document. Backends must not ignore a non-empty filter.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_vdb/_vector_store.py:206–230

search

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

asyncabstractabstractmethod

执行搜索相关操作。

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. Backends translate this into a native payload filter. Used for defense-in-depth cross-tenant scoping when an isolation strategy co-locates multiple knowledge bases inside the same collection.

返回Returns list[VectorSearchResult]

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

源码Source agentscope/rag/_vdb/_vector_store.py:237–268

list_documents

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

asyncabstractabstractmethod

列出文档。

Aggregates records by :attr:VectorRecord.document_id and returns one :class:DocumentSummary per document.

参数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. Used together with the search-time filter when an isolation strategy co-locates multiple knowledge bases inside the same collection.

返回Returns list[DocumentSummary]

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

源码Source agentscope/rag/_vdb/_vector_store.py:275–302

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。

Returns the slice [offset, offset + limit) of the document's chunks, ordered by :attr:Chunk.chunk_index ascending.

参数Parameters

名称Name	类型Type	说明Description
`collection`	str	目标集合名称。The target collection name.
`document_id`	str	文档标识（类型 `str`）。The source document whose chunks should be listed — the same value carried by :attr:`VectorRecord.document_id` at insertion time.
`offset`kw默认default `0`	int	偏移量（类型 `int`）。Number of leading chunks to skip; equals the first returned chunk's `chunk_index`.
`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`）。If provided, restrict listing to records whose `chunk.metadata` matches every `key == value` pair in this dict. Used for defense-in-depth cross-tenant scoping when an isolation strategy co-locates multiple knowledge bases inside the same collection.

返回Returns list[Chunk]

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

异常Raises

可能抛出 `Exception`。`NotImplementedError`:
可能抛出 `Exception`。If the backend does not support chunk listing.

源码Source agentscope/rag/_vdb/_vector_store.py:308–365