OMGeeky/gpt-pilot
1
0
Fork 0
mirror of https://github.com/OMGeeky/gpt-pilot.git synced 2025-12-31 16:40:03 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #81 from nalbion/feature/should_execute_step

Browse Source 
Refactored `execute_step()` as per #80
This commit is contained in:
LeonOstrez
2023-09-12 12:57:52 +02:00
committed by GitHub
parent b023205a53 5b7b621832
commit 80914f0722
6 changed files with 71 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
pilot/helpers/agents/Architect.py
View File

@@ -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,

24
pilot/helpers/agents/Developer.py
View File

@@ -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

29
pilot/helpers/agents/ProductOwner.py
View File

@@ -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']

11
pilot/helpers/agents/TechLead.py
View File

@@ -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']

19
pilot/utils/test_utils.py Normal file
View File

@@ -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

11
pilot/utils/utils.py
View File

@@ -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):