Pandas

PandasQueryEngine

Bases: BaseQueryEngine

Pandas查询引擎。

将自然语言转换为Pandas Python代码。

警告：此工具允许代理访问eval函数。在运行此工具的机器上可能会发生任意代码执行。不建议在生产环境中使用此工具，并且需要进行严格的沙盒化或虚拟机设置。

Parameters:

Name	Type	Description	Default
df	DataFrame	要使用的Pandas数据框。	required
instruction_str	Optional[str]	要使用的指令字符串。	None
output_processor	Optional[Callable[[str], str]]	输出处理器。一个可调用对象，接受输出字符串、Pandas数据框和任何输出kwargs，并返回一个字符串。	required
pandas_prompt	Optional[BasePromptTemplate]	要使用的Pandas提示。	None
head	int	表格上下文中要显示的行数。	5
llm	Optional[LLM]	要使用的语言模型。	None

示例： pip install llama-index-experimental

```python
import pandas as pd
from llama_index.experimental.query_engine.pandas import PandasQueryEngine

df = pd.DataFrame(
    {
        "city": ["Toronto", "Tokyo", "Berlin"],
        "population": [2930000, 13960000, 3645000]
    }
)

query_engine = PandasQueryEngine(df=df, verbose=True)

response = query_engine.query("What is the population of Tokyo?")
```

Source code in llama_index/experimental/query_engine/pandas/pandas_query_engine.py

class PandasQueryEngine(BaseQueryEngine):
    """Pandas查询引擎。

将自然语言转换为Pandas Python代码。

警告：此工具允许代理访问`eval`函数。在运行此工具的机器上可能会发生任意代码执行。不建议在生产环境中使用此工具，并且需要进行严格的沙盒化或虚拟机设置。

Args:
    df (pd.DataFrame): 要使用的Pandas数据框。
    instruction_str (Optional[str]): 要使用的指令字符串。
    output_processor (Optional[Callable[[str], str]]): 输出处理器。一个可调用对象，接受输出字符串、Pandas数据框和任何输出kwargs，并返回一个字符串。
    例如.kwargs["max_colwidth"] = [int] 用于设置在str(df)期间每列可以显示的文本长度。如果数据框中可能存在较长的文本，则将其设置为较大的数字。
    pandas_prompt (Optional[BasePromptTemplate]): 要使用的Pandas提示。
    head (int): 表格上下文中要显示的行数。
    llm (Optional[LLM]): 要使用的语言模型。

示例：
    `pip install llama-index-experimental`

    ```python
    import pandas as pd
    from llama_index.experimental.query_engine.pandas import PandasQueryEngine

    df = pd.DataFrame(
        {
            "city": ["Toronto", "Tokyo", "Berlin"],
            "population": [2930000, 13960000, 3645000]
        }
    )

    query_engine = PandasQueryEngine(df=df, verbose=True)

    response = query_engine.query("What is the population of Tokyo?")
    ```"""

    def __init__(
        self,
        df: pd.DataFrame,
        instruction_str: Optional[str] = None,
        instruction_parser: Optional[PandasInstructionParser] = None,
        pandas_prompt: Optional[BasePromptTemplate] = None,
        output_kwargs: Optional[dict] = None,
        head: int = 5,
        verbose: bool = False,
        service_context: Optional[ServiceContext] = None,
        llm: Optional[LLM] = None,
        synthesize_response: bool = False,
        response_synthesis_prompt: Optional[BasePromptTemplate] = None,
        **kwargs: Any,
    ) -> None:
        """初始化参数。"""
        self._df = df

        self._head = head
        self._pandas_prompt = pandas_prompt or DEFAULT_PANDAS_PROMPT
        self._instruction_str = instruction_str or DEFAULT_INSTRUCTION_STR
        self._instruction_parser = instruction_parser or PandasInstructionParser(
            df, output_kwargs or {}
        )
        self._verbose = verbose

        self._llm = llm or llm_from_settings_or_context(Settings, service_context)
        self._synthesize_response = synthesize_response
        self._response_synthesis_prompt = (
            response_synthesis_prompt or DEFAULT_RESPONSE_SYNTHESIS_PROMPT
        )

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

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

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

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

    @classmethod
    def from_index(cls, index: PandasIndex, **kwargs: Any) -> "PandasQueryEngine":
        logger.warning(
            "PandasIndex is deprecated. "
            "Directly construct PandasQueryEngine with df instead."
        )
        return cls(df=index.df, service_context=index.service_context, **kwargs)

    def _get_table_context(self) -> str:
        """获取表格上下文。"""
        return str(self._df.head(self._head))

    def _query(self, query_bundle: QueryBundle) -> Response:
        """回答一个查询。"""
        context = self._get_table_context()

        pandas_response_str = self._llm.predict(
            self._pandas_prompt,
            df_str=context,
            query_str=query_bundle.query_str,
            instruction_str=self._instruction_str,
        )

        if self._verbose:
            print_text(f"> Pandas Instructions:\n" f"```\n{pandas_response_str}\n```\n")
        pandas_output = self._instruction_parser.parse(pandas_response_str)
        if self._verbose:
            print_text(f"> Pandas Output: {pandas_output}\n")

        response_metadata = {
            "pandas_instruction_str": pandas_response_str,
            "raw_pandas_output": pandas_output,
        }
        if self._synthesize_response:
            response_str = str(
                self._llm.predict(
                    self._response_synthesis_prompt,
                    query_str=query_bundle.query_str,
                    pandas_instructions=pandas_response_str,
                    pandas_output=pandas_output,
                )
            )
        else:
            response_str = str(pandas_output)

        return Response(response=response_str, metadata=response_metadata)

    async def _aquery(self, query_bundle: QueryBundle) -> Response:
        return self._query(query_bundle)