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

Merge branch 'main' of github.com:Pythagora-io/copilot

Browse Source
This commit is contained in:
Zvonimir Sabljic
2023-07-28 10:26:42 +02:00
parent c863c8fadc abdfb4dd53
commit 3f9bc48f96
9 changed files with 199 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
euclid/const/common.py
View File

@@ -1,8 +1,16 @@
APP_TYPES = ['Web App', 'Script', 'Mobile App (unavailable)', 'Chrome Extension (unavailable)']
ROLES = {
'product_owner': ['project_summary', 'user_stories', 'user_tasks'],
'product_owner': ['project_description', 'user_stories', 'user_tasks'],
'architect': ['architecture'],
'tech_lead': ['development_planing'],
'full_stack_developer': ['create_scripts', 'coding']
}
STEPS = ['project_summary', 'user_stories', 'user_tasks', 'development_planing', 'create_scripts', 'coding']
STEPS = [
'project_description',
'user_stories',
'user_tasks',
'architecture',
'development_planing',
'create_scripts',
'coding'
]

11
euclid/database/database.py
View File

@@ -63,7 +63,8 @@ def create_tables():
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (app_id)
REFERENCES apps (app_id)
ON UPDATE CASCADE ON DELETE CASCADE
ON UPDATE CASCADE ON DELETE CASCADE,
UNIQUE (app_id, step)
)
""")
@@ -113,9 +114,14 @@ def save_progress(app_id, step, data):
if isinstance(data, dict):
data = json.dumps(data)
# INSERT the data, but on conflict (if the app_id and step combination already exists) UPDATE the data
insert = sql.SQL(
"INSERT INTO progress_steps (app_id, step, data, completed) VALUES (%s, %s, %s, false)"
"""INSERT INTO progress_steps (app_id, step, data, completed)
VALUES (%s, %s, %s, false)
ON CONFLICT (app_id, step) DO UPDATE
SET data = excluded.data, completed = excluded.completed"""
)
cursor.execute(insert, (app_id, step, data))
conn.commit()
@@ -123,6 +129,7 @@ def save_progress(app_id, step, data):
conn.close()
def get_apps_by_id(app_id):
conn = create_connection()
cursor = conn.cursor()

96
euclid/main.py
View File

@@ -1,17 +1,14 @@
# main_old.py
# main.py
from __future__ import print_function, unicode_literals
import uuid
import json
from dotenv import load_dotenv
from termcolor import colored
from utils.utils import get_arguments, execute_step, split_into_bullets, find_role_from_step
from database.database import save_progress, save_app, get_progress_steps
from utils.utils import get_arguments
from logger.logger import logger
from prompts.prompts import ask_for_app_type, ask_for_main_app_definition, get_additional_info_from_openai, \
get_additional_info_from_user, generate_messages_from_description, execute_chat_prompt
from utils.llm_connection import get_prompt
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
def init():
@@ -24,85 +21,6 @@ def init():
return arguments
def get_project_description(args):
current_step = 'project_summary'
# 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
# PROJECT DESCRIPTION
app_type = ask_for_app_type()
save_app(args['user_id'], args['app_id'], app_type)
description = ask_for_main_app_definition()
high_level_messages = get_additional_info_from_openai(generate_messages_from_description(description, app_type))
high_level_summary = execute_chat_prompt('utils/summary.prompt',
{'conversation': '\n'.join(
[f"{msg['role']}: {msg['content']}" for msg in high_level_messages])},
current_step)
app_data = {'app_id': args['app_id'], 'app_type': app_type}
args['app_type'] = app_type
save_progress(args['app_id'], current_step,
{"messages": high_level_messages, "summary": high_level_summary, "app_data": app_data})
return high_level_summary
# PROJECT DESCRIPTION END
def get_user_stories(summary, args):
current_step = 'user_stories'
role = find_role_from_step(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"User stories already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
logger.info(message)
return summary, args
# USER STORIES
print(colored(f"Generating user stories...\n", "green"))
logger.info(f"Generating user stories...")
user_stories = execute_chat_prompt('user_stories/specs.prompt',
{'summary': summary, 'app_type': args['app_type']},
current_step)
logger.info(split_into_bullets(user_stories))
user_stories = get_additional_info_from_user(split_into_bullets(user_stories), role)
logger.info(f"Final user stories: {user_stories}")
save_progress(args['app_id'], current_step, {"user_stories": user_stories})
return user_stories
# USER STORIES END
if __name__ == "__main__":
args = init()
@@ -110,6 +28,8 @@ if __name__ == "__main__":
user_stories = get_user_stories(high_level_summary, args)
user_tasks = get_user_tasks(user_stories, args)
# get architecture plan
# development

42
euclid/prompts/prompts.py
View File

@@ -2,6 +2,7 @@
import inquirer
from inquirer.themes import GreenPassion
from termcolor import colored
import questionary
from const import common
from const.llm import MAX_QUESTIONS, END_RESPONSE
@@ -67,21 +68,17 @@ def ask_for_main_app_definition():
def ask_user(question):
while True:
questions = [
inquirer.Text('answer', message=question)
]
answer = questionary.text(question).ask()
answers = inquirer.prompt(questions, theme=GreenPassion())
if answers is None:
if answer is None:
print("Exiting application.")
exit(0)
if answers['answer'].strip() == '':
if answer.strip() == '':
print("No input provided! Please try again.")
continue
else:
return answers['answer']
return answer
def get_additional_info_from_openai(messages):
@@ -115,12 +112,15 @@ def get_additional_info_from_user(messages, role):
for message in messages:
while True:
print(colored(f"Please check this message and say what needs to be changed. If everything is ok just type 'DONE'.", "yellow"))
print(colored(
f"Please check this message and say what needs to be changed. If everything is ok just type 'DONE'.",
"yellow"))
answer = ask_user(message)
if answer.lower() == 'done':
break
response = create_gpt_chat_completion(
generate_messages_from_custom_conversation(role, [get_prompt('utils/update.prompt'), message, answer], 'user'),
generate_messages_from_custom_conversation(role, [get_prompt('utils/update.prompt'), message, answer],
'user'),
'additional_info')
message = response
@@ -146,6 +146,7 @@ def generate_messages_from_description(description, app_type):
def generate_messages_from_custom_conversation(role, messages, start_role='user'):
# messages is list of strings
result = [get_sys_message(role)]
for i, message in enumerate(messages):
@@ -157,21 +158,28 @@ def generate_messages_from_custom_conversation(role, messages, start_role='user'
return result
def execute_chat_prompt(prompt_file, prompt_data, chat_type):
def execute_chat_prompt(prompt_file, prompt_data, chat_type, previous_messages=None):
# Generate a prompt for the completion type.
prompt = get_prompt(prompt_file, prompt_data)
new_message = {"role": "user", "content": prompt}
# Pass the prompt to the API.
messages = [
get_sys_message(find_role_from_step(chat_type)),
{"role": "user", "content": prompt},
]
if previous_messages:
# Use the provided previous_messages instead of the default system message.
messages = previous_messages + [new_message]
else:
# Use the default system message.
messages = [
get_sys_message(find_role_from_step(chat_type)),
new_message,
]
response = create_gpt_chat_completion(messages, chat_type)
messages.append({"role": "assistant", "content": response})
print_msg = capitalize_first_word_with_underscores(chat_type)
print(colored(f"{print_msg}:\n", "green"))
print(f"{response}")
logger.info(f"{print_msg}: {response}\n")
return response
return response, messages

48
euclid/steps/project_description/project_description.py Normal file
View File

@@ -0,0 +1,48 @@
# project_description.py
import json
from termcolor import colored
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, execute_chat_prompt
def get_project_description(args):
current_step = 'project_description'
# 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
# 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, high_level_messages = execute_chat_prompt('utils/summary.prompt',
{'conversation': '\n'.join(
[f"{msg['role']}: {msg['content']}" for msg in high_level_messages])},
current_step)
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
# PROJECT DESCRIPTION END

46
euclid/steps/user_stories/user_stories.py Normal file
View File

@@ -0,0 +1,46 @@
# user_stories.py
import json
from termcolor import colored
from utils.utils import execute_step, split_into_bullets, 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, execute_chat_prompt
def get_user_stories(summary, args):
current_step = 'user_stories'
role = find_role_from_step(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
# USER STORIES
print(colored(f"Generating user stories...\n", "green"))
logger.info(f"Generating user stories...")
user_stories, user_stories_messages = execute_chat_prompt('user_stories/specs.prompt',
{'prompt': summary, 'app_type': args['app_type']},
current_step)
logger.info(split_into_bullets(user_stories))
user_stories = get_additional_info_from_user(split_into_bullets(user_stories), role)
logger.info(f"Final user stories: {user_stories}")
save_progress(args['app_id'], current_step, {"user_stories": user_stories, "app_data": generate_app_data(args)})
return user_stories
# USER STORIES END

46
euclid/steps/user_tasks/user_tasks.py Normal file
View File

@@ -0,0 +1,46 @@
# user_tasks.py
import json
from termcolor import colored
from utils.utils import execute_step, split_into_bullets, 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, execute_chat_prompt
def get_user_tasks(summary, args):
current_step = 'user_tasks'
role = find_role_from_step(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')
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 summary
# USER TASKS
print(colored(f"Generating user tasks...\n", "green"))
logger.info(f"Generating user tasks...")
user_tasks, user_tasks_messages = execute_chat_prompt('user_tasks/specs.prompt',
{'prompt': summary, 'app_type': args['app_type']},
current_step)
logger.info(split_into_bullets(user_tasks))
user_tasks = get_additional_info_from_user(split_into_bullets(user_tasks), role)
logger.info(f"Final user tasks: {user_tasks}")
save_progress(args['app_id'], current_step, {"user_tasks": user_tasks, "app_data": generate_app_data(args)})
return user_tasks
# USER TASKS END

6
euclid/utils/utils.py
View File

@@ -38,7 +38,7 @@ def get_arguments():
if 'step' not in arguments:
arguments['step'] = None
print(f"If you wish to continue with this project in future run 'python main.py app_id={arguments['app_id']}")
print(f"If you wish to continue with this project in future run 'python main.py app_id={arguments['app_id']}'")
return arguments
@@ -149,3 +149,7 @@ def split_into_bullets(text):
split_text = re.split(pattern, text)
split_text = [bullet for bullet in split_text if bullet] # Remove any empty strings from the list
return split_text
def generate_app_data(args):
return {'app_id': args['app_id'], 'app_type': args['app_type']}

2
requirements.txt
View File

@@ -6,9 +6,11 @@ idna==3.4
inquirer==3.1.3
Jinja2==3.1.2
MarkupSafe==2.1.3
prompt-toolkit==3.0.39
psycopg2==2.9.6
python-dotenv==1.0.0
python-editor==1.0.4
questionary==1.10.0
readchar==4.0.5
regex==2023.6.3
requests==2.31.0