节点后处理器

概念

节点后处理器是一组模块，它们接收一组节点，并在返回节点之前应用某种转换或过滤。

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

LlamaIndex 提供了几种可立即使用的节点后处理器，同时还提供了一个简单的 API，用于添加自定义后处理器。

提示

对节点后处理器在流水线中的位置感到困惑？阅读有关高级概念的内容

使用模式

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

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:
    """节点后处理器。"""

    @abstractmethod
    def _postprocess_nodes(
        self, nodes: List[NodeWithScore], query_bundle: Optional[QueryBundle]
    ) -> List[NodeWithScore]:
        """后处理节点。"""

一个虚拟节点后处理器可以用几行代码实现：

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

模块

查看完整的模块列表以获取更多详细信息。