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

fix saving user and app, update user stories and user tasks to be generated one by one

Browse Source
This commit is contained in:
LeonOstrez
2023-08-03 17:24:59 +02:00
parent 3f6ea24dae
commit 208fba56a9
8 changed files with 122 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
euclid/database/database.py
View File

@@ -1,5 +1,7 @@
from playhouse.shortcuts import model_to_dict
from peewee import *
from functools import reduce
import operator
from utils.utils import hash_data
from database.models.components.base_models import database
@@ -16,20 +18,45 @@ from database.models.development import Development
from database.models.file_snapshot import FileSnapshot
def save_user(user_id, email="email", password="password"):
def save_user(user_id, email, password):
try:
user = User.get(User.id == user_id)
return user
except DoesNotExist:
return User.create(id=user_id, email=email, password=password)
try:
return User.create(id=user_id, email=email, password=password)
except IntegrityError as e:
existing_user = User.get(User.email == email)
return existing_user
def save_app(user_id, app_id, app_type):
def get_user(user_id=None, email=None):
if not user_id and not email:
raise ValueError("Either user_id or email must be provided")
query = []
if user_id:
query.append(User.id == user_id)
if email:
query.append(User.email == email)
try:
app = App.get(App.id == app_id)
user = User.get(reduce(operator.or_, query))
return user
except DoesNotExist:
user = save_user(user_id)
app = App.create(id=app_id, user=user, app_type=app_type)
raise ValueError("No user found with provided id or email")
def save_app(args):
try:
app = App.get(App.id == args['app_id'])
except DoesNotExist:
try:
user = get_user(user_id=args['user_id'])
except ValueError:
user = save_user(args['user_id'], args['email'], args['password'])
app = App.create(id=args['app_id'], user=user, app_type=args['app_type'])
return app

42
euclid/helpers/AgentConvo.py
View File

@@ -1,27 +1,32 @@
import subprocess
from termcolor import colored
from database.database import get_development_step_from_messages, save_development_step
from utils.utils import array_of_objects_to_string
from utils.llm_connection import get_prompt, create_gpt_chat_completion
from utils.utils import get_sys_message, find_role_from_step, capitalize_first_word_with_underscores
from logger.logger import logger
from termcolor import colored
from prompts.prompts import ask_user
from const.llm import END_RESPONSE
class AgentConvo:
def __init__(self, agent):
self.messages = []
self.branches = {}
self.log_to_user = True
self.agent = agent
self.high_level_step = self.agent.project.current_step
# add system message
self.messages.append(get_sys_message(self.agent.role))
def send_message(self, prompt_path, prompt_data, function_calls=None):
def send_message(self, prompt_path=None, prompt_data=None, function_calls=None):
# craft message
prompt = get_prompt(prompt_path, prompt_data)
self.messages.append({"role": "user", "content": prompt})
if prompt_path is not None and prompt_data is not None:
prompt = get_prompt(prompt_path, prompt_data)
self.messages.append({"role": "user", "content": prompt})
# check if we already have the LLM response saved
saved_checkpoint = get_development_step_from_messages(self.agent.project.args['app_id'], self.messages)
@@ -33,7 +38,7 @@ class AgentConvo:
# if we don't, get the response from LLM
response = create_gpt_chat_completion(self.messages, self.high_level_step, function_calls=function_calls)
save_development_step(self.agent.project.args['app_id'], self.messages, response)
# TODO handle errors from OpenAI
if response == {}:
raise Exception("OpenAI API error happened.")
@@ -54,7 +59,6 @@ class AgentConvo:
message_content = '\n'.join(string_response)
# TODO END
# TODO we need to specify the response when there is a function called
# TODO maybe we can have a specific function that creates the GPT response from the function call
self.messages.append({"role": "assistant", "content": message_content})
@@ -62,6 +66,25 @@ class AgentConvo:
return response
def continuous_conversation(self, prompt_path, prompt_data, function_calls=None):
self.log_to_user = False
accepted_messages = []
response = self.send_message(prompt_path, prompt_data, function_calls)
# Continue conversation until GPT response equals END_RESPONSE
while response != END_RESPONSE:
print(colored("Do you want to add anything else? If not, just press ENTER.", 'yellow'))
user_message = ask_user(response, False)
if user_message == "":
accepted_messages.append(response)
self.messages.append({"role": "user", "content": user_message})
response = self.send_message(None, None, function_calls)
self.log_to_user = True
return accepted_messages
def save_branch(self, branch_name):
self.branches[branch_name] = self.messages.copy()
@@ -83,8 +106,9 @@ class AgentConvo:
def log_message(self, content):
print_msg = capitalize_first_word_with_underscores(self.high_level_step)
print(colored(f"{print_msg}:\n", "green"))
print(f"{content}\n")
if self.log_to_user:
print(colored(f"{print_msg}:\n", "green"))
print(f"{content}\n")
logger.info(f"{print_msg}: {content}\n")
def to_playground(self):

5
euclid/helpers/agents/Architect.py
View File

@@ -10,9 +10,11 @@ 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)
self.convo_architecture = None
def get_architecture(self):
self.project.current_step = 'architecture'
@@ -34,7 +36,8 @@ class Architect(Agent):
'user_tasks': self.project.user_tasks,
'app_type': self.project.args['app_type']}, ARCHITECTURE)
architecture = get_additional_info_from_user(architecture, 'architect')
if self.project.args.get('advanced', False):
architecture = get_additional_info_from_user(architecture, 'architect')
logger.info(f"Final architecture: {architecture}")

56
euclid/helpers/agents/ProductOwner.py
View File

@@ -1,14 +1,14 @@
from helpers.Agent import Agent
import json
from termcolor import colored
from helpers.AgentConvo import AgentConvo
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
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
generate_messages_from_description
from const.llm import END_RESPONSE
class ProductOwner(Agent):
def __init__(self, project):
@@ -28,7 +28,7 @@ class ProductOwner(Agent):
# 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'])
save_app(self.project.args)
main_prompt = ask_for_main_app_definition()
@@ -54,7 +54,7 @@ class ProductOwner(Agent):
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
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):
@@ -63,29 +63,27 @@ class ProductOwner(Agent):
return step['user_stories']
# USER STORIES
print(colored(f"Generating user stories...\n", "green"))
logger.info(f"Generating user stories...")
msg = f"Generating USER STORIES...\n"
print(colored(msg, "green"))
logger.info(msg)
user_stories = self.convo_user_stories.send_message('user_stories/specs.prompt', {
self.project.user_stories = self.convo_user_stories.continuous_conversation('user_stories/specs.prompt', {
'prompt': self.project_description,
'app_type': self.project.args['app_type']
}, USER_STORIES)
'app_type': self.project.args['app_type'],
'END_RESPONSE': END_RESPONSE
})
logger.info(user_stories)
user_stories = get_additional_info_from_user(user_stories, 'product_owner')
logger.info(f"Final user stories: {user_stories}")
logger.info(f"Final user stories: {self.project.user_stories}")
save_progress(self.project.args['app_id'], self.project.current_step, {
"messages": self.convo_user_stories.messages,
"user_stories": user_stories,
"user_stories": self.project.user_stories,
"app_data": generate_app_data(self.project.args)
})
return user_stories
return self.project.user_stories
# USER STORIES END
def get_user_tasks(self):
self.project.current_step = 'user_tasks'
self.convo_user_stories.high_level_step = self.project.current_step
@@ -97,22 +95,20 @@ class ProductOwner(Agent):
return step['user_tasks']
# USER TASKS
print(colored(f"Generating user tasks...\n", "green"))
logger.info(f"Generating user tasks...")
msg = f"Generating USER TASKS...\n"
print(colored(msg, "green"))
logger.info(msg)
user_tasks = self.convo_user_stories.send_message('user_stories/user_tasks.prompt',
{}, USER_TASKS)
self.project.user_tasks = self.convo_user_stories.continuous_conversation('user_stories/user_tasks.prompt',
{ 'END_RESPONSE': END_RESPONSE })
logger.info(user_tasks)
user_tasks = get_additional_info_from_user(user_tasks, 'product_owner')
logger.info(f"Final user tasks: {user_tasks}")
logger.info(f"Final user tasks: {self.project.user_tasks}")
save_progress(self.project.args['app_id'], self.project.current_step, {
"messages": self.convo_user_stories.messages,
"user_tasks": user_tasks,
"user_tasks": self.project.user_tasks,
"app_data": generate_app_data(self.project.args)
})
return user_tasks
# USER TASKS END
return self.project.user_tasks
# USER TASKS END

1
euclid/main.py
View File

@@ -7,6 +7,7 @@ from helpers.Project import Project
from utils.utils import get_arguments
from logger.logger import logger
def init():
load_dotenv()

9
euclid/prompts/user_stories/specs.prompt
View File

@@ -1,4 +1,4 @@
I want you to create the application (let's call it Euclid) that can be described like this:
I want you to create {{ app_type }} (let's call it Euclid) that can be described like this:
```
{{ prompt }}
```
@@ -15,4 +15,9 @@ Think step by step about the description for the app Euclid and the additional q
- `user will run the script from the CLI`
- `user will get the list of all channels in a CSV file`
Return the list of user stories in a JSON array.
**IMPORTANT**
Return one user story at the time. Do not return anything else but single user story. I might ask you to modify some user stories and only when I send you empty response you can move to next user story.
**IMPORTANT**
Once you are done creating all user stories, write the response containing nothing else but this:
{{END_RESPONSE}}

10
euclid/prompts/user_stories/user_tasks.prompt
View File

@@ -1,2 +1,10 @@
Ok, great. Now, based on these stories, break down user tasks that a user needs to do to interact with the app. In the example description (`Create a script that finds Youtube channels with the word "test" inside the channel name`), user tasks could be:
- `user runs the CLI command in which they specify the keyword youtube channel needs to contain and the location where the CSV file will be saved to`
- `user runs the CLI command in which they specify the keyword youtube channel needs to contain and the location where the CSV file will be saved to`
**IMPORTANT**
Return one user task at the time. Do not return anything else but single user task. I might ask you to modify some user tasks and only when I send you empty response you can move to next user task.
**IMPORTANT**
Once you are done creating all user tasks, write the response containing nothing else but this:
{{END_RESPONSE}}

11
euclid/utils/utils.py
View File

@@ -30,12 +30,17 @@ def get_arguments():
key, value = arg.split('=', 1)
arguments[key] = value
else:
# Handle arguments without '=' (e.g., positional arguments).
pass
arguments[arg] = True
if 'user_id' not in arguments:
arguments['user_id'] = str(uuid.uuid4())
if 'email' not in arguments:
arguments['email'] = 'email'
if 'password' not in arguments:
arguments['password'] = 'password'
if 'app_id' not in arguments:
arguments['app_id'] = str(uuid.uuid4())
@@ -158,9 +163,11 @@ def step_already_finished(args, step):
def generate_app_data(args):
return {'app_id': args['app_id'], 'app_type': args['app_type']}
def array_of_objects_to_string(array):
return '\n'.join([f'{key}: {value}' for key, value in array.items()])
def hash_data(data):
serialized_data = json.dumps(data, sort_keys=True).encode('utf-8')
return hashlib.sha256(serialized_data).hexdigest()