usage: conn profile [-h] (--add | --del | --mod | --show) profile
@@ -292,6 +293,7 @@ Commands:
run Run scripts or commands on nodes
config Manage app config
api Start and stop connpy api
+ ai Make request to an AI
```
### Manage profiles
@@ -541,7 +543,7 @@ __pdoc__ = {
class ai
-(config, org=None, api_key=None, model='gpt-3.5-turbo', temp=0.7)
+(config, org=None, api_key=None, model=None, temp=0.7)
-
This class generates a ai object. Containts all the information and methods to make requests to openAI chatGPT to run actions on the application.
@@ -587,7 +589,7 @@ __pdoc__ = {
'''
- def __init__(self, config, org = None, api_key = None, model = "gpt-3.5-turbo", temp = 0.7):
+ def __init__(self, config, org = None, api_key = None, model = None, temp = 0.7):
'''
### Parameters:
@@ -627,18 +629,25 @@ __pdoc__ = {
openai.api_key = self.config.config["openai"]["api_key"]
except:
raise ValueError("Missing openai api_key")
+ if model:
+ self.model = model
+ else:
+ try:
+ self.model = self.config.config["openai"]["model"]
+ except:
+ self.model = "gpt-3.5-turbo-0613"
+ self.temp = temp
self.__prompt = {}
self.__prompt["original_system"] = """
- You are the AI assistant of a network connection manager and automation app called connpy. When provided with user input analyze the input and extract the following information:
+ You are the AI chatbot and assistant of a network connection manager and automation app called connpy. When provided with user input analyze the input and extract the following information. If user wants to chat just reply and don't call a function:
- - app_related: True if the input is related to the application's purpose and the request is understood; False if the input is not related, not understood, or if mandatory information like filter is missing. If user ask information about the app it should be false
- type: Given a user input, identify the type of request they want to make. The input will represent one of two options:
1. "command" - The user wants to get information from devices by running commands.
2. "list_nodes" - The user wants to get a list of nodes, devices, servers, or routers.
The 'type' field should reflect whether the user input is a command or a request for a list of nodes.
- - filter: One or more regex patterns indicating the device or group of devices the command should be run on, returned as a Python list (e.g., ['hostname', 'hostname@folder', '@subfolder@folder']). The filter can have different formats, such as:
+ - filter: One or more regex patterns indicating the device or group of devices the command should be run on. The filter can have different formats, such as:
- hostname
- hostname@folder
- hostname@subfolder@folder
@@ -649,41 +658,46 @@ __pdoc__ = {
The filter should be extracted from the user input exactly as it was provided.
Always preserve the exact filter pattern provided by the user, with no modifications. Do not process any regex, the application can do that.
- If no filter is specified, set it to None.
- - Expected: This field represents an expected output to search for when running the command. It's an optional value for the user.
-Set it to 'None' if no value was captured.
-The expected value should ALWAYS come from the user input explicitly.
-Users will typically use words like verify, check, make sure, or similar to refer to the expected value.
-
- - response: An optional field to be filled when app_related is False or when providing an explanation related to the app. This is where you can engage in small talk, answer questions not related to the app, or provide explanations about the extracted information.
-
- Always respond in the following format:
-
- app_related: {{app_related}}
- Type: {{command}}
- Filter: {{filter}}
- Expected: {{expected}}
- Response: {{response}}
"""
self.__prompt["original_user"] = "Get the IP addresses of loopback0 for all routers from w2az1 and e1.*(prod|dev) and check if they have the ip 192.168.1.1"
- self.__prompt["original_assistant"] = "app_related: True\nType: Command\nFilter: ['w2az1', 'e1.*(prod|dev)']\nExpected: 192.168.1.1"
+ self.__prompt["original_assistant"] = {"name": "get_network_device_info", "arguments": "{\n \"type\": \"command\",\n \"filter\": [\"w2az1\",\"e1.*(prod|dev)\"]\n}"}
+ self.__prompt["original_function"] = {}
+ self.__prompt["original_function"]["name"] = "get_network_device_info"
+ self.__prompt["original_function"]["descriptions"] = "You are the AI chatbot and assistant of a network connection manager and automation app called connpy. When provided with user input analyze the input and extract the information acording to the function, If user wants to chat just reply and don't call a function",
+ self.__prompt["original_function"]["parameters"] = {}
+ self.__prompt["original_function"]["parameters"]["type"] = "object"
+ self.__prompt["original_function"]["parameters"]["properties"] = {}
+ self.__prompt["original_function"]["parameters"]["properties"]["type"] = {}
+ self.__prompt["original_function"]["parameters"]["properties"]["type"]["type"] = "string"
+ self.__prompt["original_function"]["parameters"]["properties"]["type"]["description"] ="""
+Categorize the user's request based on the operation they want to perform on the nodes. The requests can be classified into the following categories:
+
+ 1. "command" - This represents a request to retrieve specific information or configurations from nodes. An example would be: "go to routers in @office and get the config".
+
+ 2. "list_nodes" - This is when the user wants a list of nodes. An example could be: "get me the nodes in @office".
+"""
+ self.__prompt["original_function"]["parameters"]["properties"]["type"]["enum"] = ["command", "list_nodes"]
+ self.__prompt["original_function"]["parameters"]["properties"]["filter"] = {}
+ self.__prompt["original_function"]["parameters"]["properties"]["filter"]["type"] = "array"
+ self.__prompt["original_function"]["parameters"]["properties"]["filter"]["items"] = {}
+ self.__prompt["original_function"]["parameters"]["properties"]["filter"]["items"]["type"] = "string"
+ self.__prompt["original_function"]["parameters"]["properties"]["filter"]["items"]["description"] = """One or more regex patterns indicating the device or group of devices the command should be run on. The filter should be extracted from the user input exactly as it was provided.
+ The filter can have different formats, such as:
+ - hostname
+ - hostname@folder
+ - hostname@subfolder@folder
+ - partofhostname
+ - @folder
+ - @subfolder@folder
+ - regex_pattern
+ """
+ self.__prompt["original_function"]["parameters"]["required"] = ["type", "filter"]
self.__prompt["command_system"] = """
- For each device listed below, provide the command(s) needed to perform the specified action, depending on the device OS (e.g., Cisco IOSXR router, Linux server). Always format your response as a Python list (e.g., ['command1', 'command2']).
-
+ For each device listed below, provide the command(s) needed to perform the specified action, depending on the device OS (e.g., Cisco IOSXR router, Linux server).
The application knows how to connect to devices via SSH, so you only need to provide the command(s) to run after connecting.
-
If the commands needed are not for the specific OS type, just send an empty list (e.g., []).
-
- It is crucial to always include the device name provided in your response, even when there is only one device.
-
Note: Preserving the integrity of user-provided commands is of utmost importance. If a user has provided a specific command to run, include that command exactly as it was given, even if it's not recognized or understood. Under no circumstances should you modify or alter user-provided commands.
-
- Your response has to be always like this:
- node1: ["command1", "command2"]
- node2: ["command1", "command2", "command3"]
- node1@folder: ["command1"]
- Node4@subfolder@folder: []
"""
self.__prompt["command_user"]= """
input: show me the full configuration for all this devices:
@@ -691,20 +705,44 @@ Users will typically use words like verify, check, make sure, or similar to refe
Devices:
router1: cisco ios
"""
- self.__prompt["command_assistant"]= """
- router1: ['show running-config']
+ self.__prompt["command_assistant"] = {"name": "get_commands", "arguments": "{\n \"router1\": \"show running-configuration\"\n}"}
+ self.__prompt["command_function"] = {}
+ self.__prompt["command_function"]["name"] = "get_commands"
+ self.__prompt["command_function"]["descriptions"] = """
+ For each device listed below, provide the command(s) needed to perform the specified action, depending on the device OS (e.g., Cisco IOSXR router, Linux server).
+ The application knows how to connect to devices via SSH, so you only need to provide the command(s) to run after connecting.
+ If the commands needed are not for the specific OS type, just send an empty list (e.g., []).
"""
+ self.__prompt["command_function"]["parameters"] = {}
+ self.__prompt["command_function"]["parameters"]["type"] = "object"
+ self.__prompt["command_function"]["parameters"]["properties"] = {}
self.__prompt["confirmation_system"] = """
Please analyze the user's input and categorize it as either an affirmation or negation. Based on this analysis, respond with:
- 'True' if the input is an affirmation like 'do it', 'go ahead', 'sure', etc.
- 'False' if the input is a negation.
- If the input does not fit into either of these categories, kindly express that you didn't understand and request the user to rephrase their response.
+ 'true' if the input is an affirmation like 'do it', 'go ahead', 'sure', etc.
+ 'false' if the input is a negation.
+ 'none' If the input does not fit into either of these categories.
"""
self.__prompt["confirmation_user"] = "Yes go ahead!"
self.__prompt["confirmation_assistant"] = "True"
- self.model = model
- self.temp = temp
+ self.__prompt["confirmation_function"] = {}
+ self.__prompt["confirmation_function"]["name"] = "get_confirmation"
+ self.__prompt["confirmation_function"]["descriptions"] = """
+ Analize user request and respond:
+ """
+ self.__prompt["confirmation_function"]["parameters"] = {}
+ self.__prompt["confirmation_function"]["parameters"]["type"] = "object"
+ self.__prompt["confirmation_function"]["parameters"]["properties"] = {}
+ self.__prompt["confirmation_function"]["parameters"]["properties"]["result"] = {}
+ self.__prompt["confirmation_function"]["parameters"]["properties"]["result"]["description"] = """'true' if the input is an affirmation like 'do it', 'go ahead', 'sure', etc.
+'false' if the input is a negation.
+'none' If the input does not fit into either of these categories"""
+ self.__prompt["confirmation_function"]["parameters"]["properties"]["result"]["type"] = "string"
+ self.__prompt["confirmation_function"]["parameters"]["properties"]["result"]["enum"] = ["true", "false", "none"]
+ self.__prompt["confirmation_function"]["parameters"]["properties"]["response"] = {}
+ self.__prompt["confirmation_function"]["parameters"]["properties"]["response"]["description"] = "If the user don't message is not an affiramtion or negation, kindly ask the user to rephrase."
+ self.__prompt["confirmation_function"]["parameters"]["properties"]["response"]["type"] = "string"
+ self.__prompt["confirmation_function"]["parameters"]["required"] = ["result"]
def process_string(self, s):
if s.startswith('[') and s.endswith(']') and not (s.startswith("['") and s.endswith("']")) and not (s.startswith('["') and s.endswith('"]')):
@@ -732,83 +770,37 @@ Users will typically use words like verify, check, make sure, or similar to refe
myfunction = False
return myfunction
- def _clean_original_response(self, raw_response):
- #Parse response for first request to openAI GPT.
- info_dict = {}
- info_dict["app_related"] = False
- current_key = "response"
- for line in raw_response.split("\n"):
- if line.strip() == "":
- line = "\n"
- possible_keys = ["app_related", "type", "filter", "expected", "response"]
- if ':' in line and (key := line.split(':', 1)[0].strip().lower()) in possible_keys:
- key, value = line.split(":", 1)
- key = key.strip().lower()
- value = value.strip()
- # Convert "true" or "false" (case-insensitive) to Python boolean
- if value.lower() == "true":
- value = True
- elif value.lower() == "false":
- value = False
- elif value.lower() == "none":
- value = None
- if key == "filter":
- value = self.process_string(value)
- value = ast.literal_eval(value)
- #store in dictionary
- info_dict[key] = value
- current_key = key
- else:
- if current_key == "response":
- if "response" in info_dict:
- info_dict[current_key] += "\n" + line
- else:
- info_dict[current_key] = line
-
- return info_dict
-
def _clean_command_response(self, raw_response):
#Parse response for command request to openAI GPT.
info_dict = {}
info_dict["commands"] = []
info_dict["variables"] = {}
info_dict["variables"]["__global__"] = {}
- for line in raw_response.split("\n"):
- if ":" in line:
- key, value = line.split(":", 1)
- key = key.strip()
- newvalue = {}
- pattern = r'\[.*?\]'
- match = re.search(pattern, value.strip())
- try:
- value = ast.literal_eval(match.group(0))
- for i,e in enumerate(value, start=1):
- newvalue[f"command{i}"] = e
- if f"{{command{i}}}" not in info_dict["commands"]:
- info_dict["commands"].append(f"{{command{i}}}")
- info_dict["variables"]["__global__"][f"command{i}"] = ""
- info_dict["variables"][key] = newvalue
- except:
- pass
+ for key, value in raw_response.items():
+ key = key.strip()
+ newvalue = {}
+ for i,e in enumerate(value, start=1):
+ newvalue[f"command{i}"] = e
+ if f"{{command{i}}}" not in info_dict["commands"]:
+ info_dict["commands"].append(f"{{command{i}}}")
+ info_dict["variables"]["__global__"][f"command{i}"] = ""
+ info_dict["variables"][key] = newvalue
return info_dict
- def _clean_confirmation_response(self, raw_response):
- #Parse response for confirmation request to openAI GPT.
- value = raw_response.strip()
- if value.strip(".").lower() == "true":
- value = True
- elif value.strip(".").lower() == "false":
- value = False
- return value
-
def _get_commands(self, user_input, nodes):
#Send the request for commands for each device to openAI GPT.
output_list = []
+ command_function = deepcopy(self.__prompt["command_function"])
for key, value in nodes.items():
tags = value.get('tags', {})
try:
if os_value := tags.get('os'):
output_list.append(f"{key}: {os_value}")
+ command_function["parameters"]["properties"][key] = {}
+ command_function["parameters"]["properties"][key]["type"] = "array"
+ command_function["parameters"]["properties"][key]["description"] = f"OS: {os_value}"
+ command_function["parameters"]["properties"][key]["items"] = {}
+ command_function["parameters"]["properties"][key]["items"]["type"] = "string"
except:
pass
output_str = "\n".join(output_list)
@@ -816,17 +808,20 @@ Users will typically use words like verify, check, make sure, or similar to refe
message = []
message.append({"role": "system", "content": dedent(self.__prompt["command_system"]).strip()})
message.append({"role": "user", "content": dedent(self.__prompt["command_user"]).strip()})
- message.append({"role": "assistant", "content": dedent(self.__prompt["command_assistant"]).strip()})
+ message.append({"role": "assistant", "content": None, "function_call": self.__prompt["command_assistant"]})
message.append({"role": "user", "content": command_input})
+ functions = [command_function]
response = openai.ChatCompletion.create(
model=self.model,
messages=message,
+ functions=functions,
+ function_call={"name": "get_commands"},
temperature=self.temp
)
output = {}
- output["dict_response"] = response
- output["raw_response"] = response["choices"][0]["message"]["content"]
- output["response"] = self._clean_command_response(output["raw_response"])
+ result = response["choices"][0]["message"].to_dict()
+ json_result = json.loads(result["function_call"]["arguments"])
+ output["response"] = self._clean_command_response(json_result)
return output
def _get_filter(self, user_input, chat_history = None):
@@ -834,7 +829,8 @@ Users will typically use words like verify, check, make sure, or similar to refe
message = []
message.append({"role": "system", "content": dedent(self.__prompt["original_system"]).strip()})
message.append({"role": "user", "content": dedent(self.__prompt["original_user"]).strip()})
- message.append({"role": "assistant", "content": dedent(self.__prompt["original_assistant"]).strip()})
+ message.append({"role": "assistant", "content": None, "function_call": self.__prompt["original_assistant"]})
+ functions = [self.__prompt["original_function"]]
if not chat_history:
chat_history = []
chat_history.append({"role": "user", "content": user_input})
@@ -842,36 +838,55 @@ Users will typically use words like verify, check, make sure, or similar to refe
response = openai.ChatCompletion.create(
model=self.model,
messages=message,
+ functions=functions,
+ function_call="auto",
temperature=self.temp,
top_p=1
)
+ def extract_quoted_strings(text):
+ pattern = r'["\'](.*?)["\']'
+ matches = re.findall(pattern, text)
+ return matches
+ expected = extract_quoted_strings(user_input)
output = {}
- output["dict_response"] = response
- output["raw_response"] = response["choices"][0]["message"]["content"]
- chat_history.append({"role": "assistant", "content": output["raw_response"]})
+ result = response["choices"][0]["message"].to_dict()
+ if result["content"]:
+ output["app_related"] = False
+ chat_history.append({"role": "assistant", "content": result["content"]})
+ output["response"] = result["content"]
+ else:
+ json_result = json.loads(result["function_call"]["arguments"])
+ output["app_related"] = True
+ output["filter"] = json_result["filter"]
+ output["type"] = json_result["type"]
+ chat_history.append({"role": "assistant", "content": result["content"], "function_call": {"name": result["function_call"]["name"], "arguments": json.dumps(json_result)}})
+ output["expected"] = expected
output["chat_history"] = chat_history
- clear_response = self._clean_original_response(output["raw_response"])
- output["response"] = self._clean_original_response(output["raw_response"])
return output
def _get_confirmation(self, user_input):
#Send the request to identify if user is confirming or denying the task
message = []
- message.append({"role": "system", "content": dedent(self.__prompt["confirmation_system"]).strip()})
- message.append({"role": "user", "content": dedent(self.__prompt["confirmation_user"]).strip()})
- message.append({"role": "assistant", "content": dedent(self.__prompt["confirmation_assistant"]).strip()})
message.append({"role": "user", "content": user_input})
+ functions = [self.__prompt["confirmation_function"]]
response = openai.ChatCompletion.create(
model=self.model,
messages=message,
+ functions=functions,
+ function_call={"name": "get_confirmation"},
temperature=self.temp,
top_p=1
)
+ result = response["choices"][0]["message"].to_dict()
+ json_result = json.loads(result["function_call"]["arguments"])
output = {}
- output["dict_response"] = response
- output["raw_response"] = response["choices"][0]["message"]["content"]
- output["response"] = self._clean_confirmation_response(output["raw_response"])
+ if json_result["result"] == "true":
+ output["result"] = True
+ elif json_result["result"] == "false":
+ output["result"] = False
+ elif json_result["result"] == "none":
+ output["result"] = json_result["response"]
return output
def confirm(self, user_input, max_retries=3, backoff_num=1):
@@ -894,7 +909,7 @@ Users will typically use words like verify, check, make sure, or similar to refe
'''
result = self._retry_function(self._get_confirmation, max_retries, backoff_num, user_input)
if result:
- output = result["response"]
+ output = result["result"]
else:
output = f"{self.model} api is not responding right now, please try again later."
return output
@@ -956,14 +971,14 @@ Users will typically use words like verify, check, make sure, or similar to refe
output["app_related"] = False
output["response"] = f"{self.model} api is not responding right now, please try again later."
return output
- output["app_related"] = original["response"]["app_related"]
+ output["app_related"] = original["app_related"]
output["chat_history"] = original["chat_history"]
if not output["app_related"]:
- output["response"] = original["response"]["response"]
+ output["response"] = original["response"]
else:
- type = original["response"]["type"].lower()
- if "filter" in original["response"]:
- output["filter"] = original["response"]["filter"]
+ type = original["type"]
+ if "filter" in original:
+ output["filter"] = original["filter"]
if not self.config.config["case"]:
if isinstance(output["filter"], list):
output["filter"] = [item.lower() for item in output["filter"]]
@@ -990,8 +1005,8 @@ Users will typically use words like verify, check, make sure, or similar to refe
output["args"]["commands"] = commands["response"]["commands"]
output["args"]["vars"] = commands["response"]["variables"]
output["nodes"] = [item for item in output["nodes"] if output["args"]["vars"].get(item)]
- if original["response"].get("expected"):
- output["args"]["expected"] = original["response"]["expected"]
+ if original.get("expected"):
+ output["args"]["expected"] = original["expected"]
output["action"] = "test"
else:
output["action"] = "run"
@@ -1121,14 +1136,14 @@ Users will typically use words like verify, check, make sure, or similar to refe
output["app_related"] = False
output["response"] = f"{self.model} api is not responding right now, please try again later."
return output
- output["app_related"] = original["response"]["app_related"]
+ output["app_related"] = original["app_related"]
output["chat_history"] = original["chat_history"]
if not output["app_related"]:
- output["response"] = original["response"]["response"]
+ output["response"] = original["response"]
else:
- type = original["response"]["type"].lower()
- if "filter" in original["response"]:
- output["filter"] = original["response"]["filter"]
+ type = original["type"]
+ if "filter" in original:
+ output["filter"] = original["filter"]
if not self.config.config["case"]:
if isinstance(output["filter"], list):
output["filter"] = [item.lower() for item in output["filter"]]
@@ -1155,8 +1170,8 @@ Users will typically use words like verify, check, make sure, or similar to refe
output["args"]["commands"] = commands["response"]["commands"]
output["args"]["vars"] = commands["response"]["variables"]
output["nodes"] = [item for item in output["nodes"] if output["args"]["vars"].get(item)]
- if original["response"].get("expected"):
- output["args"]["expected"] = original["response"]["expected"]
+ if original.get("expected"):
+ output["args"]["expected"] = original["expected"]
output["action"] = "test"
else:
output["action"] = "run"
@@ -1220,7 +1235,7 @@ Users will typically use words like verify, check, make sure, or similar to refe
'''
result = self._retry_function(self._get_confirmation, max_retries, backoff_num, user_input)
if result:
- output = result["response"]
+ output = result["result"]
else:
output = f"{self.model} api is not responding right now, please try again later."
return output
@@ -1868,6 +1883,13 @@ Users will typically use words like verify, check, make sure, or similar to refe
bulkparser = subparsers.add_parser("bulk", help="Add nodes in bulk")
bulkparser.add_argument("bulk", const="bulk", nargs=0, action=self._store_type, help="Add nodes in bulk")
bulkparser.set_defaults(func=self._func_others)
+ # AIPARSER
+ aiparser = subparsers.add_parser("ai", help="Make request to an AI")
+ aiparser.add_argument("ask", nargs='*', help="Ask connpy AI something")
+ aiparser.add_argument("--model", nargs=1, help="Set the OPENAI model id")
+ aiparser.add_argument("--org", nargs=1, help="Set the OPENAI organization id")
+ aiparser.add_argument("--api_key", nargs=1, help="Set the OPENAI API key")
+ aiparser.set_defaults(func=self._func_ai)
#RUNPARSER
runparser = subparsers.add_parser("run", help="Run scripts or commands on nodes", formatter_class=argparse.RawTextHelpFormatter)
runparser.add_argument("run", nargs='+', action=self._store_type, help=self._help("run"), default="run")
@@ -1889,10 +1911,12 @@ Users will typically use words like verify, check, make sure, or similar to refe
configcrud.add_argument("--keepalive", dest="idletime", nargs=1, action=self._store_type, help="Set keepalive time in seconds, 0 to disable", type=int, metavar="INT")
configcrud.add_argument("--completion", dest="completion", nargs=1, choices=["bash","zsh"], action=self._store_type, help="Get terminal completion configuration for conn")
configcrud.add_argument("--configfolder", dest="configfolder", nargs=1, action=self._store_type, help="Set the default location for config file", metavar="FOLDER")
- configcrud.add_argument("--openai", dest="openai", nargs=2, action=self._store_type, help="Set openai organization and api_key", metavar=("ORGANIZATION", "API_KEY"))
+ configcrud.add_argument("--openai-org", dest="organization", nargs=1, action=self._store_type, help="Set openai organization", metavar="ORGANIZATION")
+ configcrud.add_argument("--openai-api-key", dest="api_key", nargs=1, action=self._store_type, help="Set openai api_key", metavar="API_KEY")
+ configcrud.add_argument("--openai-model", dest="model", nargs=1, action=self._store_type, help="Set openai model", metavar="MODEL")
configparser.set_defaults(func=self._func_others)
#Manage sys arguments
- commands = ["node", "profile", "mv", "move","copy", "cp", "bulk", "ls", "list", "run", "config", "api"]
+ commands = ["node", "profile", "mv", "move","copy", "cp", "bulk", "ls", "list", "run", "config", "api", "ai"]
profilecmds = ["--add", "-a", "--del", "--rm", "-r", "--mod", "--edit", "-e", "--show", "-s"]
if len(argv) >= 2 and argv[1] == "profile" and argv[0] in profilecmds:
argv[1] = argv[0]
@@ -2035,10 +2059,14 @@ Users will typically use words like verify, check, make sure, or similar to refe
for k, v in node.items():
if isinstance(v, str):
print(k + ": " + v)
- else:
+ elif isinstance(v, list):
print(k + ":")
for i in v:
print(" - " + i)
+ elif isinstance(v, dict):
+ print(k + ":")
+ for i,d in v.items():
+ print(" - " + i + ": " + d)
def _mod(self, args):
if args.data == None:
@@ -2103,10 +2131,14 @@ Users will typically use words like verify, check, make sure, or similar to refe
for k, v in profile.items():
if isinstance(v, str):
print(k + ": " + v)
- else:
+ elif isinstance(v, list):
print(k + ":")
for i in v:
print(" - " + i)
+ elif isinstance(v, dict):
+ print(k + ":")
+ for i,d in v.items():
+ print(" - " + i + ": " + d)
def _profile_add(self, args):
matches = list(filter(lambda k: k == args.data[0], self.profiles))
@@ -2144,7 +2176,7 @@ Users will typically use words like verify, check, make sure, or similar to refe
def _func_others(self, args):
#Function called when using other commands
- actions = {"ls": self._ls, "move": self._mvcp, "cp": self._mvcp, "bulk": self._bulk, "completion": self._completion, "case": self._case, "fzf": self._fzf, "idletime": self._idletime, "configfolder": self._configfolder, "openai": self._openai}
+ actions = {"ls": self._ls, "move": self._mvcp, "cp": self._mvcp, "bulk": self._bulk, "completion": self._completion, "case": self._case, "fzf": self._fzf, "idletime": self._idletime, "configfolder": self._configfolder, "organization": self._openai, "api_key": self._openai, "model": self._openai}
return actions.get(args.command)(args)
def _ls(self, args):
@@ -2262,10 +2294,12 @@ Users will typically use words like verify, check, make sure, or similar to refe
print("Config saved")
def _openai(self, args):
- openaikeys = {}
- openaikeys["organization"] = args.data[0]
- openaikeys["api_key"] = args.data[1]
- self._change_settings(args.command, openaikeys)
+ if "openai" in self.config.config:
+ openaikeys = self.config.config["openai"]
+ else:
+ openaikeys = {}
+ openaikeys[args.command] = args.data[0]
+ self._change_settings("openai", openaikeys)
def _change_settings(self, name, value):
@@ -2279,6 +2313,115 @@ Users will typically use words like verify, check, make sure, or similar to refe
actions = {"noderun": self._node_run, "generate": self._yaml_generate, "run": self._yaml_run}
return actions.get(args.action)(args)
+ def _func_ai(self, args):
+ arguments = {}
+ if args.model:
+ arguments["model"] = args.model[0]
+ if args.org:
+ arguments["org"] = args.org[0]
+ if args.api_key:
+ arguments["api_key"] = args.api_key[0]
+ self.myai = ai(self.config, **arguments)
+ if args.ask:
+ input = " ".join(args.ask)
+ request = self.myai.ask(input, dryrun = True)
+ if not request["app_related"]:
+ mdprint(Markdown(request["response"]))
+ print("\r")
+ else:
+ if request["action"] == "list_nodes":
+ if request["filter"]:
+ nodes = self.config._getallnodes(request["filter"])
+ else:
+ nodes = self.config._getallnodes()
+ list = "\n".join(nodes)
+ print(list)
+ else:
+ yaml_data = yaml.dump(request["task"])
+ confirmation = f"I'm going to run the following task:\n```{yaml_data}```"
+ mdprint(Markdown(confirmation))
+ question = [inquirer.Confirm("task", message="Are you sure you want to continue?")]
+ print("\r")
+ confirm = inquirer.prompt(question)
+ if confirm == None:
+ exit(7)
+ if confirm["task"]:
+ script = {}
+ script["name"] = "RESULT"
+ script["output"] = "stdout"
+ script["nodes"] = request["nodes"]
+ script["action"] = request["action"]
+ if "expected" in request:
+ script["expected"] = request["expected"]
+ script.update(request["args"])
+ self._cli_run(script)
+ else:
+ history = None
+ mdprint(Markdown("**Chatbot**: Hi! How can I help you today?\n\n---"))
+ while True:
+ questions = [
+ inquirer.Text('message', message="User", validate=self._ai_validation),
+ ]
+ answers = inquirer.prompt(questions)
+ if answers == None:
+ exit(7)
+ response, history = self._process_input(answers["message"], history)
+ mdprint(Markdown(f"""**Chatbot**:\n{response}\n\n---"""))
+ return
+
+
+ def _ai_validation(self, answers, current, regex = "^.+$"):
+ #Validate ai user chat.
+ if not re.match(regex, current):
+ raise inquirer.errors.ValidationError("", reason="Can't send empty messages")
+ return True
+
+ def _process_input(self, input, history):
+ response = self.myai.ask(input , chat_history = history, dryrun = True)
+ if not response["app_related"]:
+ if not history:
+ history = []
+ history.extend(response["chat_history"])
+ return response["response"], history
+ else:
+ history = None
+ if response["action"] == "list_nodes":
+ if response["filter"]:
+ nodes = self.config._getallnodes(response["filter"])
+ else:
+ nodes = self.config._getallnodes()
+ list = "\n".join(nodes)
+ response = f"```{list}\n```"
+ else:
+ yaml_data = yaml.dump(response["task"])
+ confirmresponse = f"I'm going to run the following task:\n```{yaml_data}```\nPlease confirm"
+ while True:
+ mdprint(Markdown(f"""**Chatbot**:\n{confirmresponse}"""))
+ questions = [
+ inquirer.Text('message', message="User", validate=self._ai_validation),
+ ]
+ answers = inquirer.prompt(questions)
+ if answers == None:
+ exit(7)
+ confirmation = self.myai.confirm(answers["message"])
+ if isinstance(confirmation, bool):
+ if not confirmation:
+ response = "Request cancelled"
+ else:
+ nodes = self.connnodes(self.config.getitems(response["nodes"]), config = self.config)
+ if response["action"] == "run":
+ output = nodes.run(**response["args"])
+ response = ""
+ elif response["action"] == "test":
+ result = nodes.test(**response["args"])
+ yaml_result = yaml.dump(result,default_flow_style=False, indent=4)
+ output = nodes.output
+ response = f"This is the result for your test:\n```\n{yaml_result}\n```"
+ for k,v in output.items():
+ response += f"\n***{k}***:\n```\n{v}\n```\n"
+ break
+ return response, history
+
def _func_api(self, args):
if args.command == "stop" or args.command == "restart":
args.data = stop_api()
@@ -2324,68 +2467,67 @@ Users will typically use words like verify, check, make sure, or similar to refe
print("failed reading file {}".format(args.data[0]))
exit(10)
for script in scripts["tasks"]:
- args = {}
- try:
- action = script["action"]
- nodelist = script["nodes"]
- args["commands"] = script["commands"]
- output = script["output"]
- if action == "test":
- args["expected"] = script["expected"]
- except KeyError as e:
- print("'{}' is mandatory".format(e.args[0]))
- exit(11)
- nodes = self.connnodes(self.config.getitems(nodelist), config = self.config)
- stdout = False
- if output is None:
- pass
- elif output == "stdout":
- stdout = True
- elif isinstance(output, str) and action == "run":
- args["folder"] = output
- try:
- args["vars"] = script["variables"]
- except:
- pass
- try:
- options = script["options"]
- thisoptions = {k: v for k, v in options.items() if k in ["prompt", "parallel", "timeout"]}
- args.update(thisoptions)
- except:
- options = None
- size = str(os.get_terminal_size())
- p = re.search(r'.*columns=([0-9]+)', size)
- columns = int(p.group(1))
- if action == "run":
- nodes.run(**args)
- print(script["name"].upper() + "-" * (columns - len(script["name"])))
- for i in nodes.status.keys():
- print(" " + i + " " + "-" * (columns - len(i) - 13) + (" PASS(0)" if nodes.status[i] == 0 else " FAIL({})".format(nodes.status[i])))
- if stdout:
- for line in nodes.output[i].splitlines():
- print(" " + line)
- elif action == "test":
- nodes.test(**args)
- print(script["name"].upper() + "-" * (columns - len(script["name"])))
- for i in nodes.status.keys():
- print(" " + i + " " + "-" * (columns - len(i) - 13) + (" PASS(0)" if nodes.status[i] == 0 else " FAIL({})".format(nodes.status[i])))
+ self._cli_run(script)
+
+
+ def _cli_run(self, script):
+ args = {}
+ try:
+ action = script["action"]
+ nodelist = script["nodes"]
+ args["commands"] = script["commands"]
+ output = script["output"]
+ if action == "test":
+ args["expected"] = script["expected"]
+ except KeyError as e:
+ print("'{}' is mandatory".format(e.args[0]))
+ exit(11)
+ nodes = self.connnodes(self.config.getitems(nodelist), config = self.config)
+ stdout = False
+ if output is None:
+ pass
+ elif output == "stdout":
+ stdout = True
+ elif isinstance(output, str) and action == "run":
+ args["folder"] = output
+ if "variables" in script:
+ args["vars"] = script["variables"]
+ if "vars" in script:
+ args["vars"] = script["vars"]
+ try:
+ options = script["options"]
+ thisoptions = {k: v for k, v in options.items() if k in ["prompt", "parallel", "timeout"]}
+ args.update(thisoptions)
+ except:
+ options = None
+ size = str(os.get_terminal_size())
+ p = re.search(r'.*columns=([0-9]+)', size)
+ columns = int(p.group(1))
+ if action == "run":
+ nodes.run(**args)
+ print(script["name"].upper() + "-" * (columns - len(script["name"])))
+ for i in nodes.status.keys():
+ print(" " + i + " " + "-" * (columns - len(i) - 13) + (" PASS(0)" if nodes.status[i] == 0 else " FAIL({})".format(nodes.status[i])))
+ if stdout:
+ for line in nodes.output[i].splitlines():
+ print(" " + line)
+ elif action == "test":
+ nodes.test(**args)
+ print(script["name"].upper() + "-" * (columns - len(script["name"])))
+ for i in nodes.status.keys():
+ print(" " + i + " " + "-" * (columns - len(i) - 13) + (" PASS(0)" if nodes.status[i] == 0 else " FAIL({})".format(nodes.status[i])))
+ if nodes.status[i] == 0:
+ max_length = max(len(s) for s in nodes.result[i].keys())
+ for k,v in nodes.result[i].items():
+ print(" TEST for '{}'".format(k) + " "*(max_length - len(k) + 1) + "--> " + str(v).upper())
+ if stdout:
if nodes.status[i] == 0:
- try:
- myexpected = args["expected"].format(**args["vars"][i])
- except:
- try:
- myexpected = args["expected"].format(**args["vars"]["__global__"])
- except:
- myexpected = args["expected"]
- print(" TEST for '{}' --> ".format(myexpected) + str(nodes.result[i]).upper())
- if stdout:
- if nodes.status[i] == 0:
- print(" " + "-" * (len(myexpected) + 16 + len(str(nodes.result[i]))))
- for line in nodes.output[i].splitlines():
- print(" " + line)
- else:
- print("Wrong action '{}'".format(action))
- exit(13)
+ print(" " + "-" * (max_length + 21))
+ for line in nodes.output[i].splitlines():
+ print(" " + line)
+ else:
+ print("Wrong action '{}'".format(action))
+ exit(13)
def _choose(self, list, name, action):
#Generates an inquirer list to pick
@@ -2717,28 +2859,37 @@ Users will typically use words like verify, check, make sure, or similar to refe
if type == "usage":
return "conn [-h] [--add | --del | --mod | --show | --debug] [node|folder]\n conn {profile,move,mv,copy,cp,list,ls,bulk,config} ..."
if type == "end":
- return "Commands:\n profile Manage profiles\n move (mv) Move node\n copy (cp) Copy node\n list (ls) List profiles, nodes or folders\n bulk Add nodes in bulk\n run Run scripts or commands on nodes\n config Manage app config\n api Start and stop connpy api"
+ return "Commands:\n profile Manage profiles\n move (mv) Move node\n copy (cp) Copy node\n list (ls) List profiles, nodes or folders\n bulk Add nodes in bulk\n run Run scripts or commands on nodes\n config Manage app config\n api Start and stop connpy api\n ai Make request to an AI"
if type == "bashcompletion":
return '''
#Here starts bash completion for conn
_conn()
{
- strings="$(connpy-completion-helper ${#COMP_WORDS[@]} ${COMP_WORDS[@]})"
- COMPREPLY=($(compgen -W "$strings" -- "${COMP_WORDS[-1]}"))
+ mapfile -t strings < <(connpy-completion-helper "bash" "${#COMP_WORDS[@]}" "${COMP_WORDS[@]}")
+ local IFS=$'\\t\\n'
+ COMPREPLY=($(compgen -W "$(printf '%s' "${strings[@]}")" -- "${COMP_WORDS[-1]}"))
}
-complete -o nosort -F _conn conn
-complete -o nosort -F _conn connpy
+
+complete -o nospace -o nosort -F _conn conn
+complete -o nospace -o nosort -F _conn connpy
#Here ends bash completion for conn
'''
if type == "zshcompletion":
return '''
-
#Here starts zsh completion for conn
autoload -U compinit && compinit
_conn()
{
- strings=($(connpy-completion-helper ${#words} $words))
- compadd "$@" -- `echo $strings`
+ strings=($(connpy-completion-helper "zsh" ${#words} $words))
+ for string in "${strings[@]}"; do
+ if [[ "${string}" =~ .*/$ ]]; then
+ # If the string ends with a '/', do not append a space
+ compadd -S '' -- "$string"
+ else
+ # If the string does not end with a '/', append a space
+ compadd -S ' ' -- "$string"
+ fi
+ done
}
compdef _conn conn
compdef _conn connpy
@@ -2895,7 +3046,7 @@ tasks:
-def start(self, argv=['connpy', '--html', '-o', 'docs/', '--force'])
+def start(self, argv=['connpy', '-o', 'docs/', '--html', '--force'])
-
Parameters:
@@ -2954,6 +3105,13 @@ tasks:
bulkparser = subparsers.add_parser("bulk", help="Add nodes in bulk")
bulkparser.add_argument("bulk", const="bulk", nargs=0, action=self._store_type, help="Add nodes in bulk")
bulkparser.set_defaults(func=self._func_others)
+ # AIPARSER
+ aiparser = subparsers.add_parser("ai", help="Make request to an AI")
+ aiparser.add_argument("ask", nargs='*', help="Ask connpy AI something")
+ aiparser.add_argument("--model", nargs=1, help="Set the OPENAI model id")
+ aiparser.add_argument("--org", nargs=1, help="Set the OPENAI organization id")
+ aiparser.add_argument("--api_key", nargs=1, help="Set the OPENAI API key")
+ aiparser.set_defaults(func=self._func_ai)
#RUNPARSER
runparser = subparsers.add_parser("run", help="Run scripts or commands on nodes", formatter_class=argparse.RawTextHelpFormatter)
runparser.add_argument("run", nargs='+', action=self._store_type, help=self._help("run"), default="run")
@@ -2975,10 +3133,12 @@ tasks:
configcrud.add_argument("--keepalive", dest="idletime", nargs=1, action=self._store_type, help="Set keepalive time in seconds, 0 to disable", type=int, metavar="INT")
configcrud.add_argument("--completion", dest="completion", nargs=1, choices=["bash","zsh"], action=self._store_type, help="Get terminal completion configuration for conn")
configcrud.add_argument("--configfolder", dest="configfolder", nargs=1, action=self._store_type, help="Set the default location for config file", metavar="FOLDER")
- configcrud.add_argument("--openai", dest="openai", nargs=2, action=self._store_type, help="Set openai organization and api_key", metavar=("ORGANIZATION", "API_KEY"))
+ configcrud.add_argument("--openai-org", dest="organization", nargs=1, action=self._store_type, help="Set openai organization", metavar="ORGANIZATION")
+ configcrud.add_argument("--openai-api-key", dest="api_key", nargs=1, action=self._store_type, help="Set openai api_key", metavar="API_KEY")
+ configcrud.add_argument("--openai-model", dest="model", nargs=1, action=self._store_type, help="Set openai model", metavar="MODEL")
configparser.set_defaults(func=self._func_others)
#Manage sys arguments
- commands = ["node", "profile", "mv", "move","copy", "cp", "bulk", "ls", "list", "run", "config", "api"]
+ commands = ["node", "profile", "mv", "move","copy", "cp", "bulk", "ls", "list", "run", "config", "api", "ai"]
profilecmds = ["--add", "-a", "--del", "--rm", "-r", "--mod", "--edit", "-e", "--show", "-s"]
if len(argv) >= 2 and argv[1] == "profile" and argv[0] in profilecmds:
argv[1] = argv[0]
@@ -3166,12 +3326,12 @@ tasks:
t = open(logfile, "r").read()
else:
t = logfile
+ while t.find("\b") != -1:
+ t = re.sub('[^\b]\b', '', t)
t = t.replace("\n","",1)
t = t.replace("\a","")
t = t.replace('\n\n', '\n')
t = re.sub(r'.\[K', '', t)
- while t.find("\b") != -1:
- t = re.sub('[^\b]\b', '', t)
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/ ]*[@-~])')
t = ansi_escape.sub('', t)
t = t.lstrip(" \n\r")
@@ -3374,6 +3534,8 @@ tasks:
output = ''
if not isinstance(commands, list):
commands = [commands]
+ if not isinstance(expected, list):
+ expected = [expected]
if "screen_length_command" in self.tags:
commands.insert(0, self.tags["screen_length_command"])
self.mylog = io.BytesIO()
@@ -3391,18 +3553,25 @@ tasks:
output = self._logclean(self.mylog.getvalue().decode(), True)
self.output = output
if result in [0, 1]:
- lastcommand = commands[-1]
- if vars is not None:
- expected = expected.format(**vars)
- lastcommand = lastcommand.format(**vars)
- last_command_index = output.rfind(lastcommand)
- cleaned_output = output[last_command_index + len(lastcommand):].strip()
- if expected in cleaned_output:
- self.result = True
- else:
- self.result = False
+ # lastcommand = commands[-1]
+ # if vars is not None:
+ # lastcommand = lastcommand.format(**vars)
+ # last_command_index = output.rfind(lastcommand)
+ # cleaned_output = output[last_command_index + len(lastcommand):].strip()
+ self.result = {}
+ for e in expected:
+ if vars is not None:
+ e = e.format(**vars)
+ updatedprompt = re.sub(r'(?<!\\)\$', '', prompt)
+ newpattern = f".*({updatedprompt}).*{e}.*"
+ cleaned_output = output
+ cleaned_output = re.sub(newpattern, '', cleaned_output)
+ if e in cleaned_output:
+ self.result[e] = True
+ else:
+ self.result[e]= False
self.status = 0
- return False
+ return self.result
if result == 2:
self.result = None
self.status = 2
@@ -3776,6 +3945,8 @@ tasks:
output = ''
if not isinstance(commands, list):
commands = [commands]
+ if not isinstance(expected, list):
+ expected = [expected]
if "screen_length_command" in self.tags:
commands.insert(0, self.tags["screen_length_command"])
self.mylog = io.BytesIO()
@@ -3793,18 +3964,25 @@ tasks:
output = self._logclean(self.mylog.getvalue().decode(), True)
self.output = output
if result in [0, 1]:
- lastcommand = commands[-1]
- if vars is not None:
- expected = expected.format(**vars)
- lastcommand = lastcommand.format(**vars)
- last_command_index = output.rfind(lastcommand)
- cleaned_output = output[last_command_index + len(lastcommand):].strip()
- if expected in cleaned_output:
- self.result = True
- else:
- self.result = False
+ # lastcommand = commands[-1]
+ # if vars is not None:
+ # lastcommand = lastcommand.format(**vars)
+ # last_command_index = output.rfind(lastcommand)
+ # cleaned_output = output[last_command_index + len(lastcommand):].strip()
+ self.result = {}
+ for e in expected:
+ if vars is not None:
+ e = e.format(**vars)
+ updatedprompt = re.sub(r'(?<!\\)\$', '', prompt)
+ newpattern = f".*({updatedprompt}).*{e}.*"
+ cleaned_output = output
+ cleaned_output = re.sub(newpattern, '', cleaned_output)
+ if e in cleaned_output:
+ self.result[e] = True
+ else:
+ self.result[e]= False
self.status = 0
- return False
+ return self.result
if result == 2:
self.result = None
self.status = 2
diff --git a/requirements.txt b/requirements.txt
index d6eeaf2..9963cee 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,9 +1,10 @@
-Flask>=2.0.3
+Flask>=2.3.2
inquirer>=3.1.3
-openai>=0.27.4
+openai>=0.27.6
pexpect>=4.8.0
pycryptodome>=3.17
pyfzf>=0.3.1
PyYAML>=6.0
-setuptools>=67.6.1
+setuptools>=67.8.0
+rich>=13.4.2
waitress>=2.1.2
diff --git a/setup.cfg b/setup.cfg
index 72b12a1..fba1cff 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -32,6 +32,7 @@ install_requires =
waitress
PyYAML
openai
+ rich
[options.extras_require]
fuzzysearch = pyfzf