> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentscope.io/llms.txt
> Use this file to discover all available pages before exploring further.

> ## Agent Instructions
> For AgentScope Python, use https://docs.agentscope.io/stable/en/index for new projects. For existing projects, check the installed agentscope version and use matching versioned documentation.
> The /latest/ alias points to the newest documentation and may move ahead of /stable/ when a development version exists. Do not mix AgentScope 1.x and 2.x APIs.
> State the AgentScope version when providing installation commands or code examples. ReMe uses its own continuously updated /reme/latest/ documentation.

# RAG

> Build retrieval-augmented generation (RAG) capabilities for agents.

In AgentScope, RAG is composed of the following **independently replaceable** modules:

| Module | Description |
| - | - |
| Parser | Splits a raw file into a list of `Section` objects, where each `Section` corresponds to a natural boundary in the source (PDF page, PPTX slide, Markdown heading block, whole image, etc.) |
| Chunker | Cuts `Section`s into the final `Chunk`s to be indexed; never merges across `Section`s |
| Embedding Model | Embeds a `Chunk`'s text or multimodal content into a vector |
| Vector Store | Connects to a vector database, stores `Chunk` vectors with metadata, and supports retrieval |
| KnowledgeBase handle | Binds together an embedding model, a vector store, and a collection, exposing `insert_document` / `search` / `list_documents` / `delete_document` as the one-stop entry point |

This chapter focuses on **using RAG in non-service scenarios** — indexing files, retrieving knowledge, and integrating with an agent.

> [!TIP]
>
> For embedding models and how to configure them, see the [Embedding Model chapter](https://docs.agentscope.io/en/versions/2.0.12dev/building-blocks/model/embedding.md); for the service version of RAG (with an HTTP service, file hosting, and distributed indexing), see [RAG Service](https://docs.agentscope.io/en/versions/2.0.12dev/deploy/rag.md).

## Existing Implementations

AgentScope ships out-of-the-box default implementations for every module, all inheriting from base classes so users can easily swap them out:

### Parser

| Class | Description | Supported File Types |
| - | - | - |
| `TextParser` | Text parser: the entire file is returned as a single `Section` and split downstream by a chunker | `text/plain` `text/markdown` `text/csv` `text/html` `text/x-rst` `application/json` `application/xml` `application/x-yaml` |
| `PDFParser` | PDF parser, **one `Section` per page**; the metadata carries a `page` field that starts at 1. | `application/pdf` |
| `PPTParser` | PowerPoint (`.pptx`) parser, walks slides in order: - text/tables are merged into the same `Section`, - images are read as standalone `DataBlock`s. The metadata carries a `slide` field that starts at 1. | `application/vnd.openxmlformats-officedocument.presentationml.presentation` |
| `WordParser` | Word (`.docx`) parser, walks the document element by element: - adjacent paragraphs (and, by default, tables) are merged into the same text `Section`, - embedded images are read as standalone `DataBlock`s. | `application/vnd.openxmlformats-officedocument.wordprocessingml.document` |
| `ExcelParser` | Excel (`.xlsx` / `.xls`) parser, reads the workbook sheet by sheet: - each sheet's table is rendered as Markdown or JSON text, - embedded images (with `include_image=True`) are read as standalone `DataBlock`s. With `separate_sheet=True` the metadata carries a `sheet` field. | `application/vnd.openxmlformats-officedocument.spreadsheetml.sheet` `application/vnd.ms-excel` |
| `ImageParser` | Image parser, reads the entire image as a single `Section` | `image/png` `image/jpeg` `image/gif` `image/bmp` `image/webp` |

> [!TIP]
>
> The PDF, PPT, Word, and Excel parsers depend on additional third-party libraries; install them in one shot with `pip install agentscope[rag]`.

### Chunker

| Class | Description |
| - | - |
| `ApproxTokenChunker` | Splits text by approximate token count, without depending on any tokenizer. Approximation strategy: `len(text.encode("utf-8")) // 4`; multimodal `DataBlock`s pass through unchanged. |
| In development ... | |

Chunkers declare their tunable options through a Pydantic parameter model. For example, use `ApproxTokenChunker.Parameters` to configure chunk size and overlap:

```python theme={null}
from agentscope.rag import ApproxTokenChunker

chunker = ApproxTokenChunker(
    parameters=ApproxTokenChunker.Parameters(
        chunk_size=256,
        overlap=32,
    ),
)
```

### Embedding Model

See the [Embedding Model chapter](https://docs.agentscope.io/en/versions/2.0.12dev/building-blocks/model/embedding.md).

### Vector Database

| Class | Description |
| - | - |
| `QdrantStore` | Qdrant-based vector database implementation, supporting in-memory (`location=":memory:"`), local disk (`path=...`), and remote service (`url=...`) deployments |
| `MilvusLiteStore` | Milvus Lite-based vector database implementation, supporting local persistent `.db` files (`uri="./rag_demo.db"`) and Milvus-compatible endpoint URIs |
| `MongoDBStore` | MongoDB Vector Search-based implementation (Atlas or self-hosted); each knowledge base maps to one collection in the configured database. Fields used in `metadata_filter` must be declared via `filter_fields` at construction so they enter the vector index |
| `ElasticsearchStore` | Elasticsearch-based implementation using dense-vector indexes with approximate kNN search; each knowledge base maps to one index |

> [!TIP]
>
> `agentscope[rag]` only carries the dependencies needed for document parsing. Each vector database has its own optional extra: `agentscope[vdb-qdrant]`, `agentscope[vdb-milvus]`, `agentscope[vdb-mongodb]`, and `agentscope[vdb-elasticsearch]`.

## Using RAG

AgentScope recommends going through the **`KnowledgeBase` handle** as the entry point for RAG. It binds an embedding model, a vector store, a collection (and an optional `metadata_filter` for multi-tenant isolation) together and exposes only four operations:

| Method | Description |
| - | - |
| `insert_document(chunks, document_id=None, document_metadata=None)` | Embeds and writes a batch of `Chunk`s as a single document; returns the `document_id` |
| `search(queries, top_k=5, score_threshold=None)` | Runs vector retrieval over a list of queries (`str` / `TextBlock` / `DataBlock`), with automatic deduplication and sorting |
| `delete_document(document_id)` | Removes every chunk of one document by its `document_id` |
| `list_documents()` | Returns a list of `DocumentSummary` entries for every document in this knowledge base |

### Indexing a File

Indexing a file goes through three steps — **file parsing → chunking → embedding + insertion** — one per module. The end-to-end flow:

**1. Parse the file**

Call the parser's `parse` method to read the raw file into a list of `Section`s, where each `Section` corresponds to a natural boundary in the source (PDF page / PPT slide / image …).

The `file` parameter of `parse(file, filename)` accepts both **`bytes`** and **`str`**:

* `bytes` is treated as the raw file content;
* `str` in a binary parser (`PDFParser` / `PPTParser` / `WordParser` / `ExcelParser` / `ImageParser`) is a **filesystem path** that the parser reads from disk for you;
* `str` in `TextParser` is disambiguated at runtime — if the string names an existing file it is read and decoded with the configured `encoding`; otherwise it is used verbatim as pre-decoded text.

```python Text file theme={null}
from agentscope.rag import TextParser

parser = TextParser()

# 1) Pass bytes directly
sections = await parser.parse(
    file=b"# Cats\nCats sleep 12-16 hours per day.\n",
    filename="cats.md",
)

# 2) Pass a file path (an existing file is read from disk)
sections = await parser.parse(file="./cats.md", filename="cats.md")
```

```python PDF theme={null}
from agentscope.rag import PDFParser

parser = PDFParser()

# 1) Pass a file path
sections = await parser.parse(file="./report.pdf", filename="report.pdf")

# 2) Or pass bytes directly (e.g. from an HTTP upload / blob store)
with open("report.pdf", "rb") as f:
    sections = await parser.parse(file=f.read(), filename="report.pdf")

# Each section.metadata contains {"page": N}
```

```python PowerPoint theme={null}
from agentscope.rag import PPTParser

parser = PPTParser(
    include_image=True,          # Whether to extract embedded images
    separate_table=False,        # Whether to emit tables as their own Section
    table_format="markdown",     # Table rendering: "markdown" or "json"
    slide_prefix="<slide index={index}>",
    slide_suffix="</slide>",
)

# File path
sections = await parser.parse(file="./deck.pptx", filename="deck.pptx")
```

```python Word theme={null}
from agentscope.rag import WordParser

parser = WordParser(
    include_image=True,        # Whether to extract embedded images
    separate_table=False,      # Whether to emit tables as their own Section
    table_format="markdown",   # Table rendering: "markdown" or "json"
)

# File path
sections = await parser.parse(file="./doc.docx", filename="doc.docx")
```

```python Excel theme={null}
from agentscope.rag import ExcelParser

parser = ExcelParser(
    include_sheet_names=True,        # Prepend each sheet's name as a header
    include_cell_coordinates=False,  # Include cell coordinates like [A1]
    include_image=False,             # Whether to extract embedded images
    separate_sheet=False,            # Keep sheets in separate Sections
    table_format="markdown",         # Table rendering: "markdown" or "json"
)

# File path
sections = await parser.parse(file="./table.xlsx", filename="table.xlsx")
# With separate_sheet=True, each section.metadata contains {"sheet": "<name>"}
```

```python Image theme={null}
from agentscope.rag import ImageParser

parser = ImageParser()

# File path
sections = await parser.parse(file="./cat.png", filename="cat.png")
# The entire image is wrapped as a single DataBlock; metadata records the media_type
```

**2. Split into Chunks**

Call the chunker's `chunk` method to turn the `Section` list into the final `Chunk` list to be indexed. Conventions: never merge across `Section`s; multimodal `DataBlock`s pass through as whole chunks; `chunk_index` runs consecutively from 0; every chunk carries the same `total_chunks`.

```python theme={null}
from agentscope.rag import ApproxTokenChunker

chunker = ApproxTokenChunker(
    parameters=ApproxTokenChunker.Parameters(
        chunk_size=256,
        overlap=32,
    ),
)
chunks = await chunker.chunk(sections)
```

**3. Write to the knowledge base**

Construct a `KnowledgeBase` handle and write the chunk list — embedding and storage are taken care of by the handle. All chunks of the same document share one `document_id`, which makes whole-document deletion easy.

```python theme={null}
from agentscope.credential import DashScopeCredential
from agentscope.embedding import DashScopeEmbeddingModel
from agentscope.rag import KnowledgeBase, QdrantStore

embedding_model = DashScopeEmbeddingModel(
    credential=DashScopeCredential(api_key="YOUR_API_KEY"),
    model="text-embedding-v4",
    dimensions=1024,
)

# QdrantStore is an async context manager; entering it opens the client connection
store = QdrantStore(location=":memory:")  # or url="http://..." for a real cluster

async with store:
    knowledge = KnowledgeBase(
        name="demo-kb",
        description="A toy corpus.",
        embedding_model=embedding_model,
        vector_store=store,
        collection="demo-kb",
    )
    # The backing collection is created on first use, sized to embedding_model.dimensions
    document_id = await knowledge.insert_document(
        chunks,
        document_metadata={"filename": "cats.md"},
    )
```

> [!NOTE]
>
> `KnowledgeBase` does not open or close the vector store connection itself; enter the `VectorStoreBase` instance in an `async with` block before using it.

To use another backend, install its extra and replace only the vector store construction:

```python Milvus Lite theme={null}
from agentscope.rag import MilvusLiteStore

# Local persistent .db file; requires agentscope[vdb-milvus]
store = MilvusLiteStore(uri="./rag_demo.db")
```

```python MongoDB theme={null}
from agentscope.rag import MongoDBStore

# MongoDB Vector Search (Atlas or self-hosted); requires agentscope[vdb-mongodb]
store = MongoDBStore(
    uri="mongodb+srv://user:pass@cluster.mongodb.net",
    database="agentscope_rag",
    # Fields used in metadata_filter must be declared here,
    # so they are included in the vector search index
    filter_fields=["chunk.metadata.tenant_id"],
)
```

```python Elasticsearch theme={null}
from agentscope.rag import ElasticsearchStore

# Dense-vector indexes with approximate kNN; requires agentscope[vdb-elasticsearch]
store = ElasticsearchStore(hosts="http://localhost:9200")
```

### Vector Retrieval

Call `KnowledgeBase.search` directly with a list of query strings / `TextBlock`s / `DataBlock`s — no manual embedding required:

```python theme={null}
async with store:
    results = await knowledge.search(
        queries=["When do cats sleep?"],
        top_k=3,
        score_threshold=None,  # minimum score to keep a hit; None keeps everything
    )
    for r in results:
        print(r.score, r.document_id, r.chunk.content)
```

`search` does the following internally:

1. **Drops unusable queries**: when the bound embedding model's `supports_multimodal == False`, `DataBlock` queries are silently dropped.
2. **Batched embedding**: every query is embedded in a single batch, then the collection is searched concurrently.
3. **Deduplication**: hits are deduplicated by `(document_id, chunk_index)` keeping the highest score.
4. **Truncation**: results are sorted by descending score and truncated to `top_k`.

The return value is a list of `VectorSearchResult`s; each entry carries `score`, `document_id`, and the matched `chunk`.

> [!NOTE]
>
> A higher `score` always means a better match: stores that use a distance metric (L2, for instance) negate it before returning, so scores from those knowledge bases are negative and `score_threshold` has to be negative too. Scores are comparable only within one knowledge base, never across knowledge bases built on different embedding models or metrics.

### Document Management

`KnowledgeBase` exposes two document-level helpers:

```python theme={null}
# 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)
```

`DocumentSummary` carries the `document_id`, the original filename `source`, `chunk_count`, and the `metadata` recorded on the first chunk by the parser / uploader.

### Multi-tenant Isolation: `metadata_filter`

When multiple logical knowledge bases need to share one physical collection, pass a `metadata_filter` when constructing the `KnowledgeBase` (a typical pattern is stamping every record with `{"tenant_id": "..."}`):

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

`metadata_filter` is a **defense-in-depth** mechanism:

* `search` and `list_documents` restrict records to those matching every `key == value` pair — nothing ever escapes the scope.
* `insert_document` **forces** the same metadata fields onto every chunk, so even a buggy or malicious parser cannot rebind a record into another scope.

`None` (the default) disables filtering — appropriate for deployments where every knowledge base owns its own collection outright.

### Multimodal Support

AgentScope's RAG natively supports the ingestion and retrieval of multimodal data — the key is matching the parser's and the embedding model's capabilities: the former must be able to parse multimodal files into `DataBlock`s, the latter must be able to embed `DataBlock`s directly.

* **Check which file types a Parser supports**: every `ParserBase` subclass declares its capability via the class attribute `supported_media_types` (a list of IANA media types), which you can read directly or auto-complete in your IDE.

  ```python theme={null}
  >>> from agentscope.rag import TextParser, ImageParser
  >>> TextParser.supported_media_types
  ['text/plain', 'text/markdown', 'text/csv', 'text/html', 'text/x-rst',
   'application/json', 'application/xml', 'application/x-yaml']
  >>> ImageParser.supported_media_types
  ['image/png', 'image/jpeg', 'image/gif', 'image/bmp', 'image/webp']
  ```

* **Check which modalities an embedding model supports**: the instance attribute `embedding_model.supports_multimodal` tells whether the model can directly handle `DataBlock`s (images / video / audio).

  ```python theme={null}
  >>> embedding_model.supports_multimodal
  True
  ```

When the parser yields `Chunk`s containing multimodal content and `embedding_model.supports_multimodal == True`, the ingestion and retrieval pipelines work without any extra configuration. Text-only models silently drop `DataBlock` queries inside `KnowledgeBase.search` instead of raising.

### Integrating with an Agent

`RAGMiddleware` plugs retrieval into the `Agent` class's reasoning-acting loop. The middleware does not own the embedding model or the vector store — it consumes **a list of pre-built `KnowledgeBase` handles**, which may mix knowledge bases that use different embedding models.

`RAGMiddleware` supports two working modes (`RAGMiddleware.Parameters.mode`), which can be used individually or **combined** (by attaching two instances with different `mode`s):

| Mode | Trigger | Retrieval Query | Injection |
| - | - | - | - |
| `"static"` | Before the **first reasoning step** of each reply (`agent.state.cur_iter == 0`) | The input message of the reply method is used as the retrieval query | Retrieval results are wrapped into a `HintBlock` and injected into the context |
| `"agentic"` (default) | The model invokes the retrieval tool on its own | Decided by the model itself | Exposes a `search_knowledge` tool — the agent decides when to retrieve and what query to use |

All parameters are wrapped in the nested `RAGMiddleware.Parameters` model:

| Field | Default | Description |
| - | - | - |
| `mode` | `"agentic"` | Integration mode, see above |
| `top_k` | `5` | Maximum number of hits returned in one search, deduplicated across knowledge bases and query inputs before truncation |
| `score_threshold` | `None` | Minimum score to keep a hit; `None` keeps everything. A higher score means a better match, and it is negative under a distance metric |
| `rerank_candidate_k` | `None` | How many candidates the rerank model judges before `top_k` of them are kept; must be no smaller than `top_k`, defaults to twice it, capped at 50. Ignored when no rerank model is configured |
| `emit_hint_event` | `True` | In `static` mode, whether to additionally emit a `HintBlockEvent` so the frontend can display the matched snippets |
| `persist_hint` | `False` | In `static` mode, whether the injected block stays persistently in the context (it is removed after reasoning by default, to avoid polluting the next turn) |

In addition, in `agentic` mode `RAGMiddleware.list_tools()` returns a single `search_knowledge` tool — you must manually register it in the agent's `Toolkit` so the model can call it. The tool's description automatically lists the `name` / `description` of every attached knowledge base; the model can also restrict a search to a subset via the `knowledge_bases=[...]` argument.

Configure RAG on an agent instance with the following code:

```python static mode theme={null}
from agentscope.middleware import RAGMiddleware
from agentscope.tool import Toolkit

static_mw = RAGMiddleware(
    knowledge_bases=[knowledge],       # One or more KnowledgeBase handles
    parameters=RAGMiddleware.Parameters(
        mode="static",
        top_k=3,
        emit_hint_event=False,
    ),
)

agent = Agent(
    name="static-agent",
    system_prompt="Answer the user's question using the retrieved material.",
    model=chat_model,
    toolkit=Toolkit(),
    middlewares=[static_mw],
)
```

```python agentic mode theme={null}
from agentscope.middleware import RAGMiddleware
from agentscope.tool import Toolkit

agentic_mw = RAGMiddleware(
    knowledge_bases=[knowledge],
    parameters=RAGMiddleware.Parameters(mode="agentic", top_k=3),
)

# Note: in agentic mode you must manually inject the search_knowledge tool into the Toolkit
toolkit = Toolkit(tools=await agentic_mw.list_tools())

agent = Agent(
    name="agentic-agent",
    system_prompt="When necessary, call the search_knowledge tool to look up material.",
    model=chat_model,
    toolkit=toolkit,
    middlewares=[agentic_mw],
)
```

```python combined modes theme={null}
from agentscope.middleware import RAGMiddleware
from agentscope.tool import Toolkit

# static gives the first turn some background automatically;
# agentic lets the model fetch more on demand
static_mw = RAGMiddleware(
    knowledge_bases=[knowledge],
    parameters=RAGMiddleware.Parameters(mode="static", top_k=3),
)
agentic_mw = RAGMiddleware(
    knowledge_bases=[knowledge],
    parameters=RAGMiddleware.Parameters(mode="agentic", top_k=3),
)

# Note: in agentic mode you must manually inject the search_knowledge tool into the Toolkit
toolkit = Toolkit(tools=await agentic_mw.list_tools())

agent = Agent(
    name="hybrid-agent",
    system_prompt="Answer using the retrieved material; call search_knowledge for more when needed.",
    model=chat_model,
    toolkit=toolkit,
    middlewares=[static_mw, agentic_mw],
)
```

### Rerank the Results

Vector retrieval ranks by embedding similarity, so the chunks that answer the question best are not always on top. Pass a `rerank_model` to `RAGMiddleware` and each retrieval first pulls back `rerank_candidate_k` candidates, then lets that model read them and pick the final `top_k`:

```python Configure a rerank model theme={null}
from agentscope.middleware import RAGMiddleware
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential

rag_mw = RAGMiddleware(
    knowledge_bases=[knowledge],
    rerank_model=DashScopeChatModel(       # the chat model that does the reranking
        credential=DashScopeCredential(api_key="YOUR_API_KEY"),
        model="qwen-max",
    ),
    parameters=RAGMiddleware.Parameters(
        mode="agentic",
        top_k=3,                # keep 3 hits in the end
        rerank_candidate_k=10,  # pull 10 candidates for the rerank model first
    ),
)
```

Reranking applies to both modes. What to keep in mind:

| Aspect | Description |
| - | - |
| Candidate count | `rerank_candidate_k` must be no smaller than `top_k`, defaults to `2 × top_k`, and is capped at 50; with too few candidates the rerank model can only reorder what vector retrieval already returned |
| Failure | Reranking is best-effort: a failed model call is logged as a warning and the vector retrieval order is used instead, without interrupting the retrieval |
| Multimodal candidates | Candidates the rerank model cannot read (a modality it does not support, for example) are pushed to the end or dropped |
| Prompt | `rerank_prompt` overrides the rerank instruction; it must contain the `{query}` placeholder and may contain `{top_k}`. The framework appends the candidate contents as content blocks after the instruction |

> [!WARNING]
>
> Reranking costs one extra LLM call per retrieval. The more and the longer the candidates, the higher the latency and cost, so start from the default `2 × top_k` and tune from there.

## Custom Extensions

All RAG modules use base-class inheritance, so users can customize Parser, Chunker, Embedding Model, and Vector Store — inherit from the corresponding base class, implement its core methods, and the custom class slots seamlessly into the pipeline above.

> [!TIP]
>
> Contributions of new Parsers, Chunkers, and Vector Stores to the official AgentScope repository are welcome!

### Custom Parser

Inherit from `ParserBase`, declare the IANA media types you can handle in the class attribute `supported_media_types`, and implement `async def parse(file, filename)` to split a byte stream into a list of `Section`s:

```python theme={null}
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## "))
        ]
```

You may also override `supported_extensions()` if needed (the default reverse-lookup from `supported_media_types` produces noisy developer extensions; override explicitly when you want the front-end file picker to show only a curated set).

### Custom Chunker

Inherit from `ChunkerBase` and implement `async def chunk(sections)` to turn a list of `Section`s into the `Chunk`s to be indexed. Conventions: never merge across `Section`s; multimodal `DataBlock`s pass through as whole chunks; `chunk_index` runs consecutively from 0 across the result list; `total_chunks` stays consistent on every chunk:

```python theme={null}
from pydantic import Field

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


class FixedCharChunker(ChunkerBase):
    chunker_type = "fixed_char"

    class Parameters(ChunkerBase.Parameters):
        chunk_size: int = Field(default=1000, ge=1)

    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.parameters.chunk_size):
                chunks.append(
                    Chunk(
                        content=TextBlock(
                            text=text[
                                start : start + self.parameters.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
```

`chunker_type` identifies the chunker for persistence and reconstruction, and must be unique within an application. The JSON Schema of `Parameters` is used directly for server-side configuration forms and parameter validation.

### Custom Vector Database

Inherit from `VectorStoreBase`, implement `create_collection` / `delete_collection` / `has_collection` / `insert` / `delete` / `search` / `list_documents`, and manage the underlying connection lifecycle through `__aenter__` / `__aexit__`:

```python theme={null}
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]: ...
```

Implementation notes:

* `delete` removes **every** record belonging to a `document_id`; callers add and remove documents as a unit.
* `search` and `list_documents` must translate `metadata_filter` into a backend-native payload filter so multi-tenant isolation works.
* `insert` must persist both `VectorRecord.document_id` and the `chunk` — otherwise `delete` and `list_documents` cannot work.

## Further Reading

> **[RAG Service](https://docs.agentscope.io/en/versions/2.0.12dev/deploy/rag.md)**
>
> A multi-tenant, distributed RAG service with HTTP API, file hosting, and managed vector databases.

> **[Middleware](https://docs.agentscope.io/en/versions/2.0.12dev/building-blocks/middleware.md)**
>
> See how `RAGMiddleware` plugs into the reply / reasoning hooks.

> **[Embedding Model](https://docs.agentscope.io/en/versions/2.0.12dev/building-blocks/model/embedding.md)**
>
> Available embedding models and their parameters.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.