aboutsummaryrefslogtreecommitdiff
path: root/autogpt/logs/log_cycle.py
diff options
context:
space:
mode:
Diffstat (limited to 'autogpt/logs/log_cycle.py')
-rw-r--r--autogpt/logs/log_cycle.py88
1 files changed, 0 insertions, 88 deletions
diff --git a/autogpt/logs/log_cycle.py b/autogpt/logs/log_cycle.py
deleted file mode 100644
index f3cbf166e..000000000
--- a/autogpt/logs/log_cycle.py
+++ /dev/null
@@ -1,88 +0,0 @@
-import json
-import os
-from typing import Any, Dict, Union
-
-from .logger import logger
-
-DEFAULT_PREFIX = "agent"
-FULL_MESSAGE_HISTORY_FILE_NAME = "full_message_history.json"
-CURRENT_CONTEXT_FILE_NAME = "current_context.json"
-NEXT_ACTION_FILE_NAME = "next_action.json"
-PROMPT_SUMMARY_FILE_NAME = "prompt_summary.json"
-SUMMARY_FILE_NAME = "summary.txt"
-SUPERVISOR_FEEDBACK_FILE_NAME = "supervisor_feedback.txt"
-PROMPT_SUPERVISOR_FEEDBACK_FILE_NAME = "prompt_supervisor_feedback.json"
-USER_INPUT_FILE_NAME = "user_input.txt"
-
-
-class LogCycleHandler:
- """
- A class for logging cycle data.
- """
-
- def __init__(self):
- self.log_count_within_cycle = 0
-
- @staticmethod
- def create_directory_if_not_exists(directory_path: str) -> None:
- if not os.path.exists(directory_path):
- os.makedirs(directory_path, exist_ok=True)
-
- def create_outer_directory(self, ai_name: str, created_at: str) -> str:
- log_directory = logger.get_log_directory()
-
- if os.environ.get("OVERWRITE_DEBUG") == "1":
- outer_folder_name = "auto_gpt"
- else:
- ai_name_short = self.get_agent_short_name(ai_name)
- outer_folder_name = f"{created_at}_{ai_name_short}"
-
- outer_folder_path = os.path.join(log_directory, "DEBUG", outer_folder_name)
- self.create_directory_if_not_exists(outer_folder_path)
-
- return outer_folder_path
-
- def get_agent_short_name(self, ai_name: str) -> str:
- return ai_name[:15].rstrip() if ai_name else DEFAULT_PREFIX
-
- def create_inner_directory(self, outer_folder_path: str, cycle_count: int) -> str:
- nested_folder_name = str(cycle_count).zfill(3)
- nested_folder_path = os.path.join(outer_folder_path, nested_folder_name)
- self.create_directory_if_not_exists(nested_folder_path)
-
- return nested_folder_path
-
- def create_nested_directory(
- self, ai_name: str, created_at: str, cycle_count: int
- ) -> str:
- outer_folder_path = self.create_outer_directory(ai_name, created_at)
- nested_folder_path = self.create_inner_directory(outer_folder_path, cycle_count)
-
- return nested_folder_path
-
- def log_cycle(
- self,
- ai_name: str,
- created_at: str,
- cycle_count: int,
- data: Union[Dict[str, Any], Any],
- file_name: str,
- ) -> None:
- """
- Log cycle data to a JSON file.
-
- Args:
- data (Any): The data to be logged.
- file_name (str): The name of the file to save the logged data.
- """
- nested_folder_path = self.create_nested_directory(
- ai_name, created_at, cycle_count
- )
-
- json_data = json.dumps(data, ensure_ascii=False, indent=4)
- log_file_path = os.path.join(
- nested_folder_path, f"{self.log_count_within_cycle}_{file_name}"
- )
-
- logger.log_json(json_data, log_file_path)
- self.log_count_within_cycle += 1