diff --git a/pilot/helpers/agents/Architect.py b/pilot/helpers/agents/Architect.py index 6237c4b..e6b1e07 100644 --- a/pilot/helpers/agents/Architect.py +++ b/pilot/helpers/agents/Architect.py @@ -4,12 +4,14 @@ import json from termcolor import colored from const.function_calls import ARCHITECTURE -from utils.utils import execute_step, find_role_from_step, generate_app_data +from utils.utils import should_execute_step, find_role_from_step, generate_app_data from database.database import save_progress, get_progress_steps from logger.logger import logger from prompts.prompts import get_additional_info_from_user from helpers.AgentConvo import AgentConvo +ARCHITECTURE_STEP = 'architecture' + class Architect(Agent): def __init__(self, project): @@ -17,12 +19,11 @@ class Architect(Agent): self.convo_architecture = None def get_architecture(self): - self.project.current_step = 'architecture' - self.convo_architecture = AgentConvo(self) + self.project.current_step = ARCHITECTURE_STEP # If this app_id already did this step, just get all data from DB and don't ask user again - step = get_progress_steps(self.project.args['app_id'], self.project.current_step) - if step and not execute_step(self.project.args['step'], self.project.current_step): + step = get_progress_steps(self.project.args['app_id'], ARCHITECTURE_STEP) + if step and not should_execute_step(self.project.args['step'], ARCHITECTURE_STEP): step_already_finished(self.project.args, step) return step['architecture'] @@ -30,6 +31,7 @@ class Architect(Agent): print(colored(f"Planning project architecture...\n", "green", attrs=['bold'])) logger.info(f"Planning project architecture...") + self.convo_architecture = AgentConvo(self) architecture = self.convo_architecture.send_message('architecture/technologies.prompt', {'name': self.project.args['name'], 'prompt': self.project.project_description, diff --git a/pilot/helpers/agents/Developer.py b/pilot/helpers/agents/Developer.py index 9ff9e60..33bd3da 100644 --- a/pilot/helpers/agents/Developer.py +++ b/pilot/helpers/agents/Developer.py @@ -8,13 +8,15 @@ from helpers.agents.CodeMonkey import CodeMonkey from logger.logger import logger from helpers.Agent import Agent from helpers.AgentConvo import AgentConvo -from utils.utils import execute_step, array_of_objects_to_string, generate_app_data +from utils.utils import should_execute_step, array_of_objects_to_string, generate_app_data from helpers.cli import build_directory_tree, run_command_until_success, execute_command_and_check_cli_response, debug from const.function_calls import FILTER_OS_TECHNOLOGIES, DEVELOPMENT_PLAN, EXECUTE_COMMANDS, GET_TEST_TYPE, DEV_TASKS_BREAKDOWN, IMPLEMENT_TASK from database.database import save_progress, get_progress_steps, save_file_description from utils.utils import get_os_info from helpers.cli import execute_command +ENVIRONMENT_SETUP_STEP = 'environment_setup' + class Developer(Agent): def __init__(self, project): super().__init__('full_stack_developer', project) @@ -155,12 +157,12 @@ class Developer(Agent): def set_up_environment(self): - self.project.current_step = 'environment_setup' + self.project.current_step = ENVIRONMENT_SETUP_STEP self.convo_os_specific_tech = AgentConvo(self) # If this app_id already did this step, just get all data from DB and don't ask user again - step = get_progress_steps(self.project.args['app_id'], self.project.current_step) - if step and not execute_step(self.project.args['step'], self.project.current_step): + step = get_progress_steps(self.project.args['app_id'], ENVIRONMENT_SETUP_STEP) + if step and not should_execute_step(self.project.args['step'], ENVIRONMENT_SETUP_STEP): step_already_finished(self.project.args, step) return @@ -168,7 +170,7 @@ class Developer(Agent): while user_input.lower() != 'done': user_input = styled_text(self.project, 'Please set up your local environment so that the technologies above can be utilized. When you\'re done, write "DONE"') save_progress(self.project.args['app_id'], self.project.current_step, { - "os_specific_techologies": [], "newly_installed_technologies": [], "app_data": generate_app_data(self.project.args) + "os_specific_technologies": [], "newly_installed_technologies": [], "app_data": generate_app_data(self.project.args) }) return # ENVIRONMENT SETUP @@ -176,7 +178,7 @@ class Developer(Agent): logger.info(f"Setting up the environment...") os_info = get_os_info() - os_specific_techologies = self.convo_os_specific_tech.send_message('development/env_setup/specs.prompt', + os_specific_technologies = self.convo_os_specific_tech.send_message('development/env_setup/specs.prompt', { "name": self.project.args['name'], "app_type": self.project.args['app_type'], @@ -184,8 +186,8 @@ class Developer(Agent): "technologies": self.project.architecture }, FILTER_OS_TECHNOLOGIES) - for technology in os_specific_techologies: - # TODO move the functions definisions to function_calls.py + for technology in os_specific_technologies: + # TODO move the functions definitions to function_calls.py cli_response, llm_response = self.convo_os_specific_tech.send_message('development/env_setup/install_next_technology.prompt', { 'technology': technology}, { 'definitions': [{ @@ -200,7 +202,7 @@ class Developer(Agent): }, 'timeout': { 'type': 'number', - 'description': f'Timeout in seconds for the approcimate time this command takes to finish.', + 'description': 'Timeout in seconds for the approcimate time this command takes to finish.', } }, 'required': ['command', 'timeout'], @@ -219,10 +221,10 @@ class Developer(Agent): for cmd in installation_commands: run_command_until_success(cmd['command'], cmd['timeout'], self.convo_os_specific_tech) - logger.info('The entire tech stack neede is installed and ready to be used.') + logger.info('The entire tech stack needed is installed and ready to be used.') save_progress(self.project.args['app_id'], self.project.current_step, { - "os_specific_techologies": os_specific_techologies, "newly_installed_technologies": [], "app_data": generate_app_data(self.project.args) + "os_specific_technologies": os_specific_technologies, "newly_installed_technologies": [], "app_data": generate_app_data(self.project.args) }) # ENVIRONMENT SETUP END diff --git a/pilot/helpers/agents/ProductOwner.py b/pilot/helpers/agents/ProductOwner.py index 4f4b932..f9213eb 100644 --- a/pilot/helpers/agents/ProductOwner.py +++ b/pilot/helpers/agents/ProductOwner.py @@ -4,12 +4,16 @@ from helpers.AgentConvo import AgentConvo from helpers.Agent import Agent from logger.logger import logger from database.database import save_progress, save_app, get_progress_steps -from utils.utils import execute_step, generate_app_data, step_already_finished, clean_filename +from utils.utils import should_execute_step, generate_app_data, step_already_finished, clean_filename from utils.files import setup_workspace from prompts.prompts import ask_for_app_type, ask_for_main_app_definition, get_additional_info_from_openai, \ generate_messages_from_description, ask_user from const.llm import END_RESPONSE +PROJECT_DESCRIPTION_STEP = 'project_description' +USER_STORIES_STEP = 'user_stories' +USER_TASKS_STEP = 'user_tasks' + class ProductOwner(Agent): def __init__(self, project): @@ -17,12 +21,11 @@ class ProductOwner(Agent): def get_project_description(self): self.project.app = save_app(self.project.args) - self.project.current_step = 'project_description' - convo_project_description = AgentConvo(self) + self.project.current_step = PROJECT_DESCRIPTION_STEP # If this app_id already did this step, just get all data from DB and don't ask user again - step = get_progress_steps(self.project.args['app_id'], self.project.current_step) - if step and not execute_step(self.project.args['step'], self.project.current_step): + step = get_progress_steps(self.project.args['app_id'], PROJECT_DESCRIPTION_STEP) + if step and not should_execute_step(self.project.args['step'], PROJECT_DESCRIPTION_STEP): step_already_finished(self.project.args, step) self.project.root_path = setup_workspace(self.project.args) self.project.project_description = step['summary'] @@ -30,7 +33,8 @@ class ProductOwner(Agent): return # PROJECT DESCRIPTION - self.project.args['app_type'] = ask_for_app_type() + if 'app_type' not in self.project.args: + self.project.args['app_type'] = ask_for_app_type() if 'name' not in self.project.args: self.project.args['name'] = clean_filename(ask_user(self.project, 'What is the project name?')) @@ -45,6 +49,7 @@ class ProductOwner(Agent): generate_messages_from_description(main_prompt, self.project.args['app_type'], self.project.args['name'])) print(colored('Project Summary:\n', 'green', attrs=['bold'])) + convo_project_description = AgentConvo(self) high_level_summary = convo_project_description.send_message('utils/summary.prompt', {'conversation': '\n'.join( [f"{msg['role']}: {msg['content']}" for msg in @@ -63,12 +68,12 @@ class ProductOwner(Agent): # PROJECT DESCRIPTION END def get_user_stories(self): - self.project.current_step = 'user_stories' + self.project.current_step = USER_STORIES_STEP self.convo_user_stories = AgentConvo(self) # If this app_id already did this step, just get all data from DB and don't ask user again - step = get_progress_steps(self.project.args['app_id'], self.project.current_step) - if step and not execute_step(self.project.args['step'], self.project.current_step): + step = get_progress_steps(self.project.args['app_id'], USER_STORIES_STEP) + if step and not should_execute_step(self.project.args['step'], USER_STORIES_STEP): step_already_finished(self.project.args, step) self.convo_user_stories.messages = step['messages'] return step['user_stories'] @@ -98,12 +103,12 @@ class ProductOwner(Agent): # USER STORIES END def get_user_tasks(self): - self.project.current_step = 'user_tasks' + self.project.current_step = USER_TASKS_STEP self.convo_user_stories.high_level_step = self.project.current_step # If this app_id already did this step, just get all data from DB and don't ask user again - step = get_progress_steps(self.project.args['app_id'], self.project.current_step) - if step and not execute_step(self.project.args['step'], self.project.current_step): + step = get_progress_steps(self.project.args['app_id'], USER_TASKS_STEP) + if step and not should_execute_step(self.project.args['step'], USER_TASKS_STEP): step_already_finished(self.project.args, step) return step['user_tasks'] diff --git a/pilot/helpers/agents/TechLead.py b/pilot/helpers/agents/TechLead.py index 08f55fb..7b42251 100644 --- a/pilot/helpers/agents/TechLead.py +++ b/pilot/helpers/agents/TechLead.py @@ -6,7 +6,7 @@ from const.function_calls import DEV_STEPS from helpers.cli import build_directory_tree from helpers.AgentConvo import AgentConvo -from utils.utils import execute_step, array_of_objects_to_string, generate_app_data +from utils.utils import should_execute_step, array_of_objects_to_string, generate_app_data from database.database import save_progress, get_progress_steps from logger.logger import logger from const.function_calls import FILTER_OS_TECHNOLOGIES, DEVELOPMENT_PLAN, EXECUTE_COMMANDS @@ -14,17 +14,20 @@ from const.code_execution import MAX_COMMAND_DEBUG_TRIES from utils.utils import get_os_info from helpers.cli import execute_command +DEVELOPMENT_PLANNING_STEP = 'development_planning' + + class TechLead(Agent): def __init__(self, project): super().__init__('tech_lead', project) def create_development_plan(self): - self.project.current_step = 'development_planning' + self.project.current_step = DEVELOPMENT_PLANNING_STEP self.convo_development_plan = AgentConvo(self) # If this app_id already did this step, just get all data from DB and don't ask user again - step = get_progress_steps(self.project.args['app_id'], self.project.current_step) - if step and not execute_step(self.project.args['step'], self.project.current_step): + step = get_progress_steps(self.project.args['app_id'], DEVELOPMENT_PLANNING_STEP) + if step and not should_execute_step(self.project.args['step'], DEVELOPMENT_PLANNING_STEP): step_already_finished(self.project.args, step) return step['development_plan'] diff --git a/pilot/utils/test_utils.py b/pilot/utils/test_utils.py new file mode 100644 index 0000000..62ea60a --- /dev/null +++ b/pilot/utils/test_utils.py @@ -0,0 +1,19 @@ +from .utils import should_execute_step + + +class TestShouldExecuteStep: + def test_no_step_arg(self): + assert should_execute_step(None, 'project_description') is True + assert should_execute_step(None, 'architecture') is True + assert should_execute_step(None, 'coding') is True + + def test_skip_step(self): + assert should_execute_step('architecture', 'project_description') is False + assert should_execute_step('architecture', 'architecture') is True + assert should_execute_step('architecture', 'coding') is True + + def test_unknown_step(self): + assert should_execute_step('architecture', 'unknown') is False + assert should_execute_step('unknown', 'project_description') is False + assert should_execute_step('unknown', None) is False + assert should_execute_step(None, None) is False diff --git a/pilot/utils/utils.py b/pilot/utils/utils.py index 1a378cd..85b1f3e 100644 --- a/pilot/utils/utils.py +++ b/pilot/utils/utils.py @@ -108,11 +108,16 @@ def get_os_info(): return array_of_objects_to_string(os_info) -def execute_step(matching_step, current_step): - matching_step_index = STEPS.index(matching_step) if matching_step in STEPS else None +def should_execute_step(arg_step, current_step): + """ + :param arg_step: `project.args['step']`, may be None + :param current_step: The step that would be executed next by the calling method. + :return: True if `current_step` should be executed. + """ + arg_step_index = 0 if arg_step is None else STEPS.index(arg_step) if arg_step in STEPS else None current_step_index = STEPS.index(current_step) if current_step in STEPS else None - return matching_step_index is not None and current_step_index is not None and current_step_index >= matching_step_index + return arg_step_index is not None and current_step_index is not None and current_step_index >= arg_step_index def step_already_finished(args, step):