Spaces:
Sleeping
Sleeping
| import os | |
| import re | |
| import traceback | |
| import appdirs | |
| import json | |
| from .code_interpreter_utils.create_code_interpreter import \ | |
| create_code_interpreter | |
| from .code_interpreter_utils.language_map import language_map | |
| from .code_interpreter_utils.truncate_output import truncate_output | |
| from .tool import Tool | |
| class CodeInterpreter(Tool): | |
| """ | |
| using open interpreter to interpret code | |
| by https://github.com/KillianLucas/open-interpreter | |
| """ | |
| description = 'Executes code on the user\'s machine, **in the users local environment**, and returns the output' | |
| name = 'code_interpreter' | |
| parameters: list = [{ | |
| 'name': 'language', | |
| 'description': | |
| 'The programming language (required parameter to the `execute` function)', | |
| 'required': True | |
| }, { | |
| 'name': 'code', | |
| 'description': 'The code to execute (required)', | |
| 'required': True | |
| }] | |
| def __init__(self, cfg={}): | |
| super().__init__(cfg) | |
| self.create_code_interpreter = create_code_interpreter | |
| self.language_map = language_map | |
| self.truncate_output = truncate_output | |
| self._code_interpreters = {} | |
| self.max_output = self.cfg.get('max_output', 2000) | |
| def _local_call(self, *args, **kwargs): | |
| language, code = self._handle_input_fallback(**kwargs) | |
| try: | |
| # Fix a common error where the LLM thinks it's in a Jupyter notebook | |
| if language == 'python' and code.startswith('!'): | |
| code = code[1:] | |
| language = 'shell' | |
| if language in self.language_map: | |
| if language not in self._code_interpreters: | |
| self._code_interpreters[ | |
| language] = self.create_code_interpreter(language) | |
| code_interpreter = self._code_interpreters[language] | |
| else: | |
| # This still prints code but don't allow code to run. Let Open-Interpreter know through output message | |
| error_output = f'Error: Open Interpreter does not currently support {language}.' | |
| print(error_output) | |
| output = '\n' + error_output | |
| return {'result': output.strip()} | |
| output = '' | |
| for line in code_interpreter.run(code): | |
| if 'output' in line: | |
| output += '\n' + line['output'] | |
| # Truncate output | |
| output = self.truncate_output(output, self.max_output) | |
| except Exception as e: | |
| error = traceback.format_exc() | |
| output = ' '.join(f'{key}:{value}' | |
| for key, value in kwargs.items()) | |
| output += f'\nDetail error is {e}.\n{error}' | |
| return {'result': output.strip()} | |
| def _handle_input_fallback(self, **kwargs): | |
| """ | |
| an alternative method is to parse code in content not from function call | |
| such as: | |
| text = response['content'] | |
| code_block = re.search(r'```([\s\S]+)```', text) # noqa W^05 | |
| if code_block: | |
| result = code_block.group(1) | |
| language = result.split('\n')[0] | |
| code = '\n'.join(result.split('\n')[1:]) | |
| :param fallback_text: | |
| :return: language, cocde | |
| """ | |
| language = kwargs.get('language', None) | |
| code = kwargs.get('code', None) | |
| fallback = kwargs.get('fallback', None) | |
| if language and code: | |
| return language, code | |
| elif fallback: | |
| try: | |
| text = fallback | |
| code_block = re.search(r'```([\s\S]+)```', text) # noqa W^05 | |
| if code_block: | |
| result = code_block.group(1) | |
| # for multi code_block | |
| result = result.split('```')[0] | |
| language = result.split('\n')[0] | |
| if language == 'py' or language == 'python': | |
| # handle py case | |
| # ```py code ``` | |
| language = 'python' | |
| code = '\n'.join(result.split('\n')[1:]) | |
| return language, code | |
| if language == 'json': | |
| # handle json case | |
| # ```json {language,code}``` | |
| parameters = json.loads('\n'.join( | |
| result.split('\n')[1:]).replace('\n', '')) | |
| return parameters['language'], parameters['code'] | |
| except ValueError: | |
| return language, code | |
| else: | |
| return language, code | |