> ## 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. # RAG > 为智能体构建检索增强生成（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 功能，包括索引文件、检索知识、集成到智能体等。 嵌入模型的介绍和配置方式请见[嵌入模型章节](/versions/2.0.3/zh/building-blocks/model)；服务化版本的 RAG（带 HTTP 服务、文件托管、分布式索引）请见 [RAG 服务](/versions/2.0.3/zh/deploy/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` | | `ImageParser` | 图片解析器，将整张图片读取为一个 `Section` | `image/png`
`image/jpeg`
`image/gif`
`image/bmp`
`image/webp` | | 开发中 ... | | | PDF 与 PPT 解析依赖额外的第三方库，可以通过 `pip install agentscope[rag]` 一次性安装。 ### 切块器 | 类 | 描述 | | -------------------- | --------------------------------------------------------------------------------------------------- | | `ApproxTokenChunker` | 按近似 Token 数进行切分，不依赖任何 tokenizer。
近似策略：`len(text.encode("utf-8")) // 4`；多模态 `DataBlock` 整块透传不切分 | | 开发中 ... | | ### 嵌入模型 请见[嵌入模型章节](/versions/2.0.3/zh/building-blocks/model)。 ### 向量数据库 | 类 | 描述 | | ------------- | ------------------------------------------------------------------------------------- | | `QdrantStore` | 基于 Qdrant 的向量数据库实现，支持内存（`location=":memory:"`）、本地磁盘（`path=...`）、远程服务（`url=...`）三种部署方式 | | 开发中 ... | | ## 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` / `ImageParser`）中表示**文件路径**，解析器自动读盘； * `str` 在 `TextParser` 中按运行时判断——若该字符串指向一个存在的文件就当作路径读盘并按 `encoding` 解码，否则当作已解码的文本内容直接使用。 ```python 文本文件 theme={null} 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") ``` ```python PDF theme={null} 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} ``` ```python PowerPoint theme={null} from agentscope.rag import PPTParser parser = PPTParser( include_image=True, # 是否抽取嵌入的图片 separate_table=False, # 是否把表格独立为单独的 Section table_format="markdown", # 表格渲染格式："markdown" 或 "json" slide_prefix="", slide_suffix="", ) # 文件路径 sections = await parser.parse(file="./deck.pptx", filename="deck.pptx") ``` ```python 图片 theme={null} from agentscope.rag import ImageParser parser = ImageParser() # 文件路径 sections = await parser.parse(file="./cat.png", filename="cat.png") # 整张图片包裹为一个 DataBlock，元信息记录 media_type ``` 使用切块器的 `chunk` 方法把 `Section` 数组切成最终入库的 `Chunk` 数组。约定：不跨 `Section` 合并；多模态 `DataBlock` 整块透传；`chunk_index` 从 0 连续编号，`total_chunks` 在每个 chunk 上都是同一值。 ```python theme={null} from agentscope.rag import ApproxTokenChunker chunker = ApproxTokenChunker(chunk_size=256, overlap=32) chunks = await chunker.chunk(sections) ``` 构造一个 `KnowledgeBase` 句柄，把 chunk 数组写入即可——嵌入与写库由句柄一并完成。同一文档的所有 chunk 共享一个 `document_id`，便于后续按文档级别整体删除。 ```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 是一个 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` 上下文中。 ### 向量检索 通过 `KnowledgeBase.search` 直接传入查询字符串/`TextBlock`/`DataBlock` 即可，无需手动嵌入： ```python theme={null} 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` 内部做了以下几件事： 1. **过滤不可用查询**：若绑定的嵌入模型 `supports_multimodal == False`，会静默丢弃 `DataBlock` 类型的查询； 2. **批量嵌入**：所有查询一次性批量嵌入，再并发地对 collection 检索； 3. **去重**：按 `(document_id, chunk_index)` 去重，保留每条记录的最高分； 4. **截断**：按分数降序排序后截断为 `top_k`。 返回结果是 `VectorSearchResult` 数组，每条记录包含 `score`、`document_id` 与命中 `chunk`。 ### 文档管理 `KnowledgeBase` 暴露了文档级别的两个辅助方法： ```python theme={null} # 列出所有文档（每个文档一个 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）： ```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` 是**深度防御**机制： * `search` 与 `list_documents` 严格按这些 `key == value` 过滤记录——永远跨不出当前作用域； * `insert_document` 会**强制覆盖**每个 chunk 的同名 metadata 字段，从而即使解析器/调用方误写也不会让记录跨域泄漏。 `None`（默认值）表示禁用过滤，对应"每个知识库独占自己的 collection"的部署形态。 ### 多模态支持 AgentScope 的 RAG 原生支持多模态数据的入库与检索，关键在于解析器与嵌入模型能力的匹配——前者要能把多模态文件解析成 `DataBlock`，后者要能直接对 `DataBlock` 做嵌入： * **查看 Parser 支持的文件类型**：每个 `ParserBase` 子类通过类属性 `supported_media_types`（IANA 媒体类型列表）声明能力，可直接读取或在 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'] ``` * **查看嵌入模型支持的模态**：通过实例属性 `embedding_model.supports_multimodal` 判断模型是否能直接处理 `DataBlock`（图片 / 视频 / 音频）。 ```python theme={null} >>> 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 功能： ```python static 模式 theme={null} 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], ) ``` ```python agentic 模式 theme={null} 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], ) ``` ```python 两种模式组合 theme={null} 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`： ```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 # 按二级标题切成若干 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 上一致： ```python theme={null} 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__` 管理底层连接的生命周期： ```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]: ... ``` 实现要点： * `delete` 按 `document_id` 删除该文档的**所有**记录，调用方按文档为单位增删。 * `search` / `list_documents` 必须把 `metadata_filter` 翻译成对应后端的 payload filter，以支持多租户隔离。 * `insert` 时需要把 `VectorRecord.document_id` 与 `chunk` 都持久化下来——否则 `delete` 与 `list_documents` 都无法工作。 ## 延伸阅读 多租户、分布式的 RAG 服务，支持 HTTP API、文件托管、向量数据库托管。 了解 `RAGMiddleware` 是如何嵌入 reply / reasoning 钩子的。 可用的嵌入模型及其参数。