|
|
|
import copy |
|
import io |
|
from contextlib import redirect_stdout |
|
from typing import Any, Optional, Type |
|
|
|
from asyncer import asyncify |
|
|
|
from lagent.actions.base_action import AsyncActionMixin, BaseAction, tool_api |
|
from lagent.actions.parser import BaseParser, JsonParser |
|
from lagent.schema import ActionReturn, ActionStatusCode |
|
|
|
|
|
class GenericRuntime: |
|
GLOBAL_DICT = {} |
|
LOCAL_DICT = None |
|
HEADERS = [] |
|
|
|
def __init__(self): |
|
self._global_vars = copy.copy(self.GLOBAL_DICT) |
|
self._local_vars = copy.copy(self.LOCAL_DICT) if self.LOCAL_DICT else None |
|
|
|
for c in self.HEADERS: |
|
self.exec_code(c) |
|
|
|
def exec_code(self, code_piece: str) -> None: |
|
exec(code_piece, self._global_vars) |
|
|
|
def eval_code(self, expr: str) -> Any: |
|
return eval(expr, self._global_vars) |
|
|
|
|
|
class PythonInterpreter(BaseAction): |
|
"""A Python executor that can execute Python scripts. |
|
|
|
Args: |
|
answer_symbol (str, Optional): the answer symbol from LLM. Defaults to ``None``. |
|
answer_expr (str, Optional): the answer function name of the Python |
|
script. Defaults to ``'solution()'``. |
|
answer_from_stdout (boolean, Optional): whether the execution results is from |
|
stdout. Defaults to ``False``. |
|
timeout (int, Optional): Upper bound of waiting time for Python script execution. |
|
Defaults to ``20``. |
|
description (dict, Optional): The description of the action. Defaults to ``None``. |
|
parser (Type[BaseParser]): The parser class to process the |
|
action's inputs and outputs. Defaults to :class:`JsonParser`. |
|
""" |
|
|
|
def __init__( |
|
self, |
|
answer_symbol: Optional[str] = None, |
|
answer_expr: Optional[str] = 'solution()', |
|
answer_from_stdout: bool = False, |
|
timeout: int = 20, |
|
description: Optional[dict] = None, |
|
parser: Type[BaseParser] = JsonParser, |
|
) -> None: |
|
super().__init__(description, parser) |
|
self.answer_symbol = answer_symbol |
|
self.answer_expr = answer_expr |
|
self.answer_from_stdout = answer_from_stdout |
|
self.timeout = timeout |
|
|
|
@tool_api |
|
def run(self, command: str) -> ActionReturn: |
|
"""用来执行Python代码。代码必须是一个函数，函数名必须得是 'solution'，代码对应你的思考过程。代码实例格式如下： |
|
|
|
```python |
|
# import 依赖包 |
|
import xxx |
|
def solution(): |
|
# 初始化一些变量 |
|
variable_names_with_real_meaning = xxx |
|
# 步骤一 |
|
mid_variable = func(variable_names_with_real_meaning) |
|
# 步骤 x |
|
mid_variable = func(mid_variable) |
|
# 最后结果 |
|
final_answer = func(mid_variable) |
|
return final_answer |
|
``` |
|
|
|
Args: |
|
command (:class:`str`): Python code snippet |
|
""" |
|
from func_timeout import FunctionTimedOut, func_set_timeout |
|
|
|
self.runtime = GenericRuntime() |
|
try: |
|
tool_return = func_set_timeout(self.timeout)(self._call)(command) |
|
except FunctionTimedOut as e: |
|
tool_return = ActionReturn(type=self.name) |
|
tool_return.errmsg = repr(e) |
|
tool_return.state = ActionStatusCode.API_ERROR |
|
return tool_return |
|
|
|
def _call(self, command: str) -> ActionReturn: |
|
tool_return = ActionReturn(type=self.name) |
|
try: |
|
if '```python' in command: |
|
command = command.split('```python')[1].split('```')[0] |
|
elif '```' in command: |
|
command = command.split('```')[1].split('```')[0] |
|
tool_return.args = dict(text='```python\n' + command + '\n```') |
|
command = command.split('\n') |
|
|
|
if self.answer_from_stdout: |
|
program_io = io.StringIO() |
|
with redirect_stdout(program_io): |
|
self.runtime.exec_code('\n'.join(command)) |
|
program_io.seek(0) |
|
res = program_io.readlines()[-1] |
|
elif self.answer_symbol: |
|
self.runtime.exec_code('\n'.join(command)) |
|
res = self.runtime._global_vars[self.answer_symbol] |
|
elif self.answer_expr: |
|
self.runtime.exec_code('\n'.join(command)) |
|
res = self.runtime.eval_code(self.answer_expr) |
|
else: |
|
self.runtime.exec_code('\n'.join(command[:-1])) |
|
res = self.runtime.eval_code(command[-1]) |
|
except Exception as e: |
|
tool_return.errmsg = repr(e) |
|
tool_return.type = self.name |
|
tool_return.state = ActionStatusCode.API_ERROR |
|
return tool_return |
|
try: |
|
tool_return.result = [dict(type='text', content=str(res))] |
|
tool_return.state = ActionStatusCode.SUCCESS |
|
except Exception as e: |
|
tool_return.errmsg = repr(e) |
|
tool_return.type = self.name |
|
tool_return.state = ActionStatusCode.API_ERROR |
|
return tool_return |
|
|
|
|
|
class AsyncPythonInterpreter(AsyncActionMixin, PythonInterpreter): |
|
"""A Python executor that can execute Python scripts. |
|
|
|
Args: |
|
answer_symbol (str, Optional): the answer symbol from LLM. Defaults to ``None``. |
|
answer_expr (str, Optional): the answer function name of the Python |
|
script. Defaults to ``'solution()'``. |
|
answer_from_stdout (boolean, Optional): whether the execution results is from |
|
stdout. Defaults to ``False``. |
|
timeout (int, Optional): Upper bound of waiting time for Python script execution. |
|
Defaults to ``20``. |
|
description (dict, Optional): The description of the action. Defaults to ``None``. |
|
parser (Type[BaseParser]): The parser class to process the |
|
action's inputs and outputs. Defaults to :class:`JsonParser`. |
|
""" |
|
|
|
@tool_api |
|
@asyncify |
|
def run(self, command: str) -> ActionReturn: |
|
"""用来执行Python代码。代码必须是一个函数，函数名必须得是 'solution'，代码对应你的思考过程。代码实例格式如下： |
|
|
|
```python |
|
# import 依赖包 |
|
import xxx |
|
def solution(): |
|
# 初始化一些变量 |
|
variable_names_with_real_meaning = xxx |
|
# 步骤一 |
|
mid_variable = func(variable_names_with_real_meaning) |
|
# 步骤 x |
|
mid_variable = func(mid_variable) |
|
# 最后结果 |
|
final_answer = func(mid_variable) |
|
return final_answer |
|
``` |
|
|
|
Args: |
|
command (:class:`str`): Python code snippet |
|
""" |
|
return super().run(command) |
|
|
