变换（transformation）就是接收一组节点作为输入、并返回一组节点的东西。每个实现 Transformation 基类的组件都同时有同步的 __call__() 与异步的 acall() 定义。

目前以下组件属于 Transformation 对象：

• TextSplitter
• NodeParser
• MetadataExtractor
• Embeddingsmodel（查看我们支持的嵌入清单）

使用方式

变换最适合与 IngestionPipeline 一起使用，但也可以直接调用。

from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor

node_parser = SentenceSplitter(chunk_size=512)
extractor = TitleExtractor()

# use transforms directly
nodes = node_parser(documents)

# or use a transformation in async
nodes = await extractor.acall(nodes)

与索引结合

变换可以传入索引或全局设置，并在调用索引的 from_documents() 或 insert() 时生效。

from llama_index.core import VectorStoreIndex
from llama_index.core.extractors import (
    TitleExtractor,
    QuestionsAnsweredExtractor,
)
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import TokenTextSplitter

transformations = [
    TokenTextSplitter(chunk_size=512, chunk_overlap=128),
    TitleExtractor(nodes=5),
    QuestionsAnsweredExtractor(questions=3),
]

# global
from llama_index.core import Settings

Settings.transformations = [text_splitter, title_extractor, qa_extractor]

# per-index
index = VectorStoreIndex.from_documents(
    documents, transformations=transformations
)

自定义变换

只要实现基类，你就可以自己实现任意变换。

下面这个自定义变换会移除文本中的特殊字符与标点。

import re
from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.schema import TransformComponent

class TextCleaner(TransformComponent):
    def __call__(self, nodes, **kwargs):
        for node in nodes:
            node.text = re.sub(r"[^0-9A-Za-z ]", "", node.text)
        return nodes

之后它们既可以直接使用，也可以放进任意 IngestionPipeline。

# use in a pipeline
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TextCleaner(),
        OpenAIEmbedding(),
    ],
)

nodes = pipeline.run(documents=[Document.example()])