servers/Review/utils/parse_output.py

import json
from typing import Any, Callable

from pydantic import BaseModel


class OutputParserError(Exception):
    """
    Exception raised when the output parser fails to parse the output.
    """

    def __init__(self, message, output=None):
        self.message = message
        self.output = output
        super().__init__(self.message)

    def __str__(self):
        if self.output:
            return f"{self.message}\nProblematic output: {self.output}"
        return self.message


def find_json_in_string(string: str) -> str:
    """
    Method to extract all text in the left-most brace that appears in a string.
    Used to extract JSON from a string (note that this function does not validate the JSON).

    Example:
        string = "bla bla bla {this is {some} text{{}and it's sneaky}} because {it's} confusing"
        output = "{this is {some} text{{}and it's sneaky}}"
    """
    stack = 0
    start_index = None

    for i, c in enumerate(string):
        if c == "{":
            if stack == 0:
                start_index = i  # Start index of the first '{'
            stack += 1  # Push to stack
        elif c == "}":
            stack -= 1  # Pop stack
            if stack == 0:
                # Return the substring from the start of the first '{' to the current '}'
                return string[start_index : i + 1] if start_index is not None else ""

    # If no complete set of braces is found, return an empty string
    return ""


def parse_json_output(output: str) -> Any:
    """Take a string output and parse it as JSON"""
    # First try to load the string as JSON
    try:
        return json.loads(output)
    except json.JSONDecodeError as e:
        print(f"Failed to parse output as direct JSON: {e}")
        pass

    # If that fails, assume that the output is in a code block - remove the code block markers and try again
    parsed_output = output
    if "```" in parsed_output:
        try:
            parts = parsed_output.split("```")
            if len(parts) >= 3:
                parsed_output = parts[1]
                if parsed_output.startswith("json") or parsed_output.startswith("JSON"):
                    parsed_output = parsed_output[4:].strip()
                return json.loads(parsed_output)
        except (IndexError, json.JSONDecodeError) as e:
            print(f"Failed to parse output from code block: {e}")
            pass

    # As a last attempt, try to manually find the JSON object in the output and parse it
    parsed_output = find_json_in_string(output)
    if parsed_output:
        try:
            return json.loads(parsed_output)
        except json.JSONDecodeError as e:
            print(f"Failed to parse extracted JSON: {e}")
            # raise OutputParserError(f"Failed to parse output as JSON: {e}", output)
            raise ValueError("Failed to parse output as JSON")

    # If all fails, raise an error
    # raise OutputParserError("Failed to parse output as JSON", output)
    raise ValueError("Failed to parse output as JSON")


def create_type_parser(type: BaseModel) -> Callable[[str], BaseModel]:
    """Create a function that takes a string output and parses it as a specified Pydantic model"""

    def convert_json_string_to_type(output: str) -> BaseModel:
        """Take a string output and parse it as a Pydantic model"""
        output_dict = parse_json_output(output)
        return type.model_validate(output_dict)

    return convert_json_string_to_type