github/gpt-pilot
1
1
Fork 0
mirror of https://github.com/Pythagora-io/gpt-pilot.git synced 2026-01-10 13:37:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

add logs for all agents and actions

Browse Source
This commit is contained in:
LeonOstrez
2024-02-25 12:47:10 -08:00
parent 47dcbe25f9
commit 958479fadd
13 changed files with 77 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
pilot/database/database.py
View File

@@ -68,7 +68,7 @@ def get_created_apps_with_steps():
for dev_step in app['development_steps']:
# Filter out unwanted keys first
filtered_step = {k: v for k, v in dev_step.items() if k in {'id', 'created_at', 'prompt_path'}}
filtered_step = {k: v for k, v in dev_step.items() if k in {'id', 'prompt_path'}}
if 'breakdown' in filtered_step['prompt_path']:
filtered_step['name'] = f"Task {task_counter}"
@@ -94,6 +94,13 @@ def get_created_apps_with_steps():
# Update the dev_step in the list
new_development_steps.append(filtered_step)
last_step = get_last_development_step(app['id'])
if last_step:
new_development_steps.append({
'id': last_step['id'],
'prompt_path': last_step['prompt_path'],
'name': 'Latest Step',
})
app['development_steps'] = new_development_steps
return apps
@@ -116,6 +123,18 @@ def get_all_app_development_steps(app_id, last_step=None, loading_steps_only=Fal
return [model_to_dict(dev_step) for dev_step in query]
def get_last_development_step(app_id, last_step=None):
last_dev_step_query = DevelopmentSteps.select().where(DevelopmentSteps.app == app_id)
if last_step is not None:
last_dev_step_query = last_dev_step_query.where(DevelopmentSteps.id <= last_step)
# Order by ID in descending order to get the last step and fetch the first result
last_dev_step = last_dev_step_query.order_by(DevelopmentSteps.id.desc()).first()
# If a step is found, convert it to a dictionary, otherwise return None
return model_to_dict(last_dev_step) if last_dev_step else None
def save_user(user_id, email, password):
try:
user = User.get(User.id == user_id)

3
pilot/helpers/AgentConvo.py
View File

@@ -238,7 +238,8 @@ class AgentConvo:
if self.log_to_user:
if self.agent.project.checkpoints['last_development_step'] is not None:
dev_step_msg = f'\nDev step {str(self.agent.project.checkpoints["last_development_step"]["id"])}\n'
print(color_yellow_bold(dev_step_msg), end='')
if not self.agent.project.check_ipc():
print(color_yellow_bold(dev_step_msg), end='')
logger.info(dev_step_msg)
print(f"\n{content}\n", type='local')
logger.info(f"{print_msg}: {content}\n")

1
pilot/helpers/Debugger.py
View File

@@ -66,6 +66,7 @@ class Debugger:
user_input = answer
self.agent.project.current_task.add_user_input_to_debugging_task(user_input)
print('', type='verbose', category='agent:debugger')
llm_response = convo.send_message('dev_ops/debug.prompt',
{
'command': command['command'] if command is not None else None,

14
pilot/helpers/Project.py
View File

@@ -102,8 +102,6 @@ class Project:
self.tasks_to_load = []
self.features_to_load = []
self.dev_steps_to_load = []
if self.continuing_project:
self.setup_loading()
# end loading of project
def set_root_path(self, root_path: str):
@@ -167,9 +165,13 @@ class Project:
if not test_api_access(self):
return False
if self.continuing_project:
self.setup_loading()
self.project_manager = ProductOwner(self)
self.spec_writer = SpecWriter(self)
print('', type='verbose', category='agent:product-owner')
self.project_manager.get_project_description(self.spec_writer)
self.project_manager.get_user_stories()
# self.user_tasks = self.project_manager.get_user_tasks()
@@ -183,6 +185,7 @@ class Project:
self.developer.set_up_environment()
self.technical_writer = TechnicalWriter(self)
print('', type='verbose', category='agent:tech-lead')
self.tech_lead = TechLead(self)
self.tech_lead.create_development_plan()
@@ -210,6 +213,7 @@ class Project:
self.previous_features = get_features_by_app_id(self.args['app_id'])
if not self.skip_steps:
print('', type='verbose', category='pythagora')
feature_description = ask_user(self, "Project is finished! Do you want to add any features or changes? "
"If yes, describe it here and if no, just press ENTER",
require_some_input=False)
@@ -217,6 +221,7 @@ class Project:
if feature_description == '':
return
print('', type='verbose', category='agent:tech-lead')
self.tech_lead.create_feature_plan(feature_description)
# loading of features
@@ -248,6 +253,7 @@ class Project:
self.current_feature = feature_description
self.developer.start_coding('feature')
print('', type='verbose', category='agent:tech-lead')
self.tech_lead.create_feature_summary(feature_description)
def get_directory_tree(self, with_descriptions=False):
@@ -544,7 +550,9 @@ class Project:
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=None, cbs={}, convo=None, is_root_task=False, add_loop_button=False):
def ask_for_human_intervention(self, message, description=None, cbs={}, convo=None, is_root_task=False,
add_loop_button=False, category='human-intervention'):
print('', type='verbose', category=category)
answer = ''
question = color_yellow_bold(message)

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

@@ -45,6 +45,7 @@ class Developer(Agent):
self.debugger = Debugger(self)
def start_coding(self, task_source):
print('', type='verbose', category='agent:developer')
if not self.project.finished:
self.project.current_step = 'coding'
update_app_status(self.project.args['app_id'], self.project.current_step)
@@ -67,6 +68,7 @@ class Developer(Agent):
if current_progress_percent > threshold and threshold not in documented_thresholds:
if not self.project.skip_steps:
self.project.technical_writer.document_project(current_progress_percent)
print('', type='verbose', category='agent:developer')
documented_thresholds.add(threshold)
if self.project.tasks_to_load:
@@ -101,14 +103,14 @@ class Developer(Agent):
update_app_status(self.project.args['app_id'], self.project.current_step)
message = 'The app is DONE!!! Yay...you can use it now.\n'
logger.info(message)
print(color_green_bold(message))
print(color_green_bold(message), category='success')
if not self.project.skip_steps:
telemetry.set("end_result", "success:initial-project")
telemetry.send()
else:
message = 'Feature complete!\n'
logger.info(message)
print(color_green_bold(message))
print(color_green_bold(message), category='success')
if not self.project.skip_steps:
telemetry.set("end_result", "success:feature")
telemetry.send()
@@ -121,7 +123,7 @@ class Developer(Agent):
:param task_source: The source of the task, one of: 'app', 'feature', 'debugger', 'iteration'.
:param development_task: The task to implement.
"""
print(color_green_bold(f'Implementing task #{i + 1}: ') + color_green(f' {development_task["description"]}\n'))
print(color_green_bold(f'Implementing task #{i + 1}: ') + color_green(f' {development_task["description"]}\n'), category='agent:developer')
self.project.dot_pilot_gpt.chat_log_folder(i + 1)
convo_dev_task = AgentConvo(self)
@@ -495,6 +497,13 @@ class Developer(Agent):
"""
function_uuid = str(uuid.uuid4())
convo.save_branch(function_uuid)
agent_map = {
'app': 'agent:developer',
'feature': 'agent:developer',
'debugger': 'agent:debugger',
'troubleshooting': 'agent:troubleshooter',
'review': 'agent:reviewer',
}
for (i, step) in enumerate(task_steps):
print_step_progress(i+1, len(task_steps), step, task_source)
@@ -525,20 +534,19 @@ class Developer(Agent):
convo.load_branch(function_uuid)
if step['type'] == 'command':
print('', type='verbose', category='command run')
result = self.step_command_run(convo, task_steps, i, success_with_cli_response=need_to_see_output)
# if need_to_see_output and 'cli_response' in result:
# result['user_input'] = result['cli_response']
elif step['type'] in ['save_file', 'modify_file', 'code_change']:
print('', type='verbose', category='save file')
print('', type='verbose', category='agent:code-monkey')
result = self.step_save_file(convo, step, i, test_after_code_changes)
elif step['type'] == 'delete_file':
result = self.step_delete_file(convo, step, i, test_after_code_changes)
elif step['type'] == 'human_intervention':
print('', type='verbose', category='human intervention')
print('', type='verbose', category='human-intervention')
result = self.step_human_intervention(convo, task_steps, i)
# TODO background_command - if we run commands in background we should have way to kill processes
@@ -547,7 +555,7 @@ class Developer(Agent):
# terminate_named_process(step['kill_process'])
# result = {'success': True}
print('', type='verbose', category='agent:developer')
print('', type='verbose', category=agent_map[task_source])
logger.info(' step result: %s', result)
if (not result['success']) or (need_to_see_output and result.get("user_input") != "SKIP"):
@@ -628,7 +636,8 @@ class Developer(Agent):
return_cli_response=True, is_root_task=True)},
convo=iteration_convo,
is_root_task=True,
add_loop_button=iteration_count > 3)
add_loop_button=iteration_count > 3,
category='human-test')
logger.info('response: %s', response)
self.review_count = 0
@@ -638,6 +647,7 @@ class Developer(Agent):
return {"success": True, "user_input": user_feedback}
if user_feedback is not None:
print('', type='verbose', category='agent:troubleshooter')
user_feedback = self.bug_report_generator(user_feedback)
stuck_in_loop = user_feedback.startswith(STUCK_IN_LOOP)
if stuck_in_loop:
@@ -761,6 +771,7 @@ class Developer(Agent):
Review all task changes and refactor big files.
:return: bool - True if the task changes passed review, False if not
"""
print('', type='verbose', category='agent:reviewer')
self.review_count += 1
review_result = self.review_code_changes()
refactoring_done = self.refactor_code()

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

@@ -23,7 +23,7 @@ class ProductOwner(Agent):
def get_project_description(self, spec_writer):
print(json.dumps({
"project_stage": "project_description"
}), type='info')
}), type='info', category='agent:product-owner')
self.project.app = get_app(self.project.args['app_id'], error_if_not_found=False)

3
pilot/helpers/agents/TechnicalWriter.py
View File

@@ -12,12 +12,13 @@ class TechnicalWriter(Agent):
def document_project(self, percent):
files = self.project.get_all_coded_files()
print(f'{color_green_bold("CONGRATULATIONS!!!")}')
print(f'{color_green_bold("CONGRATULATIONS!!!")}', category='success')
print(f'You reached {color_green(str(percent) + "%")} of your project generation!\n\n')
print('For now, you have created:\n')
print(f'{color_green(len(files))} files\n')
print(f'{color_green(count_lines_of_code(files))} lines of code\n\n')
print('Before continuing, GPT Pilot will create some documentation for the project...\n')
print('', type='verbose', category='agent:tech-writer')
self.create_license()
self.create_readme()
self.create_api_documentation()

2
pilot/helpers/cli.py
View File

@@ -194,6 +194,7 @@ def execute_command(project, command, timeout=None, success_message=None, comman
If `cli_response` not None: 'was interrupted by user', 'timed out' or `None` - caller should send `cli_response` to LLM
exit_code (int): The exit code of the process.
"""
print('', type='verbose', category='exec-command')
project.finish_loading()
if timeout is not None:
if timeout < 0:
@@ -216,7 +217,6 @@ def execute_command(project, command, timeout=None, success_message=None, comman
logger.info('--------- EXECUTE COMMAND ---------- : %s', question)
answer = ask_user(project, question, False, hint='If yes, just press ENTER. Otherwise, type "no" but it will be processed as successfully executed.')
# TODO can we use .confirm(question, default='yes').ask() https://questionary.readthedocs.io/en/stable/pages/types.html#confirmation
print('answer: ' + answer)
if answer.lower() in NEGATIVE_ANSWERS:
return None, 'SKIP', None
elif answer.lower() not in AFFIRMATIVE_ANSWERS:

1
pilot/main.py
View File

@@ -126,6 +126,7 @@ if __name__ == "__main__":
print('Exit', type='exit')
except Exception as err:
print('', type='verbose', category='error')
print(color_red('---------- GPT PILOT EXITING WITH ERROR ----------'))
traceback.print_exc()
print(color_red('--------------------------------------------------'))

2
pilot/prompts/prompts.py
View File

@@ -55,7 +55,7 @@ def ask_for_main_app_definition(project):
def ask_user(project, question: str, require_some_input=True, hint: str = None, ignore_user_input_count: bool = False):
while True:
if hint is not None:
print(color_white_bold(hint) + '\n', type='hint')
print(color_white_bold(hint) + '\n')
project.finish_loading()
answer = styled_text(project, question, hint=hint, ignore_user_input_count=ignore_user_input_count)

3
pilot/utils/custom_print.py
View File

@@ -1,6 +1,7 @@
import builtins
from helpers.ipc import IPCClient
from const.ipc import MESSAGE_TYPE, LOCAL_IGNORE_MESSAGE_TYPES
from utils.print import remove_ansi_codes
def get_custom_print(args):
@@ -19,7 +20,7 @@ def get_custom_print(args):
ipc_client_instance.send({
'type': MESSAGE_TYPE[kwargs['type']],
'category': kwargs['category'] if 'category' in kwargs else '',
'content': message,
'content': remove_ansi_codes(message),
})
if kwargs['type'] == MESSAGE_TYPE['user_input_request']:
return ipc_client_instance.listen()

13
pilot/utils/print.py
View File

@@ -1,3 +1,6 @@
import re
def print_task_progress(index, num_of_tasks, description, task_source, status):
"""
Print task progress in extension.
@@ -36,3 +39,13 @@ def print_step_progress(index, num_of_steps, step, task_source):
'step': step,
'source': task_source,
}}, type='progress')
def remove_ansi_codes(s: str) -> str:
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
# Check if the input is a string
if isinstance(s, str):
return ansi_escape.sub('', s)
else:
# If the input is not a string, return the input as is
return s

11
pilot/utils/questionary.py
View File

@@ -1,14 +1,9 @@
import platform
import questionary
import re
import sys
from database.database import save_user_input
from utils.style import style_config
def remove_ansi_codes(s: str) -> str:
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
return ansi_escape.sub('', s)
from utils.print import remove_ansi_codes
def styled_select(*args, **kwargs):
@@ -29,7 +24,6 @@ def styled_text(project, question, ignore_user_input_count=False, style=None, hi
if project is not None and project.check_ipc():
response = print(question, type='user_input_request')
print(response)
else:
used_style = style if style is not None else style_config.get_style()
question = remove_ansi_codes(question) # Colorama and questionary are not compatible and styling doesn't work
@@ -39,7 +33,8 @@ def styled_text(project, question, ignore_user_input_count=False, style=None, hi
if not ignore_user_input_count:
save_user_input(project, question, response, hint)
print('\n\n', end='')
if project is not None and not project.check_ipc():
print('\n\n', end='')
return response