Prev next

节点后处理器模块。

PrevNextNodePostprocessor

Bases: BaseNodePostprocessor

上一个/下一个节点后处理器。

允许用户根据节点的关系，从文档存储中获取额外的节点。

注意：这是一个测试版功能。

Parameters:

Name	Type	Description	Default
docstore	BaseDocumentStore	文档存储。	required
num_nodes	int	要返回的节点数（默认值：1）	required
mode	str	后处理器的模式。 可以是"previous"（上一个）、"next"（下一个）或"both"（两者）。	required

Source code in llama_index/core/postprocessor/node.py

class PrevNextNodePostprocessor(BaseNodePostprocessor):
    """上一个/下一个节点后处理器。

    允许用户根据节点的关系，从文档存储中获取额外的节点。

    注意：这是一个测试版功能。

    Args:
        docstore (BaseDocumentStore): 文档存储。
        num_nodes (int): 要返回的节点数（默认值：1）
        mode (str): 后处理器的模式。
            可以是"previous"（上一个）、"next"（下一个）或"both"（两者）。
"""

    docstore: BaseDocumentStore
    num_nodes: int = Field(default=1)
    mode: str = Field(default="next")

    @validator("mode")
    def _validate_mode(cls, v: str) -> str:
        """验证模式。"""
        if v not in ["next", "previous", "both"]:
            raise ValueError(f"Invalid mode: {v}")
        return v

    @classmethod
    def class_name(cls) -> str:
        return "PrevNextNodePostprocessor"

    def _postprocess_nodes(
        self,
        nodes: List[NodeWithScore],
        query_bundle: Optional[QueryBundle] = None,
    ) -> List[NodeWithScore]:
        """后处理节点。"""
        all_nodes: Dict[str, NodeWithScore] = {}
        for node in nodes:
            all_nodes[node.node.node_id] = node
            if self.mode == "next":
                all_nodes.update(get_forward_nodes(node, self.num_nodes, self.docstore))
            elif self.mode == "previous":
                all_nodes.update(
                    get_backward_nodes(node, self.num_nodes, self.docstore)
                )
            elif self.mode == "both":
                all_nodes.update(get_forward_nodes(node, self.num_nodes, self.docstore))
                all_nodes.update(
                    get_backward_nodes(node, self.num_nodes, self.docstore)
                )
            else:
                raise ValueError(f"Invalid mode: {self.mode}")

        all_nodes_values: List[NodeWithScore] = list(all_nodes.values())
        sorted_nodes: List[NodeWithScore] = []
        for node in all_nodes_values:
            # variable to check if cand node is inserted
            node_inserted = False
            for i, cand in enumerate(sorted_nodes):
                node_id = node.node.node_id
                # prepend to current candidate
                prev_node_info = cand.node.prev_node
                next_node_info = cand.node.next_node
                if prev_node_info is not None and node_id == prev_node_info.node_id:
                    node_inserted = True
                    sorted_nodes.insert(i, node)
                    break
                # append to current candidate
                elif next_node_info is not None and node_id == next_node_info.node_id:
                    node_inserted = True
                    sorted_nodes.insert(i + 1, node)
                    break

            if not node_inserted:
                sorted_nodes.append(node)

        return sorted_nodes