From b9165a66e86e328444e42bdcfd533074a8f3f182 Mon Sep 17 00:00:00 2001 From: Zvonimir Sabljic Date: Tue, 1 Aug 2023 13:52:06 +0200 Subject: [PATCH] Big refactor to use classes --- euclid/helpers/Agent.py | 4 + euclid/helpers/Project.py | 39 ++++ euclid/helpers/agents/Architect.py | 54 +++++ euclid/helpers/agents/ProductOwner.py | 137 ++++++++++++ euclid/helpers/agents/TechLead.py | 58 ++++++ euclid/helpers/agents/__init__.py | 5 + euclid/main.py | 26 +-- euclid/steps/architecture/architecture.py | 51 ----- euclid/steps/development/development.py | 196 ------------------ .../project_description.py | 52 ----- euclid/steps/user_stories/user_stories.py | 51 ----- euclid/steps/user_tasks/user_tasks.py | 48 ----- 12 files changed, 301 insertions(+), 420 deletions(-) create mode 100644 euclid/helpers/Agent.py create mode 100644 euclid/helpers/Project.py create mode 100644 euclid/helpers/agents/Architect.py create mode 100644 euclid/helpers/agents/ProductOwner.py create mode 100644 euclid/helpers/agents/TechLead.py create mode 100644 euclid/helpers/agents/__init__.py delete mode 100644 euclid/steps/architecture/architecture.py delete mode 100644 euclid/steps/development/development.py delete mode 100644 euclid/steps/project_description/project_description.py delete mode 100644 euclid/steps/user_stories/user_stories.py delete mode 100644 euclid/steps/user_tasks/user_tasks.py diff --git a/euclid/helpers/Agent.py b/euclid/helpers/Agent.py new file mode 100644 index 0000000..da704c2 --- /dev/null +++ b/euclid/helpers/Agent.py @@ -0,0 +1,4 @@ +class Agent: + def __init__(self, role, project): + self.role = role + self.project = project \ No newline at end of file diff --git a/euclid/helpers/Project.py b/euclid/helpers/Project.py new file mode 100644 index 0000000..ec0f2dc --- /dev/null +++ b/euclid/helpers/Project.py @@ -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() \ No newline at end of file diff --git a/euclid/helpers/agents/Architect.py b/euclid/helpers/agents/Architect.py new file mode 100644 index 0000000..f55e168 --- /dev/null +++ b/euclid/helpers/agents/Architect.py @@ -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 diff --git a/euclid/helpers/agents/ProductOwner.py b/euclid/helpers/agents/ProductOwner.py new file mode 100644 index 0000000..2bcd790 --- /dev/null +++ b/euclid/helpers/agents/ProductOwner.py @@ -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 \ No newline at end of file diff --git a/euclid/helpers/agents/TechLead.py b/euclid/helpers/agents/TechLead.py new file mode 100644 index 0000000..0f7df7f --- /dev/null +++ b/euclid/helpers/agents/TechLead.py @@ -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 \ No newline at end of file diff --git a/euclid/helpers/agents/__init__.py b/euclid/helpers/agents/__init__.py new file mode 100644 index 0000000..a5aab38 --- /dev/null +++ b/euclid/helpers/agents/__init__.py @@ -0,0 +1,5 @@ +from .ProductOwner import ProductOwner +from .Developer import Developer +from .DevOps import DevOps +from .TechLead import TechLead +from .Architect import Architect diff --git a/euclid/main.py b/euclid/main.py index 82804c5..5b1f65e 100644 --- a/euclid/main.py +++ b/euclid/main.py @@ -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() \ No newline at end of file diff --git a/euclid/steps/architecture/architecture.py b/euclid/steps/architecture/architecture.py deleted file mode 100644 index 8b0817b..0000000 --- a/euclid/steps/architecture/architecture.py +++ /dev/null @@ -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 diff --git a/euclid/steps/development/development.py b/euclid/steps/development/development.py deleted file mode 100644 index 28df7ff..0000000 --- a/euclid/steps/development/development.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/euclid/steps/project_description/project_description.py b/euclid/steps/project_description/project_description.py deleted file mode 100644 index 3ba5cff..0000000 --- a/euclid/steps/project_description/project_description.py +++ /dev/null @@ -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 diff --git a/euclid/steps/user_stories/user_stories.py b/euclid/steps/user_stories/user_stories.py deleted file mode 100644 index e0affdc..0000000 --- a/euclid/steps/user_stories/user_stories.py +++ /dev/null @@ -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 \ No newline at end of file diff --git a/euclid/steps/user_tasks/user_tasks.py b/euclid/steps/user_tasks/user_tasks.py deleted file mode 100644 index 7925d5a..0000000 --- a/euclid/steps/user_tasks/user_tasks.py +++ /dev/null @@ -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 \ No newline at end of file