(Experimental) Execute the code blocks and return the result. Args: code_blocks (List[CodeBlock]): The code blocks to execute. Returns: CommandlineCodeResult: The result of the code execution.
(self, code_blocks: List[CodeBlock])
| 101 | self.latest_execution_out_str = "" |
| 102 | |
| 103 | def execute_code_blocks(self, code_blocks: List[CodeBlock]) -> CommandLineCodeResult: |
| 104 | """(Experimental) Execute the code blocks and return the result. |
| 105 | |
| 106 | Args: |
| 107 | code_blocks (List[CodeBlock]): The code blocks to execute. |
| 108 | |
| 109 | Returns: |
| 110 | CommandlineCodeResult: The result of the code execution.""" |
| 111 | |
| 112 | if len(code_blocks) == 0: |
| 113 | raise ValueError("No code blocks to execute.") |
| 114 | |
| 115 | outputs = [] |
| 116 | files = [] |
| 117 | last_exit_code = 0 |
| 118 | for code_block in code_blocks: |
| 119 | lang = self.LANGUAGE_ALIASES.get(code_block.language.lower(), code_block.language.lower()) |
| 120 | if lang not in self.DEFAULT_EXECUTION_POLICY: |
| 121 | outputs.append(f"Unsupported language {lang}\n") |
| 122 | last_exit_code = 1 |
| 123 | break |
| 124 | |
| 125 | execute_code = self.execution_policies.get(lang, False) |
| 126 | code = silence_pip(code_block.code, lang) |
| 127 | |
| 128 | # Check if there is a filename comment |
| 129 | try: |
| 130 | filename = _get_file_name_from_content(code, self._work_dir) |
| 131 | except ValueError: |
| 132 | outputs.append("Filename is not in the workspace") |
| 133 | last_exit_code = 1 |
| 134 | break |
| 135 | |
| 136 | if not filename: |
| 137 | filename = f"tmp_code_{md5(code.encode()).hexdigest()}.{lang}" |
| 138 | |
| 139 | code_path = self._work_dir / filename |
| 140 | |
| 141 | with code_path.open("w", encoding="utf-8") as fout: |
| 142 | fout.write("source /opt/miniconda3/bin/activate\n") |
| 143 | fout.write("conda activate testbed\n") |
| 144 | fout.write("cd /workspace/repository\n") |
| 145 | fout.write(code) |
| 146 | files.append(code_path) |
| 147 | |
| 148 | if not execute_code: |
| 149 | outputs.append(f"Code saved to {str(code_path)}\n") |
| 150 | continue |
| 151 | |
| 152 | command = ["timeout", str(self._timeout), _cmd(lang), filename] |
| 153 | result = self._container.exec_run(command) |
| 154 | exit_code = result.exit_code |
| 155 | output = result.output.decode("utf-8") |
| 156 | if exit_code == 124: |
| 157 | output += "\n" + TIMEOUT_MSG |
| 158 | outputs.append(output) |
| 159 | |
| 160 | last_exit_code = exit_code |