def execute()

in evals/elsuite/hr_ml_agent_bench/environment.py [0:0]

• 62 lines of code
• 16 McCabe index (conditional complexity)

    def execute(self, action: Action, max_seconds_per_step: Optional[int] = None) -> str:
        """Execute an action and return the observation."""

        trace = self._trace

        curr_step = len(trace.steps)
        action_name = action.name
        action_input = action.args

        if action_name == "Final Answer":
            observation = "end"
        elif self.is_done():
            observation = "The environment has shut down because the maximum number of steps or time has been reached. Please submit your final answer."
        elif action_name not in list(self.action_infos.keys()):
            actions = ", ".join(self.action_infos.keys())
            observation = f"Invalid action: {action_name}. Action did not execute. Please use one of the following actions:\n{actions}"
        else:
            # execute the action and get the observation
            log_file = os.path.join(
                os.path.join(self.log_dir, "tool_logs"),
                f"step_{curr_step}_tool_log.log",
            )
            usage = ",\n            ".join(
                [f"{k}: [{v}]" for k, v in self.action_infos[action_name].usage.items()]
            )
            usage = f"""{{
            {usage}
}}"""
            invalid_action_error = f"""No valid action found! Please ensure you're executing a valid action with json inputs. For example, to execute the `List Files` action, you would write:

    Action: List Files
    Action Input: {{
        "dir_path": "."
    }}

Likewise, the input for the action `{action_name}` needs to be valid json with proper entries. Please try again with the correct arguments:

    Action: {action_name}
    Action Input: {usage}"""

            if isinstance(action_input, dict):
                try:
                    if max_seconds_per_step is not None:
                        signal.signal(signal.SIGALRM, _signal_handler)
                        signal.alarm(max_seconds_per_step)

                    observation = self.action_infos[action_name].function(
                        **action_input,
                        log_file=log_file,
                        trace=trace,
                        **self.static_kwargs_for_tools,
                        solver=self.solver,
                    )
                except TooLongPromptError:
                    observation = "EnvError: too long input for the tool"
                except LLMError as e:
                    observation = "LLMError: " + e.message
                except TimeoutError:
                    observation = f"TimeoutError: action execution time exceeded the maximum time limit of {max_seconds_per_step} seconds!"
                except EnvException as e:
                    observation = "EnvError: " + e.message
                except TypeError as e:
                    logger.info(f"Step: {curr_step}")
                    logger.info(e)
                    logger.info(action_input)
                    observation = "EnvError: " + invalid_action_error
                except Exception as e:
                    # should not happen
                    logger.info(f"Step: {curr_step}")
                    logger.info(e)
                    if "Connection aborted." in str(e):
                        raise Exception("Connection aborted for crfm")
                    observation = f"EnvError: Error executing {action_name}."
                finally:
                    if max_seconds_per_step is not None:
                        signal.alarm(0)  # disable the alarm
            else:
                observation = invalid_action_error

        step_time = time.time()

        trace.steps.append(Step(action, observation, step_time))

        self.save(curr_step)

        return observation