github/OS-Copilot
1
1
Fork 0
mirror of https://github.com/OS-Copilot/OS-Copilot.git synced 2026-05-05 03:00:15 -04:00
Code Issues Packages Projects Releases Wiki Activity

update agents & utils

Browse Source
This commit is contained in:
hccngu
2024-03-31 21:59:08 +08:00
parent 3e4879d7a0
commit 259f2d6165
12 changed files with 1651 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -1,8 +1,6 @@
langchain/
guidance/
Auto-GPT/
Agents/
utils/
/log
/examples/log/*

2
oscopilot/agents/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .friday_agent import *
from .self_learning import *

81
oscopilot/agents/base_agent.py Normal file
View File

@@ -0,0 +1,81 @@
import re
import json
from oscopilot.tool_repository.basic_tools.get_os_version import get_os_version
class BaseAgent:
"""
BaseAgent serves as the foundational class for all agents types within the system.
This class initializes the core attributes common across different agents, providing
a unified interface for further specialization. Attributes include a language learning
model, the execution environments, an action library, and a maximum iteration limit for
agents operations.
Attributes:
llm: Placeholder for a language learning model, initialized as None.
environments: The execution environments for the agents, initialized as None.
action_lib: A library of actions available to the agents, initialized as None.
max_iter: The maximum number of iterations the agents can perform, initialized as None.
"""
def __init__(self):
"""
Initializes a new instance of BaseAgent with default values for its attributes.
"""
self.system_version = get_os_version()
def extract_information(self, message, begin_str='[BEGIN]', end_str='[END]'):
"""
Extracts substrings from a message that are enclosed within specified begin and end markers.
Args:
message (str): The message from which information is to be extracted.
begin_str (str): The marker indicating the start of the information to be extracted.
end_str (str): The marker indicating the end of the information to be extracted.
Returns:
list[str]: A list of extracted substrings found between the begin and end markers.
"""
result = []
_begin = message.find(begin_str)
_end = message.find(end_str)
while not (_begin == -1 or _end == -1):
result.append(message[_begin + len(begin_str):_end])
message = message[_end + len(end_str):]
_begin = message.find(begin_str)
_end = message.find(end_str)
return result
def extract_json_from_string(self, text):
"""
Identifies and extracts JSON data embedded within a given string.
This method searches for JSON data within a string, specifically looking for
JSON blocks that are marked with ```json``` notation. It attempts to parse
and return the first JSON object found.
Args:
text (str): The text containing the JSON data to be extracted.
Returns:
dict: The parsed JSON data as a dictionary if successful.
str: An error message indicating a parsing error or that no JSON data was found.
"""
# Improved regular expression to find JSON data within a string
json_regex = r'```json\s*\n\{[\s\S]*?\n\}\s*```'
# Search for JSON data in the text
matches = re.findall(json_regex, text)
# Extract and parse the JSON data if found
if matches:
# Removing the ```json and ``` from the match to parse it as JSON
json_data = matches[0].replace('```json', '').replace('```', '').strip()
try:
# Parse the JSON data
parsed_json = json.loads(json_data)
return parsed_json
except json.JSONDecodeError as e:
return f"Error parsing JSON data: {e}"
else:
return "No JSON data found in the string."

290
oscopilot/agents/friday_agent.py Normal file
View File

@@ -0,0 +1,290 @@
from oscopilot.agents.base_agent import BaseAgent
from oscopilot.tool_repository.basic_tools.get_os_version import check_os_version
import json
import logging
import sys
from oscopilot.prompts.friday_pt import prompt
from oscopilot.utils import TaskStatusCode, InnerMonologue, ExecutionState, JudgementResult, RepairingResult
class FridayAgent(BaseAgent):
"""
A FridayAgent orchestrates the execution of tasks by integrating planning, retrieving, and executing strategies.
This agent is designed to process tasks, manage errors, and refine strategies as necessary to ensure successful task completion. It supports dynamic task planning, information retrieval, execution strategy application, and employs a mechanism for self-refinement in case of execution failures.
"""
def __init__(self, planner, retriever, executor, Tool_Manager, config):
"""
Initializes the FridayAgent with specified planning, retrieving, and executing strategies, alongside configuration settings.
Args:
planner (callable): A strategy for planning the execution of tasks.
retriever (callable): A strategy for retrieving necessary information or tools related to the tasks.
executor (callable): A strategy for executing planned tasks.
config (object): Configuration settings for the agent.
Raises:
ValueError: If the OS version check fails.
"""
super().__init__()
self.config = config
tool_manager = Tool_Manager(config.generated_tool_repo_path)
self.planner = planner(prompt['planning_prompt'])
self.retriever = retriever(prompt['retrieve_prompt'], tool_manager)
self.executor = executor(prompt['execute_prompt'], tool_manager, config.max_repair_iterations)
self.score = self.config.score
self.task_status = TaskStatusCode.START
self.inner_monologue = InnerMonologue()
try:
check_os_version(self.system_version)
except ValueError as e:
print(e)
def run(self, task):
"""
Executes the given task by planning, executing, and refining as needed until the task is completed or fails.
Args:
query (object): The high-level task to be executed.
No explicit return value, but the method controls the flow of task execution and may exit the process in case of irreparable failures.
"""
self.planner.reset_plan()
sub_tasks_list = self.planning(task)
print("The task list obtained after planning is: {}".format(sub_tasks_list))
while self.planner.sub_task_list:
sub_task = self.planner.sub_task_list.pop(0)
execution_state = self.executing(sub_task, task)
isTaskCompleted, isReplan = self.self_refining(sub_task, execution_state)
if isReplan: continue
if isTaskCompleted:
print("The execution of the current sub task has been successfully completed.")
else:
print("{} not completed in repair round {}".format(sub_task, self.config.max_repair_iterations))
sys.exit()
def self_refining(self, tool_name, execution_state: ExecutionState):
"""
Analyzes and potentially refines the execution of a tool based on its current execution state.
This can involve replanning or repairing the execution strategy based on the analysis of execution errors and outcomes.
Args:
tool_name (str): The name of the tool being executed.
execution_state (ExecutionState): The current state of the tool's execution, encapsulating all relevant execution information including errors, results, and codes.
Returns:
tuple:
- isTaskCompleted (bool): Indicates whether the task associated with the tool has been successfully completed.
- isReplan (bool): Indicates whether a replan is required due to execution state analysis.
The method decides on the next steps by analyzing the type of error (if any) and the execution results, aiming to either complete the task successfully or identify the need for further action, such as replanning.
"""
isTaskCompleted = False
isReplan = False
score = 0
state, node_type, description, code, result, relevant_code = execution_state.get_all_state()
if node_type == 'Code':
judgement= self.judging(tool_name, state, code, description)
score = judgement.score
# need_repair, critique, score, reasoning, error_type
if judgement.need_repair:
if judgement.error_type == 'replan':
print("The current task requires replanning...")
new_sub_task_list = self.replanning(tool_name, judgement.reasoning)
print("The new task list obtained after planning is: {}".format(new_sub_task_list))
isReplan = True
else:
repairing_result = self.repairing(tool_name, code, description, state, judgement.critique, judgement.need_repair)
isTaskCompleted = repairing_result.isTaskCompleted
score = repairing_result.score
result = repairing_result.result
# isTaskCompleted, code, critique, score, result
# if not isTaskCompleted:
# print("{} not completed in repair round {}".format(tool, args.max_repair_iterations))
# break
else:
isTaskCompleted = True
if isTaskCompleted and score >= self.score:
self.executor.store_tool(tool_name, code)
print("{} has been stored in the tool repository.".format(tool_name))
else:
isTaskCompleted = True
if isTaskCompleted:
self.planner.update_tool(tool_name, result, relevant_code, True, node_type)
# print("The execution of the current task has been successfully completed.")
return isTaskCompleted, isReplan
def planning(self, task):
"""
Decomposes a given high-level task into a list of sub-tasks by retrieving relevant tool names and descriptions, facilitating structured execution planning.
Args:
task (object): The high-level task to be planned and executed.
Returns:
list: A list of sub-tasks generated by decomposing the high-level task, intended for sequential execution to achieve the task's goal.
This method leverages the retriever component to fetch information relevant to the task, which is then used by the planner component to decompose the task into manageable sub-tasks.
"""
retrieve_tool_name = self.retriever.retrieve_tool_name(task)
retrieve_tool_description_pair = self.retriever.retrieve_tool_description_pair(retrieve_tool_name)
# decompose task
self.planner.decompose_task(task, retrieve_tool_description_pair)
return self.planner.sub_task_list
def executing(self, tool_name, original_task):
"""
Executes a given sub-task as part of the task execution process, handling different types of tasks including code execution, API calls, and question-answering.
Args:
tool_name (str): The name of the tool associated with the sub-task.
original_task (object): The original high-level task that has been decomposed into sub-tasks.
Returns:
ExecutionState: The state of execution for the sub-task, including the result, any errors encountered, and additional execution-related information.
The method dynamically adapts the execution strategy based on the type of sub-task, utilizing the executor component for code execution, API interaction, or question-answering as appropriate.
"""
tool_node = self.planner.tool_node[tool_name]
description = tool_node.description
logging.info("The current subtask is: {subtask}".format(subtask=description))
code = ''
state = None
# The return value of the current task
result = ''
relevant_code = {}
node_type = tool_node.node_type
pre_tasks_info = self.planner.get_pre_tasks_info(tool_name)
if node_type == 'Code':
# retrieve existing tool
retrieve_name = self.retriever.retrieve_tool_name(description, 3)
relevant_code = self.retriever.retrieve_tool_code_pair(retrieve_name)
# task execute step
if node_type == 'QA':
# result = execute_agent.question_and_answer_tool(pre_tasks_info, task, task)
if self.planner.tool_num == 1:
result = self.executor.question_and_answer_tool(pre_tasks_info, original_task, original_task)
else:
result = self.executor.question_and_answer_tool(pre_tasks_info, original_task, description)
print(result)
logging.info(result)
else:
invoke = ''
if node_type == 'API':
api_path = self.executor.extract_API_Path(description)
code = self.executor.api_tool(description, api_path, pre_tasks_info)
else:
code, invoke = self.executor.generate_tool(tool_name, description, pre_tasks_info, relevant_code)
# Execute python tool class code
state = self.executor.execute_tool(code, invoke, node_type)
result = state.result
logging.info(state)
output = {
"result": state.result,
"error": state.error
}
logging.info(f"The subtask result is: {json.dumps(output)}")
return ExecutionState(state, node_type, description, code, result, relevant_code)
def judging(self, tool_name, state, code, description):
"""
Evaluates the execution of a tool based on its execution state and the provided code and description, determining whether the tool's execution was successful or requires amendment.
Args:
tool_name (str): The name of the tool being judged.
state (ExecutionState): The current execution state of the tool, including results and error information.
code (str): The source code associated with the tool's execution.
description (str): A description of the tool's intended functionality.
Returns:
JudgementResult: An object encapsulating the judgement on the tool's execution, including whether it needs repair, a critique of the execution, and an optional error type and reasoning for the judgement.
This method assesses the correctness of the executed code and its alignment with the expected outcomes, guiding potential repair or amendment actions.
"""
# Check whether the code runs correctly, if not, amend the code
reasoning = ''
error_type = ''
tool_node = self.planner.tool_node[tool_name]
next_action = tool_node.next_action
need_repair = False
critique = ''
score = 0
# If no error is reported, check whether the task is completed
if state.error == None:
critique, judge, score = self.executor.judge_tool(code, description, state, next_action)
if not judge:
print("critique: {}".format(critique))
need_repair = True
else:
# Determine whether it is caused by an error outside the code
reasoning, error_type = self.executor.analysis_tool(code, description, state)
need_repair = True
return JudgementResult(need_repair, critique, score, reasoning, error_type)
def replanning(self, tool_name, reasoning):
"""
Initiates the replanning process for a task based on new insights or failures encountered during execution, aiming to adjust the plan to better achieve the task goals.
Args:
tool_name (str): The name of the tool related to the task that requires replanning.
reasoning (str): The rationale behind the need for replanning, often based on execution failures or updated task requirements.
Returns:
list: An updated list of sub-tasks after the replanning process, intended for sequential execution to complete the task.
This method identifies alternative or additional tools and their descriptions based on the provided reasoning, updating the task plan accordingly.
"""
relevant_tool_name = self.retriever.retrieve_tool_name(reasoning)
relevant_tool_description_pair = self.retriever.retrieve_tool_description_pair(relevant_tool_name)
self.planner.replan_task(reasoning, tool_name, relevant_tool_description_pair)
return self.planner.sub_task_list
def repairing(self, tool_name, code, description, state, critique, need_repair):
"""
Attempts to repair the execution of a tool by amending its code based on the critique received and the current execution state, iterating until the code executes successfully or reaches the maximum iteration limit.
Args:
tool_name (str): The name of the tool being repaired.
code (str): The current code of the tool that requires repairs.
description (str): A description of the tool's intended functionality.
state (ExecutionState): The current execution state of the tool, including results and error information.
critique (str): Feedback on the tool's last execution attempt, identifying issues to be addressed.
need_repair (bool): A flag indicating whether the tool's code needs repairs.
Returns:
RepairingResult: An object encapsulating the result of the repair attempt, including whether the task has been completed successfully, the amended code, critique, execution score, and the execution result.
The method iterates, amending the tool's code based on feedback until the code executes correctly or the maximum number of iterations is reached. It leverages the executor component for amending the code and re-evaluating its execution.
"""
tool_node = self.planner.tool_node[tool_name]
next_action = tool_node.next_action
pre_tasks_info = self.planner.get_pre_tasks_info(tool_name)
trial_times = 0
score = 0
while (trial_times < self.executor.max_iter and need_repair == True):
trial_times += 1
print("current amend times: {}".format(trial_times))
new_code, invoke = self.executor.repair_tool(code, description, state, critique, pre_tasks_info)
critique = ''
code = new_code
# Run the current code and check for errors
state = self.executor.execute_tool(code, invoke, 'Code')
result = state.result
logging.info(state)
# print(state)
# Recheck
if state.error == None:
critique, judge, score = self.executor.judge_tool(code, description, state, next_action)
# The task execution is completed and the loop exits
if judge:
need_repair = False
break
# print("critique: {}".format(critique))
else: # The code still needs to be corrected
need_repair = True
return RepairingResult(not need_repair, code, critique, score, result)

538
oscopilot/agents/prompt.py Normal file
View File

@@ -0,0 +1,538 @@
"""
This modules contains a comprehensive `prompts` dictionary that serves as a repository of prompts for guiding the AI agents's interactions across various operational scenarios, including execution, planning, and information retrieval tasks. These prompts are meticulously crafted to instruct the AI in performing its duties, ranging from code generation and amendment to task decomposition and planning, as well as error analysis and tool usage.
The dictionary is segmented into three main categories:
1. **execute_prompt**: Contains prompts for execution-related tasks, such as code generation, invocation, amendment, and error judgment. These are further detailed for system actions and user interactions, facilitating a diverse range of programming and troubleshooting tasks.
2. **planning_prompt**: Focuses on task planning and re-planning, decomposing complex tasks into manageable sub-tasks, and adapting plans based on unforeseen issues, ensuring that the AI can assist in project management and task organization effectively.
3. **retrieve_prompt**: Dedicated to information retrieval, including filtering code snippets based on specific criteria, aiding the AI in sourcing and suggesting code solutions efficiently.
Each category comprises system and user prompts, where system prompts define the AI's task or query in detail, and user prompts typically include placeholders for dynamic information insertion, reflecting the context or specific requirements of the task at hand.
Usage:
The `prompts` dictionary is utilized by the AI agents to dynamically select appropriate prompts based on the current context or task, ensuring relevant and precise guidance for each operation. This dynamic approach allows the AI to adapt its interactions and responses to suit a wide array of programming and operational needs, enhancing its utility and effectiveness in assisting users.
Example:
.. code-block:: python
# Accessing a specific prompts for task execution
execute_prompt = prompts['execute_prompt']['_SYSTEM_SKILL_CREATE_AND_INVOKE_PROMPT']
"""
prompt = {
'execute_prompt': {
# Code generate and invoke prompts in os
'_SYSTEM_SKILL_CREATE_AND_INVOKE_PROMPT': '''
You are helpful assistant to assist in writing Python tool code for tasks completed on operating systems. Your expertise lies in creating Python classes that perform specific tasks, adhering to a predefined format and structure.
Your goal is to generate Python tool code in the form of a class. The code should be structured to perform a user-specified task on the current operating system. The class must be easy to use and understand, with clear instructions and comments.
You should only respond with a python code and a invocation statement.
Python code in the format as described below:
1. Code Structure: Begin with the necessary import statement: from oscopilot.tool_repository.basic_tools.base_action import BaseAction. Then, define the class using the class name which is the same as the task name provided by the user.
2. Initialization Code: Initialization Code: In the __init__ method of the class, only "self._description" is initialized. This attribute succinctly summarizes the main function and purpose of the class.
3. Code used to accomplish the Task: Note that you should avoid using bash for the current task if you can, and prioritize using some of python's basic libraries for the current task. If the task involves os bash operations, instruct the use of the subprocess library, particularly the run method, to execute these operations. All core code used to accomplish the task should be encapsulated within the __call__ method of the class.
4. Parameters of __call__ method: The parameter design of __call__ methods should be comprehensive and generic enough to apply to different goals in all the same task scenarios. The parameters of the __call__ method are obtained by parsing and abstracting the task description, and the goals of the specific task can not be hard-coded into the method.
5. Detailed Comments: Provide comprehensive comments throughout the code. This includes describing the purpose of the class, and the function of parameters, especially in the __call__ method.
invocation statement in the format as described below:
1. Parameter Details Interpretation: Understand the parameter details of the __call__ method. This will help select the correct parameters to fill in the invocation statement.
2. Task Description Analysis: Analyze the way the code is called based on the current task, the generated code, and the Information of Prerequisite Tasks.
3. Generating Invocation Statement: Construct the __call__ method invocation statement. This includes instantiating the class and passing the appropriate arguments to the __call__ method based on the task description. For example, if my class is called Demo, and its __call__ method takes parameters a and b, then my invocation statement should be Demo()(a,b).
4. Output Format: The final output should include the invocation statement, which must be enclosed in <invoke></invoke> tags. For example, <invoke>Demo()(a,b)</invoke>.
And the code you write should also follow the following criteria:
1. The class must start with from oscopilot.tool_repository.basic_tools.base_action import BaseAction.In addition you need to import all the third-party libraries used in your code.
2. If you reuse the code provided in the user's information, the class name should be consistent with the reused code's class name; otherwise, the class name should be the same as the user's task name.
3. In the __init__ method, only self._description should be initialized. And self._description must be Code enough to encapsulate the functionality of the current class. For example, if the current task is to change the name of the file named test in the folder called document to test1, then the content of this attribute should be written as: Rename the specified file within a designated folder to a new, predetermined filename.
4. The __call__ method must allow flexible arguments (*args, **kwargs) for different user requirements. The __call__ method can not hardcode specific task details, but rather, it should abstract them into parameters that can be passed in by the user, these parameters can be obtained by parsing and abstracting the task description. For example, if the class is meant to download and play music, the __call__ method should take parameters like the download link, destination folder, and file name, instead of having these details fixed in the code. Please ensure that the class is structured to easily accommodate different types of tasks, with a clear and flexible parameter design in the __call__ method. In addition, the parameter design should be comprehensive and versatile enough to be applicable to handling different targets under all the same task scenarios.
5. For tasks involving os bash commands, use the subprocess library to execute these commands within the Python class.
6. The code should include detailed comments explaining the purpose of the class, and the role of each parameter.
7. If a file or folder creation operation is involved, the name of the file or folder should contain only English, numbers and underscores.
8. You need to note that for different system languages, some system paths may have different names, for example, the desktop path in Chinese system languages is ~/桌面 while the desktop path in English system languages is ~/Desktop.
9. If your code involves operating (reading or writing or creating) files or folders under a specified path, be sure to change the current working directory to that specified path before performing file-related operations.
10. If the user does not specifically request it (specify an absolute path), all your file operations should be relative to the user's working directory, and all created files should be stored in that directory and its subdirectories as a matter of priority. And once a file or directory query is involved, the priority is to query from below the default initial working directory.
11. The working directory given by the user can not be hardcoded in your code, because different user can have different working directory at different time.
12. If you need to access the user's working directory, you should make the user's working directory a parameter that can be passed to the __call__ method. If the user provides a value for the working directory as a parameter, then use the path provided by the user as the working directory path. Otherwise, you can obtain it using methods like os.getcwd().
13. You only need to write the class, don't instantiate it and call the __call__ method. If you want to write an example of how to use the class, be sure to put the example in the comments.
14. The description of parameters in the __call__ method must follow a standardized format: Args: [description of input parameters], Returns: [description of the method's return value].
15. In the __call__ method, you need to print the task execution completion message if the task execution completes.
16. Please note that the code you generate is mainly used under the operating system, so it often involves system-level operations such as reading and writing files. You need to write a certain fault-tolerant mechanism to handle potential problems that may arise during these operations, such as Problems such as file non-existence and insufficient permissions.
17. If the __call__ method needs a return value to help perform the next task, for example, if a task needs to return a list or value to facilitate the next task to receive, then let the __call__ method return. Otherwise, there is no need to return
18. If the __call__ method involves file operations, then the file's path must be passed as a parameter to the __call__ method, in particular, if you are operating multiple files, pass the paths of these files as parameters in the form of a list. If it involves moving files, then both the source and destination paths must be provided as parameters to the __call__ method, since the source and destination may not be in the same directory.
19. If the current task requires the use of the return results from a preceding task, then its corresponding call method must include a parameter specifically for receiving the return results of the preceding task.
20. Please note that in the Relevant Code section of the user's information, I have provided some codes that may be capable of solving the current task. Please carefully review these codes. If there is a code that can directly solve the current task, please reuse it without making any modifications.
21. If the code involves the output of file paths, ensure that the output includes the files' absolute path.
22. When your code involves the task of file operation, please be sure to pay attention to the naming format of the file. If it is a jpg file called XXX, the name should be XXX.jpg. If it is an mp4 file called XXX, the name should be XXX.mp4. Additionally, the file name passed in may or may not have a file format suffix, and you need to handle these cases.
23. Please note that the file path provided in the task might not include the file extension. This does not necessarily mean that the path is for a folder. You are required to devise an operation to determine the type of the file, which will assist you in obtaining the complete file path including the file type.
24. Please note that when writing code to read the contents of a docx file, be sure to also read the table contents in the docx file.
25. If the generated code is used to run other code, then the result of the other code must be returned.
And the invocation statement should also follow the following criteria:
1. The __call__ method invocation must be syntactically correct as per Python standards.
2. Clearly identify any fake or placeholder parameters used in the invocation.
3. If necessary, you can use the Working Directory provided by the user as a parameter passed into the __call__ method.
4. The 'Information of Prerequisite Tasks' from User's information provides relevant information about the prerequisite tasks for the current task, encapsulated in a dictionary format. The key is the name of the prerequisite task, and the value consists of two parts: 'description', which is the description of the task, and 'return_val', which is the return information of the task.
5. If the execution of the current task's code requires the return value of a prerequisite task, the return information of the prerequisite task can assist you in generating the code execution for the current task.
6. 'Working Directory' in User's information represents the working directory. It may not necessarily be the same as the current working directory. If the files or folders mentioned in the task do not specify a particular directory, then by default, they are assumed to be in the working directory. This can help you understand the paths of files or folders in the task to facilitate your generation of the call.
7. The code comments include an example of a class invocation. You can refer to this example, but you should not directly copy it. Instead, you need to adapt and fill in the details of this invocation according to the current task and the information returned from previous tasks.
8. For code that involves text content as a parameter, you should ensure as much as possible that this text content is fully included in the function parameters when generating a call, rather than abbreviating it to save token count. For example, if you need to perform a file write operation, you cannot abbreviate the content to be written into __call__ method invocation, like origin text is 'Yao ming is a basketball player.', you can not write 'Yao ming is ...'.
9. If the string in the input parameter contains single quotes or double quotes, then the input of the parameter is wrapped in triple quotes. The following is an example of an invocation statement: <invoke>Demo()("""xx"x"xxx""" )</invoke>
10. All parameter information that needs to be filled in when calling must be filled in, and data cannot be omitted.
Now you will be provided with the following information, please write python code to accomplish the task and be compatible with system environments, versions and language according to these information.
''',
'_USER_SKILL_CREATE_AND_INVOKE_PROMPT': '''
User's information is as follows:
System Version: {system_version}
System language: simplified chinese
Working Directory: {working_dir}
Task Name: {task_name}
Task Description: {task_description}
Information of Prerequisite Tasks: {pre_tasks_info}
Relevant Code: {relevant_code}
''',
# Invoke generate prompts in os
'_SYSTEM_INVOKE_GENERATE_PROMPT': '''
You are an AI trained to assist with Python programming tasks, with a focus on class and method usage.
Your goal is to generate a Python __call__ method invocation statement based on provided class name, task descriptions, and method parameter details.
You should only respond with the python code in the format as described below:
1. Class Context: Begin by understanding the context of the Python class provided by the user. This includes grasping the class name and its intended functionality.
2. Task Description Analysis: Analyze the task description provided to determine the purpose of the class and how it is expected to operate. This will help in identifying the correct way to invoke the class.
3. Parameter Details Interpretation: Interpret the parameter details of the __call__ method. This will involve extracting the type of parameters and their role in the method.
4. Generating Invocation Statement: Construct the __call__ method invocation statement. This includes instantiating the class and passing the appropriate arguments to the __call__ method based on the task description. For example, if my class is called Demo, and its __call__ method takes parameters a and b, then my invocation statement could be Demo()(a,b).
5. Fake Parameter Identification: If the required parameter information (like a URL or file path) is not provided and a placeholder or fake parameter is used, clearly identify and list these as not being actual or valid values.All the fake paramters you list should be separated by comma.If there are no fake parameters,you should give a None.
6. Output Format: The final output should include two parts:The first one is the invocation statement, which must be enclosed in <invoke></invoke> tags.The second one is all the fake parameters you identified, which will be enclosed in <fake-params></fake-params> tags.
And the response you write should also follow the following criteria:
1. The __call__ method invocation must be syntactically correct as per Python standards.
2. Clearly identify any fake or placeholder parameters used in the invocation.
3. Encouraging generating a realistic and functional code snippet wherever possible.
4. If necessary, you can use the Working Directory provided by the user as a parameter passed into the __call__ method.
5. The 'Information of Prerequisite Tasks' from User's information provides relevant information about the prerequisite tasks for the current task, encapsulated in a dictionary format. The key is the name of the prerequisite task, and the value consists of two parts: 'description', which is the description of the task, and 'return_val', which is the return information of the task.
6. If the execution of the current task's code requires the return value of a prerequisite task, the return information of the prerequisite task can assist you in generating the code execution for the current task.
7. 'Working Directory' in User's information represents the working directory. It may not necessarily be the same as the current working directory. If the files or folders mentioned in the task do not specify a particular directory, then by default, they are assumed to be in the working directory. This can help you understand the paths of files or folders in the task to facilitate your generation of the call.
8. The code comments include an example of a class invocation. You can refer to this example, but you should not directly copy it. Instead, you need to adapt and fill in the details of this invocation according to the current task and the information returned from previous tasks.
Now you will be provided with the following information, please generate your response according to these information:
''',
'_USER_INVOKE_GENERATE_PROMPT': '''
User's information are as follows:
Class Name: {class_name}
Task Description: {task_description}
__call__ Method Parameters: {args_description}
Information of Prerequisite Tasks: {pre_tasks_info}
Working Directory: {working_dir}
''',
# Skill amend and invoke prompts in os
'_SYSTEM_SKILL_AMEND_AND_INVOKE_PROMPT': '''
You are an AI expert in Python programming, with a focus on diagnosing and resolving code issues.
Your goal is to precisely identify the reasons for failure in the existing Python code and implement effective modifications to ensure it accomplishes the intended task without errors.
You should only respond with a python code and a invocation statement.
Python code in the format as described below:
1. Modified Code: Based on the error analysis, the original code is modified to fix all the problems and provide the final correct code to the user to accomplish the target task. If the code is error free, fix and refine the code based on the Critique On The Code provided by the user to accomplish the target task.
2. Error Analysis: Conduct a step-by-step analysis to identify why the code is generating errors or failing to complete the task. This involves checking for syntax errors, logical flaws, and any other issues that might hinder execution.
3. Detailed Explanation: Offer a clear and comprehensive explanation for each identified issue, detailing why these issues are occurring and how they are impacting the code's functionality.
invocation statement in the format as described below:
1. Parameter Details Interpretation: Understand the parameter details of the __call__ method. This will help select the correct parameters to fill in the invocation statement.
2. Task Description Analysis: Analyze the way the code is called based on the current task, the generated code, and the Information of Prerequisite Tasks.
3. Generating Invocation Statement: Construct the __call__ method invocation statement. This includes instantiating the class and passing the appropriate arguments to the __call__ method based on the task description. For example, if my class is called Demo, and its __call__ method takes parameters a and b, then my invocation statement should be Demo()(a,b).
4. Output Format: The final output should include the invocation statement, which must be enclosed in <invoke></invoke> tags. For example, <invoke>Demo()(a,b)</invoke>.
And the code you write should also follow the following criteria:
1. You must keep the original code as formatted as possible, e.g. class name, methods, etc. You can only modify the relevant implementation of the __call__ method in the code.
2. Please avoid throwing exceptions in your modified code, as this may lead to consistent error reports during execution. Instead, you should handle the caught exceptions appropriately!
3. Some errors may be caused by unreasonable tasks initiated by the user, resulting in outcomes that differ from what is expected. Examples include scenarios where the file to be created already exists, or the parameters passed in are incorrect. To prevent further errors, you need to implement fault tolerance or exception handling.
4. Ensure the final code is syntactically correct, optimized for performance, and follows Python best practices. The final code should contain only the class definition; any code related to class instantiation and invocation must be commented out.
5. The python code must be enclosed between ```python and ```. For example, ```python [python code] ```.
6. The analysis and explanations must be clear, brief and easy to understand, even for those with less programming experience.
7. All modifications must address the specific issues identified in the error analysis.
8. The solution must enable the code to successfully complete the intended task without errors.
9. When Critique On The Code in User's information is empty, it means that there is an error in the code itself, you should fix the error in the code so that it can accomplish the current task.
10. In User's information, 'Working Directory' represents the root directory of the working directory, and 'Current Working Directory' represents the directory where the current task is located.
And the invocation statement should also follow the following criteria:
1. The __call__ method invocation must be syntactically correct as per Python standards.
2. Clearly identify any fake or placeholder parameters used in the invocation.
3. If necessary, you can use the Working Directory provided by the user as a parameter passed into the __call__ method.
4. The 'Information of Prerequisite Tasks' from User's information provides relevant information about the prerequisite tasks for the current task, encapsulated in a dictionary format. The key is the name of the prerequisite task, and the value consists of two parts: 'description', which is the description of the task, and 'return_val', which is the return information of the task.
5. If the execution of the current task's code requires the return value of a prerequisite task, the return information of the prerequisite task can assist you in generating the code execution for the current task.
6. 'Working Directory' in User's information represents the working directory. It may not necessarily be the same as the current working directory. If the files or folders mentioned in the task do not specify a particular directory, then by default, they are assumed to be in the working directory. This can help you understand the paths of files or folders in the task to facilitate your generation of the call.
7. The code comments include an example of a class invocation. You can refer to this example, but you should not directly copy it. Instead, you need to adapt and fill in the details of this invocation according to the current task and the information returned from previous tasks.
8. All parameter information that needs to be filled in when calling must be filled in, and data cannot be omitted.
Now you will be provided with the following information, please give your modified python code and invocation statement according to these information:
''',
'_USER_SKILL_AMEND_AND_INVOKE_PROMPT': '''
User's information are as follows:
Original Code: {original_code}
Task: {task}
Error Messages: {error}
Code Output: {code_output}
Current Working Directiory: {current_working_dir}
Working Directiory: {working_dir}
Files And Folders in Current Working Directiory: {files_and_folders}
Critique On The Code: {critique}
Information of Prerequisite Tasks: {pre_tasks_info}
''',
# Skill amend prompts in os
'_SYSTEM_SKILL_AMEND_PROMPT': '''
You are an AI expert in Python programming, with a focus on diagnosing and resolving code issues.
Your goal is to precisely identify the reasons for failure in the existing Python code and implement effective modifications to ensure it accomplishes the intended task without errors.
You should only respond with the python code in the format as described below:
1. Modified Code: Based on the error analysis, the original code is modified to fix all the problems and provide the final correct code to the user to accomplish the target task. If the code is error free, fix and refine the code based on the Critique On The Code provided by the user to accomplish the target task.
2. Error Analysis: Conduct a step-by-step analysis to identify why the code is generating errors or failing to complete the task. This involves checking for syntax errors, logical flaws, and any other issues that might hinder execution.
3. Detailed Explanation: Offer a clear and comprehensive explanation for each identified issue, detailing why these issues are occurring and how they are impacting the code's functionality.
And the code you write should also follow the following criteria:
1. You must keep the original code as formatted as possible, e.g. class name, methods, etc. You can only modify the relevant implementation of the __call__ method in the code.
2. Please avoid throwing exceptions in your modified code, as this may lead to consistent error reports during execution. Instead, you should handle the caught exceptions appropriately!
3. Some errors may be caused by unreasonable tasks initiated by the user, resulting in outcomes that differ from what is expected. Examples include scenarios where the file to be created already exists, or the parameters passed in are incorrect. To prevent further errors, you need to implement fault tolerance or exception handling.
4. Ensure the final code is syntactically correct, optimized for performance, and follows Python best practices. The final code should contain only the class definition; any code related to class instantiation and invocation must be commented out.
5. The python code must be enclosed between ```python and ```. For example, ```python [python code] ```.
6. The analysis and explanations must be clear, brief and easy to understand, even for those with less programming experience.
7. All modifications must address the specific issues identified in the error analysis.
8. The solution must enable the code to successfully complete the intended task without errors.
9. When Critique On The Code in User's information is empty, it means that there is an error in the code itself, you should fix the error in the code so that it can accomplish the current task.
10. In User's information, 'Working Directory' represents the root directory of the working directory, and 'Current Working Directory' represents the directory where the current task is located.
Now you will be provided with the following information, please give your modified python code according to these information:
''',
'_USER_SKILL_AMEND_PROMPT': '''
User's information are as follows:
Original Code: {original_code}
Task: {task}
Error Messages: {error}
Code Output: {code_output}
Current Working Directiory: {current_working_dir}
Working Directiory: {working_dir}
Files And Folders in Current Working Directiory: {files_and_folders}
Critique On The Code: {critique}
''',
# Skill create prompts in os
'_SYSTEM_SKILL_CREATE_PROMPT': '''
You are helpful assistant to assist in writing Python tool code for tasks completed on operating systems. Your expertise lies in creating Python classes that perform specific tasks, adhering to a predefined format and structure.
Your goal is to generate Python tool code in the form of a class. The code should be structured to perform a user-specified task on the current operating system. The class must be easy to use and understand, with clear instructions and comments.
You should only respond with the python code in the format as described below:
1. Code Structure: Begin with the necessary import statement: from oscopilot.tool_repository.basic_tools.base_action import BaseAction. Then, define the class using the class name which is the same as the task name provided by the user.
2. Initialization Code: Initialization Code: In the __init__ method of the class, only "self._description" is initialized. This attribute succinctly summarizes the main function and purpose of the class.
3. Code used to accomplish the Task: Note that you should avoid using bash for the current task if you can, and prioritize using some of python's basic libraries for the current task. If the task involves os bash operations, instruct the use of the subprocess library, particularly the run method, to execute these operations. All core code used to accomplish the task should be encapsulated within the __call__ method of the class.
4. Parameters of __call__ method: The parameter design of __call__ methods should be comprehensive and generic enough to apply to different goals in all the same task scenarios. The parameters of the __call__ method are obtained by parsing and abstracting the task description, and the goals of the specific task can not be hard-coded into the method.
5. Detailed Comments: Provide comprehensive comments throughout the code. This includes describing the purpose of the class, and the function of parameters, especially in the __call__ method.
And the code you write should also follow the following criteria:
1. The class must start with from oscopilot.tool_repository.basic_tools.base_action import BaseAction.In addition you need to import all the third-party libraries used in your code.
2. The class name should be the same as the user's task name.
3. In the __init__ method, only self._description should be initialized. And self._description must be Code enough to encapsulate the functionality of the current class. For example, if the current task is to change the name of the file named test in the folder called document to test1, then the content of this attribute should be written as: Rename the specified file within a designated folder to a new, predetermined filename.
4. The __call__ method must allow flexible arguments (*args, **kwargs) for different user requirements. The __call__ method can not hardcode specific task details, but rather, it should abstract them into parameters that can be passed in by the user, these parameters can be obtained by parsing and abstracting the task description. For example, if the class is meant to download and play music, the __call__ method should take parameters like the download link, destination folder, and file name, instead of having these details fixed in the code. Please ensure that the class is structured to easily accommodate different types of tasks, with a clear and flexible parameter design in the __call__ method. In addition, the parameter design should be comprehensive and versatile enough to be applicable to handling different targets under all the same task scenarios.
5. For tasks involving os bash commands, use the subprocess library to execute these commands within the Python class.
6. The code should include detailed comments explaining the purpose of the class, and the role of each parameter.
7. If a file or folder creation operation is involved, the name of the file or folder should contain only English, numbers and underscores.
8. You need to note that for different system languages, some system paths may have different names, for example, the desktop path in Chinese system languages is ~/桌面 while the desktop path in English system languages is ~/Desktop.
9. If your code involves operating (reading or writing or creating) files or folders under a specified path, be sure to change the current working directory to that specified path before performing file-related operations.
10. If the user does not specifically request it (specify an absolute path), all your file operations should be relative to the user's working directory, and all created files should be stored in that directory and its subdirectories as a matter of priority. And once a file or directory query is involved, the priority is to query from below the default initial working directory.
11. The working directory given by the user can not be hardcoded in your code, because different user can have different working directory at different time.
12. If you need to access the user's working directory, you should make the user's working directory a parameter that can be passed to the __call__ method. If the user provides a value for the working directory as a parameter, then use the path provided by the user as the working directory path. Otherwise, you can obtain it using methods like os.getcwd().
13. You only need to write the class, don't instantiate it and call the __call__ method. If you want to write an example of how to use the class, be sure to put the example in the comments.
14. The description of parameters in the __call__ method must follow a standardized format: Args: [description of input parameters], Returns: [description of the method's return value].
15. In the __call__ method, you need to print the task execution completion message if the task execution completes.
16. Please note that the code you generate is mainly used under the operating system, so it often involves system-level operations such as reading and writing files. You need to write a certain fault-tolerant mechanism to handle potential problems that may arise during these operations, such as Problems such as file non-existence and insufficient permissions.
17. If the __call__ method needs a return value to help perform the next task, for example, if a task needs to return a list or value to facilitate the next task to receive, then let the __call__ method return. Otherwise, there is no need to return
18. If the __call__ method involves file operations, then the file's path must be passed as a parameter to the __call__ method, in particular, if you are operating multiple files, pass the paths of these files as parameters in the form of a list. If it involves moving files, then both the source and destination paths must be provided as parameters to the __call__ method, since the source and destination may not be in the same directory.
19. If the current task requires the use of the return results from a preceding task, then its corresponding call method must include a parameter specifically for receiving the return results of the preceding task.
Now you will be provided with the following information, please write python code to accomplish the task and be compatible with system environments, versions and language according to these information.
''',
'_USER_SKILL_CREATE_PROMPT': '''
User's information is as follows:
System Version: {system_version}
System language: simplified chinese
Working Directory: {working_dir}
Task Name: {task_name}
Task Description: {task_description}
''',
# Task judge prompts in os
'_SYSTEM_TASK_JUDGE_PROMPT': '''
You are an AI program expert to verify Python code against a user's task requirements.
Your goal is to determine if the provided Python code accomplishes the user's specified task based on the feedback information, And score the code based on the degree of generalizability of the code.
You should only respond with the JSON result in the format as described below:
1. Analyze the provided code: Examine the user's Python code to understand its functionality and structure.
2. Compare the code with the task description: Align the objectives stated in the user's task description with the capabilities of the code.
3. Evaluate the feedback information: Review the user's feedback, Includes the output of the code and the working catalog information provided to measure the effectiveness of the code.
4. Formulate a reasoning process: Comprehensive code analysis and feedback evaluation, create a logical reasoning process regarding the effectiveness of the code in accomplishing the task and the generalizability of the code. The generality of the code can be analyzed in terms of the flexibility of the parameters in the code, the handling of errors and exceptions, the clarity of the comments, the efficiency of the code, and the security perspective.
5. Evaluating Task Completion: Determine if the task is complete based on the reasoning process, expressed as a Boolean value, with true meaning the task is complete and false meaning the task is not complete.
6. Evaluating the code's generality: based on the analysis of the code's generality by the reasoning process, the code's generality is scored by assigning an integer score between 1 and 10 to reflect the code's generality, with a score of 1-4 indicating that the code is not sufficiently generalized, and that it may be possible to write the task objective directly into the code instead of passing it in as a parameter. a score of 5-7 indicates that the code is capable of accomplishing the task for different objectives of the same task, but does not do well in aspects such as security, clarity of comments, efficiency, or error and exception handling, and a score of 8 and above indicates that the code has good versatility and performs well in security, clarity of comments, efficiency, or error and exception handling.
7. Output Format: You should only return a JSON with no extra content. The JSON should contain three keys: the first is called 'reasoning', with its value being a string that represents your reasoning process. the second is called 'judge', its value is the boolean type true or false, true indicates that the code completes the current task, false indicates that it does not. The last is called 'score', which is a number between 1 and 10, representing code generality rating based on the result of 'Evaluating the code's generality'.
And you should also follow the following criteria:
1. Ensure accurate understanding of the Python code.
2. Relate the code functionality to the user's task.
3. Assess the completion degree of the task based on the feedback information.
4. Provide clear, logical reasoning.
5. You need to aware that the code I provided does not generate errors, I am just uncertain whether it effectively accomplishes the intended task.
6. If the task involves file creation, information regarding the current working directory and all its subdirectories and files may assist you in determining whether the file has been successfully created.
7. If the Code Output contains information indicating that the task has been completed, the task can be considered completed.
8. In User's information, 'Working Directory' represents the root directory of the working directory, and 'Current Working Directory' represents the directory where the current task is located.
9. If the task is not completed, it may be because the code generation and calling did not consider the information returned by the predecessor task. This information may be used as input parameters of the __call__ method.
10. 'Next Task' in the User's information describes tasks that follow the current task and may depend on the return from the current task. If necessary, you should check the current task's code output to ensure it returns the information required for these subsequent tasks. If it does not, then the current task can be considered incomplete.
Now you will be provided with the following information, please give the result JSON according to these information:
''',
'_USER_TASK_JUDGE_PROMPT': '''
User's information are as follows:
Current Code: {current_code}
Task: {task}
Code Output: {code_output}
Current Working Directiory: {current_working_dir}
Working Directory: {working_dir}
Files And Folders in Current Working Directiory: {files_and_folders}
Next Task: {next_action}
''',
# Code error judge prompts in os
'_SYSTEM_ERROR_ANALYSIS_PROMPT': '''
You are an expert in analyzing Python code errors, you are able to make an accurate analysis of different types of errors, and your return results adhere to a predefined format and structure.
Your goal is to analyze the errors that occur in the execution of the code provided to you, and determine whether the type of error is one that requires external additions (e.g., missing dependency packages, environments configuration issues, version incompatibility, etc.) or one that only requires internal changes to the code (e.g., syntax errors, logic errors, data type errors).
You should only respond with the JSON result in the format as described below:
1. Analyze the provided code and error: Examine the user's Python code to understand its functionality and structure. Combine the code with the error message, locate the error location, and analyze the specific reason for the error step by step.
2. Evaluate the feedback information: Review the user's feedback, including Current Working Directiory, Files And Folders in Current Working Directiory, combine with the previous analysis to further analyze the cause of the error.
3. Determine the type of error: Based on the error analysis results and current task, determine the type of error, whether it belongs to External Supplementation Required Errors or Internal Code Modification Errors.
4. Output Format: You should only return a JSON with no extra content. The JSON should contain two keys: one is called 'reasoning', with its value being a string that represents your reasoning process; the other is called 'type', where the value of 'type' is assigned as 'planning' for errors that fall under External Supplementation Required Errors, and as 'amend' for errors that are classified as Internal Code Modification Errors.
And you should also follow the following criteria:
1. Ensure accurate understanding of the Python code and the error.
2. There are only two types of errors, External Supplementation Required Errors and Internal Code Modification Errors.
3. Understanding the definition of External Supplementation Required Errors: This type of error involves not only modifying the code itself, but also requiring some additional operations in the running environments of the code, this requires new tasks to complete the additional operations.
4. Understanding the definition of Internal Code Modification Errors: This type of error can be resolved by modifying the code itself without having to perform any additional steps outside of the code.
5. Provide clear, logical reasoning.
6. The value of type can only be 'replan' or 'amend'.
7. In User's information, 'Working Directory' represents the root directory of the working directory, and 'Current Working Directory' represents the directory where the current task is located.
''',
'_USER_ERROR_ANALYSIS_PROMPT': '''
User's information are as follows:
Current Code: {current_code}
Task: {task}
Code Error: {code_error}
Current Working Directiory: {current_working_dir}
Working Directiory: {working_dir}
Files And Folders in Current Working Directiory: {files_and_folders}
''',
# Tool usage prompts in os
'_SYSTEM_TOOL_USAGE_PROMPT': '''
You are a useful AI assistant capable of accessing APIs to complete user-specified tasks, according to API documentation,
by using the provided ToolRequestUtil tool. The API documentation is as follows:
{openapi_doc}
The user-specified task is as follows:
{tool_sub_task}
The context which can further help you to determine the params of the API is as follows:
{context}
You need to complete the code using the ToolRequestUtil tool to call the specified API and print the return value
of the api.
ToolRequestUtil is a utility class, and the parameters of its 'request' method are described as follows:
def request(self, api_path, method, params=None, content_type=None):
"""
:param api_path: the path of the API
:param method: get/post
:param params: the parameters of the API, can be None.You cannot pass files to 'params' parameter.All files should be passed to 'files' parameter.
:param files: files to be uploaded, can be None.Remember if the parameters of the API contain files, you need to use the 'files' parameter to upload the files.
:param content_type: the content_type of api, e.g., application/json, multipart/form-data, can be None
:return: the response from the API
"""
Please begin your code completion:
''',
'_USER_TOOL_USAGE_PROMPT': '''
from oscopilot.tool_repository.manager.tool_request_util import ToolRequestUtil
tool_request_util = ToolRequestUtil()
# TODO: your code here
''',
# QA prompts in os
'_SYSTEM_QA_PROMPT': '''
You are a helpful ai assistant that can answer the question with the help of the context provided by the user in a step by step manner. The full question may help you to solve the current question.
If you don't know how to answer the user's question, answer "I don't know." instead of making up an answer.
And you should also follow the following criteria:
1. If the pre-task does not return the information you want, but your own knowledge can answer the current question, then you try to use your own knowledge to answer it.
2. If your current solution is incorrect but you have a potential solution, please implement your potential solution directly.
3. If you lack specific knowledge but can make inferences based on relevant knowledge, you can try to infer the answer to the question.
''',
'_USER_QA_PROMPT': '''
Context: {context}
Full Question: {question}
Current Question: {current_question}
'''
},
'planning_prompt': {
# Task decompose prompts in os
'_SYSTEM_TASK_DECOMPOSE_PROMPT': '''
You are an expert in making plans.
I will give you a task and ask you to decompose this task into a series of subtasks. These subtasks can form a directed acyclic graph, and each subtask is an atomic operation. Through the execution of topological sorting of subtasks, I can complete the entire task.
You should only respond with a reasoning process and a JSON result in the format as described below:
1. Carry out step-by-step reasoning based on the given task until the task is completed. Each step of reasoning is decomposed into subtasks. For example, the current task is to reorganize the text files containing the word 'agents' in the folder called document into the folder called agents. Then the reasoning process is as follows: According to Current Working Directiory and Files And Folders in Current Working Directiory information, the folders documernt and agents exist, so firstly, retrieve the txt text in the folder call document in the working directory. If the text contains the word "agents", save the path of the text file into the list, and return. Secondly, put the retrieved files into a folder named agents based on the file path list obtained by executing the previous subtask.
2. There are three types of subtasks, the first is a subtask that requires the use of APIs to access internet resources to obtain information, such as retrieving information from the Internet, this type of task is called 'API subtask', and all available APIs are only listed in the API List. The second is a subtask that does not require the use of API tools but need to write code to complete, which is called 'Code subtask', 'Code subtask' usually only involves operating system or file operations. The third is called 'QA subtask', it neither requires writing code nor calling API to complete the task, it will analyze the current subtask description and the return results of the predecessor subtasks to get an appropriate answer.
3. Each decomposed subtask has four attributes: name, task description, and dependencies. 'name' abstracts an appropriate name based on the reasoning process of the current subtask. 'description' is the process of the current subtask, if the task includes a specific file path, then the 'description' in all subtasks must contain the path to that file. 'dependencies' refers to the list of subtask names that the current subtask depends on based on the reasoning process. These subtasks must be executed before the current subtask. 'type' indicates whether the current subtask is a Code subtask or a API subtask or a QA subtask, If it is a Code subtask, its value is 'Code', if it is a API subtask, its value is 'API', if it is a QA subtask, its value is 'QA'.
4. In JSON, each decomposed subtask contains four attributes: name, description, dependencies and type, which are obtained through reasoning about the subtask. The key of each subtask is the 'name' attribute of the subtask.
5. Continuing with the example in 1, the format of the JSON data I want to get is as follows:
```json
{
"retrieve_files" : {
"name": "retrieve_files",
"description": "retrieve the txt text in the folder call document in the working directory. If the text contains the word "agents", save the path of the text file into the list, and return.",
"dependencies": [],
"type" : "Code"
},
"organize_files" : {
"name": "organize_files",
"description": "put the retrieved files into a folder named agents based on the file path list obtained by executing the previous task.",
"dependencies": ["retrieve_files"],
"type": "Code"
}
}
```
And you should also follow the following criteria:
1. A task can be decomposed down into one or more subtasks, depending on the complexity of the task.
2. The Tool List I gave you contains the name of each tool and the corresponding operation description. These tools are all atomic code task. You can refer to these atomic operations to decompose the code subtask.
3. If it is a pure mathematical problem, you can write code to complete it, and then process a QA subtask to analyze the results of the code to solve the problem.
4. The decomposed subtasks can form a directed acyclic graph based on the dependencies between them.
5. The description information of the subtask must be detailed enough, no entity and operation information in the task can be ignored.
6. 'Current Working Directiory' and 'Files And Folders in Current Working Directiory' specify the path and directory of the current working directory. These information may help you understand and generate subtasks.
7. The tasks currently designed are compatible with and can be executed on the present version of the system.
8. The current subtask may need the return results of its predecessor subtasks. For example, the current subtask is: Move the text files containing the word 'agents' from the folder named 'document' in the working directory to a folder named 'agents'. We can decompose this task into two subtasks, the first subtask is to retrieve text files containing the word 'agents' from the folder named 'document', and return their path list. The second subtask is to move the txt files of the corresponding path to the folder named 'agents' based on the path list returned by the previous subtask.
9. If the current subtask needs to use the return result of the previous subtask, then this process should be written in the task description of the subtask.
10. Please note that the name of a Code subtask must be abstract. For instance, if the subtask is to search for the word "agents," then the subtask name should be "search_word," not "search_agent." As another example, if the subtask involves moving a file named "test," then the subtask name should be "move_file," not "move_test." Additionally, if a tool in the Tool list is described as an abstraction of the current subtask, then the name of the current subtask should directly reuse the name of that tool.
11. When generating the subtask description, you need to clearly specify whether the operation targets a single entity or multiple entities that meet certain criteria.
12. When decomposing subtasks, avoid including redundant information. For instance, if the task is to move txt files containing the word 'agents' from the folder named 'document' to a folder named 'XXX', one subtask should be to retrieve text files containing the word 'agents' from the folder named 'document', and return their path list. Then, the next subtask should be to move the txt files to the folder named 'XXX' based on the path list returned by the previous subtask, rather than moving the txt files that contain the word 'agents' to the folder named 'XXX' based on the path list returned by the previous subtask. The latter approach would result in redundant information in the subtasks.
13. User's information provided you with a API List that includes the API path and their corresponding descriptions. These APIs are designed for interacting with internet resources, such as bing search, web page information, etc.
14. When decomposing subtasks, you need to pay attention to whether the current subtask involves obtaining data from internet resources, such as finding cat pictures on the Internet, retrieving information on a certain web page, etc., then you need to select the relevant API from the API List.
15. If the current subtask is a API subtask, the description of the subtask must include the API path of the specified API to facilitate my extraction through the special format of the API path. For example, if an API subtask is to use the bing search API to find XXX, then the description of the subtask should be: "Use the "/tools/bing/searchv2' API to search for XXX".
16. Please note that QA subtasks will not be generated continuously, that is, there will be no dependency between any two QA subtasks.
17. A QA subtask can perform comprehension analysis task, such as content conversion, format transformation, information summarization or analysis, answering academic questions, language translation, creative writing, logical reasoning based on existing information, and providing daily life advice and guidance, etc.
18. If the task involves file or operating system operations, such as file reading and writing, downloading, moving, then decompose the Code subtask. If the task requires the use of APIs to access internet resources to obtain information, such as web page retrieval, obtaining web page text content, etc., then decompose the API subtask. QA subtasks usually use the results of reading files from the Code task and the content returned by the API task to help complete intermediate steps or give the final answer to the task.
19. If the task does not involve any file operations or Internet data acquisition, then only plan a QA subtask, and the 'description' of the QA subtask must be the full content of the original task.
20. If the task is to read and analyze the content of a PowerPoint presentation, it can be broken down into two subtasks. The first is a Code subtask, which involves extracting the text content of the PowerPoint slides into a list. The second is a QA subtask, which complete the task base on the text information extracted from each slide.
21. Once the task involves obtaining knowledge such as books, articles, character information, etc. you need to plan API subtasks to obtain this knowledge from the Internet.
22. When decomposing an API subtask which uses the Bing Load Page API, you need to proceed to plan a QA subtask for analyzing and summarizing the information returned by that API subtask. For example, if the task is to find information about XXX, then your task will be broken down into three subtasks. The first API subtask is to use the Bing Search API to find relevant web page links. The second API subtask is to use the Bing Load Page API to obtain the information of the web pages found in the previous subtask. The final sub-task is a QA subtask, which is used to analyze the web page information returned by the previous sub-task and complete the task.
23. When the task involves retrieving a certain detailed content, then after decomposing the API subtask using Bing Search API, you also need to decompose an API subtask using Bing Load Page API, using for more detailed content.
24. If the attached file is a png or jpg file, you should first decompose a API subtask, which uses image caption API to analyze image and solve problem. If it is necessary to obtain information from the Internet, then an API subtask should be decomposed. Otherwise, proceed with a QA subtask, which analyzes and completes task based on the return from API subtask.
25. Please note that all available APIs are only in the API List. You should not make up APIs that are not in the API List.
26. If the attached file is a mp3 file, you can only break out two subtasks! The task must first be decomposed a API subtask, which uses audio2text API to transcribe mp3 audio to text. Then proceed with a QA subtask, which analyzes and completes task based on the return from API subtask.
27. Since the analyse or the content of the file are in the return value of the first decomposed subtask(usually a Code subtask), if the following subtask requires the content or the analyse, you should add the first subtask to the dependencies of current subtask.
28. If the task is to perform operations on a specific file., then all the subtasks must write the full path of the file in the task description, so as to locate the file when executing the subtasks.
29. If a task has attributes such as Task, Input, Output, and Path, it's important to know that Task refers to the task that needs to be completed. Input and Output are the prompts for inputs and outputs while writing the code functions during the task execution phase. Path is the file path that needs to be operated on.
''',
'_USER_TASK_DECOMPOSE_PROMPT': '''
User's information are as follows:
System Version: {system_version}
Task: {task}
Tool List: {tool_list}
API List: {api_list}
Current Working Directiory: {working_dir}
Files And Folders in Current Working Directiory: {files_and_folders}
''',
# Task replan prompts in os
'_SYSTEM_TASK_REPLAN_PROMPT': '''
You are an expert at designing new tasks based on the results of your reasoning.
When I was executing the code of current task, an issue occurred that is not related to the code. The user information includes a reasoning process addressing this issue. Based on the results of this reasoning, please design a new task to resolve the problem.
You should only respond with a reasoning process and a JSON result in the format as described below:
1. Design new tasks based on the reasoning process of current task errors. For example, the inference process analyzed that the reason for the error was that there was no numpy package in the environments, causing it to fail to run. Then the reasoning process for designing a new task is: According to the reasoning process of error reporting, because there is no numpy package in the environments, we need to use the pip tool to install the numpy package.
2. There are three types of subtasks, the first is a task that requires the use of APIs to access internet resources to obtain information, such as retrieving information from the Internet, this type of task is called 'API subtask', and the second is a task that does not require the use of API tools but need to write code to complete, which is called 'Code subtask', 'Code subtask' usually only involves operating system or file operations. The third is called 'QA subtask', It neither requires writing code nor calling API to complete the task, it will analyze the current subtask description and the return results of the predecessor tasks to get an appropriate answer.
3. Each decomposed subtask has four attributes: name, task description, and dependencies. 'name' abstracts an appropriate name based on the reasoning process of the current subtask. 'description' is the process of the current subtask. 'dependencies' refers to the list of task names that the current task depends on based on the reasoning process. These tasks must be executed before the current task. 'type' indicates whether the current task is a Code task or a API task or a QA task, If it is a Code task, its value is 'Code', if it is a API task, its value is 'API', if it is a QA task, its value is 'QA'.
4. Continuing with the example in 1, the format of the JSON data I want to get is as follows:
```json
{
"install_package" : {
"name": "install_package",
"description": "Use pip to install the numpy package that is missing in the environments.",
"dependencies": [],
"type" : "Code"
}
}
```
And you should also follow the following criteria:
1. The tasks you design based on the reasoning process are all atomic operations. You may need to design more than one task to meet the requirement that each task is an atomic operation.
2. The Tool List I gave you contains the name of each tool and the corresponding operation description. These tools are all atomic operations. You can refer to these atomic operations to design new task.
3. If an atomic operation in the Tool List can be used as a new task, then the name of the decomposed sub-task should be directly adopted from the name of that atomic tool.
4. The dependency relationship between the newly added task and the current task cannot form a loop.
5. The description information of the new task must be detailed enough, no entity and operation information in the task can be ignored.
6. 'Current Working Directiory' and 'Files And Folders in Current Working Directiory' specify the path and directory of the current working directory. These information may help you understand and generate tasks.
7. The tasks currently designed are compatible with and can be executed on the present version of the system.
8. Please note that the name of a task must be abstract. For instance, if the task is to search for the word "agents," then the task name should be "search_word," not "search_agent." As another example, if the task involves moving a file named "test," then the task name should be "move_file," not "move_test.
9. Please note that QA subtasks will not be generated continuously, that is, there will be no dependency between any two QA subtasks.
10. A QA subtask can perform comprehension analysis task, such as content conversion and format transformation, information summarization or analysis, answering academic questions, language translation, creative writing, logical reasoning based on existing information, and providing daily life advice and guidance, etc.
''',
'_USER_TASK_REPLAN_PROMPT': '''
User's information are as follows:
Current Task: {current_task}
Current Task Description: {current_task_description}
System Version: {system_version}
reasoning: {reasoning}
Tool List: {tool_list}
Current Working Directiory: {working_dir}
Files And Folders in Current Working Directiory: {files_and_folders}
''',
},
'retrieve_prompt': {
# tool code filter prompts
'_SYSTEM_ACTION_CODE_FILTER_PROMPT': '''
You are an expert in analyzing python code.
I will assign you a task and provide a dictionary of tool names along with their corresponding codes. Based on the current task, please analyze the dictionary to determine if there is any tool whose code can be used to complete the task. If such a code exists, return the tool name that corresponds to the code you believe is best suited for completing the task. If no appropriate code exists, return an empty string.
You should only respond with the format as described below:
1. First, understand the requirements of the task. Next, read the code for each tool, understanding their functions and methods. Examine the methods and attributes within the class, learning about their individual purposes and return values. Finally, by combining the task with the parameters of each tool class's __call__ method, determine whether the content of the task can serve as an argument for the __call__ method, thereby arriving at an analysis result.
2. Based on the above analysis results, determine whether there is code corresponding to the tool that can complete the current task. If so, return the tool name corresponding to the code you think is the most appropriate. If not, return an empty string.
3. Output Format: The final output should include one part: the name of the selected tool or empty string, which must be enclosed in <action></action> tags.
And you should also follow the following criteria:
1. There may be multiple codes that meet the needs of completing the task, but I only need you to return the tool name corresponding to the most appropriate code.
2. If no code can complete the task, be sure to return an empty string, rather than a name of a tool corresponding to a code that is nearly but not exactly suitable.
''',
'_USER_ACTION_CODE_FILTER_PROMPT': '''
User's information are as follows:
Tool Code Pair: {tool_code_pair}
Task: {task_description}
''',
},
'self_learning_prompt' : {
# self learning prompt
'_SYSTEM_COURSE_DESIGN_PROMPT' : '''
You are an expert in designing a python course focused entirely on using a specific Python package to operate a particular software, each lesson in the course includes specific tasks for operating the software package, as well as prompts for program input and output. Students will write Python code based on the content of each lesson and the relevant prompts to complete tasks, thereby learning how to use specific package to operate software.
I will provide you with the name of the software you need to learn, the specific Python package required to operate it, and an example of course design. Additionally, there may be a provision of the software's demo file path and its contents. I want you to design a software learning course, aimed at mastering skills for performing specific software operations using specific python package. Please generate a progressively challenging course based on the information and criteria below.
Excel Course Design Example: To help you better design a course on related software, here I provide you with an example of a course design for learning to manipulate Excel files using openpyxl. Lesson 1, use openpyxl to read all the contents of sheet 'Sheet1' in demo.xlsx, the input is the path of file and the name of the sheet, the output is the contents of 'Sheet1' in 'demo.xlsx' as a list of rows, where each row contains the data from the respective row in the sheet, and demo.xlsx is located in 'working_dir/demo.xlsx'. Lesson 2, use the Python package 'openpyxl' to read all the contents of column 'Product' of sheet 'Sheet1' in demo.xlsx, the input is the path of file, sheet name and column name, the output is the contents of column 'Product' of 'Sheet1' in 'demo.xlsx' as a list, and demo.xlsx is located in 'working_dir/demo.xlsx'. Lesson 3, use openpyxl to insert a new sheet named 'new sheet' into demo.xlsx, the input is the path of file and the name of the new sheet, the output is None, and demo.xlsx is located in 'working_dir/demo.xlsx'. Lesson 3, use the Python package 'openpyxl' to copy the 'Product' column from 'Sheet1' to 'Sheet2' in demo.xlsx. input is the path of the file, sheet name1, sheet name2, column name, output is None, and demo.xlsx is located in 'working_dir/demo.xlsx'. Lesson 5, use the Python package 'openpyxl' to create a histogram that represents the data from the 'Product' and 'Sales' columns in the 'Sheet1' of demo.xlsx, the input is the path of the file, sheet name, column name1, colunm name2, the output is None, and demo.xlsx is located in 'working_dir/demo.xlsx'. lesson 6, use openpyxl to sum the values under the 'sales' column from the sheet 'Sheet1', the input is the path of the file ,sheet name and column name, the output is the sum of the 'sales' column, and demo.xlsx is located in 'working_dir/demo.xlsx'.
Note that only six lessons are listed here for demonstration purposes; you will need to design the course to include as many lessons as possible to comprehensively learn Python package manipulation in practice.
You should only respond with the format as described below:
1. Output Format: The course designed consists of lessons, all lessons designed must be organised into a JSON data format, where key is the name of the lesson and value is a detailed description of the lesson.
2. Course design: The design of the course must progress from easy to difficult, with the more complex and challenging lessons later in the course incorporating the objectives of the earlier lessons.
3. lesson's name and description: The lesson's name is a summary of its current contents, and the description of the lesson have three or four parts: Task, Input, Output, File Path(If it exists). Task is a detailed description of the course content, Input is the prompt for the input of the program, Output is the prompt for the output of the program, and File Path is the path of the corresponding operating file.
4. Continuing with the Excel Course Design Example, the format of the JSON data I want to get is as follows:
```json
{
"read_specified_sheet" : "Task: Use the Python package 'openpyxl' to read all the contents of sheet 'Sheet1' in demo.xlsx. Input: The path of file, sheet name. Output: return the contents of 'Sheet1' in 'demo.xlsx' as a list of rows, where each row contains the data from the respective row in the sheet. File Path: working_dir/demo.xlsx",
"read_specified_sheet_column" : "Task: Use the Python package 'openpyxl' to read all the contents of column 'Product' of sheet 'Sheet1' in demo.xlsx. Input: The path of file, sheet name and column name. Output: return the contents of column 'Product' of 'Sheet1' in 'demo.xlsx' as a list. File Path: working_dir/demo.xlsx",
"insert_new_sheet" : "Task: Use the Python package 'openpyxl' to insert a new sheet named 'new sheet' into demo.xlsx. Input: The path of file and the name of the new sheet. Output: None. File Path: working_dir/demo.xlsx",
"copy_column_to_another_sheet" : "Task: Use the Python package 'openpyxl' to copy the 'Product' column from 'Sheet1' to 'Sheet2' in demo.xlsx. Input: The path of the file, sheet name1, sheet name2, column name. Output: None. File Path: working_dir/demo.xlsx",
"plot_histogram_from_sheet " : "Task: Use the Python package 'openpyxl' to create a histogram that represents the data from the 'Product' and 'Sales' columns in the 'Sheet1' of demo.xlsx. Input: The path of the file, sheet name, column name1, colunm name2. Output: None. File Path: working_dir/demo.xlsx",
"sum_column_values_in_sheet" : "Task: Use the Python package 'openpyxl' to sum the values under the 'Sales' column from the sheet 'Sheet1'. Input: The path of the file ,sheet name and column name. Output: The sum of the 'sales' column in 'Sheet1'. File Path: working_dir/demo.xlsx"
}
```
And you should also follow the following criteria:
1. My goal is to learn and master all the functionalities of this package for operating the software, enabling practical solutions to real-world problems. Therefore, the course design should encompass all features of the package as comprehensively as possible.
2. Each lesson's description should include the path of the corresponding operating file, if such a file exists, to facilitate learning directly on that file.
3. Your operation is executed under the specified System Version, so you need to be aware that the generated course can be executed under that OS environment.
4. If the Demo File Path is empty, you will need to generate a appropriate course, based on your understanding of the provided software and the package.
5. If Demo File Path is not empty, you must have an in-depth understanding and analysis of File Content and design a suitable and detailed course based on File Content. For example, when learning to use openpyxl to manipulate Excel files, if the columns in File Content have numerical information, you can design lessons on arithmetic operations such as summation related to that column, and the lessons can be specific to the names of the columns that need to be operated on.
6. Please note, an output of 'None' means that when students are learning a lesson, the code they write does not need to return a value. They only need to write the code according to the lesson contents and input prompts to perform operations on the software.
7. To help students better learn the course and achieve the teaching objectives, the tasks in the lessons must be as detailed and unambiguous as possible.
8. The code written by students during their course must be sufficiently versatile. Therefore, when designing the course, you should be able to transform the key information of tasks within the lesson into function parameters. Moreover, each parameter's content should be explicitly detailed in the Input and Output sections.
''',
'_USER_COURSE_DESIGN_PROMPT' : '''
User's information are as follows:
Software Name: {software_name}
Python Package Name: {package_name}
Demo File Path: {demo_file_path}
File Content: {file_content}
System Version: {system_version}
''',
},
'text_extract_prompt' : '''
Please return all the contents of the file.
File Path: {file_path}
Tips:
1. You need to be aware that the contents of some files may be stored in different places, for example, the contents of Excel may stored in different sheets and the contents of PPT may stored in different slides. For such files, I would like to return the contents of files in a dictionary format, organized by each sheet or slide, for easy retrieval and reading.
2. You can only break down the task into one subtask. The subtask is for reading out all the contents of the file.
3. If the file is a sheet file, I would like the output to be a dictionary, the key should be the name of each sheet, and the value should be a list of lists, where each inner list contains the contents of a row from that sheet.
'''
}

57
oscopilot/agents/self_learning.py Normal file
View File

@@ -0,0 +1,57 @@
import os
import logging
from oscopilot.prompts.friday_pt import prompt
from oscopilot.utils import self_learning_print_logging, get_project_root_path
class SelfLearning:
def __init__(self, agent, learner, tool_manager, config, text_extractor=None):
super().__init__()
self.config = config
self.agent = agent
self.learner = learner(prompt['self_learning_prompt'], tool_manager)
self.course = {}
if text_extractor:
self.text_extractor = text_extractor(agent)
def self_learning(self, software_name, package_name, demo_file_path):
"""
Start the self learning process.
"""
self_learning_print_logging(self.config)
file_content = None
if demo_file_path:
if not os.path.isabs(demo_file_path):
demo_file_path = get_project_root_path() + demo_file_path # TODO: test abs path
file_content = self.text_extract(demo_file_path)
self.course = self.course_design(software_name, package_name, demo_file_path, file_content)
self.learn_course(self.course)
def text_extract(self, demo_file_path):
"""
Extract the content of the file.
"""
file_content = self.text_extractor.extract_file_content(demo_file_path)
return file_content
def course_design(self, software_name, package_name, demo_file_path, file_content=None):
"""
Design the course based on the content of the file.
"""
course = self.learner.design_course(software_name, package_name, demo_file_path, file_content)
return course
def learn_course(self, course):
"""
Learn the course designed by the learner.
"""
logging.info(f'There are {len(self.course)} lessons in the course.')
for name, lesson in course.items():
logging.info(f"The current lesson is: {name}")
logging.info(f"The current lesson content is: {lesson}")
self.agent.run(lesson)
def continuous_learning(self):
pass

3
oscopilot/utils/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .config import *
from .utils import *
from .schema import *

99
oscopilot/utils/config.py Normal file
View File

@@ -0,0 +1,99 @@
import os
import argparse
import logging
from oscopilot.utils.utils import random_string, get_project_root_path
import dotenv
import sys
dotenv.load_dotenv(override=True)
class Config:
_instance = None
@classmethod
def initialize(cls, args):
if cls._instance is None:
cls._instance = cls.__new__(cls)
cls._instance.parameters = vars(args)
@classmethod
def get_parameter(cls, key):
if cls._instance is None:
return None
return cls._instance.parameters.get(key, None)
def setup_config():
parser = argparse.ArgumentParser(description='Inputs')
parser.add_argument('--generated_tool_repo_path', type=str, default='oscopilot/tool_repository/generated_tools', help='generated tool repo path')
parser.add_argument('--working_dir', type=str, default='working_dir', help='working dir path')
parser.add_argument('--query', type=str, default=None, help='Enter your task')
parser.add_argument('--query_file_path', type=str, default='', help='Enter the path of the files for your task or leave empty if not applicable')
parser.add_argument('--max_repair_iterations', type=int, default=3, help='Sets the max number of repair attempts. Default is 3.')
parser.add_argument('--logging_filedir', type=str, default='log', help='log path')
parser.add_argument('--logging_filename', type=str, default='temp0325.log', help='log file name')
parser.add_argument('--logging_prefix', type=str, default=random_string(16), help='log file prefix')
parser.add_argument('--score', type=int, default=8, help='critic score > score => store the tool')
# for Self-Leanring
parser.add_argument('--software_name', type=str, default='Excel', help='The name of the software used for learning.')
parser.add_argument('--package_name', type=str, default='openpyxl', help='The name of the package used for learning.')
parser.add_argument('--demo_file_path', type=str, default=get_project_root_path() + 'working_dir/Invoices.xlsx', help='Entering the path of the demo file helps you design the course, or leave it empty if not applicable.')
# for GAIA
parser.add_argument('--dataset_cache', type=str, default=None, help='Path to the dataset cache folder')
parser.add_argument('--level', type=int, default=1, help='Specifies the level of the GAIA dataset to use. Valid options are 1, 2, or 3')
parser.add_argument('--dataset_type', type=str, default='test', help='Defines the type of dataset to use, either `validation` for development or `test` for testing purposes')
parser.add_argument('--gaia_task_id', type=str, default=None, help='GAIA dataset task_id')
# for SheetCopilot
parser.add_argument('--sheet_task_id', type=int, default=None, help='sheet task dataset task id')
# Check if the script is being run in a test environment
if 'pytest' in sys.modules:
# In a test environment, use default values
args = parser.parse_args([])
else:
# In a non-test environment, parse command-line arguments
args = parser.parse_args()
Config.initialize(args)
if not os.path.exists(args.logging_filedir):
os.mkdir(args.logging_filedir)
logging.basicConfig(
filename=os.path.join(args.logging_filedir, args.logging_filename),
level=logging.INFO,
format=f'[{args.logging_prefix}] %(asctime)s - %(levelname)s - %(message)s'
)
return args
def setup_pre_run(args):
task = 'Your task is: {0}'.format(args.query)
if args.query_file_path != '':
task = task + '\nThe path of the files you need to use: {0}'.format(args.query_file_path)
print('Task:\n'+task)
logging.info(task)
return task
def self_learning_print_logging(args):
task = 'Your task is: Learn to use {0} to operate {1}'.format(args.software_name, args.package_name)
if args.demo_file_path != '':
task = task + '\nThe path of the file helps you design the course: {0}'.format(args.demo_file_path)
print('Task:\n'+task)
logging.info(task)

73
oscopilot/utils/llms.py Normal file
View File

@@ -0,0 +1,73 @@
import openai
import logging
import os
from dotenv import load_dotenv
load_dotenv(override=True)
MODEL_NAME = os.getenv('MODEL_NAME')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
OPENAI_ORGANIZATION = os.getenv('OPENAI_ORGANIZATION')
BASE_URL = os.getenv('OPENAI_BASE_URL')
class OpenAI:
"""
A class for interacting with the OpenAI API, allowing for chat completion requests.
This class simplifies the process of sending requests to OpenAI's chat model by providing
a convenient interface for the chat completion API. It handles setting up the API key
and organization for the session and provides a method to send chat messages.
Attributes:
model_name (str): The name of the model to use for chat completions. Default is set
by the global `MODEL_NAME`.
api_key (str): The API key used for authentication with the OpenAI API. This should
be set through the `OPENAI_API_KEY` global variable.
organization (str): The organization ID for OpenAI. Set this through the
`OPENAI_ORGANIZATION` global variable.
"""
def __init__(self):
"""
Initializes the OpenAI object with the given configuration.
"""
self.model_name = MODEL_NAME
def chat(self, messages, temperature=0):
"""
Sends a chat completion request to the OpenAI API using the specified messages and parameters.
Args:
messages (list of dict): A list of message dictionaries, where each dictionary
should contain keys like 'role' and 'content' to
specify the role (e.g., 'system', 'user') and content of
each message.
temperature (float, optional): Controls randomness in the generation. Lower values
make the model more deterministic. Defaults to 0.
Returns:
str: The content of the first message in the response from the OpenAI API.
"""
response = openai.chat.completions.create(
model=self.model_name,
messages=messages,
temperature=temperature
)
logging.info(f"Response: {response.choices[0].message.content}")
return response.choices[0].message.content
def main():
message = [
{"role": "user", "content": 'hello'},
]
# print(OPENAI_API_KEY)
# print(BASE_URL)
llm = OpenAI()
print(llm.chat(message))
if __name__ == '__main__':
main()

80
oscopilot/utils/schema.py Normal file
View File

@@ -0,0 +1,80 @@
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Union
from enum import IntEnum
@dataclass
class RepairingResult:
"""
Stores the results and intermediate representation of the repairing process
"""
isTaskCompleted: bool = False
code: str = ''
critique: str = ''
score: str = ''
result: str = ''
@dataclass
class JudgementResult:
"""
Stores the results and intermediate representation of the judging process
"""
need_repair: bool = False
critique: str = ''
score: int = 0
reasoning: str = ''
error_type: str = ''
@dataclass
class InnerMonologue:
"""
Stores all the intermediate representation during agent running
"""
reasoning: str = ''
error_type: str = ''
critique: str = ''
isRePlan: bool = False
isTaskCompleted: bool = False
@dataclass
class EnvState:
"""
Represents the state of an environment in which commands are executed.
"""
command: List[str] = field(default_factory=list)
result: Optional[str] = None
error: Optional[str] = None
pwd: Optional[str] = None
ls: Optional[str] = None
def __str__(self):
return (f"Result: {self.result}\n"
f"Error: {self.error}\n"
f"PWD: {self.pwd}\n"
f"LS: {self.ls}")
@dataclass
class ExecutionState:
"""
Stores all the intermediate representation during agent executing.
"""
state: Optional[EnvState] = None
node_type: str = ''
description: str = ''
code: str = ''
result: str = ''
relevant_code: str = ''
def get_all_state(self):
return self.state, self.node_type, self.description, self.code, self.result, self.relevant_code
class TaskStatusCode(IntEnum):
START = 1
FAILED = 6
COMPLETED = 7

65
oscopilot/utils/server_config.py Normal file
View File

@@ -0,0 +1,65 @@
import os
class ConfigManager:
"""
A singleton class responsible for managing configuration settings across the application.
This class implements the singleton design pattern to ensure that only one instance of the
ConfigManager exists at any time. It provides methods to set, apply, and clear proxy settings
for HTTP and HTTPS traffic.
Attributes:
_instance (ConfigManager): A private class-level attribute that holds the singleton instance.
http_proxy (str): The HTTP proxy URL.
https_proxy (str): The HTTPS proxy URL.
"""
_instance = None
def __new__(cls):
"""
Overrides the default instantiation process to ensure only one instance of ConfigManager is created.
Returns:
ConfigManager: The singleton instance of the ConfigManager.
"""
if cls._instance is None:
cls._instance = super(ConfigManager, cls).__new__(cls)
cls._instance.http_proxy = "http://127.0.0.1:10809"
cls._instance.https_proxy = "http://127.0.0.1:10809"
# cls._instance.http_proxy = None
# cls._instance.https_proxy = None
return cls._instance
def set_proxies(self, http, https):
"""
Sets the HTTP and HTTPS proxy URLs.
Args:
http (str): The HTTP proxy URL.
https (str): The HTTPS proxy URL.
"""
self.http_proxy = http
self.https_proxy = https
def apply_proxies(self):
"""
Applies the configured proxy settings by setting them in the environments variables.
The method sets the 'http_proxy' and 'https_proxy' environments variables based on the
configured proxy URLs. If no proxies are configured, the environments variables are not modified.
"""
if self.http_proxy:
os.environ["http_proxy"] = self.http_proxy
if self.https_proxy:
os.environ["https_proxy"] = self.https_proxy
def clear_proxies(self):
"""
Clears the proxy settings from the environments variables.
This method removes the 'http_proxy' and 'https_proxy' entries from the environments variables,
effectively clearing any proxy settings that were previously applied.
"""
os.environ.pop("http_proxy", None)
os.environ.pop("https_proxy", None)

363
oscopilot/utils/utils.py Normal file
View File

@@ -0,0 +1,363 @@
import copy
import numpy as np
import itertools
import json
import logging
import os
import re
import string
from typing import Any
import tqdm
import re
import tiktoken
import random
from datasets import load_dataset
def random_string(length):
"""
Generates a random string of a specified length.
Args:
length (int): The desired length of the random string.
Returns:
str: A string of random characters and digits of the specified length.
"""
characters = string.ascii_letters + string.digits
random_string = ''.join(random.choice(characters) for _ in range(length))
return random_string
def num_tokens_from_string(string: str) -> int:
"""
Calculates the number of tokens in a given text string according to a specific encoding.
Args:
text (str): The text string to be tokenized.
Returns:
int: The number of tokens the string is encoded into according to the model's tokenizer.
"""
encoding = tiktoken.encoding_for_model('gpt-4-1106-preview')
num_tokens = len(encoding.encode(string))
return num_tokens
def parse_content(content, html_type="html.parser"):
"""
Parses and cleans the given HTML content, removing specified tags, ids, and classes.
Args:
content (str): The HTML content to be parsed and cleaned.
type (str, optional): The type of parser to be used by BeautifulSoup. Defaults to "html.parser".
Supported types include "html.parser", "lxml", "lxml-xml", "xml", and "html5lib".
Raises:
ValueError: If an unsupported parser type is specified.
Returns:
str: The cleaned text extracted from the HTML content.
"""
implemented = ["html.parser", "lxml", "lxml-xml", "xml", "html5lib"]
if html_type not in implemented:
raise ValueError(f"Parser type {html_type} not implemented. Please choose one of {implemented}")
from bs4 import BeautifulSoup
soup = BeautifulSoup(content, html_type)
original_size = len(str(soup.get_text()))
tags_to_exclude = [
"nav",
"aside",
"form",
"header",
"noscript",
"svg",
"canvas",
"footer",
"script",
"style",
]
for tag in soup(tags_to_exclude):
tag.decompose()
ids_to_exclude = ["sidebar", "main-navigation", "menu-main-menu"]
for id in ids_to_exclude:
tags = soup.find_all(id=id)
for tag in tags:
tag.decompose()
classes_to_exclude = [
"elementor-location-header",
"navbar-header",
"nav",
"header-sidebar-wrapper",
"blog-sidebar-wrapper",
"related-posts",
]
for class_name in classes_to_exclude:
tags = soup.find_all(class_=class_name)
for tag in tags:
tag.decompose()
content = soup.get_text()
content = clean_string(content)
cleaned_size = len(content)
if original_size != 0:
logging.info(
f"Cleaned page size: {cleaned_size} characters, down from {original_size} (shrunk: {original_size-cleaned_size} chars, {round((1-(cleaned_size/original_size)) * 100, 2)}%)" # noqa:E501
)
return content
def clean_string(text):
"""
Cleans a given string by performing various operations such as whitespace normalization,
removal of backslashes, and replacement of hash characters with spaces. It also reduces
consecutive non-alphanumeric characters to a single occurrence.
Args:
text (str): The text to be cleaned.
Returns:
str: The cleaned text after applying all the specified cleaning operations.
"""
# Replacement of newline characters:
text = text.replace("\n", " ")
# Stripping and reducing multiple spaces to single:
cleaned_text = re.sub(r"\s+", " ", text.strip())
# Removing backslashes:
cleaned_text = cleaned_text.replace("\\", "")
# Replacing hash characters:
cleaned_text = cleaned_text.replace("#", " ")
# Eliminating consecutive non-alphanumeric characters:
# This regex identifies consecutive non-alphanumeric characters (i.e., not
# a word character [a-zA-Z0-9_] and not a whitespace) in the string
# and replaces each group of such characters with a single occurrence of
# that character.
# For example, "!!! hello !!!" would become "! hello !".
cleaned_text = re.sub(r"([^\w\s])\1*", r"\1", cleaned_text)
return cleaned_text
def is_readable(s):
"""
Heuristic to determine if a string is "readable" (mostly contains printable characters and forms meaningful words)
:param s: string
:return: True if the string is more than 95% printable.
"""
try:
printable_ratio = sum(c in string.printable for c in s) / len(s)
except ZeroDivisionError:
logging.warning("Empty string processed as unreadable")
printable_ratio = 0
return printable_ratio > 0.95 # 95% of characters are printable
def format_source(source: str, limit: int = 20) -> str:
"""
Format a string to only take the first x and last x letters.
This makes it easier to display a URL, keeping familiarity while ensuring a consistent length.
If the string is too short, it is not sliced.
"""
if len(source) > 2 * limit:
return source[:limit] + "..." + source[-limit:]
return source
def is_valid_json_string(source: str):
"""
Checks if a given string is a valid JSON.
Args:
source (str): The string to be validated as JSON.
Returns:
bool: True if the given string is a valid JSON format, False otherwise.
"""
try:
_ = json.loads(source)
return True
except json.JSONDecodeError:
logging.error(
"Insert valid string format of JSON. \
Check the docs to see the supported formats - `https://docs.embedchain.ai/data-sources/json`"
)
return False
def chunks(iterable, batch_size=100, desc="Processing chunks"):
"""
Breaks an iterable into smaller chunks of a specified size, yielding each chunk in sequence.
Args:
iterable (iterable): The iterable to be chunked.
batch_size (int, optional): The size of each chunk. Defaults to 100.
desc (str, optional): Description text to be displayed alongside the progress bar. Defaults to "Processing chunks".
Yields:
tuple: A chunk of the iterable, with a maximum length of `batch_size`.
"""
it = iter(iterable)
total_size = len(iterable)
with tqdm(total=total_size, desc=desc, unit="batch") as pbar:
chunk = tuple(itertools.islice(it, batch_size))
while chunk:
yield chunk
pbar.update(len(chunk))
chunk = tuple(itertools.islice(it, batch_size))
def generate_prompt(template: str, replace_dict: dict):
"""
Generates a string by replacing placeholders in a template with values from a dictionary.
Args:
template (str): The template string containing placeholders to be replaced.
replace_dict (dict): A dictionary where each key corresponds to a placeholder in the template
and each value is the replacement for that placeholder.
Returns:
str: The resulting string after all placeholders have been replaced with their corresponding values.
"""
prompt = copy.deepcopy(template)
for k, v in replace_dict.items():
prompt = prompt.replace(k, str(v))
return prompt
def cosine_similarity(a, b):
"""
Calculates the cosine similarity between two vectors.
Args:
a (array_like): The first vector.
b (array_like): The second vector.
Returns:
float: The cosine similarity between vectors `a` and `b`.
"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))\
def send_chat_prompts(sys_prompt, user_prompt, llm):
"""
Sends a sequence of chat prompts to a language learning model (LLM) and returns the model's response.
Args:
sys_prompt (str): The system prompt that sets the context or provides instructions for the language learning model.
user_prompt (str): The user prompt that contains the specific query or command intended for the language learning model.
llm (object): The language learning model to which the prompts are sent. This model is expected to have a `chat` method that accepts structured prompts.
Returns:
The response from the language learning model, which is typically a string containing the model's answer or generated content based on the provided prompts.
The function is a utility for simplifying the process of sending structured chat prompts to a language learning model and parsing its response, useful in scenarios where dynamic interaction with the model is required.
"""
message = [
{"role": "system", "content": sys_prompt},
{"role": "user", "content": user_prompt},
]
return llm.chat(message)
def get_project_root_path():
"""
This function returns the absolute path of the project root directory. It assumes that it is being called from a file located in oscopilot/utils/.
Args:
None
Returns:
str: The absolute path of the project root directory.
"""
script_path = os.path.abspath(__file__)
# Get the directory of the script (oscopilot/utils)
script_directory = os.path.dirname(script_path)
# Get the parent directory of script_directory (oscopilot)
oscopilot_directory = os.path.dirname(script_directory)
# Get the project root directory
project_root_path = os.path.dirname(oscopilot_directory)
return project_root_path + '/'
class GAIALoader:
def __init__(self, level=1, cache_dir=None):
if cache_dir != None:
assert os.path.exists(cache_dir), f"Cache directory {cache_dir} does not exist."
self.cache_dir = cache_dir
try:
self.dataset = load_dataset("gaia-benchmark/GAIA", "2023_level{}".format(level), cache_dir=self.cache_dir)
except Exception as e:
raise Exception(f"Failed to load GAIA dataset: {e}")
else:
self.dataset = load_dataset("gaia-benchmark/GAIA", "2023_level{}".format(level))
def get_data_by_task_id(self, task_id, dataset_type):
if self.dataset is None or dataset_type not in self.dataset:
raise ValueError("Dataset not loaded or data set not available.")
data_set = self.dataset[dataset_type]
for record in data_set:
if record['task_id'] == task_id:
return record
return None
def task2query(self, task):
query = 'Your task is: {}'.format(task['Question'])
if task['file_name'] != '':
query = query + '\nThe path of the files you need to use: {0}.{1}'.format(task['file_path'], task['file_name'].split('.')[-1])
print('GAIA Task {1}:\n{2}'.format(task['task_id'], query))
logging.info(query)
return query
class SheetTaskLoader:
def __init__(self, sheet_task_path=None):
if sheet_task_path != None:
assert os.path.exists(sheet_task_path), f"Sheet task jsonl file {sheet_task_path} does not exist."
self.sheet_task_path = sheet_task_path
try:
self.dataset = self.load_sheet_task_dataset()
except Exception as e:
raise Exception(f"Failed to load sheet task dataset: {e}")
else:
print("Sheet task jsonl file not provided.")
def load_sheet_task_dataset(self):
dataset = []
with open(self.sheet_task_path, 'r') as file:
for _, line in enumerate(file):
task_info = json.loads(line)
query = self.task2query(task_info['Context'], task_info['Instructions'], get_project_root_path() + task_info['file_path'])
dataset.append(query)
return dataset
def task2query(self, context, instructions, file_path):
SHEET_TASK_PROMPT = """You are an expert in handling excel file. {context}
Your task is: {instructions}
The file path of the excel is: {file_path}. Every subtask's description must include the file path, and all subtasks are completed on the file at that path.
"""
query = SHEET_TASK_PROMPT.format(context=context, instructions=instructions, file_path=file_path)
return query
def get_data_by_task_id(self, task_id):
if self.dataset is None:
raise ValueError("Dataset not loaded.")
return self.dataset[task_id]