Citation

CitationQueryEngine

Bases: BaseQueryEngine

引文查询引擎。

Parameters:

Name	Type	Description	Default
retriever	BaseRetriever	一个检索器对象。	required
response_synthesizer	Optional[BaseSynthesizer]	一个BaseSynthesizer对象。	None
citation_chunk_size	int	引文块的大小，默认为512。用于控制来源的粒度。	DEFAULT_CITATION_CHUNK_SIZE
citation_chunk_overlap	int	引文节点的重叠，默认为20。	DEFAULT_CITATION_CHUNK_OVERLAP
text_splitter	Optional[TextSplitter]	用于创建引文来源节点的文本分割器。默认为SentenceSplitter。	None
callback_manager	Optional[CallbackManager]	回调管理器。	None
metadata_mode	MetadataMode	控制元数据如何包含在引文提示中的MetadataMode对象。	NONE

Source code in llama_index/core/query_engine/citation_query_engine.py

class CitationQueryEngine(BaseQueryEngine):
    """引文查询引擎。

    Args:
        retriever (BaseRetriever): 一个检索器对象。
        response_synthesizer (Optional[BaseSynthesizer]):
            一个BaseSynthesizer对象。
        citation_chunk_size (int):
            引文块的大小，默认为512。用于控制来源的粒度。
        citation_chunk_overlap (int): 引文节点的重叠，默认为20。
        text_splitter (Optional[TextSplitter]):
            用于创建引文来源节点的文本分割器。默认为SentenceSplitter。
        callback_manager (Optional[CallbackManager]): 回调管理器。
        metadata_mode (MetadataMode): 控制元数据如何包含在引文提示中的MetadataMode对象。"""

    def __init__(
        self,
        retriever: BaseRetriever,
        llm: Optional[LLM] = None,
        response_synthesizer: Optional[BaseSynthesizer] = None,
        citation_chunk_size: int = DEFAULT_CITATION_CHUNK_SIZE,
        citation_chunk_overlap: int = DEFAULT_CITATION_CHUNK_OVERLAP,
        text_splitter: Optional[TextSplitter] = None,
        node_postprocessors: Optional[List[BaseNodePostprocessor]] = None,
        callback_manager: Optional[CallbackManager] = None,
        metadata_mode: MetadataMode = MetadataMode.NONE,
    ) -> None:
        self.text_splitter = text_splitter or SentenceSplitter(
            chunk_size=citation_chunk_size, chunk_overlap=citation_chunk_overlap
        )
        self._retriever = retriever

        service_context = retriever.get_service_context()
        callback_manager = (
            callback_manager
            or callback_manager_from_settings_or_context(Settings, service_context)
        )
        llm = llm or llm_from_settings_or_context(Settings, service_context)

        self._response_synthesizer = response_synthesizer or get_response_synthesizer(
            llm=llm,
            service_context=service_context,
            callback_manager=callback_manager,
        )
        self._node_postprocessors = node_postprocessors or []
        self._metadata_mode = metadata_mode

        for node_postprocessor in self._node_postprocessors:
            node_postprocessor.callback_manager = callback_manager

        super().__init__(callback_manager=callback_manager)

    @classmethod
    def from_args(
        cls,
        index: BaseGPTIndex,
        llm: Optional[LLM] = None,
        response_synthesizer: Optional[BaseSynthesizer] = None,
        citation_chunk_size: int = DEFAULT_CITATION_CHUNK_SIZE,
        citation_chunk_overlap: int = DEFAULT_CITATION_CHUNK_OVERLAP,
        text_splitter: Optional[TextSplitter] = None,
        citation_qa_template: BasePromptTemplate = CITATION_QA_TEMPLATE,
        citation_refine_template: BasePromptTemplate = CITATION_REFINE_TEMPLATE,
        retriever: Optional[BaseRetriever] = None,
        node_postprocessors: Optional[List[BaseNodePostprocessor]] = None,
        # response synthesizer args
        response_mode: ResponseMode = ResponseMode.COMPACT,
        use_async: bool = False,
        streaming: bool = False,
        # class-specific args
        metadata_mode: MetadataMode = MetadataMode.NONE,
        **kwargs: Any,
    ) -> "CitationQueryEngine":
        """初始化一个CitationQueryEngine对象。

Args:
    index: (BastGPTIndex): 用于查询的索引
    llm: (Optional[LLM]): 用于生成响应的LLM对象。
    citation_chunk_size (int):
        引文块的大小，默认为512。用于控制来源的粒度。
    citation_chunk_overlap (int): 引文节点的重叠，默认为20。
    text_splitter (Optional[TextSplitter]):
        用于创建引文来源节点的文本分割器。默认为SentenceSplitter。
    citation_qa_template (BasePromptTemplate): 初始引文QA的模板
    citation_refine_template (BasePromptTemplate):
        引文细化的模板。
    retriever (BaseRetriever): 检索器对象。
    service_context (Optional[ServiceContext]): ServiceContext对象。
    node_postprocessors (Optional[List[BaseNodePostprocessor]]): 节点后处理器的列表。
    verbose (bool): 是否打印调试信息。
    response_mode (ResponseMode): ResponseMode对象。
    use_async (bool): 是否使用异步。
    streaming (bool): 是否使用流式处理。
    optimizer (Optional[BaseTokenUsageOptimizer]): BaseTokenUsageOptimizer对象。
"""
        retriever = retriever or index.as_retriever(**kwargs)

        response_synthesizer = response_synthesizer or get_response_synthesizer(
            llm=llm,
            service_context=index.service_context,
            text_qa_template=citation_qa_template,
            refine_template=citation_refine_template,
            response_mode=response_mode,
            use_async=use_async,
            streaming=streaming,
        )

        return cls(
            retriever=retriever,
            llm=llm,
            response_synthesizer=response_synthesizer,
            callback_manager=callback_manager_from_settings_or_context(
                Settings, index.service_context
            ),
            citation_chunk_size=citation_chunk_size,
            citation_chunk_overlap=citation_chunk_overlap,
            text_splitter=text_splitter,
            node_postprocessors=node_postprocessors,
            metadata_mode=metadata_mode,
        )

    def _get_prompt_modules(self) -> PromptMixinType:
        """获取提示子模块。"""
        return {"response_synthesizer": self._response_synthesizer}

    def _create_citation_nodes(self, nodes: List[NodeWithScore]) -> List[NodeWithScore]:
        """修改检索到的节点，使其成为细粒度的数据源。"""
        new_nodes: List[NodeWithScore] = []
        for node in nodes:
            text_chunks = self.text_splitter.split_text(
                node.node.get_content(metadata_mode=self._metadata_mode)
            )

            for text_chunk in text_chunks:
                text = f"Source {len(new_nodes)+1}:\n{text_chunk}\n"

                new_node = NodeWithScore(
                    node=TextNode.parse_obj(node.node), score=node.score
                )
                new_node.node.text = text
                new_nodes.append(new_node)
        return new_nodes

    def retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
        nodes = self._retriever.retrieve(query_bundle)

        for postprocessor in self._node_postprocessors:
            nodes = postprocessor.postprocess_nodes(nodes, query_bundle=query_bundle)

        return nodes

    async def aretrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
        nodes = await self._retriever.aretrieve(query_bundle)

        for postprocessor in self._node_postprocessors:
            nodes = postprocessor.postprocess_nodes(nodes, query_bundle=query_bundle)

        return nodes

    @property
    def retriever(self) -> BaseRetriever:
        """获取检索器对象。"""
        return self._retriever

    def synthesize(
        self,
        query_bundle: QueryBundle,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
    ) -> RESPONSE_TYPE:
        nodes = self._create_citation_nodes(nodes)
        return self._response_synthesizer.synthesize(
            query=query_bundle,
            nodes=nodes,
            additional_source_nodes=additional_source_nodes,
        )

    async def asynthesize(
        self,
        query_bundle: QueryBundle,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
    ) -> RESPONSE_TYPE:
        nodes = self._create_citation_nodes(nodes)
        return await self._response_synthesizer.asynthesize(
            query=query_bundle,
            nodes=nodes,
            additional_source_nodes=additional_source_nodes,
        )

    def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        """回答一个查询。"""
        with self.callback_manager.event(
            CBEventType.QUERY, payload={EventPayload.QUERY_STR: query_bundle.query_str}
        ) as query_event:
            with self.callback_manager.event(
                CBEventType.RETRIEVE,
                payload={EventPayload.QUERY_STR: query_bundle.query_str},
            ) as retrieve_event:
                nodes = self.retrieve(query_bundle)
                nodes = self._create_citation_nodes(nodes)

                retrieve_event.on_end(payload={EventPayload.NODES: nodes})

            response = self._response_synthesizer.synthesize(
                query=query_bundle,
                nodes=nodes,
            )

            query_event.on_end(payload={EventPayload.RESPONSE: response})

        return response

    async def _aquery(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        """回答一个查询。"""
        with self.callback_manager.event(
            CBEventType.QUERY, payload={EventPayload.QUERY_STR: query_bundle.query_str}
        ) as query_event:
            with self.callback_manager.event(
                CBEventType.RETRIEVE,
                payload={EventPayload.QUERY_STR: query_bundle.query_str},
            ) as retrieve_event:
                nodes = await self.aretrieve(query_bundle)
                nodes = self._create_citation_nodes(nodes)

                retrieve_event.on_end(payload={EventPayload.NODES: nodes})

            response = await self._response_synthesizer.asynthesize(
                query=query_bundle,
                nodes=nodes,
            )

            query_event.on_end(payload={EventPayload.RESPONSE: response})

        return response

retriever property

retriever: BaseRetriever

获取检索器对象。

from_args classmethod

from_args(
    index: BaseGPTIndex,
    llm: Optional[LLM] = None,
    response_synthesizer: Optional[BaseSynthesizer] = None,
    citation_chunk_size: int = DEFAULT_CITATION_CHUNK_SIZE,
    citation_chunk_overlap: int = DEFAULT_CITATION_CHUNK_OVERLAP,
    text_splitter: Optional[TextSplitter] = None,
    citation_qa_template: BasePromptTemplate = CITATION_QA_TEMPLATE,
    citation_refine_template: BasePromptTemplate = CITATION_REFINE_TEMPLATE,
    retriever: Optional[BaseRetriever] = None,
    node_postprocessors: Optional[
        List[BaseNodePostprocessor]
    ] = None,
    response_mode: ResponseMode = ResponseMode.COMPACT,
    use_async: bool = False,
    streaming: bool = False,
    metadata_mode: MetadataMode = MetadataMode.NONE,
    **kwargs: Any
) -> CitationQueryEngine

初始化一个CitationQueryEngine对象。

Parameters:

Name	Type	Description	Default
index	BaseGPTIndex	(BastGPTIndex): 用于查询的索引	required
llm	Optional[LLM]	(Optional[LLM]): 用于生成响应的LLM对象。	None
citation_chunk_size	int	引文块的大小，默认为512。用于控制来源的粒度。	DEFAULT_CITATION_CHUNK_SIZE
citation_chunk_overlap	int	引文节点的重叠，默认为20。	DEFAULT_CITATION_CHUNK_OVERLAP
text_splitter	Optional[TextSplitter]	用于创建引文来源节点的文本分割器。默认为SentenceSplitter。	None
citation_qa_template	BasePromptTemplate	初始引文QA的模板	CITATION_QA_TEMPLATE
citation_refine_template	BasePromptTemplate	引文细化的模板。	CITATION_REFINE_TEMPLATE
retriever	BaseRetriever	检索器对象。	None
service_context	Optional[ServiceContext]	ServiceContext对象。	required
node_postprocessors	Optional[List[BaseNodePostprocessor]]	节点后处理器的列表。	None
verbose	bool	是否打印调试信息。	required
response_mode	ResponseMode	ResponseMode对象。	COMPACT
use_async	bool	是否使用异步。	False
streaming	bool	是否使用流式处理。	False
optimizer	Optional[BaseTokenUsageOptimizer]	BaseTokenUsageOptimizer对象。	required

Source code in llama_index/core/query_engine/citation_query_engine.py

    @classmethod
    def from_args(
        cls,
        index: BaseGPTIndex,
        llm: Optional[LLM] = None,
        response_synthesizer: Optional[BaseSynthesizer] = None,
        citation_chunk_size: int = DEFAULT_CITATION_CHUNK_SIZE,
        citation_chunk_overlap: int = DEFAULT_CITATION_CHUNK_OVERLAP,
        text_splitter: Optional[TextSplitter] = None,
        citation_qa_template: BasePromptTemplate = CITATION_QA_TEMPLATE,
        citation_refine_template: BasePromptTemplate = CITATION_REFINE_TEMPLATE,
        retriever: Optional[BaseRetriever] = None,
        node_postprocessors: Optional[List[BaseNodePostprocessor]] = None,
        # response synthesizer args
        response_mode: ResponseMode = ResponseMode.COMPACT,
        use_async: bool = False,
        streaming: bool = False,
        # class-specific args
        metadata_mode: MetadataMode = MetadataMode.NONE,
        **kwargs: Any,
    ) -> "CitationQueryEngine":
        """初始化一个CitationQueryEngine对象。

Args:
    index: (BastGPTIndex): 用于查询的索引
    llm: (Optional[LLM]): 用于生成响应的LLM对象。
    citation_chunk_size (int):
        引文块的大小，默认为512。用于控制来源的粒度。
    citation_chunk_overlap (int): 引文节点的重叠，默认为20。
    text_splitter (Optional[TextSplitter]):
        用于创建引文来源节点的文本分割器。默认为SentenceSplitter。
    citation_qa_template (BasePromptTemplate): 初始引文QA的模板
    citation_refine_template (BasePromptTemplate):
        引文细化的模板。
    retriever (BaseRetriever): 检索器对象。
    service_context (Optional[ServiceContext]): ServiceContext对象。
    node_postprocessors (Optional[List[BaseNodePostprocessor]]): 节点后处理器的列表。
    verbose (bool): 是否打印调试信息。
    response_mode (ResponseMode): ResponseMode对象。
    use_async (bool): 是否使用异步。
    streaming (bool): 是否使用流式处理。
    optimizer (Optional[BaseTokenUsageOptimizer]): BaseTokenUsageOptimizer对象。
"""
        retriever = retriever or index.as_retriever(**kwargs)

        response_synthesizer = response_synthesizer or get_response_synthesizer(
            llm=llm,
            service_context=index.service_context,
            text_qa_template=citation_qa_template,
            refine_template=citation_refine_template,
            response_mode=response_mode,
            use_async=use_async,
            streaming=streaming,
        )

        return cls(
            retriever=retriever,
            llm=llm,
            response_synthesizer=response_synthesizer,
            callback_manager=callback_manager_from_settings_or_context(
                Settings, index.service_context
            ),
            citation_chunk_size=citation_chunk_size,
            citation_chunk_overlap=citation_chunk_overlap,
            text_splitter=text_splitter,
            node_postprocessors=node_postprocessors,
            metadata_mode=metadata_mode,
        )