SQL join

SQLJoinQueryEngine

Bases: BaseQueryEngine

SQL连接查询引擎。

此查询引擎可以将SQL数据库的结果与另一个查询引擎进行“连接”。 它可以决定是需要查询SQL数据库还是其他查询引擎。 如果它决定查询SQL数据库，它将首先查询SQL数据库，然后再决定是否使用其他查询引擎检索的结果来增强信息。

Parameters:

Name	Type	Description	Default
sql_query_tool	QueryEngineTool	用于SQL数据库的查询引擎工具。	required
other_query_tool	QueryEngineTool	其他查询引擎工具。	required
selector	Optional[Union[LLMSingleSelector, PydanticSingleSelector]]	要使用的选择器。	None
service_context	Optional[ServiceContext]	要使用的服务上下文。	None
sql_join_synthesis_prompt	Optional[BasePromptTemplate]	用于SQL连接合成的PromptTemplate。	None
sql_augment_query_transform	Optional[SQLAugmentQueryTransform]	用于SQL增强查询的转换。	None
use_sql_join_synthesis	bool	是否使用SQL连接合成。	True
callback_manager	Optional[CallbackManager]	要使用的回调管理器。	None
verbose	bool	是否打印中间结果。	True

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

class SQLJoinQueryEngine(BaseQueryEngine):
    """SQL连接查询引擎。

    此查询引擎可以将SQL数据库的结果与另一个查询引擎进行“连接”。
    它可以决定是需要查询SQL数据库还是其他查询引擎。
    如果它决定查询SQL数据库，它将首先查询SQL数据库，然后再决定是否使用其他查询引擎检索的结果来增强信息。

    Args:
        sql_query_tool (QueryEngineTool): 用于SQL数据库的查询引擎工具。
        other_query_tool (QueryEngineTool): 其他查询引擎工具。
        selector (Optional[Union[LLMSingleSelector, PydanticSingleSelector]]):
            要使用的选择器。
        service_context (Optional[ServiceContext]): 要使用的服务上下文。
        sql_join_synthesis_prompt (Optional[BasePromptTemplate]):
            用于SQL连接合成的PromptTemplate。
        sql_augment_query_transform (Optional[SQLAugmentQueryTransform]): 用于SQL增强查询的转换。
        use_sql_join_synthesis (bool): 是否使用SQL连接合成。
        callback_manager (Optional[CallbackManager]): 要使用的回调管理器。
        verbose (bool): 是否打印中间结果。"""

    def __init__(
        self,
        sql_query_tool: QueryEngineTool,
        other_query_tool: QueryEngineTool,
        selector: Optional[Union[LLMSingleSelector, PydanticSingleSelector]] = None,
        llm: Optional[LLMPredictorType] = None,
        sql_join_synthesis_prompt: Optional[BasePromptTemplate] = None,
        sql_augment_query_transform: Optional[SQLAugmentQueryTransform] = None,
        use_sql_join_synthesis: bool = True,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = True,
        # deprecated
        service_context: Optional[ServiceContext] = None,
    ) -> None:
        """初始化参数。"""
        super().__init__(callback_manager=callback_manager)
        # validate that the query engines are of the right type
        if not isinstance(
            sql_query_tool.query_engine,
            (BaseSQLTableQueryEngine, NLSQLTableQueryEngine),
        ):
            raise ValueError(
                "sql_query_tool.query_engine must be an instance of "
                "BaseSQLTableQueryEngine or NLSQLTableQueryEngine"
            )
        self._sql_query_tool = sql_query_tool
        self._other_query_tool = other_query_tool

        self._llm = llm or llm_from_settings_or_context(Settings, service_context)

        self._selector = selector or get_selector_from_llm(self._llm, is_multi=False)
        assert isinstance(self._selector, (LLMSingleSelector, PydanticSingleSelector))

        self._sql_join_synthesis_prompt = (
            sql_join_synthesis_prompt or DEFAULT_SQL_JOIN_SYNTHESIS_PROMPT
        )
        self._sql_augment_query_transform = (
            sql_augment_query_transform or SQLAugmentQueryTransform(llm=self._llm)
        )
        self._use_sql_join_synthesis = use_sql_join_synthesis
        self._verbose = verbose

    def _get_prompt_modules(self) -> PromptMixinType:
        """获取提示子模块。"""
        return {
            "selector": self._selector,
            "sql_augment_query_transform": self._sql_augment_query_transform,
        }

    def _get_prompts(self) -> PromptDictType:
        """获取提示。"""
        return {"sql_join_synthesis_prompt": self._sql_join_synthesis_prompt}

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """更新提示。"""
        if "sql_join_synthesis_prompt" in prompts:
            self._sql_join_synthesis_prompt = prompts["sql_join_synthesis_prompt"]

    def _query_sql_other(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        """查询SQL数据库 + 其他查询引擎依次执行。"""
        # first query SQL database
        sql_response = self._sql_query_tool.query_engine.query(query_bundle)
        if not self._use_sql_join_synthesis:
            return sql_response

        sql_query = (
            sql_response.metadata["sql_query"] if sql_response.metadata else None
        )
        if self._verbose:
            print_text(f"SQL query: {sql_query}\n", color="yellow")
            print_text(f"SQL response: {sql_response}\n", color="yellow")

        # given SQL db, transform query into new query
        new_query = self._sql_augment_query_transform(
            query_bundle.query_str,
            metadata={
                "sql_query": _format_sql_query(sql_query),
                "sql_query_response": str(sql_response),
            },
        )

        if self._verbose:
            print_text(
                f"Transformed query given SQL response: {new_query.query_str}\n",
                color="blue",
            )
        logger.info(f"> Transformed query given SQL response: {new_query.query_str}")
        if self._sql_augment_query_transform.check_stop(new_query):
            return sql_response

        other_response = self._other_query_tool.query_engine.query(new_query)
        if self._verbose:
            print_text(f"query engine response: {other_response}\n", color="pink")
        logger.info(f"> query engine response: {other_response}")

        response_str = self._llm.predict(
            self._sql_join_synthesis_prompt,
            query_str=query_bundle.query_str,
            sql_query_str=sql_query,
            sql_response_str=str(sql_response),
            query_engine_query_str=new_query.query_str,
            query_engine_response_str=str(other_response),
        )
        if self._verbose:
            print_text(f"Final response: {response_str}\n", color="green")
        response_metadata = {
            **(sql_response.metadata or {}),
            **(other_response.metadata or {}),
        }
        source_nodes = other_response.source_nodes
        return Response(
            response_str,
            metadata=response_metadata,
            source_nodes=source_nodes,
        )

    def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        """查询并获取响应。"""
        # TODO: see if this can be consolidated with logic in RouterQueryEngine
        metadatas = [self._sql_query_tool.metadata, self._other_query_tool.metadata]
        result = self._selector.select(metadatas, query_bundle)
        # pick sql query
        if result.ind == 0:
            if self._verbose:
                print_text(f"Querying SQL database: {result.reason}\n", color="blue")
            logger.info(f"> Querying SQL database: {result.reason}")
            return self._query_sql_other(query_bundle)
        elif result.ind == 1:
            if self._verbose:
                print_text(
                    f"Querying other query engine: {result.reason}\n", color="blue"
                )
            logger.info(f"> Querying other query engine: {result.reason}")
            response = self._other_query_tool.query_engine.query(query_bundle)
            if self._verbose:
                print_text(f"Query Engine response: {response}\n", color="pink")
            return response
        else:
            raise ValueError(f"Invalid result.ind: {result.ind}")

    async def _aquery(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
        # TODO: make async
        return self._query(query_bundle)

SQLAutoVectorQueryEngine

Bases: SQLJoinQueryEngine

SQL + 矢量索引自动检索查询引擎。

该查询引擎可以查询SQL数据库和矢量数据库。它首先决定是否需要查询SQL数据库或矢量存储。如果它决定查询SQL数据库，它还将决定是否使用从矢量存储中检索的结果来增强信息。我们使用VectorIndexAutoRetriever来检索结果。

Parameters:

Name	Type	Description	Default
sql_query_tool	QueryEngineTool	用于SQL数据库的查询引擎工具。	required
vector_query_tool	QueryEngineTool	用于矢量数据库的查询引擎工具。	required
selector	Optional[Union[LLMSingleSelector, PydanticSingleSelector]]	要使用的选择器。	None
service_context	Optional[ServiceContext]	要使用的服务上下文。	None
sql_vector_synthesis_prompt	Optional[BasePromptTemplate]	用于SQL矢量合成的提示。	None
sql_augment_query_transform	Optional[SQLAugmentQueryTransform]	用于SQL增强的查询转换。	None
use_sql_vector_synthesis	bool	是否使用SQL矢量合成。	True
callback_manager	Optional[CallbackManager]	要使用的回调管理器。	None
verbose	bool	是否打印中间结果。	True

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

class SQLAutoVectorQueryEngine(SQLJoinQueryEngine):
    """SQL + 矢量索引自动检索查询引擎。

    该查询引擎可以查询SQL数据库和矢量数据库。它首先决定是否需要查询SQL数据库或矢量存储。如果它决定查询SQL数据库，它还将决定是否使用从矢量存储中检索的结果来增强信息。我们使用VectorIndexAutoRetriever来检索结果。

    Args:
        sql_query_tool (QueryEngineTool): 用于SQL数据库的查询引擎工具。
        vector_query_tool (QueryEngineTool): 用于矢量数据库的查询引擎工具。
        selector (Optional[Union[LLMSingleSelector, PydanticSingleSelector]]):
            要使用的选择器。
        service_context (Optional[ServiceContext]): 要使用的服务上下文。
        sql_vector_synthesis_prompt (Optional[BasePromptTemplate]):
            用于SQL矢量合成的提示。
        sql_augment_query_transform (Optional[SQLAugmentQueryTransform]): 用于SQL增强的查询转换。
        use_sql_vector_synthesis (bool): 是否使用SQL矢量合成。
        callback_manager (Optional[CallbackManager]): 要使用的回调管理器。
        verbose (bool): 是否打印中间结果。"""

    def __init__(
        self,
        sql_query_tool: QueryEngineTool,
        vector_query_tool: QueryEngineTool,
        selector: Optional[Union[LLMSingleSelector, PydanticSingleSelector]] = None,
        llm: Optional[LLM] = None,
        service_context: Optional[ServiceContext] = None,
        sql_vector_synthesis_prompt: Optional[BasePromptTemplate] = None,
        sql_augment_query_transform: Optional[SQLAugmentQueryTransform] = None,
        use_sql_vector_synthesis: bool = True,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = True,
    ) -> None:
        """初始化参数。"""
        # validate that the query engines are of the right type
        if not isinstance(
            sql_query_tool.query_engine,
            (BaseSQLTableQueryEngine, NLSQLTableQueryEngine),
        ):
            raise ValueError(
                "sql_query_tool.query_engine must be an instance of "
                "BaseSQLTableQueryEngine or NLSQLTableQueryEngine"
            )
        if not isinstance(vector_query_tool.query_engine, RetrieverQueryEngine):
            raise ValueError(
                "vector_query_tool.query_engine must be an instance of "
                "RetrieverQueryEngine"
            )
        if not isinstance(
            vector_query_tool.query_engine.retriever, VectorIndexAutoRetriever
        ):
            raise ValueError(
                "vector_query_tool.query_engine.retriever must be an instance "
                "of VectorIndexAutoRetriever"
            )

        sql_vector_synthesis_prompt = (
            sql_vector_synthesis_prompt or DEFAULT_SQL_VECTOR_SYNTHESIS_PROMPT
        )
        super().__init__(
            sql_query_tool,
            vector_query_tool,
            selector=selector,
            llm=llm,
            service_context=service_context,
            sql_join_synthesis_prompt=sql_vector_synthesis_prompt,
            sql_augment_query_transform=sql_augment_query_transform,
            use_sql_join_synthesis=use_sql_vector_synthesis,
            callback_manager=callback_manager,
            verbose=verbose,
        )

    def _get_prompt_modules(self) -> PromptMixinType:
        """获取提示子模块。"""
        return {
            "selector": self._selector,
            "sql_augment_query_transform": self._sql_augment_query_transform,
        }

    def _get_prompts(self) -> PromptDictType:
        """获取提示。"""
        return {"sql_join_synthesis_prompt": self._sql_join_synthesis_prompt}

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """更新提示。"""
        if "sql_join_synthesis_prompt" in prompts:
            self._sql_join_synthesis_prompt = prompts["sql_join_synthesis_prompt"]

    @classmethod
    def from_sql_and_vector_query_engines(
        cls,
        sql_query_engine: Union[BaseSQLTableQueryEngine, NLSQLTableQueryEngine],
        sql_tool_name: str,
        sql_tool_description: str,
        vector_auto_retriever: RetrieverQueryEngine,
        vector_tool_name: str,
        vector_tool_description: str,
        selector: Optional[Union[LLMSingleSelector, PydanticSingleSelector]] = None,
        **kwargs: Any,
    ) -> "SQLAutoVectorQueryEngine":
        """来自SQL和向量查询引擎。

Args:
    sql_query_engine（BaseSQLTableQueryEngine）：SQL查询引擎。
    vector_query_engine（VectorIndexAutoRetriever）：向量检索器。
    selector（Optional[Union[LLMSingleSelector, PydanticSingleSelector]]）：
        要使用的选择器。
"""
        sql_query_tool = QueryEngineTool.from_defaults(
            sql_query_engine, name=sql_tool_name, description=sql_tool_description
        )
        vector_query_tool = QueryEngineTool.from_defaults(
            vector_auto_retriever,
            name=vector_tool_name,
            description=vector_tool_description,
        )
        return cls(sql_query_tool, vector_query_tool, selector, **kwargs)

from_sql_and_vector_query_engines classmethod

from_sql_and_vector_query_engines(
    sql_query_engine: Union[
        BaseSQLTableQueryEngine, NLSQLTableQueryEngine
    ],
    sql_tool_name: str,
    sql_tool_description: str,
    vector_auto_retriever: RetrieverQueryEngine,
    vector_tool_name: str,
    vector_tool_description: str,
    selector: Optional[
        Union[LLMSingleSelector, PydanticSingleSelector]
    ] = None,
    **kwargs: Any
) -> SQLAutoVectorQueryEngine

来自SQL和向量查询引擎。

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

    @classmethod
    def from_sql_and_vector_query_engines(
        cls,
        sql_query_engine: Union[BaseSQLTableQueryEngine, NLSQLTableQueryEngine],
        sql_tool_name: str,
        sql_tool_description: str,
        vector_auto_retriever: RetrieverQueryEngine,
        vector_tool_name: str,
        vector_tool_description: str,
        selector: Optional[Union[LLMSingleSelector, PydanticSingleSelector]] = None,
        **kwargs: Any,
    ) -> "SQLAutoVectorQueryEngine":
        """来自SQL和向量查询引擎。

Args:
    sql_query_engine（BaseSQLTableQueryEngine）：SQL查询引擎。
    vector_query_engine（VectorIndexAutoRetriever）：向量检索器。
    selector（Optional[Union[LLMSingleSelector, PydanticSingleSelector]]）：
        要使用的选择器。
"""
        sql_query_tool = QueryEngineTool.from_defaults(
            sql_query_engine, name=sql_tool_name, description=sql_tool_description
        )
        vector_query_tool = QueryEngineTool.from_defaults(
            vector_auto_retriever,
            name=vector_tool_name,
            description=vector_tool_description,
        )
        return cls(sql_query_tool, vector_query_tool, selector, **kwargs)