Spaces:
Runtime error
Runtime error
File size: 2,831 Bytes
58d33f0 |
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 |
"""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)
@root_validator(pre=True)
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")
|