rmrkl/output_parser.py

import re
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union

import langchain
from langchain import LLMChain
from langchain.agents.agent import AgentOutputParser
from langchain.schema import AgentAction, AgentFinish, OutputParserException

from .prompts import (FINAL_ANSWER_ACTION, FORMAT_INSTRUCTIONS,
                      QUESTION_PROMPT, SUFFIX)


class ChatZeroShotOutputParser(AgentOutputParser):
    def get_format_instructions(self) -> str:
        return FORMAT_INSTRUCTIONS

    def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
        if FINAL_ANSWER_ACTION in text:
            return AgentFinish(
                {"output": text.split(FINAL_ANSWER_ACTION)[-1].strip()}, text
            )


        # Remove 'Thought' SUFFIX
        if text.startswith('Thought:'):
            text = text[8:]

        # \s matches against tab/newline/whitespace
        regex = (
            r"Action\s*\d*\s*:[\s]*(.*?)[\s]*Action\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)"
        )
        match = re.search(regex, text, re.DOTALL)
        if not match:
            raise OutputParserException(f"{text}")
        action = match.group(1).strip()
        action_input = match.group(2)
        return AgentAction(action, action_input.strip(" ").strip('"'), text.strip())