Using PrevNextNodePostprocessor with a vector store?

[ShawonAshraf]

The default PrevNextNodePostprocessor implementation requires a doctsore to be passed as a default arg. However, if one uses a vector store for index, the docstore gets disabled. Is there a workaround to use the PrevNextNodePostprocessor with a vector store?

[dosubot[bot]]

Hello, @ShawonAshraf! I'm here to help you with any questions or issues you have. Let's work together to find a solution while we wait for a human maintainer.

Yes, there is a workaround to use the PrevNextNodePostprocessor with a vector store. You can create a custom implementation that mimics the behavior of the PrevNextNodePostprocessor but uses the vector store for fetching nodes instead of the document store.

Here is an example of how you might implement this:

class VectorStorePrevNextNodePostprocessor(BaseNodePostprocessor):
    """Previous/Next Node post-processor for vector store.

    Allows users to fetch additional nodes from the vector store,
    based on the relationships of the nodes.

    Args:
        vector_store (BaseVectorStore): The vector store.
        num_nodes (int): The number of nodes to return (default: 1)
        mode (str): The mode of the post-processor.
            Can be "previous", "next", or "both".
    """

    vector_store: BaseVectorStore
    num_nodes: int = Field(default=1)
    mode: str = Field(default="next")

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

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

    def _postprocess_nodes(
        self,
        nodes: List[NodeWithScore],
        query_bundle: Optional[QueryBundle] = None,
    ) -> List[NodeWithScore]:
        """Postprocess nodes."""
        all_nodes: Dict[str, NodeWithScore] = {}
        for node in nodes:
            all_nodes[node.node.node_id] = node
            if self.mode == "next":
                all_nodes.update(self.get_forward_nodes(node, self.num_nodes))
            elif self.mode == "previous":
                all_nodes.update(self.get_backward_nodes(node, self.num_nodes))
            elif self.mode == "both":
                all_nodes.update(self.get_forward_nodes(node, self.num_nodes))
                all_nodes.update(self.get_backward_nodes(node, self.num_nodes))
            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:
            node_inserted = False
            for i, cand in enumerate(sorted_nodes):
                node_id = node.node.node_id
                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
                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

    def get_forward_nodes(self, node_with_score: NodeWithScore, num_nodes: int) -> Dict[str, NodeWithScore]:
        """Get forward nodes from vector store."""
        node = node_with_score.node
        nodes: Dict[str, NodeWithScore] = {node.node_id: node_with_score}
        cur_count = 0
        while cur_count < num_nodes:
            if NodeRelationship.NEXT not in node.relationships:
                break

            next_node_info = node.next_node
            if next_node_info is None:
                break

            next_node_id = next_node_info.node_id
            next_node = self.vector_store.get_node(next_node_id)
            nodes[next_node.node_id] = NodeWithScore(node=next_node)
            node = next_node
            cur_count += 1
        return nodes

    def get_backward_nodes(self, node_with_score: NodeWithScore, num_nodes: int) -> Dict[str, NodeWithScore]:
        """Get backward nodes from vector store."""
        node = node_with_score.node
        nodes: Dict[str, NodeWithScore] = {node.node_id: node_with_score}
        cur_count = 0
        while cur_count < num_nodes:
            prev_node_info = node.prev_node
            if prev_node_info is None:
                break
            prev_node_id = prev_node_info.node_id
            prev_node = self.vector_store.get_node(prev_node_id)
            if prev_node is None:
                break
            nodes[prev_node.node_id] = NodeWithScore(node=prev_node)
            node = prev_node
            cur_count += 1
        return nodes

This custom postprocessor uses a vector_store instead of a docstore to fetch the previous and next nodes. You would need to implement the BaseVectorStore class and its get_node method to interact with your vector store.

For a practical example of how to use the PrevNextNodePostprocessor with a document store, you can refer to the test_forward_back_processor function in the llama-index-core/tests/postprocessor/test_base.py file ^[1].

To continue talking to Dosu, mention @dosu.

[ShawonAshraf]

This doesn't work with a QdrantVectorStore as QdrantVectorStore doesn't have a get_nodes implementation.

[dosubot[bot]]

To modify the QdrantVectorStore to work with the PrevNextNodePostprocessor, you need to ensure that the QdrantVectorStore has a get_nodes method implemented. Based on the provided context, the QdrantVectorStore already has a get_nodes method implemented. Here is the relevant code snippet:

def get_nodes(
    self,
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
) -> List[BaseNode]:
    """
    Get nodes from the index.

    Args:
        node_ids (Optional[List[str]]): List of node IDs to retrieve.
        filters (Optional[MetadataFilters]): Metadata filters to apply.

    Returns:
        List[BaseNode]: List of nodes retrieved from the index.
    """
    should = []
    if node_ids is not None:
        should = [
            HasIdCondition(
                has_id=node_ids,
            )
        ]

    if filters is not None:
        filter = self._build_subfilter(filters)
        if filter.should is None:
            filter.should = should
        else:
            filter.should.extend(should)
    else:
        filter = Filter(should=should)

    response = self._client.scroll(
        collection_name=self.collection_name,
        limit=9999,
        scroll_filter=filter,
    )

    return self.parse_to_query_result(response[0]).nodes

This method retrieves nodes from the Qdrant index based on the provided node IDs or metadata filters. Therefore, you do not need to modify the QdrantVectorStore further for it to work with the PrevNextNodePostprocessor ^[1].

The PrevNextNodePostprocessor class is designed to fetch additional nodes from the document store based on the relationships of the nodes. It supports modes like "previous", "next", or "both" to determine the direction of node retrieval. Here is a brief overview of its implementation:

class PrevNextNodePostprocessor(BaseNodePostprocessor):
    """Previous/Next Node post-processor.

    Allows users to fetch additional nodes from the document store,
    based on the relationships of the nodes.

    NOTE: this is a beta feature.

    Args:
        docstore (BaseDocumentStore): The document store.
        num_nodes (int): The number of nodes to return (default: 1)
        mode (str): The mode of the post-processor.
            Can be "previous", "next", or "both".

    """

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

    @validator("mode")
    def _validate_mode(cls, v: str) -> str:
        """Validate mode."""
        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]:
        """Postprocess nodes."""
        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:
            node_inserted = False
            for i, cand in enumerate(sorted_nodes):
                node_id = node.node.node_id
                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
                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

This class uses helper functions get_forward_nodes and get_backward_nodes to fetch nodes based on their relationships. The QdrantVectorStore's get_nodes method can be used within these helper functions to retrieve the necessary nodes from the Qdrant index ^[2][3].

To continue talking to Dosu, mention @dosu.

[ShawonAshraf]

This is just wrong. There's no such method implemented for QdrantVectorStore.

[dosubot[bot]]

To make QdrantVectorStore compatible with the PrevNextNodePostprocessor, you need to implement the get_nodes method. This method retrieves nodes from the index based on node IDs or metadata filters. Here is the implementation of the get_nodes method:

def get_nodes(
    self,
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
) -> List[BaseNode]:
    """
    Get nodes from the index.

    Args:
        node_ids (Optional[List[str]]): List of node IDs to retrieve.
        filters (Optional[MetadataFilters]): Metadata filters to apply.

    Returns:
        List[BaseNode]: List of nodes retrieved from the index.
    """
    should = []
    if node_ids is not None:
        should = [
            HasIdCondition(
                has_id=node_ids,
            )
        ]

    if filters is not None:
        filter = self._build_subfilter(filters)
        if filter.should is None:
            filter.should = should
        else:
            filter.should.extend(should)
    else:
        filter = Filter(should=should)

    response = self._client.scroll(
        collection_name=self.collection_name,
        limit=9999,
        scroll_filter=filter,
    )

    return self.parse_to_query_result(response[0]).nodes

This method uses the _build_subfilter method to construct the necessary filters and then uses the scroll method of the Qdrant client to retrieve the nodes. The parse_to_query_result method is used to convert the response into a list of BaseNode objects ^[1].

To continue talking to Dosu, mention @dosu.

[AutoScrape123TX]

@ShawonAshraf Hello I got the exact same issue did you find a workaround or implemented VectorStorePrevNextNodePostprocessor ?

[AutoScrape123TX]

Can you share it here please ? I also use Qdrant as it's the fastest out there. I struggle implementing manual node relationships and I'm disappointed viewing that VectorStore don't have such NodePostProcessor like : PrevNextNodePostprocessor or AutoPrevNextNodePostprocessor... I will send you a tip if you want

[ShawonAshraf]

Please give me a few moments. I'll create a gist and share post the link here :)

[AutoScrape123TX]

Thanks you a lot.

[ShawonAshraf]

You may have to add some debug logs and check if the previous and next nodes are actually being retrieved for your use case.

Also, a fun fact with qdrant is that every node comes with a relationships property which holds the node ids for the related nodes. You can simplify the forward and backward node getter using that.

https://gist.github.com/ShawonAshraf/a9f3b632cb16667213c0fa0ac341e9d1

[AutoScrape123TX]

Thanks you so much ! I struggle for this since 20 hours ^^ It was hard to understand that every node postprocessor are not tailored for vector store but raw document store