pandas_dataframe.py — langchain Source File

Source Code

import re
from typing import Any

from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers.base import BaseOutputParser
from pydantic import field_validator
from typing_extensions import override

from langchain_classic.output_parsers.format_instructions import (
    PANDAS_DATAFRAME_FORMAT_INSTRUCTIONS,
)


class PandasDataFrameOutputParser(BaseOutputParser[dict[str, Any]]):
    """Parse an output using Pandas DataFrame format."""

    """The Pandas DataFrame to parse."""
    dataframe: Any

    @field_validator("dataframe")
    @classmethod
    def _validate_dataframe(cls, val: Any) -> Any:
        import pandas as pd

        if issubclass(type(val), pd.DataFrame):
            return val
        if pd.DataFrame(val).empty:
            msg = "DataFrame cannot be empty."
            raise ValueError(msg)

        msg = "Wrong type for 'dataframe', must be a subclass \
                of Pandas DataFrame (pd.DataFrame)"
        raise TypeError(msg)

    def parse_array(
        self,
        array: str,
        original_request_params: str,
    ) -> tuple[list[int | str], str]:
        """Parse the array from the request parameters.

        Args:
            array: The array string to parse.
            original_request_params: The original request parameters string.

        Returns:
            A tuple containing the parsed array and the stripped request parameters.

        Raises:
            OutputParserException: If the array format is invalid or cannot be parsed.
        """
        parsed_array: list[int | str] = []

        # Check if the format is [1,3,5]
        if re.match(r"\[\d+(,\s*\d+)*\]", array):
            parsed_array = [int(i) for i in re.findall(r"\d+", array)]
        # Check if the format is [1..5]
        elif re.match(r"\[(\d+)\.\.(\d+)\]", array):
            match = re.match(r"\[(\d+)\.\.(\d+)\]", array)
            if match:
// ... (112 more lines)