mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-08 15:48:29 +00:00
90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
import re
|
|
from typing import Any, Union
|
|
|
|
from langchain.agents.output_parsers import ReActSingleInputOutputParser
|
|
from langchain_core.agents import AgentAction, AgentFinish
|
|
from langchain_core.exceptions import OutputParserException
|
|
|
|
from crewai.utilities import I18N
|
|
|
|
FINAL_ANSWER_ACTION = "Final Answer:"
|
|
MISSING_ACTION_AFTER_THOUGHT_ERROR_MESSAGE = "I did it wrong. Invalid Format: I missed the 'Action:' after 'Thought:'. I will do right next, and don't use a tool I have already used.\n"
|
|
MISSING_ACTION_INPUT_AFTER_ACTION_ERROR_MESSAGE = "I did it wrong. Invalid Format: I missed the 'Action Input:' after 'Action:'. I will do right next, and don't use a tool I have already used.\n"
|
|
FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE = "I did it wrong. Tried to both perform Action and give a Final Answer at the same time, I must do one or the other"
|
|
|
|
|
|
class CrewAgentParser(ReActSingleInputOutputParser):
|
|
"""Parses ReAct-style LLM calls that have a single tool input.
|
|
|
|
Expects output to be in one of two formats.
|
|
|
|
If the output signals that an action should be taken,
|
|
should be in the below format. This will result in an AgentAction
|
|
being returned.
|
|
|
|
Thought: agent thought here
|
|
Action: search
|
|
Action Input: what is the temperature in SF?
|
|
|
|
If the output signals that a final answer should be given,
|
|
should be in the below format. This will result in an AgentFinish
|
|
being returned.
|
|
|
|
Thought: agent thought here
|
|
Final Answer: The temperature is 100 degrees
|
|
"""
|
|
|
|
_i18n: I18N = I18N()
|
|
agent: Any = None
|
|
|
|
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:
|
|
if includes_answer:
|
|
raise OutputParserException(
|
|
f"{FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE}: {text}"
|
|
)
|
|
action = action_match.group(1).strip()
|
|
action_input = action_match.group(2)
|
|
tool_input = action_input.strip(" ")
|
|
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):
|
|
self.agent.increment_formatting_errors()
|
|
raise OutputParserException(
|
|
f"Could not parse LLM output: `{text}`",
|
|
observation=f"{MISSING_ACTION_AFTER_THOUGHT_ERROR_MESSAGE}\n{self._i18n.slice('final_answer_format')}",
|
|
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
|
|
):
|
|
self.agent.increment_formatting_errors()
|
|
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:
|
|
format = self._i18n.slice("format_without_tools")
|
|
error = f"{format}"
|
|
self.agent.increment_formatting_errors()
|
|
raise OutputParserException(
|
|
error,
|
|
observation=error,
|
|
llm_output=text,
|
|
send_to_llm=True,
|
|
)
|