Repository navigation
Expand file tree
/
Copy pathcommands.py
More file actions
127 lines (101 loc) · 4.03 KB
/
Copy pathcommands.py
File metadata and controls
127 lines (101 loc) · 4.03 KB
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
"""
This module offers a collection of static methods that can execute different commands.
"""
import os
import subprocess
from io import StringIO
from contextlib import redirect_stdout
from duckduckgo_search import DDGS
from parallel_search import search as parallel_search
# pylint: disable=broad-exception-caught, exec-used, unspecified-encoding
class Commands:
"""
A collection of static methods that can execute different commands.
"""
@staticmethod
def execute_command(command, arg, search_provider=None, search_session_id=None) -> str:
"""
Executes the command corresponding to the provided command string.
Args:
command (str): The command string representing the command to be executed.
arg (str): The argument to be passed to the command.
search_provider (str, optional): Provider override for web_search.
search_session_id (str, optional): Research identity for related searches.
Returns:
str: The result of the command execution, or an error
message if an exception is raised during execution.
"""
try:
match command:
case "memorize_thoughts":
result = Commands.memorize_thoughts(arg)
case "execute_python":
result = Commands.execute_python(arg)
case "execute_shell":
result = Commands.execute_shell(arg)
case "web_search":
result = Commands.web_search(arg, search_provider, search_session_id)
case _:
result = f"Unknown command: {command}"
except Exception as exception:
result = f"Command returned an error:\n{str(exception)}"
return result
@staticmethod
def memorize_thoughts(arg: str) -> str:
"""
Simply returns the input string. Used to 'memorize' a thought.
Args:
arg (str): The input string.
Returns:
str: The input string.
"""
return arg
@staticmethod
def execute_python(arg: str) -> str:
"""
Executes the input Python code and returns the stdout.
Args:
arg (str): The input Python code.
Returns:
str: The stdout produced by the executed Python code.
"""
_stdout = StringIO()
with redirect_stdout(_stdout):
exec(arg)
return _stdout.getvalue()
@staticmethod
def execute_shell(arg: str) -> str:
"""
Executes the input shell command and returns the stdout and stderr.
Args:
arg (str): The input shell command.
Returns:
str: The stdout and stderr produced by the executed shell command.
"""
result = subprocess.run(arg, capture_output=True, shell=True, check=False)
stdout = result.stdout.decode("utf-8")
stderr = result.stderr.decode("utf-8")
return f"STDOUT:\n{stdout}\nSTDERR:\n{stderr}"
@staticmethod
def web_search(arg: str, provider=None, session_id=None) -> str:
"""
Searches with the selected provider; DuckDuckGo remains the default.
Args:
arg (str): The search query.
provider (str, optional): Override SEARCH_PROVIDER without changing the environment.
session_id (str, optional): Research identity shared by one agent objective.
Returns:
str: The search results.
"""
if provider is None:
provider = os.getenv("SEARCH_PROVIDER", "duckduckgo")
provider = provider.strip().lower()
if provider == "none":
return "Web search is disabled (SEARCH_PROVIDER=none)."
if provider == "parallel":
return parallel_search(arg, session_id)
if provider != "duckduckgo":
raise ValueError("SEARCH_PROVIDER must be duckduckgo, parallel, or none")
ddgs = DDGS()
ddgs_text_gen = ddgs.text(arg)
return str(list(ddgs_text_gen)[:5])