refactor(tools): overhaul Python REPL component with modern tool mode (#5463)

* refactor(tools): overhaul Python REPL component with modern tool mode

BREAKING CHANGE: Complete redesign of PythonREPL component architecture

- Replace legacy tool mode with modern tool_mode=true implementation
- Add automatic import detection for both global and from-imports
- Remove manual global_imports field in favor of automatic detection
- Implement CodeInput type support with FieldTypes.CODE
- Update schema to handle new parameters for agent compatibility
- Improve error handling and logging for import failures

The component now automatically handles imports without manual configuration,
supports modern tool mode, and provides better integration with the agent
system through updated schema definitions.

* refactor(python-repl): remove name parameter

* feat(security): add required global imports validation

- Add explicit global imports input for security control
- Set both global_imports and python_code as required fields
- Remove AST-based import analysis in favor of explicit imports

* [autofix.ci] apply automated fixes

* fix: remove generic exception handling in python_repl

Removed overly broad exception handling to comply with linting rules.

* fix

* test(python-repl): update component test to match current implementation

- Remove assertions for deprecated outputs (api_run_model, api_build_tool)
- Add assertions for current 'results' output
- Add detailed input validation tests for global_imports and python_code
- Verify input configurations including type, default values, and required status
- Ensure component template structure matches frontend requirements

This change aligns the test suite with the current PythonREPLToolComponent
implementation, improving test coverage and maintaining component reliability.

* revert(tools): restore Python REPL component to original implementation

Due to backward compatibility concerns, reverting the Python REPL component
to its initial implementation state to maintain stability and prevent breaking changes.

* feat(tools): mark Python REPL component as deprecated & Legacy

* feat(tools): add Python REPL Core component

* [autofix.ci] apply automated fixes

* fix(tests): mark Python REPL Core as unreleased component

* fix(__init__): fix

* feat(python-repl): improve example clarity for printing the results

* fix(tests): resolve KeyError in test_component_versions

* style(python-repl): format description string to follow PEP 8 guidelines

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Edwin Jose <edwin.jose@datastax.com>
This commit is contained in:
VICTOR CORREA GOMES 2025-01-20 08:54:32 -03:00 • committed by GitHub
commit 6c79a3d6aa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 153 additions and 27 deletions

View file

@ -15,6 +15,7 @@ from .google_serper_api_core import GoogleSerperAPICore
from .mcp_stdio import MCPStdio
from .python_code_structured_tool import PythonCodeStructuredTool
from .python_repl import PythonREPLToolComponent
from .python_repl_core import PythonREPLComponent
from .search_api import SearchAPIComponent
from .searxng import SearXNGToolComponent
from .serp import SerpComponent
@ -47,6 +48,7 @@ __all__ = [
"GoogleSerperAPICore",
"MCPStdio",
"PythonCodeStructuredTool",
"PythonREPLComponent",
"PythonREPLToolComponent",
"SearXNGToolComponent",
"SearchAPIComponent",

View file

@ -13,10 +13,11 @@ from langflow.schema import Data
class PythonREPLToolComponent(LCToolComponent):
display_name = "Python REPL"
display_name = "Python REPL [DEPRECATED]"
description = "A tool for running Python code in a REPL environment."
name = "PythonREPLTool"
icon = "Python"
legacy = True
inputs = [
StrInput(

View file

@ -0,0 +1,99 @@
import importlib
from langchain_experimental.utilities import PythonREPL
from langflow.custom import Component
from langflow.io import CodeInput, Output, StrInput
from langflow.schema import Data
class PythonREPLComponent(Component):
display_name = "Python REPL"
description = (
"A Python code executor that lets you run Python code with specific imported modules. "
"Remember to always use print() to see your results. Example: print(df.head())"
)
icon = "Python"
inputs = [
StrInput(
name="global_imports",
display_name="Global Imports",
info="A comma-separated list of modules to import globally, e.g. 'math,numpy,pandas'.",
value="math,pandas",
required=True,
),
CodeInput(
name="python_code",
display_name="Python Code",
info="The Python code to execute. Only modules specified in Global Imports can be used.",
value="print('Hello, World!')",
tool_mode=True,
required=True,
),
]
outputs = [
Output(
display_name="Results",
name="results",
type_=Data,
method="run_python_repl",
),
]
def get_globals(self, global_imports: str | list[str]) -> dict:
"""Create a globals dictionary with only the specified allowed imports."""
global_dict = {}
try:
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}: {e!s}"
raise ImportError(msg) from e
except Exception as e:
self.log(f"Error in global imports: {e!s}")
raise
else:
self.log(f"Successfully imported modules: {list(global_dict.keys())}")
return global_dict
def run_python_repl(self) -> Data:
try:
globals_ = self.get_globals(self.global_imports)
python_repl = PythonREPL(_globals=globals_)
result = python_repl.run(self.python_code)
result = result.strip() if result else ""
self.log("Code execution completed successfully")
return Data(data={"result": result})
except ImportError as e:
error_message = f"Import Error: {e!s}"
self.log(error_message)
return Data(data={"error": error_message})
except SyntaxError as e:
error_message = f"Syntax Error: {e!s}"
self.log(error_message)
return Data(data={"error": error_message})
except (NameError, TypeError, ValueError) as e:
error_message = f"Error during execution: {e!s}"
self.log(error_message)
return Data(data={"error": error_message})
def build(self):
return self.run_python_repl

View file

@ -15,6 +15,7 @@ _convert_field_type_to_type: dict[FieldTypes, type] = {
FieldTypes.TABLE: dict,
FieldTypes.FILE: str,
FieldTypes.PROMPT: str,
FieldTypes.CODE: str,
FieldTypes.OTHER: str,
}