Spaces:
Sleeping
Sleeping
| import re | |
| from typing import Union | |
| from langchain.agents.output_parsers import ReActSingleInputOutputParser | |
| from langchain_core.agents import AgentAction, AgentFinish | |
| from .cache import CacheHandler, CacheHit | |
| from .exceptions import TaskRepeatedUsageException | |
| from .tools_handler import ToolsHandler | |
| FINAL_ANSWER_ACTION = "Final Answer:" | |
| FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE = ( | |
| "Parsing LLM output produced both a final answer and a parse-able action:" | |
| ) | |
| class CrewAgentOutputParser(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 | |
| ``` | |
| It also prevents tools from being reused in a roll. | |
| """ | |
| class Config: | |
| arbitrary_types_allowed = True | |
| tools_handler: ToolsHandler | |
| cache: CacheHandler | |
| def parse(self, text: str) -> Union[AgentAction, AgentFinish, CacheHit]: | |
| 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: | |
| action = action_match.group(1).strip() | |
| action_input = action_match.group(2) | |
| tool_input = action_input.strip(" ") | |
| tool_input = tool_input.strip('"') | |
| last_tool_usage = self.tools_handler.last_used_tool | |
| if last_tool_usage: | |
| usage = { | |
| "tool": action, | |
| "input": tool_input, | |
| } | |
| if usage == last_tool_usage: | |
| raise TaskRepeatedUsageException(tool=action, tool_input=tool_input) | |
| result = self.cache.read(action, tool_input) | |
| if result: | |
| action = AgentAction(action, tool_input, text) | |
| return CacheHit(action=action, cache=self.cache) | |
| return super().parse(text) | |