节点后处理器

概念

节点后处理器是一组模块，它们接收一组节点，在返回节点前进行某种转换或过滤操作。

在LlamaIndex中，节点后处理器最常用于查询引擎中，位于节点检索步骤之后和响应合成步骤之前。

LlamaIndex提供了多个开箱即用的节点后处理器，同时也提供了简单的API用于添加自定义后处理器。

Tip

不清楚节点后处理器在RAG工作流中的位置？请阅读高层概念

使用模式

以下是使用节点后处理器的示例：

from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.postprocessor.cohere_rerank import CohereRerank
from llama_index.core.data_structs import Node
from llama_index.core.schema import NodeWithScore

nodes = [
    NodeWithScore(node=Node(text="text1"), score=0.7),
    NodeWithScore(node=Node(text="text2"), score=0.8),
]

# 相似度后处理器：过滤相似度分数低于0.75的节点
processor = SimilarityPostprocessor(similarity_cutoff=0.75)
filtered_nodes = processor.postprocess_nodes(nodes)

# cohere重排序：使用训练模型根据查询重排节点
reranker = CohereRerank(api_key="<COHERE_API_KEY>", top_n=2)
reranker.postprocess_nodes(nodes, query_str="<user_query>")

注意postprocess_nodes可以接收query_str或query_bundle(QueryBundle)，但不能同时接收两者。

使用模式

最常见的用法是在查询引擎中使用节点后处理器，它们会被应用于检索器返回的节点上，并在响应合成步骤之前执行。

与查询引擎配合使用

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.postprocessor import TimeWeightedPostprocessor

documents = SimpleDirectoryReader("./data").load_data()

index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine(
    node_postprocessors=[
        TimeWeightedPostprocessor(
            time_decay=0.5, time_access_refresh=False, top_k=1
        )
    ]
)

# 每次查询时都会应用所有节点后处理器
response = query_engine.query("query string")

与检索节点配合使用

也可以作为独立对象用于过滤检索到的节点：

from llama_index.core.postprocessor import SimilarityPostprocessor

nodes = index.as_retriever().retrieve("test query str")

# 过滤相似度分数低于0.75的节点
processor = SimilarityPostprocessor(similarity_cutoff=0.75)
filtered_nodes = processor.postprocess_nodes(nodes)

与自定义节点配合使用

如你所见，后处理器接收NodeWithScore对象作为输入，这只是一个包含Node和score值的包装类。

from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.core.data_structs import Node
from llama_index.core.schema import NodeWithScore

nodes = [
    NodeWithScore(node=Node(text="text"), score=0.7),
    NodeWithScore(node=Node(text="text"), score=0.8),
]

# 过滤相似度分数低于0.75的节点
processor = SimilarityPostprocessor(similarity_cutoff=0.75)
filtered_nodes = processor.postprocess_nodes(nodes)

(custom-node-postprocessor)=

自定义节点后处理器

基类是BaseNodePostprocessor，其API接口非常简单：

class BaseNodePostprocessor:
    """Node postprocessor."""

    @abstractmethod
    def _postprocess_nodes(
        self, nodes: List[NodeWithScore], query_bundle: Optional[QueryBundle]
    ) -> List[NodeWithScore]:
        """Postprocess nodes."""

只需几行代码即可实现一个虚拟的节点后处理器：

from llama_index.core import QueryBundle
from llama_index.core.postprocessor.types import BaseNodePostprocessor
from llama_index.core.schema import NodeWithScore


class DummyNodePostprocessor(BaseNodePostprocessor):
    def _postprocess_nodes(
        self, nodes: List[NodeWithScore], query_bundle: Optional[QueryBundle]
    ) -> List[NodeWithScore]:
        # 将所有节点的分数减1
        for n in nodes:
            n.score -= 1

        return nodes

模块

查看完整的模块列表获取更多细节。