| | import os |
| | import io |
| | import sys |
| | import uuid |
| | import base64 |
| | import traceback |
| | import contextlib |
| | import tempfile |
| | import subprocess |
| | import sqlite3 |
| | from typing import Dict, List, Any, Optional, Union |
| | import numpy as np |
| | import pandas as pd |
| | import matplotlib.pyplot as plt |
| | from PIL import Image |
| | from langchain_core.tools import tool |
| |
|
| | class CodeInterpreter: |
| | def __init__(self, allowed_modules=None, max_execution_time=30, working_directory=None): |
| | """Initialize the code interpreter with safety measures.""" |
| | |
| | self.allowed_modules = allowed_modules or ["numpy", "pandas", "matplotlib", "scipy", "sklearn", "math", "random", "statistics", "datetime", "collections", |
| | "itertools", "functools", "operator", "re", "json", "sympy", "networkx", "nltk", "PIL", "pytesseract", "cmath", "uuid", "tempfile", "requests", "urllib"] |
| | |
| | self.max_execution_time = max_execution_time |
| | self.working_directory = working_directory or os.path.join(os.getcwd()) |
| | if not os.path.exists(self.working_directory): |
| | os.makedirs(self.working_directory) |
| | |
| | self.globals = {"__builtins__": __builtins__, "np": np, "pd": pd, "plt": plt, "Image": Image} |
| | self.temp_sqlite_db = os.path.join(tempfile.gettempdir(), "code_exec.db") |
| |
|
| | def execute_code(self, code: str, language: str = "python") -> Dict[str, Any]: |
| | """Execute the provided code in the selected programming language.""" |
| | language = language.lower() |
| | execution_id = str(uuid.uuid4()) |
| | |
| | result = {"execution_id": execution_id, "status": "error", "stdout": "", "stderr": "", "result": None, "plots": [], "dataframes": []} |
| | |
| | try: |
| | if language == "python": |
| | return self._execute_python(code, execution_id) |
| | elif language == "bash": |
| | return self._execute_bash(code, execution_id) |
| | elif language == "sql": |
| | return self._execute_sql(code, execution_id) |
| | elif language == "c": |
| | return self._execute_c(code, execution_id) |
| | elif language == "java": |
| | return self._execute_java(code, execution_id) |
| | else: |
| | result["stderr"] = f"Unsupported language: {language}" |
| | except Exception as e: |
| | result["stderr"] = str(e) |
| | |
| | return result |
| |
|
| | def _execute_python(self, code: str, execution_id: str) -> dict: |
| | output_buffer = io.StringIO() |
| | error_buffer = io.StringIO() |
| | result = {"execution_id": execution_id, "status": "error", "stdout": "", "stderr": "", "result": None, "plots": [], "dataframes": []} |
| | |
| | try: |
| | exec_dir = os.path.join(self.working_directory, execution_id) |
| | os.makedirs(exec_dir, exist_ok=True) |
| | plt.switch_backend('Agg') |
| | |
| | with contextlib.redirect_stdout(output_buffer), contextlib.redirect_stderr(error_buffer): |
| | exec_result = exec(code, self.globals) |
| |
|
| | if plt.get_fignums(): |
| | for i, fig_num in enumerate(plt.get_fignums()): |
| | fig = plt.figure(fig_num) |
| | img_path = os.path.join(exec_dir, f"plot_{i}.png") |
| | fig.savefig(img_path) |
| | with open(img_path, "rb") as img_file: |
| | img_data = base64.b64encode(img_file.read()).decode('utf-8') |
| | result["plots"].append({"figure_number": fig_num, "data": img_data}) |
| |
|
| | for var_name, var_value in self.globals.items(): |
| | if isinstance(var_value, pd.DataFrame) and len(var_value) > 0: |
| | result["dataframes"].append({"name": var_name, "head": var_value.head().to_dict(), "shape": var_value.shape, "dtypes": str(var_value.dtypes)}) |
| | |
| | result["status"] = "success" |
| | result["stdout"] = output_buffer.getvalue() |
| | result["result"] = exec_result |
| | |
| | except Exception as e: |
| | result["status"] = "error" |
| | result["stderr"] = f"{error_buffer.getvalue()}\n{traceback.format_exc()}" |
| | |
| | return result |
| |
|
| | def _execute_bash(self, code: str, execution_id: str) -> dict: |
| | try: |
| | completed = subprocess.run(code, shell=True, capture_output=True, text=True, timeout=self.max_execution_time) |
| | return {"execution_id": execution_id, "status": "success" if completed.returncode == 0 else "error", "stdout": completed.stdout, "stderr": completed.stderr, "result": None, "plots": [], "dataframes": []} |
| | except subprocess.TimeoutExpired: |
| | return {"execution_id": execution_id, "status": "error", "stdout": "", "stderr": "Execution timed out.", "result": None, "plots": [], "dataframes": []} |
| |
|
| | def _execute_sql(self, code: str, execution_id: str) -> dict: |
| | result = {"execution_id": execution_id, "status": "error", "stdout": "", "stderr": "", "result": None, "plots": [], "dataframes": []} |
| | try: |
| | conn = sqlite3.connect(self.temp_sqlite_db) |
| | cur = conn.cursor() |
| | cur.execute(code) |
| | if code.strip().lower().startswith("select"): |
| | columns = [description[0] for description in cur.description] |
| | rows = cur.fetchall() |
| | df = pd.DataFrame(rows, columns=columns) |
| | result["dataframes"].append({"name": "query_result", "head": df.head().to_dict(), "shape": df.shape, "dtypes": str(df.dtypes)}) |
| | else: |
| | conn.commit() |
| | result["status"] = "success" |
| | result["stdout"] = "Query executed successfully." |
| |
|
| | except Exception as e: |
| | result["stderr"] = str(e) |
| | finally: |
| | conn.close() |
| |
|
| | return result |
| |
|
| | def _execute_c(self, code: str, execution_id: str) -> dict: |
| | temp_dir = tempfile.mkdtemp() |
| | source_path = os.path.join(temp_dir, "program.c") |
| | binary_path = os.path.join(temp_dir, "program") |
| |
|
| | try: |
| | with open(source_path, "w") as f: |
| | f.write(code) |
| |
|
| | compile_proc = subprocess.run(["gcc", source_path, "-o", binary_path], capture_output=True, text=True, timeout=self.max_execution_time) |
| | if compile_proc.returncode != 0: |
| | return {"execution_id": execution_id, "status": "error", "stdout": compile_proc.stdout, "stderr": compile_proc.stderr, "result": None, "plots": [], "dataframes": []} |
| |
|
| | run_proc = subprocess.run([binary_path], capture_output=True, text=True, timeout=self.max_execution_time) |
| | return {"execution_id": execution_id, "status": "success" if run_proc.returncode == 0 else "error", "stdout": run_proc.stdout, "stderr": run_proc.stderr, "result": None, "plots": [], "dataframes": []} |
| | except Exception as e: return {"execution_id": execution_id, "status": "error", "stdout": "", "stderr": str(e), "result": None, "plots": [], "dataframes": []} |
| |
|
| | def _execute_java(self, code: str, execution_id: str) -> dict: |
| | temp_dir = tempfile.mkdtemp() |
| | source_path = os.path.join(temp_dir, "Main.java") |
| |
|
| | try: |
| | with open(source_path, "w") as f: |
| | f.write(code) |
| |
|
| | compile_proc = subprocess.run(["javac", source_path], capture_output=True, text=True, timeout=self.max_execution_time) |
| | if compile_proc.returncode != 0: |
| | return {"execution_id": execution_id, "status": "error", "stdout": compile_proc.stdout, "stderr": compile_proc.stderr, "result": None, "plots": [], "dataframes": []} |
| |
|
| | run_proc = subprocess.run(["java", "-cp", temp_dir, "Main"], capture_output=True, text=True, timeout=self.max_execution_time) |
| | return {"execution_id": execution_id, "status": "success" if run_proc.returncode == 0 else "error", "stdout": run_proc.stdout, "stderr": run_proc.stderr, "result": None, "plots": [], "dataframes": []} |
| | except Exception as e: |
| | return {"execution_id": execution_id, "status": "error", "stdout": "", "stderr": str(e), "result": None, "plots": [], "dataframes": []} |
| |
|
| | interpreter_instance = CodeInterpreter() |
| |
|
| | @tool |
| | def execute_code_multilang(code: str, language: str = "python") -> str: |
| | """Execute code in multiple languages (Python, Bash, SQL, C, Java) and return results. |
| | Args: |
| | code (str): The source code to execute. |
| | language (str): The language of the code. Supported: "python", "bash", "sql", "c", "java". |
| | Returns: |
| | A string summarizing the execution results (stdout, stderr, errors, plots, dataframes if any). |
| | """ |
| | supported_languages = ["python", "bash", "sql", "c", "java"] |
| | language = language.lower() |
| |
|
| | if language not in supported_languages: |
| | return f"❌ Unsupported language: {language}. Supported languages are: {', '.join(supported_languages)}" |
| |
|
| | result = interpreter_instance.execute_code(code, language=language) |
| |
|
| | response = [] |
| |
|
| | if result["status"] == "success": |
| | response.append(f"✅ Code executed successfully in **{language.upper()}**") |
| |
|
| | if result.get("stdout"): |
| | response.append("\n**Standard Output:**\n```\n" + result["stdout"].strip() + "\n```") |
| |
|
| | if result.get("stderr"): |
| | response.append( |
| | "\n**Standard Error (if any):**\n```\n" |
| | + result["stderr"].strip() + "\n```") |
| |
|
| | if result.get("result") is not None: |
| | response.append( |
| | "\n**Execution Result:**\n```\n" |
| | + str(result["result"]).strip() + "\n```") |
| |
|
| | if result.get("dataframes"): |
| | for df_info in result["dataframes"]: |
| | response.append(f"\n**DataFrame `{df_info['name']}` (Shape: {df_info['shape']})**") |
| | df_preview = pd.DataFrame(df_info["head"]) |
| | response.append("First 5 rows:\n```\n" + str(df_preview) + "\n```") |
| |
|
| | if result.get("plots"): |
| | response.append(f"\n**Generated {len(result['plots'])} plot(s)** (Image data returned separately)") |
| |
|
| | else: |
| | response.append(f"❌ Code execution failed in **{language.upper()}**") |
| | if result.get("stderr"): |
| | response.append("\n**Error Log:**\n```\n" + result["stderr"].strip() + "\n```") |
| |
|
| | return "\n".join(response) |
