Using PrevNextNodePostprocessor with a vector store? #13930
Replies: 2 comments 10 replies
-
|
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 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 nodesThis custom postprocessor uses a For a practical example of how to use the |
Beta Was this translation helpful? Give feedback.
-
|
This doesn't work with a |
Beta Was this translation helpful? Give feedback.
-
|
To modify the 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]).nodesThis method retrieves nodes from the Qdrant index based on the provided node IDs or metadata filters. Therefore, you do not need to modify the The 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_nodesThis class uses helper functions |
Beta Was this translation helpful? Give feedback.
-
|
This is just wrong. There's no such method implemented for |
Beta Was this translation helpful? Give feedback.
-
|
To make 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]).nodesThis method uses the |
Beta Was this translation helpful? Give feedback.
-
|
@ShawonAshraf Hello I got the exact same issue did you find a workaround or implemented |
Beta Was this translation helpful? Give feedback.
-
|
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 : |
Beta Was this translation helpful? Give feedback.
-
|
Please give me a few moments. I'll create a gist and share post the link here :) |
Beta Was this translation helpful? Give feedback.
-
|
Thanks you a lot. |
Beta Was this translation helpful? Give feedback.
-
|
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 |
Beta Was this translation helpful? Give feedback.
-
|
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 |
Beta Was this translation helpful? Give feedback.
-
The default
PrevNextNodePostprocessorimplementation 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 thePrevNextNodePostprocessorwith a vector store?Beta Was this translation helpful? Give feedback.
All reactions