Auto prev next

节点后处理器模块。

AutoPrevNextNodePostprocessor

Bases: BaseNodePostprocessor

前/后节点后处理器。

允许用户从文档存储中获取额外的节点，基于节点之间的前/后关系。

注意：与PrevNextPostprocessor的区别在于，这个推断了前进/后退的方向。

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

Parameters:

Name	Type	Description	Default
docstore	BaseDocumentStore	文档存储。	required
num_nodes	int	要返回的节点数（默认值：1）	required
infer_prev_next_tmpl	str	用于推断的模板。 必需字段为{context_str}和{query_str}。	required

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

class AutoPrevNextNodePostprocessor(BaseNodePostprocessor):
    """前/后节点后处理器。

允许用户从文档存储中获取额外的节点，基于节点之间的前/后关系。

注意：与PrevNextPostprocessor的区别在于，这个推断了前进/后退的方向。

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

Args:
    docstore (BaseDocumentStore): 文档存储。
    num_nodes (int): 要返回的节点数（默认值：1）
    infer_prev_next_tmpl (str): 用于推断的模板。
        必需字段为{context_str}和{query_str}。"""

    docstore: BaseDocumentStore
    service_context: Optional[ServiceContext] = None
    llm: Optional[LLM] = None
    num_nodes: int = Field(default=1)
    infer_prev_next_tmpl: str = Field(default=DEFAULT_INFER_PREV_NEXT_TMPL)
    refine_prev_next_tmpl: str = Field(default=DEFAULT_REFINE_INFER_PREV_NEXT_TMPL)
    verbose: bool = Field(default=False)
    response_mode: ResponseMode = Field(default=ResponseMode.TREE_SUMMARIZE)

    class Config:
        """此为pydantic对象的配置。"""

        arbitrary_types_allowed = True

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

    def _parse_prediction(self, raw_pred: str) -> str:
        """解析预测。"""
        pred = raw_pred.strip().lower()
        if "previous" in pred:
            return "previous"
        elif "next" in pred:
            return "next"
        elif "none" in pred:
            return "none"
        raise ValueError(f"Invalid prediction: {raw_pred}")

    def _postprocess_nodes(
        self,
        nodes: List[NodeWithScore],
        query_bundle: Optional[QueryBundle] = None,
    ) -> List[NodeWithScore]:
        """后处理节点。"""
        llm = self.llm or llm_from_settings_or_context(Settings, self.service_context)

        if query_bundle is None:
            raise ValueError("Missing query bundle.")

        infer_prev_next_prompt = PromptTemplate(
            self.infer_prev_next_tmpl,
        )
        refine_infer_prev_next_prompt = PromptTemplate(self.refine_prev_next_tmpl)

        all_nodes: Dict[str, NodeWithScore] = {}
        for node in nodes:
            all_nodes[node.node.node_id] = node
            # use response builder instead of llm directly
            # to be more robust to handling long context
            response_builder = get_response_synthesizer(
                llm=llm,
                text_qa_template=infer_prev_next_prompt,
                refine_template=refine_infer_prev_next_prompt,
                response_mode=self.response_mode,
            )
            raw_pred = response_builder.get_response(
                text_chunks=[node.node.get_content()],
                query_str=query_bundle.query_str,
            )
            raw_pred = cast(str, raw_pred)
            mode = self._parse_prediction(raw_pred)

            logger.debug(f"> Postprocessor Predicted mode: {mode}")
            if self.verbose:
                print(f"> Postprocessor Predicted mode: {mode}")

            if mode == "next":
                all_nodes.update(get_forward_nodes(node, self.num_nodes, self.docstore))
            elif mode == "previous":
                all_nodes.update(
                    get_backward_nodes(node, self.num_nodes, self.docstore)
                )
            elif mode == "none":
                pass
            else:
                raise ValueError(f"Invalid mode: {mode}")

        sorted_nodes = sorted(all_nodes.values(), key=lambda x: x.node.node_id)
        return list(sorted_nodes)

Config

此为pydantic对象的配置。

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

class Config:
    """此为pydantic对象的配置。"""

    arbitrary_types_allowed = True