类class agentscope.rag

KnowledgeBase

class KnowledgeBase

源码Source agentscope/rag/_knowledge.py:44–454

KnowledgeBase 把嵌入模型、向量库和集合名绑定在一起，提供文档写入、向量检索、文档与分块管理。它既可以作为工具暴露给模型，也可以由 RAGMiddleware 自动注入检索结果。位于 agentscope.rag 包。核心成员包括 ensure_collection、search、insert_document、delete_document、list_documents、list_chunks。

Binds an embedding model and a vector-store collection together so callers can retrieve / insert / delete / list documents without repeating the wiring. Cheap to construct (no I/O); the collection itself is created lazily on the first operation, so a fresh deployment "just works" without an explicit setup step.

示例Example

kb = KnowledgeBase(
    name="company-handbook",
    description="Internal HR and onboarding documents.",
    embedding_model=embedding_model,
    vector_store=vector_store,
    collection="handbook",
)
await kb.insert_document(chunks)
results = await kb.search(["What is the PTO policy?"])

核心精讲Deep dive

KnowledgeBase：文档检索KnowledgeBase: document retrieval

KnowledgeBase 把嵌入模型、向量库和集合名绑定在一起，提供文档写入、向量检索、文档与分块管理。它既可以作为工具暴露给模型，也可以由 RAGMiddleware 自动注入检索结果。

KnowledgeBase binds an embedding model, a vector store and a collection name, providing document ingestion, vector search and document/chunk management. It can be exposed as a tool or used by RAGMiddleware to inject retrieved context.

insert_document 接收 Chunk 列表，并返回稳定的 document_id。insert_document accepts a list of Chunk objects and returns a stable document_id.
search 支持文本或 DataBlock 查询，可设置 top_k 与分数阈值。search accepts text or DataBlock queries and supports top_k and score thresholds.
list_documents、list_chunks 与 delete_document 覆盖知识库运维。list_documents, list_chunks and delete_document cover knowledge-base operations.

本地 Milvus Lite 知识库A local Milvus Lite knowledge base

import asyncio
import os

from agentscope.credential import OpenAICredential
from agentscope.embedding import OpenAIEmbeddingModel
from agentscope.message import TextBlock
from agentscope.rag import Chunk, KnowledgeBase, MilvusLiteStore


async def main() -> None:
    async with MilvusLiteStore(uri="./handbook.db") as store:
        kb = KnowledgeBase(
            name="handbook",
            description="Internal engineering handbook",
            embedding_model=OpenAIEmbeddingModel(
                credential=OpenAICredential(
                    api_key=os.environ["OPENAI_API_KEY"],
                ),
                model="text-embedding-3-small",
                dimensions=1536,
            ),
            vector_store=store,
            collection="handbook",
        )
        await kb.insert_document([
            Chunk(
                content=TextBlock(
                    type="text",
                    text="AgentScope uses a ReAct loop.",
                ),
                source="intro.md",
                chunk_index=0,
                total_chunks=1,
            )
        ])
        results = await kb.search(
            ["How does AgentScope reason?"],
            top_k=3,
        )
        for item in results:
            print(item)


asyncio.run(main())

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(name: str, description: str, embedding_model:` `EmbeddingModelBase, vector_store: VectorStoreBase, collectio…`	绑定嵌入模型、向量库、集合名与元数据过滤条件。 Initialize the runtime handle.

属性摘要Property Summary

属性Property	说明Description
`embedding_model`	获取向量化模型。 The bound embedding model.
`vector_store`	获取向量存储。 The bound vector store.
`collection`	获取集合。 The physical collection backing this knowledge base.
`metadata_filter`	获取元数据过滤器。 The defense-in-depth payload filter, or ``None``.

属性摘要Attribute Summary

属性Attribute	类型Type	说明Description	值Value
`name`	str	名称字段。	—
`description`	str	`description` 字段。	—

方法摘要Method Summary

方法Method	说明Description
`async def ensure_collection() -> None`async	确保底层向量集合已创建。 Idempotently create the backing collection if missing.
`async def search(queries: list[str \|` `TextBlock \| DataBlock], top_k: int = 5, score_threshold: float \| None = None) -> l…`async	对查询文本或多模态块执行相似度检索。 Search the knowledge base with one or more queries.
`async def insert_document(chunks: list[Chunk], document_id: str \| None = None, document_metadata: dict \| None = None) -…`async	写入一组文档分块并返回文档 ID。 Embed and insert a list of chunks as a single source document.
`async def delete_document(document_id: str) -> None`async	按文档 ID 删除全部相关分块。 Remove every record for one source document within this scope.
`async def list_documents() -> list['DocumentSummary']`async	列出知识库中的文档摘要。 List all distinct source documents in this knowledge base.
`async def list_chunks(document_id: str, *, offset: int = 0, limit: int = 30) -> list[Chunk]`async	分页列出指定文档的分块。 List one document's chunks ordered by ``chunk_index``.

构造器详情Constructor Detail

KnowledgeBase

def __init__(name: str, description: str, embedding_model: EmbeddingModelBase, vector_store: VectorStoreBase, collection: str, metadata_filter: dict | None = None) -> None

初始化 KnowledgeBase 实例。

Initialize the runtime handle.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	知识库名称。Agent-oriented knowledge base name. Surfaced to the LLM (via tool descriptions) and to the front-end.
`description`	str	知识库用途说明。Agent-oriented description. Should answer "what is in this knowledge base and when should I search it?" — the LLM uses it to decide whether to call the search tool in agentic mode.
`embedding_model`	EmbeddingModelBase	把文本转换为向量的嵌入模型。The embedding model used to embed both queries and inserted chunks. Must be the same model used at indexing time and at retrieval time, otherwise vectors will not be comparable.
`vector_store`	VectorStoreBase	存储向量与元数据的向量库。The shared vector-store connection. The store must already be entered (its own `__aenter__` already called) before any operation on this handle runs.
`collection`	str	向量库中的集合名。The physical collection backing this knowledge base. Created lazily on the first operation; see :meth:`ensure_collection`.
`metadata_filter`默认default `None`	dict \| None	检索时附加的元数据过滤条件。Defense-in-depth payload filter. When set: - :meth:`search` and :meth:`list_documents` restrict results to records whose payload matches every `key == value` pair; - :meth:`insert_document` forces these keys onto every inserted chunk's metadata, overriding caller-supplied values, so records cannot leak into another scope. - :meth:`delete_document` removes only records matching both the document id and every filter pair. `None` disables filtering — the default for deployments where every knowledge base owns its collection outright.

源码Source agentscope/rag/_knowledge.py:74–133

属性详情Attribute Detail

name

str

名称字段。声明为 str。

The name field.

源码Source agentscope/rag/_knowledge.py:66

description

str

description 字段。声明为 str。

The description field.

源码Source agentscope/rag/_knowledge.py:70

属性详情Property Detail

embedding_model

def embedding_model() -> EmbeddingModelBase

获取向量化模型。

The bound embedding model.

返回Returns EmbeddingModelBase

返回EmbeddingModelBase。Returns embedding model base.

源码Source agentscope/rag/_knowledge.py:140–142

vector_store

def vector_store() -> VectorStoreBase

获取向量存储。

The bound vector store.

返回Returns VectorStoreBase

返回VectorStoreBase。Returns vector store base.

源码Source agentscope/rag/_knowledge.py:145–147

collection

def collection() -> str

获取集合。

The physical collection backing this knowledge base.

返回Returns str

返回字符串结果。Returns a string.

源码Source agentscope/rag/_knowledge.py:150–152

metadata_filter

def metadata_filter() -> dict | None

获取元数据过滤器。

The defense-in-depth payload filter, or None.

返回Returns dict | None

返回dict|None。Returns dict none.

源码Source agentscope/rag/_knowledge.py:155–157

方法详情Method Detail

ensure_collection

async def ensure_collection() -> None

async

返回确保集合。

Called transparently at the top of every public operation — callers should not need to invoke it themselves.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_knowledge.py:163–191

search

async def search(queries: list[str | TextBlock | DataBlock], top_k: int = 5, score_threshold: float | None = None) -> list[VectorSearchResult]

async

执行搜索相关操作。

All queries are embedded in a single batch, then searched concurrently against the bound collection (with :attr:metadata_filter applied).

参数Parameters

名称Name	类型Type	说明Description
`queries`	list[str \| TextBlock \| DataBlock]	一个或多个查询文本或多模态块。Query inputs. Text may be either bare `str` or :class:`TextBlock`; :class:`DataBlock` items are silently dropped when the bound embedding model does not declare `supports_multimodal` — text-only models would otherwise reject them. Callers can therefore pass a mixed list without per-KB filtering.
`top_k`默认default `5`	int	返回的最大命中数量。Maximum number of results returned across all queries (after dedup).
`score_threshold`默认default `None`	float \| None	可选的相似度分数下限。Minimum :attr:`VectorSearchResult.score` for a hit to be retained. The scale depends on the store's metric — a distance metric reports negated distances, so the threshold is negative there. `None` disables filtering.

返回Returns list[VectorSearchResult]

按相关度排序的检索结果列表。`list[VectorSearchResult]`:

源码Source agentscope/rag/_knowledge.py:197–280

insert_document

async def insert_document(chunks: list[Chunk], document_id: str | None = None, document_metadata: dict | None = None) -> str

async

插入文档。

All chunks share the resolved document_id; :meth:delete_document later removes them as a unit.

参数Parameters

名称Name	类型Type	说明Description
`chunks`	list[Chunk]	分块列表（类型 `list[Chunk]`）。The pre-chunked document content (already produced by a parser + chunker pipeline). An empty list is a no-op.
`document_id`默认default `None`	str \| None	文档标识（类型 `str \| None`）。The document identifier. When `None` a fresh UUID hex is generated and returned so the caller can record it for future :meth:`delete_document` calls.
`document_metadata`默认default `None`	dict \| None	文档元数据（类型 `dict \| None`）。Document-level metadata (filename, media type, size, upload time, ...). Merged into each chunk's `metadata`.

返回Returns str

返回字符串结果。`str`:

异常Raises

可能抛出 `Exception`。`RuntimeError`:
可能抛出 `Exception`。If the embedding model returns a number of vectors
可能抛出 `Exception`。that does not match the number of chunks.

源码Source agentscope/rag/_knowledge.py:286–372

delete_document

async def delete_document(document_id: str) -> None

async

删除文档。

Records must also match :attr:metadata_filter when set.

参数Parameters

名称Name	类型Type	说明Description
`document_id`	str	文档标识（类型 `str`）。The source document id whose records should be removed.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/rag/_knowledge.py:374–398

list_documents

async def list_documents() -> list['DocumentSummary']

async

列出文档。

Filtered by :attr:metadata_filter when set, so callers only ever see documents within their own scope.

返回Returns list['DocumentSummary']

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

源码Source agentscope/rag/_knowledge.py:400–414

list_chunks

async def list_chunks(document_id: str, *, offset: int = 0, limit: int = 30) -> list[Chunk]

async

列出 list_chunks。

Filtered by :attr:metadata_filter when set, so callers only ever see chunks within their own scope.

参数Parameters

名称Name	类型Type	说明Description
`document_id`	str	文档标识（类型 `str`）。The source document whose chunks should be listed.
`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.

返回Returns list[Chunk]

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

异常Raises

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

源码Source agentscope/rag/_knowledge.py:416–454