使用 LlamaIndex 自定义 Transformation 组件实现节点元数据提取
使用 LlamaIndex 自定义 Transformation 组件实现节点元数据提取
文章目录
- 使用 LlamaIndex 自定义 Transformation 组件实现节点元数据提取
- 引言：什么是 Transformations？
- 一、Transformation 的基本概念
- 二、实战：自定义 PrefixExtractor 组件
- 三、代码解析与运行说明
引言：什么是 Transformations？
在构建基于文档的 RAG（Retrieval-Augmented Generation）系统时，数据处理流程中最终要的一步就是文档的摄取与转换。LlamaIndex 提供了一个强大的工具——IngestionPipeline，它允许我们将原始文档一步步转换为可用于检索和生成的结构化节点（Node）。
其中，Transformation 是整个管道最核心的概念之一。它是指一组节点作为输入，并返回节点列表的操作。每个实现 Transformation 基类的组件都包含同步 call() 定义或异步 acall() 定义的两种实现方式。
一、Transformation 的基本概念
- 接口定义
from llama_index.core.schema import TransformComponentclass MyCustomTransform(TransformComponent):def __call__(self, nodes, **kwargs):# 同步处理逻辑return nodesasync def acall(self, nodes, **kwargs):# 异步处理逻辑（可选）return nodes
__call__：这是每个Transformation必须实现的方法，用于同步地对输入的节点列表进行变换。acall()：如果你希望支持异步操作（例如调用远程API），可以实现该方法。
- 标准内置组件
LlamaIndex 提供了一些常用的TransformComponent实现，包括：
SentenceSplitter: 文本分块；TitleExtractor: 自动提取标题；Embedding模型: 生成向量嵌入；KeywordExtractor: 关键词抽取；
二、实战：自定义 PrefixExtractor 组件
接下来我们通过一个具体示例，展示如何继承 TransformComponenet 并实现一个自定义的节点处理器：PrefixExtractor, 它的作用是提取每个节点文本的前两个字符，并将其作为metadata 字段保存。
from llama_index.core import Document
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.dashscope import DashScopeEmbedding
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.schema import TransformComponentclass PrefixExtractor(TransformComponent):def __call__(self, nodes, **kwargs):for node in nodes:if len(node.text) >= 2:prefix = node.text[:2]node.metadata["prefix"] = prefixelse:node.metadata["prefix"] = node.textreturn nodesdocstore = SimpleDocumentStore()pipeline = IngestionPipeline(transformations=[SentenceSplitter(chunk_size=50,chunk_overlap=5,),PrefixExtractor(),DashScopeEmbedding(model="text-embedding-v3",api_key=os.getenv("ALI_API_KEY"),api_base=os.getenv("ALI_API_BASE"),),],docstore=docstore,
)nodes = pipeline.run(documents=[Document.example()]
)pipeline.persist("./pipeline_storage")for node in nodes:print("Text: ", node.text)print("Metadata: ", node.metadata)
三、代码解析与运行说明
PrefixExtractor类详解
- 我们定义了一个类
PrefixExtractor，继承自TransformComponent； - 在
__call__方法中遍历所有传入的节点； - 在每个节点提取其
.text属性的前两个字符，并存入.metadata中； - 最后返回更新后的节点列表。
- 构建
IngestionPipeline
- 首先使用
SentenceSplitter将文档分割成小块； - 然后通过我们自定义的
PrefixExtractor添加前缀字段； - 在使用
DashScopeEmbedding为每个节点生成向量； - 最后奖结果持久化到本地文件夹中。
- 输出结果实例
- 假设节点的内容为：
这是一个测试文档。
则输出部分内容如下：
{"text": "这是一个测试文档。","metadata": {"prefix": "这是"}
}