| import subprocess |
| import sys |
| from typing import Optional |
|
|
| from langchain.tools import tool |
| from langchain_experimental.tools import PythonREPLTool |
|
|
|
|
| @tool |
| def execute_python_code(intent: str, code: Optional[str] = None, file_reference: Optional[str] = None) -> str: |
| """ |
| Executes the provided python code snippet or python file identified by its reference and returns the outcome of |
| the execution |
| :param intent: this parameter should be set to either code_snippet or file_execution depending on |
| the intent of the user |
| :param code: if the intent is code_snippet, this parameter should be populated with the |
| python code snippet to be executed |
| :param file_reference: if the intent is file_execution, this parameter should |
| be populated with the reference of the file to be executed |
| :return: the outcome of the python code execution |
| """ |
| if intent == "code_snippet": |
| python_tool = PythonREPLTool() |
| if not code: |
| raise Exception("Invalid arguments. Tool intent is code_snippet but no value provided for code argument") |
| return python_tool.run(code, verbose=True) |
|
|
| if intent == "file_execution": |
| return subprocess_python_exec(file_reference) |
| raise Exception("Invalid arguments. Invalid value for intent parameter") |
|
|
|
|
| def subprocess_python_exec(file_reference: str) -> str: |
| """Execute Python code in a subprocess for better isolation""" |
| try: |
|
|
| |
| result = subprocess.run( |
| [sys.executable, file_reference], |
| capture_output=True, |
| text=True, |
| timeout=90 |
| ) |
|
|
| if result.returncode == 0: |
| return result.stdout if result.stdout else "Code executed successfully" |
| else: |
| return f"Error: {result.stderr}" |
| except subprocess.TimeoutExpired: |
| return "Error: Code execution timed out" |
| except Exception as e: |
| return f"Error: {str(e)}" |
|
|
