| 94 | |
| 95 | # @register_tool('python_executor') # Do not register this tool by default because it is dangerous. |
| 96 | class PythonExecutor(BaseTool): |
| 97 | name = 'python_executor' |
| 98 | description = 'For executing python code. Not sandboxed. Do not use it for production purposes.' |
| 99 | parameters = { |
| 100 | 'type': 'object', |
| 101 | 'properties': { |
| 102 | 'code': { |
| 103 | 'description': 'The python code.', |
| 104 | 'type': 'string', |
| 105 | } |
| 106 | }, |
| 107 | 'required': ['code'], |
| 108 | } |
| 109 | |
| 110 | def __init__(self, cfg: Optional[Dict] = None): |
| 111 | _check_deps_for_python_executor() |
| 112 | import multiprocess |
| 113 | from multiprocess import Pool |
| 114 | super().__init__(cfg) |
| 115 | |
| 116 | runtime: Optional[Any] = self.cfg.get('runtime', None) |
| 117 | get_answer_symbol: Optional[str] = self.cfg.get('get_answer_symbol', None) |
| 118 | get_answer_expr: Optional[str] = self.cfg.get('get_answer_expr', None) |
| 119 | get_answer_from_stdout: bool = self.cfg.get('get_answer_from_stdout', True) |
| 120 | timeout_length: int = self.cfg.get('timeout_length', 20) |
| 121 | |
| 122 | self.runtime = runtime if runtime else GenericRuntime() |
| 123 | self.answer_symbol = get_answer_symbol |
| 124 | self.answer_expr = get_answer_expr |
| 125 | self.get_answer_from_stdout = get_answer_from_stdout |
| 126 | self.pool = Pool(multiprocess.cpu_count()) |
| 127 | self.timeout_length = timeout_length |
| 128 | |
| 129 | def call(self, params: Union[str, dict], **kwargs) -> list: |
| 130 | try: |
| 131 | params = json5.loads(params) |
| 132 | code = params['code'] |
| 133 | except Exception: |
| 134 | code = extract_code(params) |
| 135 | |
| 136 | if not code.strip(): |
| 137 | return ['', ''] |
| 138 | |
| 139 | predictions = self.apply(code) |
| 140 | return predictions |
| 141 | |
| 142 | def apply(self, code: str) -> list: |
| 143 | return self.batch_apply([code])[0] |
| 144 | |
| 145 | def process_generation_to_code(self, gens: str): |
| 146 | return [g.split('\n') for g in gens] |
| 147 | |
| 148 | @staticmethod |
| 149 | def execute( |
| 150 | code, |
| 151 | get_answer_from_stdout=None, |
| 152 | runtime=None, |
| 153 | answer_symbol=None, |