|
|
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]: |
|
|
"""Parse the output from the agent into |
|
|
an AgentAction or AgentFinish object. |
|
|
|
|
|
Args: |
|
|
text: The text to parse. |
|
|
|
|
|
Returns: |
|
|
An AgentAction or AgentFinish object. |
|
|
|
|
|
Raises: |
|
|
OutputParserException: If the output could not be parsed. |
|
|
""" |
|
|
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)): |
|
|
|
|
|
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(" ") |
|
|
|
|
|
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" |
|
|
|
