Spaces:
Paused
Paused
File size: 6,316 Bytes
d61265e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
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()
@tool
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) |