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

Big refactor to use classes

Browse Source
This commit is contained in:
Zvonimir Sabljic
2023-08-01 13:52:06 +02:00
parent 1f55f899b5
commit b9165a66e8
12 changed files with 301 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
euclid/helpers/Agent.py Normal file
View File

@@ -0,0 +1,4 @@
class Agent:
def __init__(self, role, project):
self.role = role
self.project = project

39
euclid/helpers/Project.py Normal file
View File

@@ -0,0 +1,39 @@
from helpers.agents import Developer, DevOps, TechLead, Architect, ProductOwner
class Project:
def __init__(self, args, name=None, description=None, user_stories=None, user_tasks=None, architecture=None, development_plan=None, current_step=None):
self.args = args
if current_step != None:
self.current_step = current_step
if name != None:
self.name = name
if description != None:
self.description = description
if user_stories != None:
self.user_stories = user_stories
if user_tasks != None:
self.user_tasks = user_tasks
if architecture != None:
self.architecture = architecture
if development_plan != None:
self.development_plan = development_plan
def start(self):
self.project_manager = ProductOwner(self)
self.high_level_summary = self.project_manager.get_project_description()
self.user_stories = self.project_manager.get_user_stories()
self.user_tasks = self.project_manager.get_user_tasks()
self.architect = Architect(self)
self.architecture = self.architect.get_architecture()
self.tech_lead = TechLead(self)
self.development_plan = self.tech_lead.create_development_plan()
self.developer = Developer(self)
self.developer.set_up_environment();
self.developer.start_coding()

54
euclid/helpers/agents/Architect.py Normal file
View File

@@ -0,0 +1,54 @@
from helpers.Agent import Agent
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 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
class Architect(Agent):
def __init__(self, project):
super().__init__('architect', project)
def get_architecture(self):
self.project.current_step = 'architecture'
self.convo_architecture = AgentConvo(self)
# If this app_id already did this step, just get all data from DB and don't ask user again
steps = get_progress_steps(self.project.args['app_id'], self.project.current_step)
if steps and not execute_step(self.project.args['step'], self.project.current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
architecture = data.get('architecture')
message = f"Architecture already done for this app_id: {self.project.args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return architecture
# ARCHITECTURE
print(colored(f"Planning project architecture...\n", "green"))
logger.info(f"Planning project architecture...")
architecture = self.convo_architecture.send_message('architecture/technologies.prompt',
{'prompt': self.project.high_level_summary,
'user_stories': self.project.user_stories,
'user_tasks': self.project.user_tasks,
'app_type': self.project.args['app_type']}, ARCHITECTURE)
architecture = get_additional_info_from_user(architecture, 'architect')
logger.info(f"Final architecture: {architecture}")
save_progress(self.project.args['app_id'], self.project.current_step, {
"messages": self.convo_architecture.get_messages(),
"architecture": architecture,
"app_data": generate_app_data(self.project.args)
})
return architecture
# ARCHITECTURE END

137
euclid/helpers/agents/ProductOwner.py Normal file
View File

@@ -0,0 +1,137 @@
from helpers.Agent import Agent
import json
from termcolor import colored
from helpers.AgentConvo import AgentConvo
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
from prompts.prompts import ask_for_app_type, ask_for_main_app_definition, get_additional_info_from_openai, \
generate_messages_from_description, get_additional_info_from_user
from const.function_calls import USER_STORIES, USER_TASKS
class ProductOwner(Agent):
def __init__(self, project):
super().__init__('product_owner', project)
def get_project_description(self):
self.project.current_step = 'project_description'
convo_project_description = AgentConvo(self)
# If this app_id already did this step, just get all data from DB and don't ask user again
steps = get_progress_steps(self.project.args['app_id'], self.project.current_step)
if steps and not execute_step(self.project.args['step'], self.project.current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
summary = data.get('summary')
app_data = data.get('app_data')
self.project.args.update(app_data)
message = f"Project summary already done for this app_id: {self.project.args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
self.project_description = summary
return summary
# PROJECT DESCRIPTION
self.project.args['app_type'] = ask_for_app_type()
save_app(self.project.args['user_id'], self.project.args['app_id'], self.project.args['app_type'])
description = ask_for_main_app_definition()
high_level_messages = get_additional_info_from_openai(
generate_messages_from_description(description, self.project.args['app_type']))
high_level_summary = convo_project_description.send_message('utils/summary.prompt',
{'conversation': '\n'.join(
[f"{msg['role']}: {msg['content']}" for msg in high_level_messages])})
save_progress(self.project.args['app_id'], self.project.current_step,
{"messages": high_level_messages, "summary": high_level_summary, "app_data": generate_app_data(self.project.args)
})
self.project_description = high_level_summary
return high_level_summary
# PROJECT DESCRIPTION END
def get_user_stories(self):
self.project.current_step = 'user_stories'
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
steps = get_progress_steps(self.project.args['app_id'], self.project.current_step)
if steps and not execute_step(self.project.args['step'], self.project.current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
user_stories = data.get('user_stories')
app_data = data.get('app_data')
if app_data is not None:
self.project.args.update(app_data)
message = f"User stories already done for this app_id: {self.project.args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return user_stories
# USER STORIES
print(colored(f"Generating user stories...\n", "green"))
logger.info(f"Generating user stories...")
user_stories = self.convo_user_stories.send_message('user_stories/specs.prompt',
{'prompt': self.project_description, 'app_type': self.project.args['app_type']},
USER_STORIES)
logger.info(user_stories)
user_stories = get_additional_info_from_user(user_stories, 'product_owner')
logger.info(f"Final user stories: {user_stories}")
save_progress(self.project.args['app_id'], self.project.current_step, {
"messages": self.convo_user_stories.get_messages(), "user_stories": user_stories, "app_data": generate_app_data(self.project.args)
})
return user_stories
# USER STORIES END
def get_user_tasks(self):
current_step = 'user_tasks'
# If this app_id already did this step, just get all data from DB and don't ask user again
steps = get_progress_steps(self.project.args['app_id'], current_step)
if steps and not execute_step(self.project.args['step'], current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
app_data = data.get('app_data')
if app_data is not None:
self.project.args.update(app_data)
message = f"User tasks already done for this app_id: {self.project.args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return data.get('user_tasks')
# USER TASKS
print(colored(f"Generating user tasks...\n", "green"))
logger.info(f"Generating user tasks...")
user_tasks = self.convo_user_stories.send_message('user_stories/user_tasks.prompt',
{}, USER_TASKS)
logger.info(user_tasks)
user_tasks = get_additional_info_from_user(user_tasks, 'product_owner')
logger.info(f"Final user tasks: {user_tasks}")
save_progress(self.project.args['app_id'], current_step, {
"messages": self.convo_user_stories.get_messages(),"user_tasks": user_tasks, "app_data": generate_app_data(self.project.args)
})
return user_tasks
# USER TASKS END

58
euclid/helpers/agents/TechLead.py Normal file
View File

@@ -0,0 +1,58 @@
from helpers.Agent import Agent
import json
from termcolor import colored
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 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
from const.code_execution import MAX_COMMAND_DEBUG_TRIES
from utils.utils import get_os_info
from helpers.cli import execute_command
class TechLead(Agent):
def __init__(self, project):
super().__init__('tech_lead', project)
def create_development_plan(self):
self.project.current_step = 'development_planning'
self.convo_development_plan = AgentConvo(self)
steps = get_progress_steps(self.project.args['app_id'], self.project.current_step)
if steps and not execute_step(self.project.args['step'], self.project.current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
app_data = data.get('app_data')
if app_data is not None:
self.project.args.update(app_data)
message = f"Plan for development is already done for this app_id: {self.project.args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return data.get('development_plan')
# DEVELOPMENT PLANNING
print(colored(f"Starting to create the action plan for development...\n", "green"))
logger.info(f"Starting to create the action plan for development...")
# TODO add clarifications
self.development_plan = self.convo_development_plan.send_message('development/plan.prompt',
{
"app_summary": self.project.high_level_summary,
"clarification": [],
"user_stories": self.project.user_stories,
"user_tasks": self.project.user_tasks,
"technologies": self.project.architecture
}, DEVELOPMENT_PLAN)
logger.info('Plan for development is created.')
save_progress(self.project.args['app_id'], self.project.current_step, {
"development_plan": self.development_plan, "app_data": generate_app_data(self.project.args)
})
return self.development_plan

5
euclid/helpers/agents/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from .ProductOwner import ProductOwner
from .Developer import Developer
from .DevOps import DevOps
from .TechLead import TechLead
from .Architect import Architect

26
euclid/main.py
View File

@@ -2,19 +2,11 @@
from __future__ import print_function, unicode_literals
from dotenv import load_dotenv
from helpers.Project import Project
from utils.utils import get_arguments
from logger.logger import logger
from steps.project_description.project_description import get_project_description
from steps.user_stories.user_stories import get_user_stories
from steps.user_tasks.user_tasks import get_user_tasks
from steps.architecture.architecture import get_architecture
from steps.development.development import create_development_plan
from steps.development.development import set_up_environment
from steps.development.development import start_development
def init():
load_dotenv()
@@ -28,16 +20,6 @@ def init():
if __name__ == "__main__":
args = init()
high_level_summary, high_level_messages = get_project_description(args)
user_stories, convo = get_user_stories(high_level_summary, args)
user_tasks = get_user_tasks(convo, args)
architecture = get_architecture(high_level_summary, user_stories, user_tasks, args)
development_plan = create_development_plan(high_level_summary, user_stories, user_tasks, architecture, args)
set_up_environment(architecture, args);
start_development(user_stories, user_tasks, architecture, args)
# TODO get checkpoint from database and fill the project with it
project = Project(args)
project.start()

51
euclid/steps/architecture/architecture.py
View File

@@ -1,51 +0,0 @@
# user_stories.py
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 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
def get_architecture(high_level_summary, user_stories, user_tasks, args):
current_step = 'architecture'
convo_architecture = AgentConvo(current_step)
# If this app_id already did this step, just get all data from DB and don't ask user again
steps = get_progress_steps(args['app_id'], current_step)
if steps and not execute_step(args['step'], current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
architecture = data.get('architecture')
message = f"Architecture already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return architecture
# ARCHITECTURE
print(colored(f"Planning project architecture...\n", "green"))
logger.info(f"Planning project architecture...")
architecture = convo_architecture.send_message('architecture/technologies.prompt',
{'prompt': high_level_summary,
'user_stories': user_stories,
'user_tasks': user_tasks,
'app_type': args['app_type']}, ARCHITECTURE)
architecture = get_additional_info_from_user(architecture, 'architect')
logger.info(f"Final architecture: {architecture}")
save_progress(args['app_id'], current_step, {
"messages": convo_architecture.get_messages(),
"architecture": architecture,
"app_data": generate_app_data(args)
})
return architecture
# ARCHITECTURE END

196
euclid/steps/development/development.py
View File

@@ -1,196 +0,0 @@
import json
from termcolor import colored
from helpers.AgentConvo import AgentConvo
from utils.utils import execute_step, find_role_from_step, 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
from const.code_execution import MAX_COMMAND_DEBUG_TRIES
from utils.utils import get_os_info
from helpers.cli import execute_command
def environment_setup():
# env_setup/specs.prompt
# loop through returned array
# install_next_technology.prompt
# cli_response.prompt
# unsuccessful_installation.prompt
# OR
execute_command();
def implement_task(task):
# development/task/breakdown.prompt
# loop through returned array
# development/task/step/specs.prompt
pass
def execute_command_and_check_cli_response(command, timeout, convo):
cli_response = execute_command(command, timeout)
response = convo.send_message('dev_ops/ran_command.prompt',
{ 'cli_response': cli_response, 'command': command })
return response
def run_command_until_success(command, timeout, convo):
command_executed = False
for _ in range(MAX_COMMAND_DEBUG_TRIES):
cli_response = execute_command(command, timeout)
response = convo.send_message('dev_ops/ran_command.prompt',
{'cli_response': cli_response, 'command': command})
command_executed = response == 'DONE'
if command_executed:
break
command = response
if not command_executed:
# TODO ask user to debug and press enter to continue
pass
def set_up_environment(technologies, args):
current_step = 'environment_setup'
convo_os_specific_tech = AgentConvo(current_step)
steps = get_progress_steps(args['app_id'], current_step)
if steps and not execute_step(args['step'], current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
app_data = data.get('app_data')
if app_data is not None:
args.update(app_data)
message = f"Environment setup is already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return
# ENVIRONMENT SETUP
print(colored(f"Setting up the environment...\n", "green"))
logger.info(f"Setting up the environment...")
# TODO: remove this once the database is set up properly
# previous_messages[2]['content'] = '\n'.join(previous_messages[2]['content'])
# TODO END
os_info = get_os_info()
os_specific_techologies = convo_os_specific_tech.send_message('development/env_setup/specs.prompt',
{ "os_info": os_info, "technologies": technologies }, FILTER_OS_TECHNOLOGIES)
for technology in os_specific_techologies:
llm_response = convo_os_specific_tech.send_message('development/env_setup/install_next_technology.prompt',
{ 'technology': technology}, {
'definitions': [{
'name': 'execute_command',
'description': f'Executes a command that should check if {technology} is installed on the machine. ',
'parameters': {
'type': 'object',
'properties': {
'command': {
'type': 'string',
'description': f'Command that needs to be executed to check if {technology} is installed on the machine.',
},
'timeout': {
'type': 'number',
'description': f'Timeout in seconds for the approcimate time this command takes to finish.',
}
},
'required': ['command', 'timeout'],
},
}],
'functions': {
'execute_command': execute_command_and_check_cli_response
},
'send_convo': True
})
if not llm_response == 'DONE':
installation_commands = convo_os_specific_tech.send_message('development/env_setup/unsuccessful_installation.prompt',
{ 'technology': technology }, {
'definitions': [{
'name': 'execute_commands',
'description': f'Executes a list of commands that should install the {technology} on the machine. ',
'parameters': {
'type': 'object',
'properties': {
'commands': {
'type': 'array',
'description': f'List of commands that need to be executed to install {technology} on the machine.',
'items': {
'type': 'object',
'properties': {
'command': {
'type': 'string',
'description': f'Command that needs to be executed as a step to install {technology} on the machine.',
},
'timeout': {
'type': 'number',
'description': f'Timeout in seconds for the approcimate time this command takes to finish.',
}
}
}
}
},
'required': ['commands'],
},
}],
'functions': {
'execute_commands': lambda commands: commands
}
})
if installation_commands is not None:
for cmd in installation_commands:
run_command_until_success(cmd['command'], cmd['timeout'], convo_os_specific_tech)
logger.info('The entire tech stack neede is installed and ready to be used.')
save_progress(args['app_id'], current_step, {
"os_specific_techologies": os_specific_techologies, "newly_installed_technologies": [], "app_data": generate_app_data(args)
})
# ENVIRONMENT SETUP END
def create_development_plan(high_level_summary, user_stories, user_tasks, technologies_to_use, args):
current_step = 'development_planning'
convo_development_plan = AgentConvo(current_step)
steps = get_progress_steps(args['app_id'], current_step)
if steps and not execute_step(args['step'], current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
app_data = data.get('app_data')
if app_data is not None:
args.update(app_data)
message = f"Plan for development is already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return data.get('development_plan')
# DEVELOPMENT PLANNING
print(colored(f"Starting to create the action plan for development...\n", "green"))
logger.info(f"Starting to create the action plan for development...")
# TODO add clarifications
development_plan = convo_development_plan.send_message('development/plan.prompt',
{
"app_summary": high_level_summary,
"clarification": [],
"user_stories": user_stories,
"user_tasks": user_tasks,
"technologies": technologies_to_use
}, DEVELOPMENT_PLAN)
logger.info('Plan for development is created.')
save_progress(args['app_id'], current_step, {
"development_plan": development_plan, "app_data": generate_app_data(args)
})
return development_plan
def start_development(user_stories, user_tasks, technologies_to_use, args):
pass

52
euclid/steps/project_description/project_description.py
View File

@@ -1,52 +0,0 @@
# project_description.py
import json
from termcolor import colored
from helpers.AgentConvo import AgentConvo
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
from prompts.prompts import ask_for_app_type, ask_for_main_app_definition, get_additional_info_from_openai, \
generate_messages_from_description
def get_project_description(args):
current_step = 'project_description'
convo_project_description = AgentConvo(current_step)
# If this app_id already did this step, just get all data from DB and don't ask user again
steps = get_progress_steps(args['app_id'], current_step)
if steps and not execute_step(args['step'], current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
summary = data.get('summary')
app_data = data.get('app_data')
args.update(app_data)
message = f"Project summary already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return summary, data.get('messages')
# PROJECT DESCRIPTION
args['app_type'] = ask_for_app_type()
save_app(args['user_id'], args['app_id'], args['app_type'])
description = ask_for_main_app_definition()
high_level_messages = get_additional_info_from_openai(
generate_messages_from_description(description, args['app_type']))
high_level_summary = convo_project_description.send_message('utils/summary.prompt',
{'conversation': '\n'.join(
[f"{msg['role']}: {msg['content']}" for msg in high_level_messages])})
save_progress(args['app_id'], current_step,
{"messages": high_level_messages, "summary": high_level_summary, "app_data": generate_app_data(args)
})
return high_level_summary, high_level_messages
# PROJECT DESCRIPTION END

51
euclid/steps/user_stories/user_stories.py
View File

@@ -1,51 +0,0 @@
# user_stories.py
import json
from termcolor import colored
from helpers.AgentConvo import AgentConvo
from utils.utils import 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 const.function_calls import USER_STORIES
def get_user_stories(summary, args):
current_step = 'user_stories'
convo_user_stories = AgentConvo(current_step)
# If this app_id already did this step, just get all data from DB and don't ask user again
steps = get_progress_steps(args['app_id'], current_step)
if steps and not execute_step(args['step'], current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
user_stories = data.get('user_stories')
app_data = data.get('app_data')
if app_data is not None:
args.update(app_data)
message = f"User stories already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return user_stories, data.get('messages')
# USER STORIES
print(colored(f"Generating user stories...\n", "green"))
logger.info(f"Generating user stories...")
user_stories = convo_user_stories.send_message('user_stories/specs.prompt',
{'prompt': summary, 'app_type': args['app_type']},
USER_STORIES)
logger.info(user_stories)
user_stories = get_additional_info_from_user(user_stories, 'product_owner')
logger.info(f"Final user stories: {user_stories}")
save_progress(args['app_id'], current_step, {
"messages": convo_user_stories.get_messages(), "user_stories": user_stories, "app_data": generate_app_data(args)
})
return user_stories, convo_user_stories
# USER STORIES END

48
euclid/steps/user_tasks/user_tasks.py
View File

@@ -1,48 +0,0 @@
# user_tasks.py
import json
from termcolor import colored
from helpers.AgentConvo import AgentConvo
from utils.utils import 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 const.function_calls import USER_TASKS
def get_user_tasks(convo, args):
current_step = 'user_tasks'
# If this app_id already did this step, just get all data from DB and don't ask user again
steps = get_progress_steps(args['app_id'], current_step)
if steps and not execute_step(args['step'], current_step):
first_step = steps[0]
data = json.loads(first_step['data'])
app_data = data.get('app_data')
if app_data is not None:
args.update(app_data)
message = f"User tasks already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return data.get('user_tasks')
# USER TASKS
print(colored(f"Generating user tasks...\n", "green"))
logger.info(f"Generating user tasks...")
user_tasks = convo.send_message('user_stories/user_tasks.prompt',
{}, USER_TASKS)
logger.info(user_tasks)
user_tasks = get_additional_info_from_user(user_tasks, 'product_owner')
logger.info(f"Final user tasks: {user_tasks}")
save_progress(args['app_id'], current_step, {
"messages": convo.get_messages(),"user_tasks": user_tasks, "app_data": generate_app_data(args)
})
return user_tasks
# USER TASKS END