OMGeeky/gpt-pilot
1
0
Fork 0
mirror of https://github.com/OMGeeky/gpt-pilot.git synced 2026-01-18 00:17:08 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #37 from rajveer43/doc-update

Browse Source 
docs update
This commit is contained in:
LeonOstrez
2023-09-04 22:54:16 +02:00
committed by GitHub
parent c4a5bbf1f2 c4299dac34
commit 38d56277ff
3 changed files with 154 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
pilot/helpers/AgentConvo.py
View File

@@ -11,6 +11,12 @@ from const.llm import END_RESPONSE
class AgentConvo:
"""
Represents a conversation with an agent.
Args:
agent: An instance of the agent participating in the conversation.
"""
def __init__(self, agent):
self.messages = []
self.branches = {}
@@ -22,6 +28,17 @@ class AgentConvo:
self.messages.append(get_sys_message(self.agent.role))
def send_message(self, prompt_path=None, prompt_data=None, function_calls=None):
"""
Sends a message in the conversation.
Args:
prompt_path: The path to a prompt.
prompt_data: Data associated with the prompt.
function_calls: Optional function calls to be included in the message.
Returns:
The response from the agent.
"""
# craft message
self.construct_and_add_message_from_prompt(prompt_path, prompt_data)
@@ -83,6 +100,17 @@ class AgentConvo:
return response
def continuous_conversation(self, prompt_path, prompt_data, function_calls=None):
"""
Conducts a continuous conversation with the agent.
Args:
prompt_path: The path to a prompt.
prompt_data: Data associated with the prompt.
function_calls: Optional function calls to be included in the conversation.
Returns:
List of accepted messages in the conversation.
"""
self.log_to_user = False
accepted_messages = []
response = self.send_message(prompt_path, prompt_data, function_calls)
@@ -111,6 +139,16 @@ class AgentConvo:
return len([msg for msg in self.messages if msg['role'] != 'system'])
def postprocess_response(self, response, function_calls):
"""
Post-processes the response from the agent.
Args:
response: The response from the agent.
function_calls: Optional function calls associated with the response.
Returns:
The post-processed response.
"""
if 'function_calls' in response and function_calls is not None:
if 'send_convo' in function_calls:
response['function_calls']['arguments']['convo'] = self
@@ -121,6 +159,12 @@ class AgentConvo:
return response
def log_message(self, content):
"""
Logs a message in the conversation.
Args:
content: The content of the message to be logged.
"""
print_msg = capitalize_first_word_with_underscores(self.high_level_step)
if self.log_to_user:
if self.agent.project.checkpoints['last_development_step'] is not None:

52
pilot/helpers/Project.py
View File

@@ -21,6 +21,19 @@ from utils.files import get_parent_folder
class Project:
def __init__(self, args, name=None, description=None, user_stories=None, user_tasks=None, architecture=None,
development_plan=None, current_step=None):
"""
Initialize a project.
Args:
args (dict): Project arguments.
name (str, optional): Project name. Default is None.
description (str, optional): Project description. Default is None.
user_stories (list, optional): List of user stories. Default is None.
user_tasks (list, optional): List of user tasks. Default is None.
architecture (str, optional): Project architecture. Default is None.
development_plan (str, optional): Development plan. Default is None.
current_step (str, optional): Current step in the project. Default is None.
"""
self.args = args
self.llm_req_num = 0
self.command_runs_count = 0
@@ -52,6 +65,9 @@ class Project:
# self.development_plan = development_plan
def start(self):
"""
Start the project.
"""
self.project_manager = ProductOwner(self)
self.project_manager.get_project_description()
self.user_stories = self.project_manager.get_user_stories()
@@ -81,6 +97,15 @@ class Project:
self.developer.start_coding()
def get_directory_tree(self, with_descriptions=False):
"""
Get the directory tree of the project.
Args:
with_descriptions (bool, optional): Whether to include descriptions. Default is False.
Returns:
dict: The directory tree.
"""
files = {}
if with_descriptions and False:
files = File.select().where(File.app_id == self.args['app_id'])
@@ -88,15 +113,36 @@ class Project:
return build_directory_tree(self.root_path + '/', ignore=IGNORE_FOLDERS, files=files, add_descriptions=False)
def get_test_directory_tree(self):
"""
Get the directory tree of the tests.
Returns:
dict: The directory tree of tests.
"""
# TODO remove hardcoded path
return build_directory_tree(self.root_path + '/tests', ignore=IGNORE_FOLDERS)
def get_all_coded_files(self):
"""
Get all coded files in the project.
Returns:
list: A list of coded files.
"""
files = File.select().where(File.app_id == self.args['app_id'])
files = self.get_files([file.path + '/' + file.name for file in files])
return files
def get_files(self, files):
"""
Get file contents.
Args:
files (list): List of file paths.
Returns:
list: A list of files with content.
"""
files_with_content = []
for file in files:
# TODO this is a hack, fix it
@@ -113,6 +159,12 @@ class Project:
return files_with_content
def save_file(self, data):
"""
Save a file.
Args:
data (dict): File data.
"""
# TODO fix this in prompts
if ' ' in data['name'] or '.' not in data['name']:
data['name'] = data['path'].rsplit('/', 1)[1]

58
pilot/helpers/cli.py
View File

@@ -24,6 +24,19 @@ def enqueue_output(out, q):
out.close()
def run_command(command, root_path, q_stdout, q_stderr, pid_container):
"""
Execute a command in a subprocess.
Args:
command (str): The command to run.
root_path (str): The directory in which to run the command.
q_stdout (Queue): A queue to capture stdout.
q_stderr (Queue): A queue to capture stderr.
pid_container (list): A list to store the process ID.
Returns:
subprocess.Popen: The subprocess object.
"""
if platform.system() == 'Windows': # Check the operating system
process = subprocess.Popen(
command,
@@ -68,6 +81,18 @@ def terminate_process(pid):
pass
def execute_command(project, command, timeout=None, force=False):
"""
Execute a command and capture its output.
Args:
project: The project associated with the command.
command (str): The command to run.
timeout (int, optional): The maximum execution time in milliseconds. Default is None.
force (bool, optional): Whether to execute the command without confirmation. Default is False.
Returns:
str: The command output.
"""
if timeout is not None:
if timeout < 1000:
timeout *= 1000
@@ -210,12 +235,33 @@ def build_directory_tree(path, prefix="", ignore=None, is_last=False, files=None
return output
def execute_command_and_check_cli_response(command, timeout, convo):
"""
Execute a command and check its CLI response.
Args:
command (str): The command to run.
timeout (int): The maximum execution time in milliseconds.
convo (AgentConvo): The conversation object.
Returns:
tuple: A tuple containing the CLI response and the agent's response.
"""
cli_response = execute_command(convo.agent.project, command, timeout)
response = convo.send_message('dev_ops/ran_command.prompt',
{ 'cli_response': cli_response, 'command': command })
return cli_response, response
def run_command_until_success(command, timeout, convo, additional_message=None, force=False):
"""
Run a command until it succeeds or reaches a timeout.
Args:
command (str): The command to run.
timeout (int): The maximum execution time in milliseconds.
convo (AgentConvo): The conversation object.
additional_message (str, optional): Additional message to include in the response.
force (bool, optional): Whether to execute the command without confirmation. Default is False.
"""
cli_response = execute_command(convo.agent.project, command, timeout, force)
response = convo.send_message('dev_ops/ran_command.prompt',
{'cli_response': cli_response, 'command': command, 'additional_message': additional_message})
@@ -230,6 +276,18 @@ def run_command_until_success(command, timeout, convo, additional_message=None,
def debug(convo, command=None, user_input=None, issue_description=None):
"""
Debug a conversation.
Args:
convo (AgentConvo): The conversation object.
command (dict, optional): The command to debug. Default is None.
user_input (str, optional): User input for debugging. Default is None.
issue_description (str, optional): Description of the issue to debug. Default is None.
Returns:
bool: True if debugging was successful, False otherwise.
"""
function_uuid = str(uuid.uuid4())
convo.save_branch(function_uuid)
success = False