Spaces:
Paused
Paused
| import subprocess | |
| import sys | |
| import os | |
| import re | |
| import json | |
| import pandas as pd | |
| import tempfile | |
| from duckduckgo_search import DDGS | |
| from langchain_core.tools import tool | |
| class CodeInterpreter: | |
| def __init__(self): | |
| self.sessions = {} | |
| def execute_code(self, code: str, language: str = "python") -> dict: | |
| if language == "python": | |
| return self._execute_python(code) | |
| elif language == "bash": | |
| return self._execute_bash(code) | |
| elif language == "sql": | |
| return self._execute_sql(code) | |
| elif language == "c": | |
| return self._execute_c(code) | |
| elif language == "java": | |
| return self._execute_java(code) | |
| else: | |
| return {"status": "error", "stderr": f"Unsupported language: {language}"} | |
| def _execute_python(self, code: str) -> dict: | |
| try: | |
| # Use a temporary file to execute the Python code | |
| with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.py') as temp_file: | |
| temp_file.write(code) | |
| temp_file_path = temp_file.name | |
| process = subprocess.run( | |
| [sys.executable, temp_file_path], | |
| capture_output=True, | |
| text=True, | |
| timeout=60 | |
| ) | |
| os.remove(temp_file_path) # Clean up the temporary file | |
| return { | |
| "status": "success" if process.returncode == 0 else "error", | |
| "stdout": process.stdout, | |
| "stderr": process.stderr, | |
| "result": None, # Python execution doesn't typically have a single 'result' like an expression | |
| "dataframes": [], # Placeholder for dataframe detection | |
| "plots": [] # Placeholder for plot detection | |
| } | |
| except subprocess.TimeoutExpired: | |
| os.remove(temp_file_path) | |
| return {"status": "error", "stderr": "Python execution timed out."} | |
| except Exception as e: | |
| return {"status": "error", "stderr": str(e)} | |
| def _execute_bash(self, code: str) -> dict: | |
| try: | |
| process = subprocess.run( | |
| ["bash", "-c", code], | |
| capture_output=True, | |
| text=True, | |
| timeout=60 | |
| ) | |
| return { | |
| "status": "success" if process.returncode == 0 else "error", | |
| "stdout": process.stdout, | |
| "stderr": process.stderr, | |
| } | |
| except subprocess.TimeoutExpired: | |
| return {"status": "error", "stderr": "Bash execution timed out."} | |
| except Exception as e: | |
| return {"status": "error", "stderr": str(e)} | |
| def _execute_sql(self, code: str) -> dict: | |
| # This is a placeholder. A real implementation would connect to a database. | |
| return {"status": "error", "stderr": "SQL execution not fully implemented. Requires database connection."} | |
| def _execute_c(self, code: str) -> dict: | |
| # This is a placeholder. A real implementation would compile and run C code. | |
| return {"status": "error", "stderr": "C execution not fully implemented. Requires C compiler."} | |
| def _execute_java(self, code: str) -> dict: | |
| # This is a placeholder. A real implementation would compile and run Java code. | |
| return {"status": "error", "stderr": "Java execution not fully implemented. Requires Java compiler."} | |
| interpreter_instance = CodeInterpreter() | |
| def execute_code_multilang(code: str, language: str = "python") -> str: | |
| """Executes code in a variety of languages and returns the output. | |
| This powerful tool can run code in Python, Bash, SQL, C, and Java. | |
| It is your primary tool for any task that requires code execution, such as: | |
| - Data analysis and manipulation (e.g., using pandas in Python). | |
| - File system operations (e.g., using Bash commands). | |
| - Complex calculations and algorithms. | |
| When using Python, you have access to a rich set of pre-installed libraries, | |
| including pandas, numpy, and scikit-learn. | |
| Args: | |
| code (str): The source code to be executed. | |
| language (str): The programming language of the code. | |
| Supported languages: "python", "bash", "sql", "c", "java". | |
| Returns: | |
| A string summarizing the execution results, including standard output, | |
| standard error, and any generated plots or dataframes. | |
| """ | |
| 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) |