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

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

Browse Source
This commit is contained in:
LeonOstrez
2023-08-14 23:48:30 +02:00
parent 29ceedf700 8c8254b594
commit f6e64d49db
16 changed files with 240 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
euclid/const/function_calls.py
View File

@@ -43,6 +43,7 @@ def return_array_from_prompt(name_plural, name_singular, return_var_name):
def command_definition(description_command=f'A single command that needs to be executed.', description_timeout=f'Timeout in milliseconds that represent the approximate time this command takes to finish. If you need to run a command that doesnt\'t finish by itself (eg. a command to run an app), put the timeout to 3000 milliseconds.'):
return {
'type': 'object',
'description': 'Command that needs to be run to complete the current task. This should be used only if the task is of a type "command".',
'properties': {
'command': {
'type': 'string',
@@ -154,6 +155,64 @@ DEV_TASKS_BREAKDOWN = {
},
}
IMPLEMENT_TASK = {
'definitions': [
{
'name': 'parse_development_task',
'description': 'Breaks down the development task into smaller steps that need to be done to implement the entire task.',
'parameters': {
'type': 'object',
"properties": {
"tasks": {
'type': 'array',
'description': 'List of smaller development steps that need to be done to complete the entire task.',
'items': {
'type': 'object',
'description': 'A smaller development step that needs to be done to complete the entire task. Remember, if you need to run a command that doesnt\'t finish by itself (eg. a command to run an app), put the timeout to 3000 milliseconds.',
'properties': {
'type': {
'type': 'string',
'enum': ['command', 'code_change', 'human_invervention'],
'description': 'Type of the development step that needs to be done to complete the entire task.',
},
'command': command_definition(),
'code_change': {
'type': 'object',
'description': 'A code change that needs to be implemented. This should be used only if the task is of a type "code_change".',
'properties': {
'name': {
'type': 'string',
'description': 'Name of the file that needs to be implemented.',
},
'path': {
'type': 'string',
'description': 'Full path of the file with the file name that needs to be implemented.',
},
'content': {
'type': 'string',
'description': 'Full content of the file that needs to be implemented.',
},
},
'required': ['name', 'path', 'content'],
},
'human_intervention_description': {
'type': 'string',
'description': 'Description of a step in debugging this issue when there is a human intervention needed. This should be used only if the task is of a type "human_intervention".',
},
},
'required': ['type'],
}
}
},
"required": ['tasks'],
},
},
],
'functions': {
'parse_development_task': lambda tasks: tasks
},
}
DEV_STEPS = {
'definitions': [
{

11
euclid/database/database.py
View File

@@ -154,10 +154,9 @@ def get_progress_steps(app_id, step=None):
return steps
def get_db_model_from_hash_id(data_to_hash, model, app_id, previous_step):
hash_id = hash_data(data_to_hash)
def get_db_model_from_hash_id(model, app_id, previous_step):
try:
db_row = model.get((model.hash_id == hash_id) & (model.app == app_id) & (model.previous_step == previous_step))
db_row = model.get((model.app == app_id) & (model.previous_step == previous_step))
except DoesNotExist:
return None
return db_row
@@ -217,7 +216,7 @@ def get_development_step_from_hash_id(project, prompt_path, prompt_data, llm_req
'prompt_data': {} if prompt_data is None else {k: v for k, v in prompt_data.items() if k not in PROMPT_DATA_TO_IGNORE},
'llm_req_num': llm_req_num
}
development_step = get_db_model_from_hash_id(data_to_hash, DevelopmentSteps, project.args['app_id'], project.checkpoints['last_development_step'])
development_step = get_db_model_from_hash_id(DevelopmentSteps, project.args['app_id'], project.checkpoints['last_development_step'])
return development_step
def save_command_run(project, command, cli_response):
@@ -240,7 +239,7 @@ def get_command_run_from_hash_id(project, command):
'command': command,
'command_runs_count': project.command_runs_count
}
command_run = get_db_model_from_hash_id(data_to_hash, CommandRuns, project.args['app_id'], project.checkpoints['last_command_run'])
command_run = get_db_model_from_hash_id(CommandRuns, project.args['app_id'], project.checkpoints['last_command_run'])
return command_run
def save_user_input(project, query, user_input):
@@ -262,7 +261,7 @@ def get_user_input_from_hash_id(project, query):
'query': query,
'user_inputs_count': project.user_inputs_count
}
user_input = get_db_model_from_hash_id(data_to_hash, UserInputs, project.args['app_id'], project.checkpoints['last_user_input'])
user_input = get_db_model_from_hash_id(UserInputs, project.args['app_id'], project.checkpoints['last_user_input'])
return user_input
def delete_all_subsequent_steps(project):

11
euclid/helpers/AgentConvo.py
View File

@@ -23,9 +23,7 @@ class AgentConvo:
def send_message(self, prompt_path=None, prompt_data=None, function_calls=None):
# craft message
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})
self.construct_and_add_message_from_prompt(prompt_path, prompt_data)
if function_calls is not None and 'function_calls' in function_calls:
self.messages[-1]['content'] += '\nMAKE SURE THAT YOU RESPOND WITH A CORRECT JSON FORMAT!!!'
@@ -134,4 +132,9 @@ class AgentConvo:
process.communicate(content.replace('{{messages}}', str(self.messages)).encode('utf-8'))
def remove_last_x_messages(self, x):
self.messages = self.messages[:-x]
self.messages = self.messages[:-x]
def construct_and_add_message_from_prompt(self, prompt_path, prompt_data):
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})

42
euclid/helpers/Project.py
View File

@@ -66,7 +66,7 @@ class Project:
if 'skip_until_dev_step' in self.args:
self.skip_until_dev_step = self.args['skip_until_dev_step']
if self.args['skip_until_dev_step'] == '0':
clear_directory(self.root_path, IGNORE_FOLDERS)
clear_directory(self.root_path)
delete_all_app_development_data(self.args['app_id'])
self.skip_steps = False
else:
@@ -117,18 +117,18 @@ class Project:
data['path'], data['full_path'] = self.get_full_file_path(data['path'], data['name'])
update_file(data['full_path'], data['content'])
file_in_db, created = File.get_or_create(
app=self.app,
name=data['name'],
path=data['path'],
full_path=data['full_path'],
)
(File.insert(app=self.app, path=data['path'], name=data['name'], full_path=data['full_path'])
.on_conflict(
conflict_target=[File.app, File.name, File.path],
preserve=[],
update={ 'name': data['name'], 'path': data['path'], 'full_path': data['full_path'] })
.execute())
def get_full_file_path(self, file_path, file_name):
file_path = file_path.replace('./', '', 1).rstrip(file_name)
if not file_path.endswith('/'):
file_path = file_path + '/'
if file_path.endswith('/'):
file_path = file_path.rstrip('/')
if file_name.startswith('/'):
file_name = file_name[1:]
@@ -136,7 +136,7 @@ class Project:
if not file_path.startswith('/'):
file_path = '/' + file_path
return (file_path, self.root_path + file_path + file_name)
return (file_path, self.root_path + file_path + '/' + file_name)
def save_files_snapshot(self, development_step_id):
files = get_files_content(self.root_path, ignore=IGNORE_FOLDERS)
@@ -166,25 +166,29 @@ class Project:
clear_directory(self.root_path, IGNORE_FOLDERS)
for file_snapshot in file_snapshots:
full_path = self.root_path + file_snapshot.file.path + '/' + file_snapshot.file.name
# Ensure directory exists
os.makedirs(os.path.dirname(full_path), exist_ok=True)
# Write/overwrite the file with its content
with open(full_path, 'w', encoding='utf-8') as f:
f.write(file_snapshot.content)
update_file(file_snapshot.file.full_path, file_snapshot.content);
def delete_all_steps_except_current_branch(self):
delete_unconnected_steps_from(self.checkpoints['last_development_step'], 'previous_step')
delete_unconnected_steps_from(self.checkpoints['last_command_run'], 'previous_step')
delete_unconnected_steps_from(self.checkpoints['last_user_input'], 'previous_step')
def ask_for_human_intervention(self, message, description):
def ask_for_human_intervention(self, message, description=None):
print(colored(message, "yellow"))
print(description)
if description is not None:
print(description)
answer = ''
while answer != 'continue':
answer = styled_text(
self,
'Once you are ready, type "continue" to continue.',
)
if answer != '' and answer != 'continue':
confirmation = styled_text(
self,
'Do you want me to debug this by your instructions? If you mistyped and just want to continue, type "continue" and if you want me to debug this, just press ENTER',
)
if confirmation == '':
print(colored('Ok, just a second.', "yellow"))
return answer

4
euclid/helpers/agents/CodeMonkey.py
View File

@@ -22,10 +22,12 @@ class CodeMonkey(Agent):
changes = convo.send_message('development/implement_changes.prompt', {
"instructions": code_changes_description,
"step_description": code_changes_description,
"step_index": step_index,
"directory_tree": self.project.get_directory_tree(True),
"files": self.project.get_files(files_needed),
}, IMPLEMENT_CHANGES)
convo.remove_last_x_messages(1)
for file_data in changes:
self.project.save_file(file_data)

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

@@ -1,6 +1,7 @@
import json
import uuid
from termcolor import colored
from utils.questionary import styled_text
from helpers.files import update_file
from utils.utils import step_already_finished
from helpers.agents.CodeMonkey import CodeMonkey
@@ -8,9 +9,9 @@ from logger.logger import logger
from helpers.Agent import Agent
from helpers.AgentConvo import AgentConvo
from utils.utils import execute_step, array_of_objects_to_string, generate_app_data
from helpers.cli import build_directory_tree, run_command_until_success, execute_command_and_check_cli_response
from const.function_calls import FILTER_OS_TECHNOLOGIES, DEVELOPMENT_PLAN, EXECUTE_COMMANDS, DEV_STEPS, GET_TEST_TYPE, DEV_TASKS_BREAKDOWN
from database.database import save_progress, get_progress_steps
from helpers.cli import build_directory_tree, run_command_until_success, execute_command_and_check_cli_response, debug
from const.function_calls import FILTER_OS_TECHNOLOGIES, DEVELOPMENT_PLAN, EXECUTE_COMMANDS, GET_TEST_TYPE, DEV_TASKS_BREAKDOWN, IMPLEMENT_TASK
from database.database import save_progress, get_progress_steps, save_file_description
from utils.utils import get_os_info
from helpers.cli import execute_command
@@ -25,21 +26,22 @@ class Developer(Agent):
print(colored(f"Ok, great, now, let's start with the actual development...\n", "green"))
logger.info(f"Starting to create the actual code...")
for i, dev_task in enumerate(self.project.development_plan):
self.implement_task(self.project.development_plan, i)
self.implement_task()
# DEVELOPMENT END
logger.info('The app is DONE!!! Yay...you can use it now.')
def implement_task(self, sibling_tasks, current_task_index, parent_task=None):
def implement_task(self):
print(colored('-------------------------', 'green', attrs=['bold']))
print(colored(f"Implementing task {current_task_index + 1}...\n", "green", attrs=['bold']))
print(colored(sibling_tasks[current_task_index]['description'], 'green', attrs=['bold']))
# print(colored(f"Implementing task {current_task_index + 1}...\n", "green", attrs=['bold']))
print(colored(f"Implementing task...\n", "green", attrs=['bold']))
# print(colored(sibling_tasks[current_task_index]['description'], 'green', attrs=['bold']))
# print(colored(task_explanation, 'green', attrs=['bold']))
print(colored('-------------------------', 'green', attrs=['bold']))
convo_dev_task = AgentConvo(self)
task_steps = convo_dev_task.send_message('development/task/breakdown.prompt', {
task_description = convo_dev_task.send_message('development/task/breakdown.prompt', {
"name": self.project.args['name'],
"app_summary": self.project.project_description,
"clarification": [],
@@ -48,11 +50,13 @@ class Developer(Agent):
"technologies": self.project.architecture,
"array_of_objects_to_string": array_of_objects_to_string,
"directory_tree": self.project.get_directory_tree(True),
"current_task_index": current_task_index,
"sibling_tasks": sibling_tasks,
"parent_task": parent_task,
}, DEV_TASKS_BREAKDOWN)
# "current_task_index": current_task_index,
# "sibling_tasks": sibling_tasks,
# "parent_task": parent_task,
})
task_steps = convo_dev_task.send_message('development/parse_task.prompt', {}, IMPLEMENT_TASK)
convo_dev_task.remove_last_x_messages(2)
self.execute_task(convo_dev_task, task_steps)
def execute_task(self, convo, task_steps, test_command=None, reset_convo=True, test_after_code_changes=True):
@@ -64,32 +68,71 @@ class Developer(Agent):
convo.load_branch(function_uuid)
if step['type'] == 'command':
run_command_until_success(step['command']['command'], step['command']['timeout'], convo)
additional_message = 'Let\'s start with the step #0:\n\n' if i == 0 else f'So far, steps { ", ".join(f"#{j}" for j in range(i)) } are finished so let\'s do step #{i + 1} now.\n\n'
run_command_until_success(step['command']['command'], step['command']['timeout'], convo, additional_message=additional_message)
elif step['type'] == 'code_change':
elif step['type'] == 'code_change' and 'code_change_description' in step:
# TODO this should be refactored so it always uses the same function call
print(f'Implementing code changes for `{step["code_change_description"]}`')
code_monkey = CodeMonkey(self.project, self)
updated_convo = code_monkey.implement_code_changes(convo, step['code_change_description'], i)
if test_after_code_changes:
self.test_code_changes(code_monkey, updated_convo)
elif step['type'] == 'code_change':
self.project.save_file(step['code_change'])
# self.project.save_file(step if 'code_change' not in step else step['code_change'])
elif step['type'] == 'human_intervention':
self.project.ask_for_human_intervention('I need your help! Can you try debugging this yourself and let me take over afterwards? Here are the details about the issue:', step['human_intervention_description'])
if test_command is not None and step.get('check_if_fixed'):
user_feedback = self.project.ask_for_human_intervention('I need your help! Can you try debugging this yourself and let me take over afterwards? Here are the details about the issue:', step['human_intervention_description'])
if user_feedback is not None:
debug(convo, user_input=user_feedback, issue_description=step['human_intervention_description'])
if test_command is not None and ('check_if_fixed' not in step or step['check_if_fixed']):
should_rerun_command = convo.send_message('dev_ops/should_rerun_command.prompt',
test_command)
if should_rerun_command == 'NO':
return True
elif should_rerun_command == 'YES':
response = execute_command_and_check_cli_response(test_command['command'], test_command['timeout'], convo)
if response == 'NEEDS_DEBUGGING':
cli_response, llm_response = execute_command_and_check_cli_response(test_command['command'], test_command['timeout'], convo)
if llm_response == 'NEEDS_DEBUGGING':
print(colored(f'Got incorrect CLI response:', 'red'))
print(response)
print(cli_response)
print(colored('-------------------', 'red'))
if response == 'DONE':
if llm_response == 'DONE':
return True
self.continue_development()
def continue_development(self):
while True:
user_feedback = self.project.ask_for_human_intervention('Can you check if all this works?')
if user_feedback == 'DONE':
return True
if user_feedback is not None:
iteration_convo = AgentConvo(self)
iteration_convo.send_message('development/iteration.prompt', {
"name": self.project.args['name'],
"app_summary": self.project.project_description,
"clarification": [],
"user_stories": self.project.user_stories,
"user_tasks": self.project.user_tasks,
"technologies": self.project.architecture,
"array_of_objects_to_string": array_of_objects_to_string,
"directory_tree": self.project.get_directory_tree(True),
"files": self.project.get_all_coded_files(),
"user_input": user_feedback,
})
# debug(iteration_convo, user_input=user_feedback)
task_steps = iteration_convo.send_message('development/parse_task.prompt', {}, IMPLEMENT_TASK)
iteration_convo.remove_last_x_messages(2)
self.execute_task(iteration_convo, task_steps)
def set_up_environment(self):
self.project.current_step = 'environment_setup'
self.convo_os_specific_tech = AgentConvo(self)
@@ -100,6 +143,10 @@ class Developer(Agent):
step_already_finished(self.project.args, step)
return
user_input = ''
while user_input != 'DONE':
user_input = styled_text(self.project, 'Please set up your local environment so that the technologies above can be utilized. When you\'re done, write "DONE"', 'yellow')
return
# ENVIRONMENT SETUP
print(colored(f"Setting up the environment...\n", "green"))
logger.info(f"Setting up the environment...")
@@ -109,7 +156,8 @@ class Developer(Agent):
{ "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',
# TODO move the functions definisions to function_calls.py
cli_response, llm_response = self.convo_os_specific_tech.send_message('development/env_setup/install_next_technology.prompt',
{ 'technology': technology}, {
'definitions': [{
'name': 'execute_command',

48
euclid/helpers/cli.py
View File

@@ -154,41 +154,55 @@ def execute_command_and_check_cli_response(command, timeout, convo):
cli_response = execute_command(convo.agent.project, command, timeout)
response = convo.send_message('dev_ops/ran_command.prompt',
{ 'cli_response': cli_response, 'command': command })
return response
return cli_response, response
def run_command_until_success(command, timeout, convo):
def run_command_until_success(command, timeout, convo, additional_message=None):
cli_response = execute_command(convo.agent.project, command, timeout)
response = convo.send_message('dev_ops/ran_command.prompt',
{'cli_response': cli_response, 'command': command})
command_successfully_executed = response == 'DONE'
{'cli_response': cli_response, 'command': command, 'additional_message': additional_message})
function_uuid = str(uuid.uuid4())
convo.save_branch(function_uuid)
for i in range(MAX_COMMAND_DEBUG_TRIES):
if command_successfully_executed:
break
convo.load_branch(function_uuid)
if response != 'DONE':
print(colored(f'Got incorrect CLI response:', 'red'))
print(cli_response)
print(colored('-------------------', 'red'))
debug(convo, {'command': command, 'timeout': timeout})
def debug(convo, command=None, user_input=None, issue_description=None):
function_uuid = str(uuid.uuid4())
convo.save_branch(function_uuid)
success = False
for i in range(MAX_COMMAND_DEBUG_TRIES):
if success:
break
convo.load_branch(function_uuid)
debugging_plan = convo.send_message('dev_ops/debug.prompt',
{ 'command': command, 'debugging_try_num': i },
{ 'command': command['command'] if command is not None else None, 'user_input': user_input, 'issue_description': issue_description },
DEBUG_STEPS_BREAKDOWN)
# TODO refactor to nicely get the developer agent
command_successfully_executed = convo.agent.project.developer.execute_task(
success = convo.agent.project.developer.execute_task(
convo,
debugging_plan,
{'command': command, 'timeout': timeout},
command,
False,
False)
if not command_successfully_executed:
if not success:
# TODO explain better how should the user approach debugging
# we can copy the entire convo to clipboard so they can paste it in the playground
convo.agent.project.ask_for_human_intervention(
'It seems like I cannot debug this problem by myself. Can you please help me and try debugging it yourself?',
user_input = convo.agent.project.ask_for_human_intervention(
'It seems like I cannot debug this problem by myself. Can you please help me and try debugging it yourself?' if user_input is None else f'Can you check this again:\n{issue_description}?',
command
)
if user_input == 'continue':
success = True
return success

5
euclid/helpers/files.py
View File

@@ -28,10 +28,11 @@ def get_files_content(directory, ignore=[]):
with open(path, 'r', encoding='utf-8', errors='ignore') as f:
file_content = f.read()
file_name = path.replace(directory + '/', '')
file_name = os.path.basename(path)
relative_path = path.replace(directory, '').replace('/' + file_name, '')
return_array.append({
'name': file_name,
'path': '/' + file.replace(file_name, ''),
'path': relative_path,
'content': file_content,
'full_path': path,
})

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

@@ -27,4 +27,6 @@ Here are user tasks that specify what users need to do to interact with "{{ name
{% 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 "{{ name }}". 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.
You do not need to list any technologies related to automated tests like Jest, Cypress, Mocha, Selenium, etc.

11
euclid/prompts/dev_ops/debug.prompt
View File

@@ -1,7 +1,14 @@
Ok, we need to debug this issue so we can execute `{{ command }}` successfully. I want you to debug this issue by yourself and I will give you 2 functions that you can use - `run_command` and `implement_code_changes`.
{% if issue_description %}
You wanted me to check this - `{{ issue_description }}` but there was a problem{% else %}Ok, we need to debug this issue{% endif %}{% if command %} and we need to be able to execute `{{ command }}` successfully. {% else %}. Here is a brief explanation of what's happening:
```
{{ user_input }}
```
{% endif %}I want you to debug this issue by yourself and I will give you 2 functions that you can use - `run_command` and `implement_code_changes`.
`run_command` function will run a command on the machine and will return the CLI output to you so you can see what to do next.
`implement_code_changes` function will change the code where you just need to thoroughly describe what needs to be implmemented, I will implement the requested changes and let you know.
After this, you need to deside what to do next. You can rerun the command `{{ command }}` to check if the problem is fixed or run another command with `run_command` or change more code with `implement_code_changes`.
Return a list of steps that are needed to debug this issue. By the time we execute the last step, the issue should be fixed completely. Also, make sure that at least the last step has `check_if_fixed` set to TRUE.
{# After this, you need to decide what to do next. You can rerun the command `{{ command }}` to check if the problem is fixed or run another command with `run_command` or change more code with `implement_code_changes`. #}

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

@@ -4,15 +4,7 @@ Here is a high level description of "{{ name }}":
```
{{ app_summary }}
```
{#
Here are some additional questions and answers to clarify the apps description:
```
{% for clarification in clarifications %}
Q: {{ clarification.question }}
A: {{ clarification.answer }}
{% endfor %}
```
#}
Here are user stories that specify how users use "{{ name }}":
```{% for story in user_stories %}
- {{ story }}{% endfor %}
@@ -39,50 +31,8 @@ We've broken it down to these subtasks:
```
{% endif %}
{% if current_task_index != 0 %}
So far, these tasks are done:
```
{% for task in sibling_tasks[0:current_task_index] %}
#{{ loop.index }} {{ task['description'] }}
{% endfor %}
```
so let's the next task which is this:
{% else %}
Let's start with the task #0:
{% endif %}
```
{{ array_of_objects_to_string(sibling_tasks[current_task_index]) }}
```
{#
Think step by step about what needs to be done to complete this task.
{% if sibling_tasks[current_task_index]['type'] == 'COMMAND' %}
Respond with all commands that need to be run to fulfill this step.
{% elif sibling_tasks[current_task_index]['type'] == 'CODE_CHANGE' %}
First, you need to know the code that's currently written so that you can appropriately write new or update the existing code. Here are all the file that are written so far in a file tree format:
```
{{ directory_tree }}
```
Now, tell me all the code that needs to be written to implement this app and have it fully working and all commands that need to be run to implement this app.
You can get the list of files by calling `get_files` function.
{% else %}
#}
This should be a simple version of the app so you don't need to aim to provide a production ready code but rather something that a developer can run locally and play with the implementation. Do not leave any parts of the code to be written afterwards. Make sure that all the code you provide is working and does as outlined in the description area above.
Here are all the file that are written so far in a file tree format:
```
{{ directory_tree }}
```
First, just make a list of steps we need to do to fulfill this task. It should be in a JSON array. Every step must NOT contain both a command that needs to be run and the code that needs to be changed. It can be either command (or multiple commands) that need to be run or a change in the code.
{#
Each step must start with a keyword `command` in case the step consists of commands that need to be run or `code_change` in case it consists of changes in the code. After the keyword, write a description of what will be done in that step. Do not write what needs to be done for each step but only list them in an array.
#}
If a step involves changing the code, it must include code changes for only one single file. In case you need to change multiple files, split it into multiple steps where each involves changes for a single file.
You can also request a human intervention if needed. For example, if you need an API key to some service, you should make one task for human to get an API key and put it where you think is appropriate. Just make sure that you describe the task is enough details so that a person reading it can be completely sure what to do.
You have all the technologies installed on the machine (not dependencies but the technologies like databases, etc.) and you have a folder set up for this project. All commands that you specify will be ran within that folder.
Also, keep in mind that you also need to write test (or tests) that will programmatically verify that your task is complete.
{#
{% endif %}
#}
Remember, I'm currently in an empty folder where I will start writing files that you tell me. Also, tell me how can I test the app to see if it's working or not. You do not need to make any automated tests work.

4
euclid/prompts/development/task/next_step.prompt
View File

@@ -9,11 +9,11 @@ This step by step about what needs to be done to fulfill this step.
{% if step_type == 'COMMAND' %}
Respond with all commands that need to be run to fulfill this step. In case this is a complex step that can't be completed by only running commands (maybe it requires some code changes or user review), respond with an array with only one item `COMPLEX`. Like this - `[{"command": "COMPLEX", "timeout": 0}]`
{% elif step_type == 'CODE_CHANGE' %}
First, you need to know the code that's currently written so that you can appropriately write new or update the existing code. Here are all the file that are written so far in a file tree format:
First, you need to know the code that's currently written so that you can appropriately write new or update the existing code. {# Here are all the file that are written so far in a file tree format:
```
{{ directory_tree }}
```
#}
Respond with a list of files that you need to see before you can write the code for the current step. This list needs to be in a JSON array where each item is a file name. Do not respond with anything other than the mentioned JSON array.
{% endif %}

6
euclid/prompts/development/task/request_files_for_code_changes.prompt
View File

@@ -9,11 +9,11 @@ Let's start with the
{% endif %}
step #{{ step_index }} - `{{ step_description }}`.
I will give you each file that needs to be changed and you will implement changes from the instructions. To do this, you will need to see the currently implemented files so first, filter the files outlined above that are relevant for the instructions. Then, tell me files that you need to see so that you can make appropriate changes to the code. If no files are needed (eg. if you need to create a file), just return an empty array.
{# I will give you each file that needs to be changed and you will implement changes from the instructions. #}To do this, you will need to see the currently implemented files so first, filter the files outlined above that are relevant for the instructions. Then, tell me files that you need to see so that you can make appropriate changes to the code. If no files are needed (eg. if you need to create a file), just return an empty array.
{#
Here is the current folder tree:
```
{{ directory_tree }}
```
#}
Remember, ask for files relative to the project root. For example, if you need a file with path `{project_root}/models/model.py`, you need to request the file `models/model.py`.

2
euclid/prompts/system_messages/full_stack_developer.prompt
View File

@@ -1 +1 @@
You are a full stack software developer who works in a software development agency. You write very modular code and you practice TDD (test driven development) whenever is suitable to use it. Your job is to implement tasks that your tech lead assigns you. Each task has a description of what needs to be implemented, a programmatic goal that will determine if a task can be marked as done from a programmatic perspective (this is basically a blueprint for an automated test that is run before you send the task for a review to your tech lead) and user-review goal that will determine if a task is done or not but from a user perspective since it will be reviewed by a human.
{#You are a full stack software developer who works in a software development agency. You write very modular code and you practice TDD (test driven development) whenever is suitable to use it. Your job is to implement tasks that your tech lead assigns you. Each task has a description of what needs to be implemented, a programmatic goal that will determine if a task can be marked as done from a programmatic perspective (this is basically a blueprint for an automated test that is run before you send the task for a review to your tech lead) and user-review goal that will determine if a task is done or not but from a user perspective since it will be reviewed by a human.#}

13
euclid/utils/llm_connection.py
View File

@@ -11,7 +11,8 @@ from jinja2 import Environment, FileSystemLoader
from const.llm import MIN_TOKENS_FOR_GPT_RESPONSE, MAX_GPT_MODEL_TOKENS, MAX_QUESTIONS, END_RESPONSE
from logger.logger import logger
from termcolor import colored
from utils.utils import get_prompt_components, fix_json_newlines
from utils.utils import get_prompt_components, fix_json
from utils.spinner import spinner_start, spinner_stop
def connect_to_llm():
@@ -96,6 +97,10 @@ def create_gpt_chat_completion(messages: List[dict], req_type, min_tokens=MIN_TO
'model': 'gpt-4',
'n': 1,
'max_tokens': min(4096, MAX_GPT_MODEL_TOKENS - tokens_in_messages),
'temperature': 1,
'top_p': 1,
'presence_penalty': 0,
'frequency_penalty': 0,
'messages': messages,
'stream': True
}
@@ -200,15 +205,17 @@ def stream_gpt_completion(data, req_type):
continue
try:
json_line = load_data_to_json(line)
json_line = json.loads(line)
if 'error' in json_line:
logger.error(f'Error in LLM response: {json_line}')
raise ValueError(f'Error in LLM response: {json_line["error"]["message"]}')
if json_line['choices'][0]['finish_reason'] == 'function_call':
function_calls['arguments'] = load_data_to_json(function_calls['arguments'])
return return_result({'function_calls': function_calls}, lines_printed);
json_line = json_line['choices'][0]['delta']
except json.JSONDecodeError:
logger.error(f'Unable to decode line: {line}')
continue # skip to the next line
@@ -250,4 +257,4 @@ def postprocessing(gpt_response, req_type):
def load_data_to_json(string):
return json.loads(fix_json_newlines(string))
return json.loads(fix_json(string))

6
euclid/utils/utils.py
View File

@@ -145,6 +145,12 @@ def replace_functions(obj):
else:
return obj
def fix_json(s):
s = s.replace('True', 'true')
s = s.replace('False', 'false')
# s = s.replace('`', '"')
return fix_json_newlines(s)
def fix_json_newlines(s):
pattern = r'("(?:\\\\n|\\.|[^"\\])*")'