| 1607 | |
| 1608 | |
| 1609 | class LocalPythonExecutor(PythonExecutor): |
| 1610 | def __init__( |
| 1611 | self, |
| 1612 | additional_authorized_imports: List[str], |
| 1613 | max_print_outputs_length: Optional[int] = None, |
| 1614 | set_timeout: bool = False |
| 1615 | ): |
| 1616 | self.custom_tools = {} |
| 1617 | self.state = {} |
| 1618 | self.max_print_outputs_length = max_print_outputs_length |
| 1619 | if max_print_outputs_length is None: |
| 1620 | self.max_print_outputs_length = DEFAULT_MAX_LEN_OUTPUT |
| 1621 | self.additional_authorized_imports = additional_authorized_imports |
| 1622 | self.authorized_imports = list(set(BASE_BUILTIN_MODULES) | set(self.additional_authorized_imports)) |
| 1623 | # TODO: assert self.authorized imports are all installed locally |
| 1624 | self.static_tools = None |
| 1625 | self.set_timeout = set_timeout |
| 1626 | |
| 1627 | def __call__(self, code_action: str) -> Tuple[Any, str, bool]: |
| 1628 | output, is_final_answer = evaluate_python_code( |
| 1629 | code_action, |
| 1630 | static_tools=self.static_tools, |
| 1631 | custom_tools=self.custom_tools, |
| 1632 | state=self.state, |
| 1633 | authorized_imports=self.authorized_imports, |
| 1634 | max_print_outputs_length=self.max_print_outputs_length, |
| 1635 | set_timeout=self.set_timeout, |
| 1636 | ) |
| 1637 | logs = str(self.state["_print_outputs"]) |
| 1638 | return output, logs, is_final_answer |
| 1639 | |
| 1640 | def send_variables(self, variables: dict): |
| 1641 | self.state.update(variables) |
| 1642 | |
| 1643 | def send_tools(self, tools: Dict[str, Tool]): |
| 1644 | self.static_tools = {**tools, **BASE_PYTHON_TOOLS.copy()} |
| 1645 | |
| 1646 | |
| 1647 | __all__ = ["evaluate_python_code", "LocalPythonExecutor"] |
no outgoing calls
no test coverage detected