RAG

async with store:
    results = await knowledge.search(
        queries=["When do cats sleep?"],
        top_k=3,
        score_threshold=None,  # only meaningful for cosine / dot-product
    )
    for r in results:
        print(r.score, r.document_id, r.chunk.content)

# List every document (one DocumentSummary per document)
summaries = await knowledge.list_documents()
for s in summaries:
    print(s.document_id, s.source, s.chunk_count, s.metadata)

# Delete every chunk belonging to one document
await knowledge.delete_document(document_id)

knowledge = KnowledgeBase(
    name="tenant-a-kb",
    description="...",
    embedding_model=embedding_model,
    vector_store=store,
    collection="shared",
    metadata_filter={"tenant_id": "tenant-a"},
)

from agentscope.message import TextBlock
from agentscope.rag import ParserBase, Section


class MyMarkdownParser(ParserBase):
    supported_media_types = ["text/markdown"]

    async def parse(
        self,
        file: bytes | str,
        filename: str,
    ) -> list[Section]:
        text = file.decode("utf-8") if isinstance(file, bytes) else file
        # Split by H2 headings into multiple Sections, preserving the source
        return [
            Section(
                content=TextBlock(text=block),
                source=filename,
                metadata={"index": index},
            )
            for index, block in enumerate(text.split("\n## "))
        ]

from agentscope.message import TextBlock
from agentscope.rag import Chunk, ChunkerBase, Section


class FixedCharChunker(ChunkerBase):
    def __init__(self, chunk_size: int = 1000) -> None:
        self._chunk_size = chunk_size

    async def chunk(self, sections: list[Section]) -> list[Chunk]:
        chunks: list[Chunk] = []
        for section in sections:
            # Multimodal content is not split — pass through as a whole chunk
            if not isinstance(section.content, TextBlock):
                chunks.append(
                    Chunk(
                        content=section.content,
                        source=section.source,
                        chunk_index=0,
                        total_chunks=0,
                        metadata=dict(section.metadata),
                    ),
                )
                continue
            text = section.content.text
            for start in range(0, len(text), self._chunk_size):
                chunks.append(
                    Chunk(
                        content=TextBlock(
                            text=text[start : start + self._chunk_size],
                        ),
                        source=section.source,
                        chunk_index=0,
                        total_chunks=0,
                        metadata=dict(section.metadata),
                    ),
                )
        # Renumber consistently
        for index, chunk in enumerate(chunks):
            chunk.chunk_index = index
            chunk.total_chunks = len(chunks)
        return chunks

from typing import Any
from agentscope.rag import (
    DocumentSummary,
    VectorRecord,
    VectorSearchResult,
    VectorStoreBase,
)


class MyVectorStore(VectorStoreBase):
    async def __aenter__(self) -> "MyVectorStore":
        self._client = await connect_my_backend(...)
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        await self._client.close()

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

    async def delete_collection(self, name: str) -> None: ...

    async def has_collection(self, name: str) -> bool: ...

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

    async def delete(self, collection: str, document_id: str) -> None: ...

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

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