RAG

AgentScope 中的 RAG 由如下可独立替换的功能模块组成：

功能模块	描述
解析器 Parser	把原始文件拆分成若干 `Section` 对象，每个 `Section` 对应文件的一个自然边界（PDF 页、PPTX 幻灯片、Markdown 标题段、整张图片等）
切块器 Chunker	把 `Section` 切成最终入库的 `Chunk`，并保证不跨 `Section` 合并
嵌入模型 Embedding Model	把 `Chunk` 的文本或多模态内容嵌入为向量
向量库 Vector Store	连接向量数据库，存储 `Chunk` 的向量与元信息，并支持检索
知识库句柄 KnowledgeBase	把嵌入模型 + 向量库 + collection 绑定在一起，封装 `insert_document` / `search` / `list_documents` / `delete_document` 四个一站式接口

本章主要介绍在非服务化场景下使用 RAG 功能，包括索引文件、检索知识、集成到智能体等。

嵌入模型的介绍和配置方式请见嵌入模型章节；服务化版本的 RAG（带 HTTP 服务、文件托管、分布式索引）请见 RAG 服务。

现有实现

AgentScope 为每个模块提供了开箱即用的默认实现，全部基于基类继承，方便用户替换：

解析器

类	描述	支持文件类型
`TextParser`	文本数据解析器，整文件作为一个 `Section` 返回，由下游切块器切分	`text/plain` `text/markdown` `text/csv` `text/html` `text/x-rst` `application/json` `application/xml` `application/x-yaml`
`PDFParser`	PDF 解析器，每页一个 `Section`，元信息中携带从 1 开始的页码 `page` 字段。	`application/pdf`
`PPTParser`	PowerPoint (`.pptx`) 解析器，按幻灯片顺序遍历形状： - 文本/表格合并为同一 `Section`， - 图片作为独立 `DataBlock` 读取。元信息中携带从 1 开始的幻灯片序号 `slide` 字段。	`application/vnd.openxmlformats-officedocument.presentationml.presentation`
`WordParser`	Word（`.docx`）解析器，按文档元素顺序遍历： - 相邻段落（默认含表格）合并为同一文本 `Section`， - 内嵌图片作为独立 `DataBlock` 读取。	`application/vnd.openxmlformats-officedocument.wordprocessingml.document`
`ExcelParser`	Excel（`.xlsx` / `.xls`）解析器，按工作表顺序读取： - 每张表的内容渲染为 Markdown 或 JSON 文本， - 内嵌图片（`include_image=True` 时）作为独立 `DataBlock` 读取。 `separate_sheet=True` 时元信息携带 `sheet` 字段。	`application/vnd.openxmlformats-officedocument.spreadsheetml.sheet` `application/vnd.ms-excel`
`ImageParser`	图片解析器，将整张图片读取为一个 `Section`	`image/png` `image/jpeg` `image/gif` `image/bmp` `image/webp`

PDF、PPT、Word、Excel 解析依赖额外的第三方库，可以通过 pip install agentscope[rag] 一次性安装。

切块器

类	描述
`ApproxTokenChunker`	按近似 Token 数进行切分，不依赖任何 tokenizer。近似策略：`len(text.encode("utf-8")) // 4`；多模态 `DataBlock` 整块透传不切分
开发中 …

嵌入模型

请见嵌入模型章节。

向量数据库

类	描述
`QdrantStore`	基于 Qdrant 的向量数据库实现，支持内存（`location=":memory:"`）、本地磁盘（`path=...`）、远程服务（`url=...`）三种部署方式
`MilvusLiteStore`	基于 Milvus Lite 的向量数据库实现，支持本地持久化 `.db` 文件（`uri="./rag_demo.db"`）以及 Milvus 兼容的服务端 URI
`MongoDBStore`	基于 MongoDB Vector Search 的实现（Atlas 或自建集群），每个知识库对应指定数据库中的一个集合。`metadata_filter` 会用到的字段必须在构造时通过 `filter_fields` 声明，才会进入向量索引
`ElasticsearchStore`	基于 Elasticsearch 稠密向量索引的实现，使用近似 kNN 检索，每个知识库对应一个索引

Qdrant 支持已包含在 pip install agentscope[rag] 中。其余后端各有独立的可选依赖：agentscope[milvuslite]、agentscope[mongodb]、agentscope[elasticsearch]。

RAG 使用

AgentScope 推荐通过知识库句柄 KnowledgeBase 作为入口使用 RAG。它把嵌入模型、向量库、collection（以及可选的 metadata_filter 多租户隔离配置）绑定在一起，对外只暴露四个动作：

方法	描述
`insert_document(chunks, document_id=None, document_metadata=None)`	把一批 `Chunk` 作为同一个文档批量嵌入并写入，返回 `document_id`
`search(queries, top_k=5, score_threshold=None)`	对一组查询（`str` / `TextBlock` / `DataBlock`）执行向量检索，自动去重排序
`delete_document(document_id)`	按 `document_id` 删除整个文档的所有 chunk
`list_documents()`	返回当前知识库中所有文档的 `DocumentSummary` 列表

索引文件

索引文件需要经过 文件解析 → 切块 → 嵌入入库 三步，对应上述三个模块。下面展示端到端的索引流程。

解析文件

使用解析器的 parse 方法把原始文件读取成 Section 数组，每个 Section 对应文件的一个自然边界（PDF 页 / PPT 幻灯片 / 图片 …）。parse(file, filename) 的 file 参数同时支持 bytes 与 str：

bytes 直接作为原始文件内容；
str 在二进制解析器（PDFParser / PPTParser / WordParser / ExcelParser / ImageParser）中表示文件路径，解析器自动读盘；
str 在 TextParser 中按运行时判断——若该字符串指向一个存在的文件就当作路径读盘并按 encoding 解码，否则当作已解码的文本内容直接使用。

from agentscope.rag import TextParser

parser = TextParser()

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

# 2) 传入文件路径（存在的文件会自动读盘）
sections = await parser.parse(file="./cats.md", filename="cats.md")

from agentscope.rag import PDFParser

parser = PDFParser()

# 1) 直接传入文件路径
sections = await parser.parse(file="./report.pdf", filename="report.pdf")

# 2) 也可以直接传 bytes（例如来自 HTTP 上传 / blob 存储）
with open("report.pdf", "rb") as f:
    sections = await parser.parse(file=f.read(), filename="report.pdf")

# 每个 section.metadata 包含 {"page": N}

from agentscope.rag import PPTParser

parser = PPTParser(
    include_image=True,          # 是否抽取嵌入的图片
    separate_table=False,        # 是否把表格独立为单独的 Section
    table_format="markdown",     # 表格渲染格式："markdown" 或 "json"
    slide_prefix="<slide index={index}>",
    slide_suffix="</slide>",
)

# 文件路径
sections = await parser.parse(file="./deck.pptx", filename="deck.pptx")

from agentscope.rag import WordParser

parser = WordParser(
    include_image=True,        # 是否抽取嵌入的图片
    separate_table=False,      # 是否把表格独立为单独的 Section
    table_format="markdown",   # 表格渲染格式："markdown" 或 "json"
)

# 文件路径
sections = await parser.parse(file="./doc.docx", filename="doc.docx")

from agentscope.rag import ExcelParser

parser = ExcelParser(
    include_sheet_names=True,        # 是否把工作表名作为标题前置
    include_cell_coordinates=False,  # 是否携带 [A1] 这样的单元格坐标
    include_image=False,             # 是否抽取嵌入的图片
    separate_sheet=False,            # 是否让各工作表的内容互不混合
    table_format="markdown",         # 表格渲染格式："markdown" 或 "json"
)

# 文件路径
sections = await parser.parse(file="./table.xlsx", filename="table.xlsx")
# separate_sheet=True 时，各 section.metadata 携带 {"sheet": "<工作表名>"}

from agentscope.rag import ImageParser

parser = ImageParser()

# 文件路径
sections = await parser.parse(file="./cat.png", filename="cat.png")
# 整张图片包裹为一个 DataBlock，元信息记录 media_type

切分 Chunk

使用切块器的 chunk 方法把 Section 数组切成最终入库的 Chunk 数组。约定：不跨 Section 合并；多模态 DataBlock 整块透传；chunk_index 从 0 连续编号，total_chunks 在每个 chunk 上都是同一值。

from agentscope.rag import ApproxTokenChunker

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

写入知识库

构造一个 KnowledgeBase 句柄，把 chunk 数组写入即可——嵌入与写库由句柄一并完成。同一文档的所有 chunk 共享一个 document_id，便于后续按文档级别整体删除。

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 是一个 async context manager；进入时打开客户端连接
store = QdrantStore(location=":memory:")  # 或 url="http://..." 指向真实集群

async with store:
    knowledge = KnowledgeBase(
        name="demo-kb",
        description="A toy corpus.",
        embedding_model=embedding_model,
        vector_store=store,
        collection="demo-kb",
    )
    # backing 的 collection 会在首次操作时按 embedding_model.dimensions 自动建立
    document_id = await knowledge.insert_document(
        chunks,
        document_metadata={"filename": "cats.md"},
    )

KnowledgeBase 不会自己创建 / 关闭向量库连接，使用前需要先把 VectorStoreBase 实例置于 async with 上下文中。

如果希望使用其他后端，安装对应的可选依赖后，只需要替换向量库的构造方式：

from agentscope.rag import MilvusLiteStore

# 本地持久化 .db 文件；需要 agentscope[milvuslite]
store = MilvusLiteStore(uri="./rag_demo.db")

from agentscope.rag import MongoDBStore

# MongoDB Vector Search（Atlas 或自建集群）；需要 agentscope[mongodb]
store = MongoDBStore(
    uri="mongodb+srv://user:pass@cluster.mongodb.net",
    database="agentscope_rag",
    # metadata_filter 会用到的字段必须在这里声明，
    # 才会被纳入向量检索索引
    filter_fields=["chunk.metadata.tenant_id"],
)

from agentscope.rag import ElasticsearchStore

# 稠密向量索引 + 近似 kNN 检索；需要 agentscope[elasticsearch]
store = ElasticsearchStore(hosts="http://localhost:9200")

向量检索

通过 KnowledgeBase.search 直接传入查询字符串/TextBlock/DataBlock 即可，无需手动嵌入：

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

search 内部做了以下几件事：

过滤不可用查询：若绑定的嵌入模型 supports_multimodal == False，会静默丢弃 DataBlock 类型的查询；
批量嵌入：所有查询一次性批量嵌入，再并发地对 collection 检索；
去重：按 (document_id, chunk_index) 去重，保留每条记录的最高分；
截断：按分数降序排序后截断为 top_k。

返回结果是 VectorSearchResult 数组，每条记录包含 score、document_id 与命中 chunk。

文档管理

KnowledgeBase 暴露了文档级别的两个辅助方法：

# 列出所有文档（每个文档一个 DocumentSummary）
summaries = await knowledge.list_documents()
for s in summaries:
    print(s.document_id, s.source, s.chunk_count, s.metadata)

# 按 document_id 删除整个文档对应的所有 chunk
await knowledge.delete_document(document_id)

DocumentSummary 包含 document_id、原始文件名 source、chunk_count，以及由解析器/上传方写入第一条 chunk 的 metadata。

多租户隔离：`metadata_filter`

如果多个逻辑知识库需要共用同一个物理 collection，可以在构造 KnowledgeBase 时传入 metadata_filter（典型场景：每条记录都带一个 {"tenant_id": "..."} payload）：

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

metadata_filter 是深度防御机制：

search 与 list_documents 严格按这些 key == value 过滤记录——永远跨不出当前作用域；
insert_document 会强制覆盖每个 chunk 的同名 metadata 字段，从而即使解析器/调用方误写也不会让记录跨域泄漏。

None（默认值）表示禁用过滤，对应”每个知识库独占自己的 collection”的部署形态。

多模态支持

AgentScope 的 RAG 原生支持多模态数据的入库与检索，关键在于解析器与嵌入模型能力的匹配——前者要能把多模态文件解析成 DataBlock，后者要能直接对 DataBlock 做嵌入：

查看 Parser 支持的文件类型：每个 ParserBase 子类通过类属性 supported_media_types（IANA 媒体类型列表）声明能力，可直接读取或在 IDE 中自动补全。

>>> 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']

查看嵌入模型支持的模态：通过实例属性 embedding_model.supports_multimodal 判断模型是否能直接处理 DataBlock（图片 / 视频 / 音频）。
>>> embedding_model.supports_multimodal True

当解析器输出带有多模态内容的 Chunk、且 embedding_model.supports_multimodal == True 时，入库与检索链路无需额外配置即可工作。文本模型遇到 DataBlock 查询时会在 KnowledgeBase.search 内部被静默丢弃，不会报错。

集成到智能体

通过 RAGMiddleware 把检索接入智能体类 Agent 的推理-行动循环。中间件不拥有嵌入模型或向量库——它消费的是一组已经构造好的 KnowledgeBase 句柄，可以混合多个使用不同嵌入模型的知识库。 RAGMiddleware 支持两种工作模式（RAGMiddleware.Parameters.mode），可以单独使用，也可以叠加使用（同时挂两个不同 mode 的实例）：

模式	触发时机	检索关键词	注入方式
`"static"`	每次 reply 的首次推理前	reply 方法的输入消息作为检索关键词，支持多模态数据	检索结果被包装成 `HintBlock` 并注入到上下文中
`"agentic"`（默认）	模型自主调用检索工具	模型自主给出	暴露 `search_knowledge` 工具，由智能体自己判断何时检索、用什么 query

参数全部封装在嵌套的 RAGMiddleware.Parameters 模型中：

字段	默认值	描述
`mode`	`"agentic"`	集成模式，详见上表
`top_k`	`5`	单次检索返回的最大命中数；跨所有知识库与 query 输入去重后截断
`score_threshold`	`None`	命中相似度下限，仅在 cosine / dot-product 下有意义
`emit_hint_event`	`True`	`static` 模式下是否额外发出 `HintBlockEvent` 供前端展示命中片段
`persist_hint`	`False`	`static` 模式下注入块是否持久留在上下文（默认推理后即移除，避免污染下一轮）

此外，RAGMiddleware.list_tools() 在 agentic 模式下会返回一个 search_knowledge 工具——需要手动把它注册到智能体的 Toolkit 里，模型才能调用。该工具的描述里会自动列出已挂载的所有知识库的 name / description，模型也可以通过 knowledge_bases=[...] 参数把搜索限定到指定子集。通过如下代码给智能体实例配置 RAG 功能：

from agentscope.middleware import RAGMiddleware
from agentscope.tool import Toolkit

static_mw = RAGMiddleware(
    knowledges=[knowledge],            # 一个或多个 KnowledgeBase 句柄
    parameters=RAGMiddleware.Parameters(
        mode="static",
        top_k=3,
        emit_hint_event=False,
    ),
)

agent = Agent(
    name="static-agent",
    system_prompt="使用检索到的资料回答用户问题。",
    model=chat_model,
    toolkit=Toolkit(),
    middlewares=[static_mw],
)

from agentscope.middleware import RAGMiddleware
from agentscope.tool import Toolkit

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

# 注意：agentic 模式下需要手动把 search_knowledge 工具注入到 Toolkit
toolkit = Toolkit(tools=await agentic_mw.list_tools())

agent = Agent(
    name="agentic-agent",
    system_prompt="必要时调用 search_knowledge 工具查询资料。",
    model=chat_model,
    toolkit=toolkit,
    middlewares=[agentic_mw],
)

from agentscope.middleware import RAGMiddleware
from agentscope.tool import Toolkit

# static 自动给首轮一个背景；agentic 让模型在需要时再补检索
static_mw = RAGMiddleware(
    knowledges=[knowledge],
    parameters=RAGMiddleware.Parameters(mode="static", top_k=3),
)
agentic_mw = RAGMiddleware(
    knowledges=[knowledge],
    parameters=RAGMiddleware.Parameters(mode="agentic", top_k=3),
)

# 注意：agentic 模式下需要手动把 search_knowledge 工具注入到 Toolkit
toolkit = Toolkit(tools=await agentic_mw.list_tools())

agent = Agent(
    name="hybrid-agent",
    system_prompt="使用检索资料回答；必要时主动调用 search_knowledge 补充。",
    model=chat_model,
    toolkit=toolkit,
    middlewares=[static_mw, agentic_mw],
)

自定义拓展

RAG 的所有模块都采用基类继承的方式，用户可以自定义 Parser、Chunker、Embedding Model、Vector Store——只要继承对应基类、实现核心方法，就能被无缝接入上面的流水线。

欢迎贡献新的 Parser、Chunker、Vector Store 到 AgentScope 官方仓库！

自定义解析器

继承 ParserBase，在类属性 supported_media_types 中声明能处理的 IANA 媒体类型，并实现 async def parse(file, filename) 把字节流拆成若干 Section：

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
        # 按二级标题切成若干 Section，保留来源
        return [
            Section(
                content=TextBlock(text=block),
                source=filename,
                metadata={"index": index},
            )
            for index, block in enumerate(text.split("\n## "))
        ]

需要时也可以重写 supported_extensions()（默认由 supported_media_types 反查得到；如果你的解析器希望前端文件选择器只展示某几个扩展名，建议显式覆盖）。

自定义切块器

继承 ChunkerBase，实现 async def chunk(sections) 把若干 Section 切成入库的 Chunk。约定：不跨 Section 合并；多模态 DataBlock 整块透传；chunk_index 在结果内从 0 连续编号；total_chunks 在每个 chunk 上一致：

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:
            # 多模态内容不切分，整块透传
            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),
                    ),
                )
        # 统一编号
        for index, chunk in enumerate(chunks):
            chunk.chunk_index = index
            chunk.total_chunks = len(chunks)
        return chunks

自定义向量数据库

继承 VectorStoreBase，实现 create_collection / delete_collection / has_collection / insert / delete / search / list_documents，并通过 __aenter__ / __aexit__ 管理底层连接的生命周期：

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]: ...

实现要点：

delete 按 document_id 删除该文档的所有记录，调用方按文档为单位增删。
search / list_documents 必须把 metadata_filter 翻译成对应后端的 payload filter，以支持多租户隔离。
insert 时需要把 VectorRecord.document_id 与 chunk 都持久化下来——否则 delete 与 list_documents 都无法工作。

RAG 服务

多租户、分布式的 RAG 服务，支持 HTTP API、文件托管、向量数据库托管。

中间件

了解 RAGMiddleware 是如何嵌入 reply / reasoning 钩子的。

嵌入模型

可用的嵌入模型及其参数。

快速开始

核心模块

智能体即服务

其他

现有实现

解析器

切块器

嵌入模型

向量数据库

RAG 使用

索引文件

向量检索

文档管理

多租户隔离：`metadata_filter`

多模态支持

集成到智能体

自定义拓展

自定义解析器

自定义切块器

自定义向量数据库

延伸阅读

RAG 服务

中间件

嵌入模型

​现有实现

​解析器

​切块器

​嵌入模型

​向量数据库

​RAG 使用

​索引文件

​向量检索

​文档管理

​多租户隔离：metadata_filter

​多模态支持

​集成到智能体

​自定义拓展

​自定义解析器

​自定义切块器

​自定义向量数据库

​延伸阅读

RAG 服务

中间件

嵌入模型

现有实现

解析器

切块器

嵌入模型

向量数据库

RAG 使用

索引文件

向量检索

文档管理

多租户隔离：`metadata_filter`

多模态支持

集成到智能体

自定义拓展

自定义解析器

自定义切块器

自定义向量数据库

延伸阅读