import time from typing import Any, Dict, Iterator, List, Optional, Tuple, Union from langchain.agents import AgentExecutor from langchain.agents.agent import ExceptionTool from langchain.agents.tools import InvalidTool from langchain.callbacks.manager import CallbackManagerForChainRun from langchain_core.agents import AgentAction, AgentFinish, AgentStep from langchain_core.exceptions import OutputParserException from langchain_core.pydantic_v1 import root_validator from langchain_core.tools import BaseTool from langchain_core.utils.input import get_color_mapping from crewai.agents.cache.cache_hit import CacheHit from crewai.tools.cache_tools import CacheTools from crewai.utilities import I18N class CrewAgentExecutor(AgentExecutor): i18n: I18N = I18N() iterations: int = 0 request_within_rpm_limit: Any = None max_iterations: Optional[int] = 15 force_answer_max_iterations: Optional[int] = None @root_validator() def set_force_answer_max_iterations(cls, values: Dict) -> Dict: values["force_answer_max_iterations"] = values["max_iterations"] - 2 return values def _should_force_answer(self) -> bool: return True if self.iterations == self.force_answer_max_iterations else False def _force_answer(self, output: AgentAction): return AgentStep( action=output, observation=self.i18n.errors("force_final_answer") ) def _call( self, inputs: Dict[str, str], run_manager: Optional[CallbackManagerForChainRun] = None, ) -> Dict[str, Any]: """Run text through and get agent response.""" # Construct a mapping of tool name to tool for easy lookup name_to_tool_map = {tool.name: tool for tool in self.tools} # We construct a mapping from each tool to a color, used for logging. color_mapping = get_color_mapping( [tool.name for tool in self.tools], excluded_colors=["green", "red"] ) intermediate_steps: List[Tuple[AgentAction, str]] = [] # Let's start tracking the number of iterations and time elapsed self.iterations = 0 time_elapsed = 0.0 start_time = time.time() # We now enter the agent loop (until it returns something). while self._should_continue(self.iterations, time_elapsed): if not self.request_within_rpm_limit or self.request_within_rpm_limit(): next_step_output = self._take_next_step( name_to_tool_map, color_mapping, inputs, intermediate_steps, run_manager=run_manager, ) if isinstance(next_step_output, AgentFinish): return self._return( next_step_output, intermediate_steps, run_manager=run_manager ) intermediate_steps.extend(next_step_output) if len(next_step_output) == 1: next_step_action = next_step_output[0] # See if tool should return directly tool_return = self._get_tool_return(next_step_action) if tool_return is not None: return self._return( tool_return, intermediate_steps, run_manager=run_manager ) self.iterations += 1 time_elapsed = time.time() - start_time output = self.agent.return_stopped_response( self.early_stopping_method, intermediate_steps, **inputs ) return self._return(output, intermediate_steps, run_manager=run_manager) def _iter_next_step( self, name_to_tool_map: Dict[str, BaseTool], color_mapping: Dict[str, str], inputs: Dict[str, str], intermediate_steps: List[Tuple[AgentAction, str]], run_manager: Optional[CallbackManagerForChainRun] = None, ) -> Iterator[Union[AgentFinish, AgentAction, AgentStep]]: """Take a single step in the thought-action-observation loop. Override this to take control of how the agent makes and acts on choices. """ try: intermediate_steps = self._prepare_intermediate_steps(intermediate_steps) # Call the LLM to see what to do. output = self.agent.plan( intermediate_steps, callbacks=run_manager.get_child() if run_manager else None, **inputs, ) if self._should_force_answer(): if isinstance(output, AgentAction) or isinstance(output, AgentFinish): output = output elif isinstance(output, CacheHit): output = output.action else: raise ValueError( f"Unexpected output type from agent: {type(output)}" ) yield self._force_answer(output) return except OutputParserException as e: if isinstance(self.handle_parsing_errors, bool): raise_error = not self.handle_parsing_errors else: raise_error = False if raise_error: raise ValueError( "An output parsing error occurred. " "In order to pass this error back to the agent and have it try " "again, pass `handle_parsing_errors=True` to the AgentExecutor. " f"This is the error: {str(e)}" ) text = str(e) if isinstance(self.handle_parsing_errors, bool): if e.send_to_llm: observation = str(e.observation) text = str(e.llm_output) else: observation = "Invalid or incomplete response" elif isinstance(self.handle_parsing_errors, str): observation = self.handle_parsing_errors elif callable(self.handle_parsing_errors): observation = self.handle_parsing_errors(e) else: raise ValueError("Got unexpected type of `handle_parsing_errors`") output = AgentAction("_Exception", observation, text) if run_manager: run_manager.on_agent_action(output, color="green") tool_run_kwargs = self.agent.tool_run_logging_kwargs() observation = ExceptionTool().run( output.tool_input, verbose=self.verbose, color=None, callbacks=run_manager.get_child() if run_manager else None, **tool_run_kwargs, ) if self._should_force_answer(): yield self._force_answer(output) return yield AgentStep(action=output, observation=observation) return # If the tool chosen is the finishing tool, then we end and return. if isinstance(output, AgentFinish): yield output return # Override tool usage to use CacheTools if isinstance(output, CacheHit): cache = output.cache action = output.action tool = CacheTools(cache_handler=cache).tool() output = action.copy() output.tool_input = f"tool:{action.tool}|input:{action.tool_input}" output.tool = tool.name name_to_tool_map[tool.name] = tool color_mapping[tool.name] = color_mapping[action.tool] actions: List[AgentAction] actions = [output] if isinstance(output, AgentAction) else output yield from actions for agent_action in actions: if run_manager: run_manager.on_agent_action(agent_action, color="green") # Otherwise we lookup the tool if agent_action.tool in name_to_tool_map: tool = name_to_tool_map[agent_action.tool] return_direct = tool.return_direct color = color_mapping[agent_action.tool] tool_run_kwargs = self.agent.tool_run_logging_kwargs() if return_direct: tool_run_kwargs["llm_prefix"] = "" # We then call the tool on the tool input to get an observation observation = tool.run( agent_action.tool_input, verbose=self.verbose, color=color, callbacks=run_manager.get_child() if run_manager else None, **tool_run_kwargs, ) else: tool_run_kwargs = self.agent.tool_run_logging_kwargs() observation = InvalidTool().run( { "requested_tool_name": agent_action.tool, "available_tool_names": list(name_to_tool_map.keys()), }, verbose=self.verbose, color=None, callbacks=run_manager.get_child() if run_manager else None, **tool_run_kwargs, ) yield AgentStep(action=agent_action, observation=observation)