JSONalayze

JSONalyzeQueryEngine

Bases: BaseQueryEngine

JSON列表形状数据分析查询引擎。

将自然语言统计查询转换为内存中的SQLite查询。

list_of_dict(List[Dict[str, Any]]): 要查询的字典列表。 service_context (ServiceContext): 服务上下文 jsonalyze_prompt (BasePromptTemplate): 要使用的JSONalyze提示。 use_async (bool): 是否使用异步。 analyzer (Callable): 执行查询的分析器。 sql_parser (BaseSQLParser): 确保从llm输出中解析出有效SQL的SQL解析器。 synthesize_response (bool): 是否合成响应。 response_synthesis_prompt (BasePromptTemplate): 要使用的响应合成提示。 table_name (str): 要使用的表名。 verbose (bool): 是否打印详细输出。

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

class JSONalyzeQueryEngine(BaseQueryEngine):
    """JSON列表形状数据分析查询引擎。

    将自然语言统计查询转换为内存中的SQLite查询。

    list_of_dict(List[Dict[str, Any]]): 要查询的字典列表。
    service_context (ServiceContext): 服务上下文
    jsonalyze_prompt (BasePromptTemplate): 要使用的JSONalyze提示。
    use_async (bool): 是否使用异步。
    analyzer (Callable): 执行查询的分析器。
    sql_parser (BaseSQLParser): 确保从llm输出中解析出有效SQL的SQL解析器。
    synthesize_response (bool): 是否合成响应。
    response_synthesis_prompt (BasePromptTemplate): 要使用的响应合成提示。
    table_name (str): 要使用的表名。
    verbose (bool): 是否打印详细输出。"""

    def __init__(
        self,
        list_of_dict: List[Dict[str, Any]],
        service_context: Optional[ServiceContext] = None,
        llm: Optional[LLM] = None,
        jsonalyze_prompt: Optional[BasePromptTemplate] = None,
        use_async: bool = False,
        analyzer: Optional[Callable] = None,
        sql_parser: Optional[BaseSQLParser] = None,
        synthesize_response: bool = True,
        response_synthesis_prompt: Optional[BasePromptTemplate] = None,
        table_name: str = DEFAULT_TABLE_NAME,
        verbose: bool = False,
        **kwargs: Any,
    ) -> None:
        """初始化参数。"""
        self._list_of_dict = list_of_dict
        self._llm = llm or llm_from_settings_or_context(Settings, service_context)
        self._jsonalyze_prompt = jsonalyze_prompt or DEFAULT_JSONALYZE_PROMPT
        self._use_async = use_async
        self._analyzer = load_jsonalyzer(use_async, analyzer)
        self._sql_parser = sql_parser or DefaultSQLParser()
        self._synthesize_response = synthesize_response
        self._response_synthesis_prompt = (
            response_synthesis_prompt or DEFAULT_RESPONSE_SYNTHESIS_PROMPT
        )
        self._table_name = table_name
        self._verbose = verbose

        super().__init__(
            callback_manager=callback_manager_from_settings_or_context(
                Settings, service_context
            )
        )

    def _get_prompts(self) -> Dict[str, Any]:
        """获取提示。"""
        return {
            "jsonalyze_prompt": self._jsonalyze_prompt,
            "response_synthesis_prompt": self._response_synthesis_prompt,
        }

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

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

    def _query(self, query_bundle: QueryBundle) -> Response:
        """回答关于JSON列表的分析查询。"""
        query = query_bundle.query_str
        if self._verbose:
            print_text(f"Query: {query}\n", color="green")

        # Perform the analysis
        sql_query, table_schema, results = self._analyzer(
            self._list_of_dict,
            query_bundle,
            self._llm,
            table_name=self._table_name,
            prompt=self._jsonalyze_prompt,
            sql_parser=self._sql_parser,
        )
        if self._verbose:
            print_text(f"SQL Query: {sql_query}\n", color="blue")
            print_text(f"Table Schema: {table_schema}\n", color="cyan")
            print_text(f"SQL Response: {results}\n", color="yellow")

        if self._synthesize_response:
            response_str = self._llm.predict(
                self._response_synthesis_prompt,
                sql_query=sql_query,
                table_schema=table_schema,
                sql_response=results,
                query_str=query_bundle.query_str,
            )
            if self._verbose:
                print_text(f"Response: {response_str}", color="magenta")
        else:
            response_str = str(results)
        response_metadata = {"sql_query": sql_query, "table_schema": str(table_schema)}

        return Response(response=response_str, metadata=response_metadata)

    async def _aquery(self, query_bundle: QueryBundle) -> Response:
        """回答关于JSON列表的分析查询。"""
        query = query_bundle.query_str
        if self._verbose:
            print_text(f"Query: {query}", color="green")

        # Perform the analysis
        sql_query, table_schema, results = self._analyzer(
            self._list_of_dict,
            query,
            self._llm,
            table_name=self._table_name,
            prompt=self._jsonalyze_prompt,
        )
        if self._verbose:
            print_text(f"SQL Query: {sql_query}\n", color="blue")
            print_text(f"Table Schema: {table_schema}\n", color="cyan")
            print_text(f"SQL Response: {results}\n", color="yellow")

        if self._synthesize_response:
            response_str = await self._llm.apredict(
                self._response_synthesis_prompt,
                sql_query=sql_query,
                table_schema=table_schema,
                sql_response=results,
                query_str=query_bundle.query_str,
            )
            if self._verbose:
                print_text(f"Response: {response_str}", color="magenta")
        else:
            response_str = json.dumps(
                {
                    "sql_query": sql_query,
                    "table_schema": table_schema,
                    "sql_response": results,
                }
            )
        response_metadata = {"sql_query": sql_query, "table_schema": str(table_schema)}

        return Response(response=response_str, metadata=response_metadata)