Run text through and get agent response.
(self, inputs: Dict[str, str])
| 27 | |
| 28 | class Executor(AgentExecutorWithTranslation): |
| 29 | def _call(self, inputs: Dict[str, str]) -> Dict[str, Any]: |
| 30 | """Run text through and get agent response.""" |
| 31 | # Construct a mapping of tool name to tool for easy lookup |
| 32 | name_to_tool_map = {tool.name: tool for tool in self.tools} |
| 33 | # We construct a mapping from each tool to a color, used for logging. |
| 34 | color_mapping = get_color_mapping( |
| 35 | [tool.name for tool in self.tools], excluded_colors=["green"] |
| 36 | ) |
| 37 | intermediate_steps: List[Tuple[AgentAction, str]] = [] |
| 38 | # Let's start tracking the iterations the agent has gone through |
| 39 | iterations = 0 |
| 40 | time_elapsed = 0.0 |
| 41 | start_time = time.time() |
| 42 | # We now enter the agent loop (until it returns something). |
| 43 | while self._should_continue(iterations, time_elapsed): |
| 44 | next_step_output = self._take_next_step( |
| 45 | name_to_tool_map, color_mapping, inputs, intermediate_steps |
| 46 | ) |
| 47 | if isinstance(next_step_output, AgentFinish): |
| 48 | yield self._return(next_step_output, intermediate_steps) |
| 49 | return |
| 50 | |
| 51 | for i, output in enumerate(next_step_output): |
| 52 | agent_action = output[0] |
| 53 | tool_logo = None |
| 54 | for tool in self.tools: |
| 55 | if tool.name == agent_action.tool: |
| 56 | tool_logo = tool.tool_logo_md |
| 57 | if isinstance(output[1], types.GeneratorType): |
| 58 | logo = f"{tool_logo}" if tool_logo is not None else "" |
| 59 | yield (AgentAction("", agent_action.tool_input, agent_action.log), f"Further use other tool {logo} to answer the question.") |
| 60 | for out in output[1]: |
| 61 | yield out |
| 62 | next_step_output[i] = (agent_action, out) |
| 63 | else: |
| 64 | for tool in self.tools: |
| 65 | if tool.name == agent_action.tool: |
| 66 | yield (AgentAction(tool_logo, agent_action.tool_input, agent_action.log), output[1]) |
| 67 | |
| 68 | intermediate_steps.extend(next_step_output) |
| 69 | if len(next_step_output) == 1: |
| 70 | next_step_action = next_step_output[0] |
| 71 | # See if tool should return directly |
| 72 | tool_return = self._get_tool_return(next_step_action) |
| 73 | if tool_return is not None: |
| 74 | yield self._return(tool_return, intermediate_steps) |
| 75 | return |
| 76 | iterations += 1 |
| 77 | time_elapsed = time.time() - start_time |
| 78 | output = self.agent.return_stopped_response( |
| 79 | self.early_stopping_method, intermediate_steps, **inputs |
| 80 | ) |
| 81 | yield self._return(output, intermediate_steps) |
| 82 | return |
| 83 | |
| 84 | def __call__( |
| 85 | self, inputs: Union[Dict[str, Any], Any], return_only_outputs: bool = False |