Spaces:
Running
Running
| import importlib | |
| from langchain.tools import StructuredTool | |
| from langchain_core.tools import ToolException | |
| from langchain_experimental.utilities import PythonREPL | |
| from loguru import logger | |
| from pydantic import BaseModel, Field | |
| from langflow.base.langchain_utilities.model import LCToolComponent | |
| from langflow.field_typing import Tool | |
| from langflow.inputs import StrInput | |
| from langflow.schema import Data | |
| class PythonREPLToolComponent(LCToolComponent): | |
| display_name = "Python REPL" | |
| description = "A tool for running Python code in a REPL environment." | |
| name = "PythonREPLTool" | |
| icon = "Python" | |
| inputs = [ | |
| StrInput( | |
| name="name", | |
| display_name="Tool Name", | |
| info="The name of the tool.", | |
| value="python_repl", | |
| ), | |
| StrInput( | |
| name="description", | |
| display_name="Tool Description", | |
| info="A description of the tool.", | |
| value="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(...)`.", | |
| ), | |
| StrInput( | |
| name="global_imports", | |
| display_name="Global Imports", | |
| info="A comma-separated list of modules to import globally, e.g. 'math,numpy'.", | |
| value="math", | |
| ), | |
| StrInput( | |
| name="code", | |
| display_name="Python Code", | |
| info="The Python code to execute.", | |
| value="print('Hello, World!')", | |
| ), | |
| ] | |
| class PythonREPLSchema(BaseModel): | |
| code: str = Field(..., description="The Python code to execute.") | |
| def get_globals(self, global_imports: str | list[str]) -> dict: | |
| global_dict = {} | |
| if isinstance(global_imports, str): | |
| modules = [module.strip() for module in global_imports.split(",")] | |
| elif isinstance(global_imports, list): | |
| modules = global_imports | |
| else: | |
| msg = "global_imports must be either a string or a list" | |
| raise TypeError(msg) | |
| for module in modules: | |
| try: | |
| imported_module = importlib.import_module(module) | |
| global_dict[imported_module.__name__] = imported_module | |
| except ImportError as e: | |
| msg = f"Could not import module {module}" | |
| raise ImportError(msg) from e | |
| return global_dict | |
| def build_tool(self) -> Tool: | |
| globals_ = self.get_globals(self.global_imports) | |
| python_repl = PythonREPL(_globals=globals_) | |
| def run_python_code(code: str) -> str: | |
| try: | |
| return python_repl.run(code) | |
| except Exception as e: | |
| logger.opt(exception=True).debug("Error running Python code") | |
| raise ToolException(str(e)) from e | |
| tool = StructuredTool.from_function( | |
| name=self.name, | |
| description=self.description, | |
| func=run_python_code, | |
| args_schema=self.PythonREPLSchema, | |
| ) | |
| self.status = f"Python REPL Tool created with global imports: {self.global_imports}" | |
| return tool | |
| def run_model(self) -> list[Data]: | |
| tool = self.build_tool() | |
| result = tool.run(self.code) | |
| return [Data(data={"result": result})] | |