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

Refactored all prints to be colored with fabulous and not termcolor

Browse Source
This commit is contained in:
Zvonimir Sabljic
2023-09-01 18:28:20 +02:00
parent ca58c4958d
commit 85ac7e8276
14 changed files with 54 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pilot/database/database.py
View File

@@ -1,6 +1,6 @@
from playhouse.shortcuts import model_to_dict
from peewee import *
from termcolor import colored
from fabulous.color import yellow, red
from functools import reduce
import operator
import psycopg2
@@ -210,7 +210,7 @@ def hash_and_save_step(Model, app_id, hash_data_args, data_fields, message):
.execute())
record = Model.get_by_id(inserted_id)
logger.debug(colored(f"{message} with id {record.id}", "yellow"))
logger.debug(yellow(f"{message} with id {record.id}"))
except IntegrityError:
print(f"A record with hash_id {hash_id} already exists for {Model.__name__}.")
return None
@@ -308,7 +308,7 @@ def delete_all_subsequent_steps(project):
def delete_subsequent_steps(model, step):
if step is None:
return
logger.info(colored(f"Deleting subsequent {model.__name__} steps after {step.id}", "red"))
logger.info(red(f"Deleting subsequent {model.__name__} steps after {step.id}"))
subsequent_steps = model.select().where(model.previous_step == step.id)
for subsequent_step in subsequent_steps:
if subsequent_step:
@@ -344,7 +344,7 @@ def delete_unconnected_steps_from(step, previous_step_field_name):
).order_by(DevelopmentSteps.id.desc())
for unconnected_step in unconnected_steps:
print(colored(f"Deleting unconnected {step.__class__.__name__} step {unconnected_step.id}", "red"))
print(red(f"Deleting unconnected {step.__class__.__name__} step {unconnected_step.id}"))
unconnected_step.delete_instance()

8
pilot/helpers/AgentConvo.py
View File

@@ -1,5 +1,5 @@
import subprocess
from termcolor import colored
from fabulous.color import yellow, bold
from database.database import get_development_step_from_hash_id, save_development_step, delete_all_subsequent_steps
from utils.utils import array_of_objects_to_string
@@ -34,7 +34,7 @@ class AgentConvo:
development_step = get_development_step_from_hash_id(self.agent.project, prompt_path, prompt_data, self.agent.project.llm_req_num)
if development_step is not None and self.agent.project.skip_steps:
# if we do, use it
print(colored(f'Restoring development step with id {development_step.id}', 'yellow'))
print(yellow(f'Restoring development step with id {development_step.id}'))
self.agent.project.checkpoints['last_development_step'] = development_step
self.agent.project.restore_files(development_step.id)
response = development_step.llm_response
@@ -90,7 +90,7 @@ class AgentConvo:
# Continue conversation until GPT response equals END_RESPONSE
while response != END_RESPONSE:
print(colored("Do you want to add anything else? If not, ", 'yellow') + colored('just press ENTER.', 'yellow', attrs=['bold']))
print(yellow("Do you want to add anything else? If not, ") + yellow(bold('just press ENTER.')))
user_message = ask_user(self.agent.project, response, False)
if user_message == "":
@@ -125,7 +125,7 @@ class AgentConvo:
print_msg = capitalize_first_word_with_underscores(self.high_level_step)
if self.log_to_user:
if self.agent.project.checkpoints['last_development_step'] is not None:
print(colored("\nDev step ", 'yellow') + colored(self.agent.project.checkpoints['last_development_step'], 'yellow', attrs=['bold']) + '\n', end='')
print(yellow("\nDev step ") + yellow(bold(self.agent.project.checkpoints['last_development_step'])) + '\n', end='')
print(f"\n{content}\n")
logger.info(f"{print_msg}: {content}\n")

5
pilot/helpers/Project.py
View File

@@ -3,7 +3,6 @@ import os
import time
from fabulous.color import bold, green, yellow
from termcolor import colored
from const.common import IGNORE_FOLDERS
from database.models.app import App
from database.database import get_app, delete_unconnected_steps_from, delete_all_app_development_data
@@ -214,9 +213,9 @@ class Project:
delete_unconnected_steps_from(self.checkpoints['last_user_input'], 'previous_step')
def ask_for_human_intervention(self, message, description=None, cbs={}):
self.log(yellow(bold(message)))
print(yellow(bold(message)))
if description is not None:
self.log(description)
print(description)
answer = ''
while answer != 'continue':
answer = styled_text(

4
pilot/helpers/agents/Architect.py
View File

@@ -1,7 +1,7 @@
from utils.utils import step_already_finished
from helpers.Agent import Agent
import json
from termcolor import colored
from fabulous.color import green, bold
from const.function_calls import ARCHITECTURE
from utils.utils import execute_step, find_role_from_step, generate_app_data
@@ -27,7 +27,7 @@ class Architect(Agent):
return step['architecture']
# ARCHITECTURE
print(colored(f"Planning project architecture...\n", "green", attrs=['bold']))
print(green(bold(f"Planning project architecture...\n")))
logger.info(f"Planning project architecture...")
architecture = self.convo_architecture.send_message('architecture/technologies.prompt',

12
pilot/helpers/agents/Developer.py
View File

@@ -1,6 +1,6 @@
import json
import uuid
from termcolor import colored
from fabulous.color import yellow, green, red, bold
from utils.questionary import styled_text
from helpers.files import update_file
from utils.utils import step_already_finished
@@ -26,7 +26,7 @@ class Developer(Agent):
self.project.skip_steps = False if ('skip_until_dev_step' in self.project.args and self.project.args['skip_until_dev_step'] == '0') else True
# DEVELOPMENT
print(colored(f"Ok, great, now, let's start with the actual development...\n", "green", attrs=['bold']))
print(green(bold(f"Ok, great, now, let's start with the actual development...\n")))
logger.info(f"Starting to create the actual code...")
self.implement_task()
@@ -100,9 +100,9 @@ class Developer(Agent):
elif should_rerun_command == 'YES':
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(red(f'Got incorrect CLI response:'))
print(cli_response)
print(colored('-------------------', 'red'))
print(red('-------------------'))
if llm_response == 'DONE':
return True
@@ -119,7 +119,7 @@ class Developer(Agent):
while True:
# TODO add description about how can the user check if the app works
user_feedback = self.project.ask_for_human_intervention(
'Can you check if the app works?\nIf you want to run the app, ' + colored('just type "r" and press ENTER', 'yellow', attrs=['bold']),
'Can you check if the app works?\nIf you want to run the app, ' + yellow(bold('just type "r" and press ENTER')),
cbs={ 'r': lambda: run_command_until_success(self.run_command, None, iteration_convo, force=True) })
if user_feedback == 'continue':
@@ -165,7 +165,7 @@ class Developer(Agent):
})
return
# ENVIRONMENT SETUP
print(colored(f"Setting up the environment...\n", "green"))
print(green(f"Setting up the environment...\n"))
logger.info(f"Setting up the environment...")
os_info = get_os_info()

7
pilot/helpers/agents/ProductOwner.py
View File

@@ -1,4 +1,3 @@
from termcolor import colored
from fabulous.color import bold, green, yellow
from helpers.AgentConvo import AgentConvo
@@ -44,7 +43,7 @@ class ProductOwner(Agent):
self.project,
generate_messages_from_description(main_prompt, self.project.args['app_type'], self.project.args['name']))
self.project.log(green(bold('Project Summary:\n')))
print(green(bold('Project Summary:\n')))
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])})
@@ -74,7 +73,7 @@ class ProductOwner(Agent):
# USER STORIES
msg = f"User Stories:\n"
self.project.log(green(bold(msg)))
print(green(bold(msg)))
logger.info(msg)
self.project.user_stories = self.convo_user_stories.continuous_conversation('user_stories/specs.prompt', {
@@ -108,7 +107,7 @@ class ProductOwner(Agent):
# USER TASKS
msg = f"User Tasks:\n"
self.project.log(green(bold(msg)))
print(green(bold(msg)))
logger.info(msg)
self.project.user_tasks = self.convo_user_stories.continuous_conversation('user_stories/user_tasks.prompt',

4
pilot/helpers/agents/TechLead.py
View File

@@ -1,7 +1,7 @@
from utils.utils import step_already_finished
from helpers.Agent import Agent
import json
from termcolor import colored
from fabulous.color import green, bold
from const.function_calls import DEV_STEPS
from helpers.cli import build_directory_tree
from helpers.AgentConvo import AgentConvo
@@ -29,7 +29,7 @@ class TechLead(Agent):
return step['development_plan']
# DEVELOPMENT PLANNING
print(colored(f"Starting to create the action plan for development...\n", "green", attrs=['bold']))
print(green(bold(f"Starting to create the action plan for development...\n")))
logger.info(f"Starting to create the action plan for development...")
# TODO add clarifications

20
pilot/helpers/cli.py
View File

@@ -7,7 +7,7 @@ import time
import uuid
import platform
from termcolor import colored
from fabulous.color import yellow, green, white, red, bold
from database.database import get_command_run_from_hash_id, save_command_run
from const.function_calls import DEBUG_STEPS_BREAKDOWN
@@ -74,8 +74,8 @@ def execute_command(project, command, timeout=None, force=False):
timeout = min(max(timeout, MIN_COMMAND_RUN_TIME), MAX_COMMAND_RUN_TIME)
if not force:
print(colored(f'\n--------- EXECUTE COMMAND ----------', 'yellow', attrs=['bold']))
print(colored(f'Can i execute the command: `') + colored(command, 'yellow', attrs=['bold']) + colored(f'` with {timeout}ms timeout?'))
print(yellow(bold(f'\n--------- EXECUTE COMMAND ----------')))
print(f'Can i execute the command: `' + yellow(bold(command)) + f'` with {timeout}ms timeout?')
answer = styled_text(
project,
@@ -87,7 +87,7 @@ def execute_command(project, command, timeout=None, force=False):
if command_run is not None and project.skip_steps:
# if we do, use it
project.checkpoints['last_command_run'] = command_run
print(colored(f'Restoring command run response id {command_run.id}:\n```\n{command_run.cli_response}```', 'yellow'))
print(yellow(f'Restoring command run response id {command_run.id}:\n```\n{command_run.cli_response}```'))
return command_run.cli_response
return_value = None
@@ -105,7 +105,7 @@ def execute_command(project, command, timeout=None, force=False):
while True and return_value is None:
elapsed_time = time.time() - start_time
if timeout is not None:
print(colored(f'\rt: {round(elapsed_time * 1000)}ms : ', 'white', attrs=['bold']), end='', flush=True)
print(white(bold(f'\rt: {round(elapsed_time * 1000)}ms : ')), end='', flush=True)
# Check if process has finished
if process.poll() is not None:
@@ -114,7 +114,7 @@ def execute_command(project, command, timeout=None, force=False):
while not q.empty():
output_line = q.get_nowait()
if output_line not in output:
print(colored('CLI OUTPUT:', 'green') + output_line, end='')
print(green('CLI OUTPUT:') + output_line, end='')
output += output_line
break
@@ -131,7 +131,7 @@ def execute_command(project, command, timeout=None, force=False):
if line:
output += line
print(colored('CLI OUTPUT:', 'green') + line, end='')
print(green('CLI OUTPUT:') + line, end='')
# Read stderr
try:
@@ -141,7 +141,7 @@ def execute_command(project, command, timeout=None, force=False):
if stderr_line:
stderr_output += stderr_line
print(colored('CLI ERROR:', 'red') + stderr_line, end='') # Print with different color for distinction
print(red('CLI ERROR:') + stderr_line, end='') # Print with different color for distinction
except (KeyboardInterrupt, TimeoutError) as e:
interrupted = True
@@ -215,9 +215,9 @@ def run_command_until_success(command, timeout, convo, additional_message=None,
{'cli_response': cli_response, 'command': command, 'additional_message': additional_message})
if response != 'DONE':
print(colored(f'Got incorrect CLI response:', 'red'))
print(red(f'Got incorrect CLI response:'))
print(cli_response)
print(colored('-------------------', 'red'))
print(red('-------------------'))
debug(convo, {'command': command, 'timeout': timeout})

4
pilot/helpers/files.py
View File

@@ -1,4 +1,4 @@
from termcolor import colored
from fabulous.color import green
import os
@@ -11,7 +11,7 @@ def update_file(path, new_content):
# Write content to the file
with open(path, 'w') as file:
file.write(new_content)
print(colored(f"Updated file {path}", "green"))
print(green(f"Updated file {path}"))
def get_files_content(directory, ignore=[]):
return_array = []

9
pilot/prompts/prompts.py
View File

@@ -1,6 +1,6 @@
# prompts/prompts.py
from termcolor import colored
from fabulous.color import yellow
import questionary
from const import common
@@ -102,14 +102,13 @@ def get_additional_info_from_user(project, messages, role):
while True:
if isinstance(message, dict) and 'text' in message:
message = message['text']
print(colored(
f"Please check this message and say what needs to be changed. If everything is ok just press ENTER",
"yellow"))
print(yellow(f"Please check this message and say what needs to be changed. If everything is ok just press ENTER",))
answer = ask_user(project, message, False)
if answer.lower() == '':
break
response = create_gpt_chat_completion(
generate_messages_from_custom_conversation(role, [get_prompt('utils/update.prompt'), message, answer], 'user'), 'additional_info')
generate_messages_from_custom_conversation(role, [get_prompt('utils/update.prompt'), message, answer], 'user'),
'additional_info')
message = response

2
pilot/utils/arguments.py
View File

@@ -1,8 +1,6 @@
import sys
import uuid
from termcolor import colored
from database.database import get_app

21
pilot/utils/llm_connection.py
View File

@@ -10,15 +10,11 @@ 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 fabulous.color import red
from utils.utils import get_prompt_components, fix_json
from utils.spinner import spinner_start, spinner_stop
def connect_to_llm():
pass
def get_prompt(prompt_name, data=None):
if data is None:
data = {}
@@ -115,8 +111,7 @@ def create_gpt_chat_completion(messages: List[dict], req_type, min_tokens=MIN_TO
response = stream_gpt_completion(gpt_data, req_type)
return response
except Exception as e:
print(
'The request to OpenAI API failed. Here is the error message:')
print('The request to OpenAI API failed. Here is the error message:')
print(e)
@@ -139,7 +134,7 @@ def retry_on_exception(func):
try:
return func(*args, **kwargs)
except Exception as e:
print(colored(f'There was a problem with request to openai API:', 'red'))
print(red(f'There was a problem with request to openai API:'))
print(str(e))
user_message = questionary.text(
"Do you want to try make the same request again? If yes, just press ENTER. Otherwise, type 'no'.",
@@ -168,8 +163,7 @@ def stream_gpt_completion(data, req_type):
delete_last_n_lines(lines_printed)
return result_data
# spinner = spinner_start(colored("Waiting for OpenAI API response...", 'yellow'))
# print(colored("Stream response from OpenAI:", 'yellow'))
# spinner = spinner_start(yellow("Waiting for OpenAI API response..."))
api_key = os.getenv("OPENAI_API_KEY")
logger.info(f'Request data: {data}')
@@ -191,6 +185,7 @@ def stream_gpt_completion(data, req_type):
gpt_response = ''
function_calls = {'name': '', 'arguments': ''}
for line in response.iter_lines():
# Ignore keep-alive new lines
if line:
@@ -226,7 +221,7 @@ def stream_gpt_completion(data, req_type):
if 'arguments' in json_line['function_call']:
function_calls['arguments'] += json_line['function_call']['arguments']
print(json_line['function_call']['arguments'], end='', flush=True)
print(json_line['function_call']['arguments'], type='stream', end='', flush=True)
if 'content' in json_line:
content = json_line.get('content')
@@ -239,9 +234,9 @@ def stream_gpt_completion(data, req_type):
buffer = "" # reset the buffer
gpt_response += content
print(content, end='', flush=True)
print(content, type='stream', end='', flush=True)
print('\n')
print('\n', type='stream')
if function_calls['arguments'] != '':
logger.info(f'Response via function call: {function_calls["arguments"]}')
function_calls['arguments'] = load_data_to_json(function_calls['arguments'])

10
pilot/utils/questionary.py
View File

@@ -1,6 +1,6 @@
from prompt_toolkit.styles import Style
import questionary
from termcolor import colored
from fabulous.color import yellow, bold
from database.database import save_user_input, get_user_input_from_hash_id
from const.ipc import MESSAGE_TYPE
@@ -25,8 +25,8 @@ def styled_text(project, question):
if user_input is not None and user_input.user_input is not None and project.skip_steps:
# if we do, use it
project.checkpoints['last_user_input'] = user_input
print(colored(f'Restoring user input id {user_input.id}: ', 'yellow'), end='')
print(colored(f'{user_input.user_input}', 'yellow', attrs=['bold']))
print(yellow(bold(f'Restoring user input id {user_input.id}: ')), end='')
print(yellow(bold(f'{user_input.user_input}')))
return user_input.user_input
if project.ipc_client_instance is None or project.ipc_client_instance.client is None:
@@ -35,8 +35,8 @@ def styled_text(project, question):
}
response = questionary.text(question, **config).unsafe_ask() # .ask() is included here
else:
response = project.log(question, MESSAGE_TYPE['user_input_request'])
project.log(response, MESSAGE_TYPE['verbose'])
response = print(question, type='user_input_request')
print(response)
user_input = save_user_input(project, question, response)

4
pilot/utils/utils.py
View File

@@ -9,7 +9,7 @@ import json
import hashlib
import re
from jinja2 import Environment, FileSystemLoader
from termcolor import colored
from fabulous.color import green
from const.llm import MAX_QUESTIONS, END_RESPONSE
from const.common import ROLES, STEPS
@@ -121,7 +121,7 @@ def step_already_finished(args, step):
args.update(step['app_data'])
message = f"{capitalize_first_word_with_underscores(step['step'])} already done for this app_id: {args['app_id']}. Moving to next step..."
print(colored(message, "green"))
print(green(message))
logger.info(message)