github/gpt-pilot
1
1
Fork 0
mirror of https://github.com/Pythagora-io/gpt-pilot.git synced 2026-01-10 05:27:54 -05: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-08-07 09:03:45 +02:00
parent 0a29d5f08b 48b7f19b92
commit 9276317bb6
19 changed files with 131 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
euclid/database/database.py
View File

@@ -33,7 +33,6 @@ def save_user(user_id, email, password):
return existing_user
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")
@@ -59,7 +58,7 @@ def save_app(args):
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'])
app = App.create(id=args['app_id'], user=user, app_type=args['app_type'], name=args['name'])
return app
@@ -146,11 +145,11 @@ def save_development_step(app_id, prompt_path, prompt_data, llm_req_num, message
})
try:
inserted_id = (DevelopmentSteps
.insert(app=app, hash_id=hash_id, messages=messages, llm_response=response)
.on_conflict(conflict_target=[DevelopmentSteps.app, DevelopmentSteps.hash_id],
preserve=[DevelopmentSteps.messages, DevelopmentSteps.llm_response],
update={})
.execute())
.insert(app=app, hash_id=hash_id, messages=messages, llm_response=response)
.on_conflict(conflict_target=[DevelopmentSteps.app, DevelopmentSteps.hash_id],
preserve=[DevelopmentSteps.messages, DevelopmentSteps.llm_response],
update={})
.execute())
dev_step = DevelopmentSteps.get_by_id(inserted_id)
print(colored(f"Saved DEV step => {dev_step.id}", "yellow"))
@@ -159,6 +158,7 @@ def save_development_step(app_id, prompt_path, prompt_data, llm_req_num, message
return None
return dev_step
def get_db_model_from_hash_id(data_to_hash, model, app_id):
hash_id = hash_data(data_to_hash)
try:
@@ -167,6 +167,7 @@ def get_db_model_from_hash_id(data_to_hash, model, app_id):
return None
return db_row
def hash_and_save_step(Model, app_id, hash_data_args, data_fields, message):
app = get_app(app_id)
hash_id = hash_data(hash_data_args)
@@ -180,11 +181,11 @@ def hash_and_save_step(Model, app_id, hash_data_args, data_fields, message):
try:
inserted_id = (Model
.insert(**data_to_insert)
.on_conflict(conflict_target=[Model.app, Model.hash_id],
preserve=[field for field in data_fields.keys()],
update={})
.execute())
.insert(**data_to_insert)
.on_conflict(conflict_target=[Model.app, Model.hash_id],
preserve=[field for field in data_fields.keys()],
update={})
.execute())
record = Model.get_by_id(inserted_id)
print(colored(f"{message} with id {record.id}", "yellow"))
@@ -193,6 +194,7 @@ def hash_and_save_step(Model, app_id, hash_data_args, data_fields, message):
return None
return record
def save_command_run(project, command, cli_response):
hash_data_args = {
'command': command,
@@ -204,6 +206,7 @@ def save_command_run(project, command, cli_response):
}
return hash_and_save_step(CommandRuns, project.args['app_id'], hash_data_args, data_fields, "Saved Command Run")
def get_command_run_from_hash_id(project, command):
data_to_hash = {
'command': command,
@@ -282,7 +285,6 @@ def drop_tables():
database.execute_sql(f'DROP TABLE IF EXISTS "{table._meta.table_name}" CASCADE')
if __name__ == "__main__":
drop_tables()
create_tables()

1
euclid/database/models/app.py
View File

@@ -7,4 +7,5 @@ from database.models.user import User
class App(BaseModel):
user = ForeignKeyField(User, backref='apps')
app_type = CharField()
name = CharField()
status = CharField(default='started')

32
euclid/helpers/Project.py
View File

@@ -12,9 +12,12 @@ from helpers.agents.ProductOwner import ProductOwner
from database.models.development_steps import DevelopmentSteps
from database.models.file_snapshot import FileSnapshot
from utils.files import get_parent_folder
class Project:
def __init__(self, args, name=None, description=None, user_stories=None, user_tasks=None, architecture=None, development_plan=None, current_step=None):
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
self.llm_req_num = 0
self.command_runs_count = 0
@@ -22,22 +25,23 @@ class Project:
self.skip_steps = False if ('skip_until_dev_step' in args and args['skip_until_dev_step'] == '0') else True
self.skip_until_dev_step = args['skip_until_dev_step'] if 'skip_until_dev_step' in args else None
# TODO make flexible
# self.root_path = get_parent_folder('euclid')
self.root_path = ''
self.restore_files({dev_step_id_to_start_from})
# self.restore_files({dev_step_id_to_start_from})
if current_step != None:
if current_step is not None:
self.current_step = current_step
if name != None:
if name is not None:
self.name = name
if description != None:
if description is not None:
self.description = description
if user_stories != None:
if user_stories is not None:
self.user_stories = user_stories
if user_tasks != None:
if user_tasks is not None:
self.user_tasks = user_tasks
if architecture != None:
if architecture is not None:
self.architecture = architecture
if development_plan != None:
if development_plan is not None:
self.development_plan = development_plan
def start(self):
@@ -54,7 +58,7 @@ class Project:
self.developer = Developer(self)
self.developer.set_up_environment();
self.developer.start_coding()
def get_directory_tree(self):
@@ -63,7 +67,7 @@ class Project:
def get_test_directory_tree(self):
# TODO remove hardcoded path
return build_directory_tree(self.root_path + '/tests', ignore=IGNORE_FOLDERS)
def get_files(self, files):
files_with_content = []
for file in files:
@@ -72,7 +76,7 @@ class Project:
"content": open(self.get_full_file_path(file), 'r').read()
})
return files_with_content
def get_full_file_path(self, file_name):
return self.root_path + '/' + file_name
@@ -86,7 +90,7 @@ class Project:
name=file['name'],
defaults={'content': file.get('content', '')}
)
file_snapshot.content = content=file['content']
file_snapshot.content = content = file['content']
file_snapshot.save()
def restore_files(self, development_step_id):
@@ -111,4 +115,4 @@ class Project:
answer = styled_text(
self,
'Once you are ready, type "continue" to continue.',
)
)

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

@@ -31,10 +31,11 @@ class Architect(Agent):
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)
{'name': self.project.args['name'],
'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)
if self.project.args.get('advanced', False):
architecture = get_additional_info_from_user(self.project, architecture, 'architect')

3
euclid/helpers/agents/Developer.py
View File

@@ -37,6 +37,7 @@ class Developer(Agent):
print(colored('-------------------------', 'green'))
convo_dev_task = AgentConvo(self)
task_steps = convo_dev_task.send_message('development/task/breakdown.prompt', {
"name": self.project.args['name'],
"app_summary": self.project.high_level_summary,
"clarification": [],
"user_stories": self.project.user_stories,
@@ -94,7 +95,7 @@ class Developer(Agent):
os_info = get_os_info()
os_specific_techologies = self.convo_os_specific_tech.send_message('development/env_setup/specs.prompt',
{ "os_info": os_info, "technologies": self.project.architecture }, FILTER_OS_TECHNOLOGIES)
{ "name": self.project.args['name'], "os_info": os_info, "technologies": self.project.architecture }, FILTER_OS_TECHNOLOGIES)
for technology in os_specific_techologies:
llm_response = self.convo_os_specific_tech.send_message('development/env_setup/install_next_technology.prompt',

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

@@ -4,9 +4,10 @@ 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 utils.utils import 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
generate_messages_from_description, ask_user
from const.llm import END_RESPONSE
@@ -27,6 +28,9 @@ class ProductOwner(Agent):
# PROJECT DESCRIPTION
self.project.args['app_type'] = ask_for_app_type()
self.project.args['name'] = clean_filename(ask_user('What is the project name?'))
setup_workspace(self.project.root_path, self.project.args['name'])
save_app(self.project.args)
@@ -34,7 +38,7 @@ class ProductOwner(Agent):
high_level_messages = get_additional_info_from_openai(
self.project,
generate_messages_from_description(main_prompt, self.project.args['app_type']))
generate_messages_from_description(main_prompt, self.project.args['app_type'], self.project.args['name']))
high_level_summary = convo_project_description.send_message('utils/summary.prompt',
{'conversation': '\n'.join(
@@ -69,6 +73,7 @@ class ProductOwner(Agent):
logger.info(msg)
self.project.user_stories = self.convo_user_stories.continuous_conversation('user_stories/specs.prompt', {
'name': self.project.args['name'],
'prompt': self.project_description,
'app_type': self.project.args['app_type'],
'END_RESPONSE': END_RESPONSE

1
euclid/helpers/agents/TechLead.py
View File

@@ -35,6 +35,7 @@ class TechLead(Agent):
# TODO add clarifications
self.development_plan = self.convo_development_plan.send_message('development/plan.prompt',
{
"name": self.project.args['name'],
"app_summary": self.project.high_level_summary,
"clarification": [],
"user_stories": self.project.user_stories,

1
euclid/main.py
View File

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

10
euclid/prompts/architecture/technologies.prompt
View File

@@ -1,6 +1,6 @@
You are working in a software development agency and a project manager approached you telling you that you're assigned to work on a new project. You are working on a {{app_type}} called Euclid and you need to create specifications on what technologies should be used in this project.
You are working in a software development agency and a project manager approached you telling you that you're assigned to work on a new project. You are working on a {{app_type}} called "{{ name }}" and you need to create specifications on what technologies should be used in this project.
Here is a high level description of Euclid:
Here is a high level description of "{{ name }}":
```
{{ prompt }}
```
@@ -13,18 +13,18 @@ A: {{ clarification.answer }}
{% endfor %}
```
Here are user stories that specify how users use Euclid:
Here are user stories that specify how users use "{{ name }}":
```
{% for story in user_stories %}
- {{ story }}
{% endfor %}
```
Here are user tasks that specify what users need to do to interact with Euclid:
Here are user tasks that specify what users need to do to interact with "{{ name }}":
```
{% for task in user_tasks %}
- {{ task }}
{% endfor %}
```
Now, based on the app's description, user stories and user tasks, think step by step and write up all technologies that will be used by your development team to create the app Euclid. Do not write any explanations behind your choices but only a list of technologies that will be used.
Now, based on the app's description, user stories and user tasks, think step by step and write up all technologies that will be used by your development team to create the app "{{ name }}". Do not write any explanations behind your choices but only a list of technologies that will be used.

2
euclid/prompts/development/env_setup/specs.prompt
View File

@@ -1,4 +1,4 @@
You are working in a software development agency and a project manager and software architect approach you telling you that you're assigned to work on a new project. You are working on a web app called Euclid and your first job is to set up the environment on a computer.
You are working in a software development agency and a project manager and software architect approach you telling you that you're assigned to work on a new project. You are working on a web app called "{{ name }}" and your first job is to set up the environment on a computer.
Here are the technologies that you need to use for this project:
```

8
euclid/prompts/development/plan.prompt
View File

@@ -1,6 +1,6 @@
You are working in a software development agency and a project manager and software architect approach you telling you that you're assigned to work on a new project. You are working on a web app called Euclid and you need to create a detailed development plan so that developers can start developing the app.
You are working in a software development agency and a project manager and software architect approach you telling you that you're assigned to work on a new project. You are working on a web app called "{{ name }}" and you need to create a detailed development plan so that developers can start developing the app.
Here is a high level description of Euclid:
Here is a high level description of "{{ name }}":
```
{{ app_summary }}
```
@@ -13,14 +13,14 @@ A: {{ clarification.answer }}
{% endfor %}
```
Here are user stories that specify how users use Euclid:
Here are user stories that specify how users use "{{ name }}":
```
{% for story in user_stories %}
- {{ story }}
{% endfor %}
```
Here are user tasks that specify what users need to do to interact with Euclid:
Here are user tasks that specify what users need to do to interact with "{{ name }}":
```
{% for task in user_tasks %}
- {{ task }}

8
euclid/prompts/development/task/breakdown.prompt
View File

@@ -1,6 +1,6 @@
You are working on a web app called Euclid and you need to write code for the entire application based on the tasks that the tech lead gives you. So that you understand better what you're working on, you're given other specs for Euclid as well.
You are working on a web app called "{{ name }}" and you need to write code for the entire application based on the tasks that the tech lead gives you. So that you understand better what you're working on, you're given other specs for "{{ name }}" as well.
Here is a high level description of Euclid:
Here is a high level description of "{{ name }}":
```
{{ app_summary }}
```
@@ -13,12 +13,12 @@ A: {{ clarification.answer }}
{% endfor %}
```
#}
Here are user stories that specify how users use Euclid:
Here are user stories that specify how users use "{{ name }}":
```{% for story in user_stories %}
- {{ story }}{% endfor %}
```
Here are user tasks that specify what users need to do to interact with Euclid:
Here are user tasks that specify what users need to do to interact with "{{ name }}":
```{% for task in user_tasks %}
- {{ task }}{% endfor %}
```

4
euclid/prompts/high_level_questions/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 the application (let's call it "{{ name }}") that can be described like this:
```
{{ prompt }}
```
@@ -19,6 +19,6 @@ Here is an overview of the tasks that you need to do:
3. Break down user tasks. In this task, you will think about the app description, answers from step #1 and the user stories from the step #2 and create a list of user tasks that a user needs to do to interact with the app. In the example description, 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`
Let's start with the task #1 Getting additional answers. Think about the description for the app Euclid and ask questions that you would like to get cleared before going onto breaking down the user stories.
Let's start with the task #1 Getting additional answers. Think about the description for the app "{{ name }}" and ask questions that you would like to get cleared before going onto breaking down the user stories.
{{single_question}}

3
euclid/prompts/prompts.py
View File

@@ -134,8 +134,9 @@ def get_additional_info_from_user(project, messages, role):
return updated_messages
def generate_messages_from_description(description, app_type):
def generate_messages_from_description(description, app_type, name):
prompt = get_prompt('high_level_questions/specs.prompt', {
'name': name,
'prompt': description,
'app_type': app_type,
'MAX_QUESTIONS': MAX_QUESTIONS

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

@@ -1,4 +1,4 @@
I want you to create {{ app_type }} (let's call it Euclid) that can be described like this:
I want you to create {{ app_type }} (let's call it "{{ name }}") that can be described like this:
```
{{ prompt }}
```
@@ -11,7 +11,7 @@ A: {{ clarification.answer }}
{% endfor %}
```
Think step by step about the description for the app Euclid and the additional questions and answers and break down user stories. You will think about the app description and the answers listed and create a list of all user stories. A user story is a description of how a user can interact with the app. For example, if an app's description is `Create a script that finds Youtube channels with the word "test" inside the channel name`, user stories could be:
Think step by step about the description for the app "{{ name }}" and the additional questions and answers and break down user stories. You will think about the app description and the answers listed and create a list of all user stories. A user story is a description of how a user can interact with the app. For example, if an app's description is `Create a script that finds Youtube channels with the word "test" inside the channel name`, user stories could be:
- `user will run the script from the CLI`
- `user will get the list of all channels in a CSV file`

25
euclid/utils/files.py Normal file
View File

@@ -0,0 +1,25 @@
import os
from pathlib import Path
def get_parent_folder(folder_name):
current_path = Path(os.path.abspath(__file__)) # get the path of the current script
while current_path.name != folder_name: # while the current folder name is not 'folder_name'
current_path = current_path.parent # go up one level
return current_path.parent
def setup_workspace(root, project_name):
create_directory(root, 'workspace')
project_path = create_directory(os.path.join(root, 'workspace'), project_name)
create_directory(project_path, 'tests')
return
def create_directory(parent_directory, new_directory):
new_directory_path = os.path.join(parent_directory, new_directory)
os.makedirs(new_directory_path, exist_ok=True)
return new_directory_path

21
euclid/utils/llm_connection.py
View File

@@ -9,6 +9,7 @@ from const.llm import MIN_TOKENS_FOR_GPT_RESPONSE, MAX_GPT_MODEL_TOKENS, MAX_QUE
from logger.logger import logger
from termcolor import colored
from utils.utils import get_prompt_components, escape_json_special_chars
from utils.spinner import spinner_start, spinner_stop
def connect_to_llm():
@@ -98,7 +99,7 @@ def create_gpt_chat_completion(messages: List[dict], req_type, min_tokens=MIN_TO
if len(function_calls['definitions']) > 1:
gpt_data['function_call'] = 'auto'
else:
gpt_data['function_call'] = { 'name': function_calls['definitions'][0]['name'] }
gpt_data['function_call'] = {'name': function_calls['definitions'][0]['name']}
try:
response = stream_gpt_completion(gpt_data, req_type)
@@ -111,7 +112,12 @@ def create_gpt_chat_completion(messages: List[dict], req_type, min_tokens=MIN_TO
def stream_gpt_completion(data, req_type):
print(colored("Waiting for OpenAI API response...", 'yellow'))
def return_result(result_data):
# spinner_stop(spinner)
return result_data
# spinner = spinner_start(colored("Waiting for OpenAI API response...", 'yellow'))
colored("Waiting for OpenAI API response...", 'yellow')
api_key = os.getenv("OPENAI_API_KEY")
logger.info(f'Request data: {data}')
@@ -129,10 +135,10 @@ def stream_gpt_completion(data, req_type):
if response.status_code != 200:
print(f'problem with request: {response.text}')
logger.debug(f'problem with request: {response.text}')
return {}
return return_result({})
gpt_response = ''
function_calls = { 'name': '', 'arguments': '' }
function_calls = {'name': '', 'arguments': ''}
for line in response.iter_lines():
# Ignore keep-alive new lines
@@ -150,7 +156,7 @@ def stream_gpt_completion(data, req_type):
json_line = json_loads_with_escape(line)
if json_line['choices'][0]['finish_reason'] == 'function_call':
function_calls['arguments'] = json_loads_with_escape(function_calls['arguments'])
return { 'function_calls': function_calls };
return return_result({'function_calls': function_calls});
json_line = json_line['choices'][0]['delta']
except json.JSONDecodeError:
@@ -174,15 +180,16 @@ def stream_gpt_completion(data, req_type):
if function_calls['arguments'] != '':
logger.info(f'Response via function call: {function_calls["arguments"]}')
function_calls['arguments'] = json_loads_with_escape(function_calls['arguments'])
return { 'function_calls': function_calls };
return return_result({'function_calls': function_calls});
logger.info(f'Response message: {gpt_response}')
new_code = postprocessing(gpt_response, req_type) # TODO add type dynamically
return { 'text': new_code }
return return_result({'text': new_code})
def postprocessing(gpt_response, req_type):
return gpt_response
def json_loads_with_escape(str):
# return json.loads(escape_json_special_chars(str))
return json.loads(str)

12
euclid/utils/spinner.py Normal file
View File

@@ -0,0 +1,12 @@
from yaspin import yaspin
from yaspin.spinners import Spinners
def spinner_start(text="Processing..."):
spinner = yaspin(Spinners.line, text=text)
spinner.start()
return spinner
def spinner_stop(spinner):
spinner.stop()

11
euclid/utils/utils.py
View File

@@ -196,4 +196,13 @@ def escape_json_special_chars(s):
for char, replacement in replacements.items():
s = s.replace(char, replacement)
return s
return s
def clean_filename(filename):
# Remove invalid characters
cleaned_filename = re.sub(r'[<>:"/\\|?*]', '', filename)
# Replace whitespace with underscore
cleaned_filename = re.sub(r'\s', '_', cleaned_filename)
return cleaned_filename