Spaces:
Runtime error
Runtime error
"""A tool for running python code in a REPL.""" | |
import ast | |
import sys | |
from typing import Dict, Optional | |
from pydantic import Field, root_validator | |
from langchain.python import PythonREPL | |
from langchain.tools.base import BaseTool | |
def _get_default_python_repl() -> PythonREPL: | |
return PythonREPL(_globals=globals(), _locals=None) | |
class PythonREPLTool(BaseTool): | |
"""A tool for running python code in a REPL.""" | |
name = "Python REPL" | |
description = ( | |
"A Python shell. Use this to execute python commands. " | |
"Input should be a valid python command. " | |
"If you want to see the output of a value, you should print it out " | |
"with `print(...)`." | |
) | |
python_repl: PythonREPL = Field(default_factory=_get_default_python_repl) | |
def _run(self, query: str) -> str: | |
"""Use the tool.""" | |
return self.python_repl.run(query) | |
async def _arun(self, query: str) -> str: | |
"""Use the tool asynchronously.""" | |
raise NotImplementedError("PythonReplTool does not support async") | |
class PythonAstREPLTool(BaseTool): | |
"""A tool for running python code in a REPL.""" | |
name = "python_repl_ast" | |
description = ( | |
"A Python shell. Use this to execute python commands. " | |
"Input should be a valid python command. " | |
"When using this tool, sometimes output is abbreviated - " | |
"make sure it does not look abbreviated before using it in your answer." | |
) | |
globals: Optional[Dict] = Field(default_factory=dict) | |
locals: Optional[Dict] = Field(default_factory=dict) | |
def validate_python_version(cls, values: Dict) -> Dict: | |
"""Validate valid python version.""" | |
if sys.version_info < (3, 9): | |
raise ValueError( | |
"This tool relies on Python 3.9 or higher " | |
"(as it uses new functionality in the `ast` module, " | |
f"you have Python version: {sys.version}" | |
) | |
return values | |
def _run(self, query: str) -> str: | |
"""Use the tool.""" | |
try: | |
tree = ast.parse(query) | |
module = ast.Module(tree.body[:-1], type_ignores=[]) | |
exec(ast.unparse(module), self.globals, self.locals) # type: ignore | |
module_end = ast.Module(tree.body[-1:], type_ignores=[]) | |
module_end_str = ast.unparse(module_end) # type: ignore | |
try: | |
return eval(module_end_str, self.globals, self.locals) | |
except Exception: | |
exec(module_end_str, self.globals, self.locals) | |
return "" | |
except Exception as e: | |
return str(e) | |
async def _arun(self, query: str) -> str: | |
"""Use the tool asynchronously.""" | |
raise NotImplementedError("PythonReplTool does not support async") | |