Spaces:
Sleeping
Sleeping
File size: 6,741 Bytes
e679d69 |
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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
# flake8: noqa: E501
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)
|