Toggle Menu
Prev Up Next

# Source code for langchain.agents.mrkl.output_parser

    import re
    from typing import Union

    from langchain_core.agents import AgentAction, AgentFinish
    from langchain_core.exceptions import OutputParserException

    from langchain.agents.agent import AgentOutputParser
    from langchain.agents.mrkl.prompt import FORMAT_INSTRUCTIONS

    FINAL_ANSWER_ACTION = "Final Answer:"
    MISSING_ACTION_AFTER_THOUGHT_ERROR_MESSAGE = (
        "Invalid Format: Missing 'Action:' after 'Thought:"
    )
    MISSING_ACTION_INPUT_AFTER_ACTION_ERROR_MESSAGE = (
        "Invalid Format: Missing 'Action Input:' after 'Action:'"
    )
    FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE = (
        "Parsing LLM output produced both a final answer and a parse-able action:"
    )

    class MRKLOutputParser(AgentOutputParser):
        """MRKL Output parser for the chat agent."""

        format_instructions: str = FORMAT_INSTRUCTIONS
        """Default formatting instructions"""

        def get_format_instructions(self) -> str:
            """Returns formatting instructions for the given output parser."""
            return self.format_instructions

        def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
            includes_answer = FINAL_ANSWER_ACTION in text
            regex = (
                r"Action\s*\d*\s*:[\s]*(.*?)[\s]*Action\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)"
            )
            action_match = re.search(regex, text, re.DOTALL)
            if action_match and includes_answer:
                if text.find(FINAL_ANSWER_ACTION) < text.find(action_match.group(0)):
                    # if final answer is before the hallucination, return final answer
                    start_index = text.find(FINAL_ANSWER_ACTION) + len(FINAL_ANSWER_ACTION)
                    end_index = text.find("\n\n", start_index)
                    return AgentFinish(
                        {"output": text[start_index:end_index].strip()}, text[:end_index]
                    )
                else:
                    raise OutputParserException(
                        f"{FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE}: {text}"
                    )

            if action_match:
                action = action_match.group(1).strip()
                action_input = action_match.group(2)
                tool_input = action_input.strip(" ")
                # ensure if its a well formed SQL query we don't remove any trailing " chars
                if tool_input.startswith("SELECT ") is False:
                    tool_input = tool_input.strip('"')

                return AgentAction(action, tool_input, text)

            elif includes_answer:
                return AgentFinish(
                    {"output": text.split(FINAL_ANSWER_ACTION)[-1].strip()}, text
                )

            if not re.search(r"Action\s*\d*\s*:[\s]*(.*?)", text, re.DOTALL):
                raise OutputParserException(
                    f"Could not parse LLM output: `{text}`",
                    observation=MISSING_ACTION_AFTER_THOUGHT_ERROR_MESSAGE,
                    llm_output=text,
                    send_to_llm=True,
                )
            elif not re.search(
                r"[\s]*Action\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)", text, re.DOTALL
            ):
                raise OutputParserException(
                    f"Could not parse LLM output: `{text}`",
                    observation=MISSING_ACTION_INPUT_AFTER_ACTION_ERROR_MESSAGE,
                    llm_output=text,
                    send_to_llm=True,
                )
            else:
                raise OutputParserException(f"Could not parse LLM output: `{text}`")

        @property
        def _type(self) -> str:
            return "mrkl"